diff --git a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts index 07c3bd731633f..9125060d811cb 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts @@ -536,7 +536,7 @@ const makeInstructionCompactionHarness = Effect.fnUntraced(function* ( flavor: { driver: ACP_TEST_DRIVER, capabilities: AcpProviderCapabilitiesV2, - supportsCompaction: true, + compaction: "marker", makeRuntime: makeMockRuntime({ childProcessSpawner, mockAgentPath, @@ -3353,6 +3353,78 @@ describe("AcpAdapterV2", () => { }).pipe(Effect.provide(layerTest), Effect.scoped), ); + it.effect.each([ + ["marker", ["Context compacted"]], + ["native", []], + ] as const)( + "%s compaction: /compact without compaction_update projects %j", + ([compaction, expectedCompactions]) => + Effect.gen(function* () { + const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const path = yield* Path.Path; + const mockAgentPath = yield* path.fromFileUrl( + new URL("../../../scripts/acp-mock-agent.ts", import.meta.url), + ); + const instanceId = ProviderInstanceId.make(`acp-test-compaction-${compaction}`); + const adapter = yield* makeAcpAdapterV2({ + instanceId, + flavor: { + driver: ACP_TEST_DRIVER, + capabilities: AcpProviderCapabilitiesV2, + compaction, + makeRuntime: makeMockRuntime({ childProcessSpawner, mockAgentPath }), + }, + selfInvocation: yield* resolveSelfInvocation(), + }); + const threadId = ThreadId.make(`thread-acp-compaction-${compaction}`); + const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({ + runtimeMode: "full-access", + interactionMode: "default", + cwd: process.cwd(), + }); + const modelSelection = { instanceId, model: "default" } as const; + const runtime = yield* adapter.openSession({ + threadId, + providerSessionId: ProviderSessionId.make(`provider-session-compaction-${compaction}`), + modelSelection, + runtimePolicy, + }); + const providerThread = yield* runtime.ensureThread({ + threadId, + modelSelection, + runtimePolicy, + }); + assert.isDefined(runtime.compactThread); + yield* runtime.compactThread!( + makeTurnInput({ + threadId, + providerThread, + instanceId, + runtimePolicy, + now: yield* DateTime.now, + messageText: "/compact", + }), + ); + const events = Array.from( + yield* runtime.events.pipe( + Stream.takeUntil((event) => event.type === "turn.terminal"), + Stream.runCollect, + ), + ); + assert.isTrue( + events.some((event) => event.type === "turn.terminal" && event.status === "completed"), + ); + assert.deepEqual( + events.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "compaction" + ? [event.turnItem.title] + : [], + ), + [...expectedCompactions], + ); + }).pipe(Effect.provide(layerTest), Effect.scoped), + ); + it.effect("projects only the tail of long ACP terminal output", () => Effect.gen(function* () { const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner; diff --git a/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts index fa5b15bbc5fdf..5b102769e758e 100644 --- a/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/AcpRegistryAdapterV2.test.ts @@ -471,6 +471,12 @@ describe("AcpRegistryAdapterV2", () => { readonly storedModePick?: string; readonly turnModels?: ReadonlyArray; readonly rejectedModel?: string; + readonly agentId?: string; + readonly turnText?: string; + readonly expectedCompaction?: { readonly status: string; readonly summary: string }; + readonly expectedRunStatus?: string; + readonly interruptCompaction?: boolean; + readonly terminateCompaction?: boolean; }) { const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; // Scope release gives the replay agent at most a second to exit, so the status @@ -513,25 +519,32 @@ describe("AcpRegistryAdapterV2", () => { ); const instanceId = ProviderInstanceId.make("acp-registry-stored-picks"); let published: AcpRegistryLiveConfiguration | undefined; + let terminateRuntime = Effect.void; + const scriptPath = yield* path.fromFileUrl( + new URL("../../../scripts/acp-replay-agent.ts", import.meta.url), + ); const adapter = yield* makeAcpRegistryAdapterV2({ selfInvocation: yield* resolveSelfInvocation(), instanceId, settings: yield* decodeAcpRegistryAdapterSettings({ - agentId: "fixture-agent", + agentId: input.agentId ?? "fixture-agent", authMethodId: "test", customModels: input.customModels ?? [], }), environment: {}, - makeRuntime: makeAcpReplayRuntime({ - customModels: input.customModels, - transcript, - statusPath, - scriptPath: yield* path.fromFileUrl( - new URL("../../../scripts/acp-replay-agent.ts", import.meta.url), - ), - childProcessSpawner, - fileSystem, - }), + makeRuntime: (runtimeInput) => { + terminateRuntime = runtimeInput.onTermination( + new EffectAcpErrors.AcpRequestError({ code: -32603, errorMessage: "transport exited" }), + ); + return makeAcpReplayRuntime({ + customModels: input.customModels, + transcript, + statusPath, + scriptPath, + childProcessSpawner, + fileSystem, + })(runtimeInput); + }, }).pipe( Effect.provide( Layer.mergeAll( @@ -592,15 +605,48 @@ describe("AcpRegistryAdapterV2", () => { runtimePolicy, now, ordinal: index + 1, + ...(input.turnText === undefined ? {} : { messageText: input.turnText }), modelSelection: target, }), ); + let compactionObserved = false; yield* runtime.events.pipe( Stream.takeUntil((event) => event.type === "turn.terminal"), Stream.runForEach((event) => - Effect.sync(() => { + Effect.gen(function* () { + if ( + event.type === "turn_item.updated" && + event.turnItem.type === "compaction" && + event.turnItem.status === "running" + ) { + if (input.terminateCompaction) yield* terminateRuntime; + if (input.interruptCompaction) + yield* runtime.interruptTurn({ + providerThread, + providerTurnId: event.turnItem.providerTurnId!, + }); + } + if ( + event.type === "turn_item.updated" && + event.turnItem.type === "compaction" && + event.turnItem.status !== "running" + ) { + assert.equal(event.turnItem.status, input.expectedCompaction?.status); + assert.equal(event.turnItem.summary, input.expectedCompaction?.summary); + assert.isNotNull(event.turnItem.completedAt); + compactionObserved = true; + } if (event.type === "turn.terminal") { - assert.equal(event.status, "completed", event.failure?.message); + assert.equal( + event.status, + input.expectedRunStatus ?? "completed", + event.failure?.message, + ); + if (input.expectedCompaction) + assert.isTrue( + compactionObserved, + "native compaction must persist before turn settlement", + ); } }), ), @@ -626,6 +672,92 @@ describe("AcpRegistryAdapterV2", () => { return { configurations }; }); + it.effect.each([ + { ordering: "before", outcome: "completed", runStatus: "completed" }, + { ordering: "after", outcome: "completed", runStatus: "completed" }, + { ordering: "terminal-first", outcome: "completed", runStatus: "completed" }, + { ordering: "after", outcome: "failed", runStatus: "failed" }, + { ordering: "after", outcome: "cancelled", runStatus: "cancelled" }, + { ordering: "prompt-error", outcome: "failed", runStatus: "failed" }, + { ordering: "prompt-cancel", outcome: "cancelled", runStatus: "cancelled" }, + { ordering: "transport-exit", outcome: "failed", runStatus: "failed" }, + { ordering: "interrupt", outcome: "interrupted", runStatus: "interrupted" }, + ])( + "preserves Devin native compaction: $ordering / $outcome", + ({ ordering, outcome, runStatus }) => { + const notice = (status: string) => ({ + type: "emit_inbound", + frame: { + kind: "notification", + method: "_cognition.ai/compaction", + params: { sessionId: "agent-session", status, summary: "Retained native summary" }, + }, + }); + const response = + ordering === "prompt-error" + ? { + type: "emit_inbound", + frame: { + kind: "response", + method: "session/prompt", + error: { code: -32603, message: "compact failed" }, + }, + } + : answer("session/prompt", { + stopReason: + ordering === "prompt-error" + ? "error" + : ordering === "prompt-cancel" + ? "cancelled" + : "end_turn", + }); + const updates = + ordering === "terminal-first" + ? [notice("started"), notice(outcome), response] + : ordering === "transport-exit" + ? [response, notice("started")] + : ordering === "interrupt" + ? [ + notice("started"), + { + type: "expect_outbound", + frame: { + kind: "notification", + method: "session/cancel", + params: { sessionId: "agent-session" }, + }, + }, + response, + notice("completed"), + ] + : ordering.startsWith("prompt-") + ? [notice("started"), response] + : [ + ...(ordering === "before" + ? [notice("started"), response] + : [response, notice("started")]), + notice(outcome), + ]; + return openWithStoredPicks({ + setup: {}, + agentId: "devin", + turnModels: ["default"], + turnText: "/compact", + expectedCompaction: { status: outcome, summary: "Retained native summary" }, + expectedRunStatus: runStatus, + interruptCompaction: ordering === "interrupt", + terminateCompaction: ordering === "transport-exit", + pickFrames: [ + outbound("session/prompt", { + sessionId: "agent-session", + prompt: [{ type: "text", text: "/compact" }], + }), + ...updates, + ], + }).pipe(Effect.provide(layerTest), Effect.scoped); + }, + ); + it.effect("switches an agent that only advertises modes with session/set_mode", () => openWithStoredPicks({ setup: { diff --git a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts index 8c5ccb48b149d..b544dd0a69db6 100644 --- a/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/AntigravityAdapterV2.ts @@ -271,7 +271,6 @@ export function makeAntigravityAcpAdapterFlavor( extractSubagentUpdate: extractAntigravitySubagentUpdate, supportsImagePrompts: true, // ACP 1.3.0 treats /compact as ordinary model input. - supportsCompaction: false, requiresSignIn: isAntigravitySignInRequiredError, // 1.4.5 returns the root trajectory's Gemini usageMetadata delta; 1.4.1 returns none. reportsPerTurnPromptUsage: true, diff --git a/apps/server/src/orchestration-v2/testkit/ReplayRuntimeInstructions.ts b/apps/server/src/orchestration-v2/testkit/ReplayRuntimeInstructions.ts index 0c3d59e93398b..53ca5b4db0451 100644 --- a/apps/server/src/orchestration-v2/testkit/ReplayRuntimeInstructions.ts +++ b/apps/server/src/orchestration-v2/testkit/ReplayRuntimeInstructions.ts @@ -48,6 +48,19 @@ export function materializeReplayTranscriptRuntimeInstructions( Array.isArray(frame.params.prompt) ) { const prompt: ReadonlyArray = frame.params.prompt; + if ( + prompt.some( + (part) => + typeof part === "object" && + part !== null && + "type" in part && + part.type === "text" && + "text" in part && + typeof part.text === "string" && + part.text.trimStart().startsWith("/"), + ) + ) + return entry; const lastPart = prompt.at(-1); if ( typeof lastPart === "object" && diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/acp_registry_compaction/input.ts b/apps/server/src/orchestration-v2/testkit/fixtures/acp_registry_compaction/input.ts new file mode 100644 index 0000000000000..b0bbd05b65faf --- /dev/null +++ b/apps/server/src/orchestration-v2/testkit/fixtures/acp_registry_compaction/input.ts @@ -0,0 +1,16 @@ +import type { OrchestratorFixtureInput } from "../shared.ts"; + +export const ACP_REGISTRY_COMPACTION_SUMMARY = "Retained the fixture decisions."; + +/** + * A registry agent gets one normal turn, then a bare `/compact`: the message + * the Compact context action sends, or a user, MCP tool or schedule types. + */ +export function acpRegistryCompactionInput(): OrchestratorFixtureInput { + return { + steps: [ + { type: "message", text: "Respond with exactly: compaction fixture ready" }, + { type: "message", text: "/compact" }, + ], + }; +} diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/acp_registry_compaction/output.ts b/apps/server/src/orchestration-v2/testkit/fixtures/acp_registry_compaction/output.ts new file mode 100644 index 0000000000000..60a095e9eea96 --- /dev/null +++ b/apps/server/src/orchestration-v2/testkit/fixtures/acp_registry_compaction/output.ts @@ -0,0 +1,95 @@ +import { assert } from "@effect/vitest"; +import type { ProviderReplayTranscript } from "@t3tools/contracts"; + +import type { OrchestratorV2ScenarioResult } from "../../OrchestratorScenario.ts"; +import { + assertBaseProjection, + assertSemanticProjectionIntegrity, + projectionFor, +} from "../shared.ts"; +import { ACP_REGISTRY_COMPACTION_SUMMARY } from "./input.ts"; + +/** + * A bare `/compact` reaches the registry agent as a `session/prompt` (the + * transcript's `compact.prompt` expectation) and settles as one completed + * compaction row carrying the agent's own summary, not a duplicate T3 marker. + */ +export function assertAcpRegistryCompactionOutput( + result: OrchestratorV2ScenarioResult, + transcript: ProviderReplayTranscript, +) { + assertBaseProjection({ + result, + transcript, + runCount: 2, + runStatuses: ["completed", "completed"], + }); + const projection = projectionFor(result, transcript.scenario); + assertSemanticProjectionIntegrity(projection); + const compactionRun = projection.runs[1]; + assert.deepEqual( + projection.turnItems + .filter((item) => item.type === "compaction") + .map((item) => [item.runId, item.status, item.summary]), + [[compactionRun?.id ?? null, "completed", ACP_REGISTRY_COMPACTION_SUMMARY]], + ); +} + +/** + * An agent without `/compact` answers it like any prompt. Its reply is the + * result; T3 must not add a "Context compacted" row the agent never reported. + */ +export function assertAcpRegistryUnsupportedCompactionOutput( + result: OrchestratorV2ScenarioResult, + transcript: ProviderReplayTranscript, +) { + assertBaseProjection({ + result, + transcript, + runCount: 2, + runStatuses: ["completed", "completed"], + }); + const projection = projectionFor(result, transcript.scenario); + assertSemanticProjectionIntegrity(projection); + const compactionRun = projection.runs[1]; + assert.deepEqual( + projection.turnItems.filter((item) => item.type === "compaction"), + [], + ); + assert.deepEqual( + projection.turnItems.flatMap((item) => + item.runId === compactionRun?.id && item.type === "assistant_message" ? [item.text] : [], + ), + ["Unknown command: /compact"], + ); +} + +/** + * Only the agent's own `completed` report means compaction happened. A turn + * that ends while the agent's compaction is still in progress settles the row + * as interrupted and keeps the reported context usage. + */ +export function assertAcpRegistryUnfinishedCompactionOutput( + result: OrchestratorV2ScenarioResult, + transcript: ProviderReplayTranscript, +) { + assertBaseProjection({ + result, + transcript, + runCount: 2, + runStatuses: ["completed", "completed"], + }); + const projection = projectionFor(result, transcript.scenario); + assertSemanticProjectionIntegrity(projection); + const compactionRun = projection.runs[1]; + assert.deepEqual( + projection.turnItems + .filter((item) => item.type === "compaction") + .map((item) => [item.runId, item.status]), + [[compactionRun?.id ?? null, "interrupted"]], + ); + assert.deepEqual( + projection.providerThreads.map((thread) => thread.contextUsage?.usedTokens), + [420], + ); +} diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/acp_registry_compaction/registry_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/acp_registry_compaction/registry_transcript.ndjson new file mode 100644 index 0000000000000..36b3572c799b7 --- /dev/null +++ b/apps/server/src/orchestration-v2/testkit/fixtures/acp_registry_compaction/registry_transcript.ndjson @@ -0,0 +1,14 @@ +{"type":"transcript_start","provider":"acpRegistry","protocol":"acp.ndjson-jsonrpc","version":"1","scenario":"acp_registry_compaction","metadata":{"generatedBy":"protocol-semantic-fixture","nativeSessionId":"registry-replay-session-1"}} +{"type":"expect_outbound","label":"initialize","frame":{"kind":"request","method":"initialize","params":{"protocolVersion":2,"info":"","capabilities":"","clientCapabilities":"","clientInfo":""}}} +{"type":"emit_inbound","label":"initialized","frame":{"kind":"response","method":"initialize","result":{"protocolVersion":1,"agentCapabilities":{"loadSession":true,"mcpCapabilities":{"http":true,"sse":true},"promptCapabilities":{"audio":false,"embeddedContext":true,"image":false}},"authMethods":[{"id":"replay","name":"Replay"}],"agentInfo":{"name":"registry-replay","version":"1"}}}} +{"type":"expect_outbound","label":"session.new","frame":{"kind":"request","method":"session/new","params":{"cwd":"","mcpServers":""}}} +{"type":"emit_inbound","label":"session.created","frame":{"kind":"response","method":"session/new","result":{"sessionId":"registry-replay-session-1","models":{"currentModelId":"grok-build","availableModels":[{"modelId":"grok-build","name":"Grok Build"}]}}}} +{"type":"emit_inbound","label":"commands.available","frame":{"kind":"notification","method":"session/update","params":{"sessionId":"registry-replay-session-1","update":{"sessionUpdate":"available_commands_update","availableCommands":[{"name":"compact","description":"Compact the conversation context"}]}}}} +{"type":"expect_outbound","label":"first.prompt","frame":{"kind":"request","method":"session/prompt","params":{"sessionId":"registry-replay-session-1","prompt":[{"type":"text","text":"\n\n\n\n\nRespond with exactly: compaction fixture ready\n"}]}}} +{"type":"emit_inbound","label":"first.assistant","frame":{"kind":"notification","method":"session/update","params":{"sessionId":"registry-replay-session-1","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"compaction fixture ready"}}}}} +{"type":"emit_inbound","label":"first.completed","frame":{"kind":"response","method":"session/prompt","result":{"stopReason":"end_turn"}}} +{"type":"expect_outbound","label":"compact.prompt","frame":{"kind":"request","method":"session/prompt","params":{"sessionId":"registry-replay-session-1","prompt":[{"type":"text","text":"/compact"}]}}} +{"type":"emit_inbound","label":"compact.started","frame":{"kind":"notification","method":"session/update","params":{"sessionId":"registry-replay-session-1","update":{"sessionUpdate":"compaction_update","compactionId":"compaction-1","status":"in_progress"}}}} +{"type":"emit_inbound","label":"compact.completed","frame":{"kind":"notification","method":"session/update","params":{"sessionId":"registry-replay-session-1","update":{"sessionUpdate":"compaction_update","compactionId":"compaction-1","status":"completed","summary":[{"type":"text","text":"Retained the fixture decisions."}]}}}} +{"type":"emit_inbound","label":"compact.turn.completed","frame":{"kind":"response","method":"session/prompt","result":{"stopReason":"end_turn"}}} +{"type":"runtime_exit","status":"success"} diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/acp_registry_compaction/unfinished_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/acp_registry_compaction/unfinished_transcript.ndjson new file mode 100644 index 0000000000000..ffdc6fcf1dc32 --- /dev/null +++ b/apps/server/src/orchestration-v2/testkit/fixtures/acp_registry_compaction/unfinished_transcript.ndjson @@ -0,0 +1,14 @@ +{"type":"transcript_start","provider":"acpRegistry","protocol":"acp.ndjson-jsonrpc","version":"1","scenario":"acp_registry_compaction_unfinished","metadata":{"generatedBy":"protocol-semantic-fixture","nativeSessionId":"registry-replay-session-1"}} +{"type":"expect_outbound","label":"initialize","frame":{"kind":"request","method":"initialize","params":{"protocolVersion":2,"info":"","capabilities":"","clientCapabilities":"","clientInfo":""}}} +{"type":"emit_inbound","label":"initialized","frame":{"kind":"response","method":"initialize","result":{"protocolVersion":1,"agentCapabilities":{"loadSession":true,"mcpCapabilities":{"http":true,"sse":true},"promptCapabilities":{"audio":false,"embeddedContext":true,"image":false}},"authMethods":[{"id":"replay","name":"Replay"}],"agentInfo":{"name":"registry-replay","version":"1"}}}} +{"type":"expect_outbound","label":"session.new","frame":{"kind":"request","method":"session/new","params":{"cwd":"","mcpServers":""}}} +{"type":"emit_inbound","label":"session.created","frame":{"kind":"response","method":"session/new","result":{"sessionId":"registry-replay-session-1","models":{"currentModelId":"grok-build","availableModels":[{"modelId":"grok-build","name":"Grok Build"}]}}}} +{"type":"emit_inbound","label":"commands.available","frame":{"kind":"notification","method":"session/update","params":{"sessionId":"registry-replay-session-1","update":{"sessionUpdate":"available_commands_update","availableCommands":[{"name":"compact","description":"Compact the conversation context"}]}}}} +{"type":"expect_outbound","label":"first.prompt","frame":{"kind":"request","method":"session/prompt","params":{"sessionId":"registry-replay-session-1","prompt":[{"type":"text","text":"\n\n\n\n\nRespond with exactly: compaction fixture ready\n"}]}}} +{"type":"emit_inbound","label":"first.assistant","frame":{"kind":"notification","method":"session/update","params":{"sessionId":"registry-replay-session-1","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"compaction fixture ready"}}}}} +{"type":"emit_inbound","label":"first.completed","frame":{"kind":"response","method":"session/prompt","result":{"stopReason":"end_turn"}}} +{"type":"expect_outbound","label":"compact.prompt","frame":{"kind":"request","method":"session/prompt","params":{"sessionId":"registry-replay-session-1","prompt":[{"type":"text","text":"/compact"}]}}} +{"type":"emit_inbound","label":"usage.before.compact","frame":{"kind":"notification","method":"session/update","params":{"sessionId":"registry-replay-session-1","update":{"sessionUpdate":"usage_update","used":420,"size":1000}}}} +{"type":"emit_inbound","label":"compact.started","frame":{"kind":"notification","method":"session/update","params":{"sessionId":"registry-replay-session-1","update":{"sessionUpdate":"compaction_update","compactionId":"compaction-1","status":"in_progress"}}}} +{"type":"emit_inbound","label":"compact.turn.completed","frame":{"kind":"response","method":"session/prompt","result":{"stopReason":"end_turn"}}} +{"type":"runtime_exit","status":"success"} diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/acp_registry_compaction/unsupported_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/acp_registry_compaction/unsupported_transcript.ndjson new file mode 100644 index 0000000000000..ad8c8a5367203 --- /dev/null +++ b/apps/server/src/orchestration-v2/testkit/fixtures/acp_registry_compaction/unsupported_transcript.ndjson @@ -0,0 +1,13 @@ +{"type":"transcript_start","provider":"acpRegistry","protocol":"acp.ndjson-jsonrpc","version":"1","scenario":"acp_registry_compaction_unsupported","metadata":{"generatedBy":"protocol-semantic-fixture","nativeSessionId":"registry-replay-session-1"}} +{"type":"expect_outbound","label":"initialize","frame":{"kind":"request","method":"initialize","params":{"protocolVersion":2,"info":"","capabilities":"","clientCapabilities":"","clientInfo":""}}} +{"type":"emit_inbound","label":"initialized","frame":{"kind":"response","method":"initialize","result":{"protocolVersion":1,"agentCapabilities":{"loadSession":true,"mcpCapabilities":{"http":true,"sse":true},"promptCapabilities":{"audio":false,"embeddedContext":true,"image":false}},"authMethods":[{"id":"replay","name":"Replay"}],"agentInfo":{"name":"registry-replay","version":"1"}}}} +{"type":"expect_outbound","label":"session.new","frame":{"kind":"request","method":"session/new","params":{"cwd":"","mcpServers":""}}} +{"type":"emit_inbound","label":"session.created","frame":{"kind":"response","method":"session/new","result":{"sessionId":"registry-replay-session-1","models":{"currentModelId":"grok-build","availableModels":[{"modelId":"grok-build","name":"Grok Build"}]}}}} +{"type":"emit_inbound","label":"commands.available","frame":{"kind":"notification","method":"session/update","params":{"sessionId":"registry-replay-session-1","update":{"sessionUpdate":"available_commands_update","availableCommands":[{"name":"review","description":"Review the current changes"}]}}}} +{"type":"expect_outbound","label":"first.prompt","frame":{"kind":"request","method":"session/prompt","params":{"sessionId":"registry-replay-session-1","prompt":[{"type":"text","text":"\n\n\n\n\nRespond with exactly: compaction fixture ready\n"}]}}} +{"type":"emit_inbound","label":"first.assistant","frame":{"kind":"notification","method":"session/update","params":{"sessionId":"registry-replay-session-1","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"compaction fixture ready"}}}}} +{"type":"emit_inbound","label":"first.completed","frame":{"kind":"response","method":"session/prompt","result":{"stopReason":"end_turn"}}} +{"type":"expect_outbound","label":"compact.prompt","frame":{"kind":"request","method":"session/prompt","params":{"sessionId":"registry-replay-session-1","prompt":[{"type":"text","text":"/compact"}]}}} +{"type":"emit_inbound","label":"compact.unknown","frame":{"kind":"notification","method":"session/update","params":{"sessionId":"registry-replay-session-1","update":{"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":"Unknown command: /compact"}}}}} +{"type":"emit_inbound","label":"compact.turn.completed","frame":{"kind":"response","method":"session/prompt","result":{"stopReason":"end_turn"}}} +{"type":"runtime_exit","status":"success"} diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/index.ts b/apps/server/src/orchestration-v2/testkit/fixtures/index.ts index 811598474bac5..1f38a980d7ba7 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/index.ts +++ b/apps/server/src/orchestration-v2/testkit/fixtures/index.ts @@ -87,6 +87,12 @@ import { assertMuseWorkflowOutput } from "./muse_workflow/output.ts"; import { assertMuseSubagentV2Output } from "./subagent_v2/muse_output.ts"; import { assertMuseMultiTurnOutput } from "./multi_turn/muse_output.ts"; import { assertMuseTurnInterruptOutput } from "./turn_interrupt/muse_output.ts"; +import { acpRegistryCompactionInput } from "./acp_registry_compaction/input.ts"; +import { + assertAcpRegistryCompactionOutput, + assertAcpRegistryUnfinishedCompactionOutput, + assertAcpRegistryUnsupportedCompactionOutput, +} from "./acp_registry_compaction/output.ts"; import { piCompactionInput } from "./pi_compaction/input.ts"; import { assertPiCompactionOutput } from "./pi_compaction/output.ts"; import { providerThreadResumeInput } from "./provider_thread_resume/input.ts"; @@ -1544,6 +1550,51 @@ export const ORCHESTRATOR_REPLAY_FIXTURES: ReadonlyArray entry.slug), + compaction: registryAgentId === "devin" ? "native-async" : "native", promptFailure: (cause) => acpRegistryPromptFailure(registryAgentId, cause), ...(registryAgentId === "factory-droid" ? { extractElicitationForm: extractDroidElicitationForm } @@ -242,6 +244,7 @@ export const makeAcpRegistryAdapterV2 = Effect.fn("makeAcpRegistryAdapterV2")(fu "cognition.ai/subagentSupport": true, "cognition.ai/messageGrouping": true, }, + registerExtensions: registerDevinCompaction, normalizeSessionUpdate: normalizeDevinSessionUpdate, normalizeToolCall: normalizeDevinToolCall, extractSubagentUpdate: extractDevinSubagentUpdate, diff --git a/packages/provider-acp-registry/src/server/devinAcp.ts b/packages/provider-acp-registry/src/server/devinAcp.ts index 4368a4cfaf437..30dd09b9e56ae 100644 --- a/packages/provider-acp-registry/src/server/devinAcp.ts +++ b/packages/provider-acp-registry/src/server/devinAcp.ts @@ -1,3 +1,5 @@ +import * as Schema from "effect/Schema"; +import type { AcpAdapterV2ExtensionContext } from "@t3tools/provider-acp/server/adapter"; import type * as AcpSchema from "effect-acp/compat"; import type { AcpToolCallState } from "@t3tools/provider-acp/server/runtimeModel"; import type { AcpAdapterV2SubagentUpdate } from "@t3tools/provider-acp/server/adapter"; @@ -54,3 +56,15 @@ export function extractDevinSubagentUpdate( result: typeof completed?.summary === "string" ? completed.summary : null, }; } + +export function registerDevinCompaction(context: AcpAdapterV2ExtensionContext) { + return context.runtime.handleExtNotification( + "_cognition.ai/compaction", + Schema.Struct({ + sessionId: Schema.String, + status: Schema.Literals(["started", "completed", "failed", "cancelled"]), + summary: Schema.optionalKey(Schema.String), + }), + context.reportCompaction, + ); +} diff --git a/packages/provider-acp/src/server/adapter.ts b/packages/provider-acp/src/server/adapter.ts index 2110f9616de54..85ed1cc2ca828 100644 --- a/packages/provider-acp/src/server/adapter.ts +++ b/packages/provider-acp/src/server/adapter.ts @@ -178,6 +178,11 @@ export interface AcpAdapterV2UserInputRequest { export interface AcpAdapterV2ExtensionContext { readonly runtime: AcpSessionRuntime.AcpSessionRuntime["Service"]; + readonly reportCompaction: (notice: { + readonly sessionId: string; + readonly status: "started" | "completed" | "failed" | "cancelled"; + readonly summary?: string | undefined; + }) => Effect.Effect; readonly reportProviderRetry: (input: { readonly sessionId: string; readonly failure: OrchestrationV2ProviderFailure; @@ -416,7 +421,8 @@ export interface AcpAdapterV2Flavor { event: AcpSessionRuntime.AcpSessionRuntimeEvent, ) => Effect.Effect; readonly subagentsIdleOnTurnCompletion?: boolean; - readonly supportsCompaction?: boolean; + // Marker agents compact silently; native agents must report their own outcome. + readonly compaction?: "marker" | "native" | "native-async"; /** * Native steering: `send` queues a message into the running turn and returns the * agent's id for it, which `injectedSteerId` reads from the update that delivers it. @@ -1017,7 +1023,7 @@ function commandExitCode(value: unknown): number | undefined { * interrupted tools must not retain that stale success code. */ function acpProjectedCommandExitCode( - status: "pending" | "running" | "waiting" | "completed" | "failed" | "interrupted", + status: ProjectedToolStatus, rawOutput: unknown, ): number | undefined { if (status !== "completed" && status !== "failed") { @@ -1224,7 +1230,7 @@ function toolStatus( } } -type ProjectedToolStatus = ReturnType | "interrupted"; +type ProjectedToolStatus = ReturnType | "interrupted" | "cancelled"; /** * Some agents answer Stop's `session/cancel` by failing the tools and subagents @@ -1243,7 +1249,12 @@ function nodeStatus(status: ProjectedToolStatus): OrchestrationV2ExecutionNode[" } function completedAtForStatus(status: ProjectedToolStatus, now: DateTime.Utc): DateTime.Utc | null { - return status === "completed" || status === "failed" || status === "interrupted" ? now : null; + return status === "completed" || + status === "failed" || + status === "interrupted" || + status === "cancelled" + ? now + : null; } // Copilot advertises URL ids such as `https://agentclientprotocol.com/protocol/session-modes#plan`. @@ -1364,6 +1375,7 @@ interface ActiveAcpTurn { interrupted: boolean; finalized: boolean; finalizedStatus: "completed" | "interrupted" | "failed" | "cancelled" | null; + compactionStatus?: "running" | "completed" | "failed" | "cancelled" | undefined; promptSettled: boolean; promptSettledStatus: "completed" | "interrupted" | "failed" | "cancelled" | null; workAccepted: boolean; @@ -2247,7 +2259,17 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( runtimeGeneration, Effect.gen(function* () { yield* Ref.set(runtimeRestartRequired, true); - if (!followsAgentWorkState) return; + if (!followsAgentWorkState) { + const context = yield* Ref.get(activeTurn); + if (context?.compactionStatus === "running") { + yield* finalizeTurn( + context, + "failed", + makeProviderFailure({ class: "transport_error" }), + ); + } + return; + } agentWork = { sessionUpdate: "state_update", state: "idle", stopReason: "error" }; for (const child of nativeChildren.values()) { if (child.status !== "running") continue; @@ -6664,6 +6686,43 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( if (flavor.registerExtensions !== undefined) { yield* flavor.registerExtensions({ runtime: targetRuntime, + reportCompaction: (notice) => + runRuntimeCallbackAtGeneration( + handlerGeneration, + Effect.gen(function* () { + const context = yield* Ref.get(activeTurn); + if ( + context === null || + context.finalized || + context.interrupted || + context.compactionStatus === undefined || + context.nativeThreadId !== notice.sessionId + ) + return; + context.compactionStatus = + notice.status === "started" ? "running" : notice.status; + const toolCallId = `acp-compaction:${context.nativeTurnId}`; + yield* emitTool( + context, + { + toolCallId, + title: "Compact context", + kind: "think", + status: + notice.status === "started" + ? "inProgress" + : notice.status === "completed" + ? "completed" + : "failed", + data: { toolCallId, rawOutput: notice.summary }, + }, + context.compactionStatus, + ); + if (context.promptSettled && context.compactionStatus !== "running") { + yield* finalizeTurn(context, context.compactionStatus); + } + }), + ).pipe(Effect.asVoid), reportProviderRetry: (notice) => runRuntimeCallbackAtGeneration(handlerGeneration, reportProviderRetry(notice)).pipe( Effect.asVoid, @@ -7422,7 +7481,13 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( for (const tool of context.tools.values()) { if (!acpCompletedTurnShouldTerminalizeTool(tool, { ...flavor, extractSubagentUpdate })) continue; - yield* emitTool(context, tool, status); + yield* emitTool( + context, + tool, + status === "completed" && tool.toolCallId.startsWith("acp-compaction:") + ? "interrupted" + : status, + ); } }); @@ -7474,6 +7539,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ): Effect.fn.Return { if (context.finalized) return; const settledStatus = context.interrupted ? "interrupted" : status; + if (context.compactionStatus === "running") yield* Ref.set(runtimeRestartRequired, true); context.finalizedStatus = settledStatus; context.finalized = true; const directStopQuarantine = yield* Ref.get(stoppedRunQuarantine); @@ -7498,15 +7564,28 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( terminalizeSubagents: directStopQuarantine, }); } + for (const tool of context.tools.values()) { + if ( + tool.toolCallId.startsWith("acp-compaction:") && + (tool.status === "pending" || tool.status === "inProgress") + ) { + yield* emitTool( + context, + tool, + settledStatus === "completed" ? "interrupted" : settledStatus, + ); + } + } yield* closeTextStreams(context); if (settledStatus !== "failed") { yield* emitProviderRetry(context, settledStatus); } const now = yield* DateTime.now; if ( - (flavor.supportsCompaction === true || flavor.compactSession !== undefined) && + (flavor.compaction === "marker" || flavor.compactSession !== undefined) && isCompactionMessage(context.input.message) && - settledStatus === "completed" + settledStatus === "completed" && + !Array.from(context.tools.keys()).some((id) => id.startsWith("acp-compaction:")) ) { const nativeItemId = `${context.nativeTurnId}:compaction`; yield* Ref.update(compactionMarkers, (current) => { @@ -7971,6 +8050,12 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( interrupted: false, finalized: false, finalizedStatus: null, + compactionStatus: + flavor.compaction === "native-async" && + turnInput.message.text.trim() === "/compact" && + turnInput.message.attachments.length === 0 + ? "running" + : undefined, promptSettled: false, promptSettledStatus: null, workAccepted: false, @@ -8227,7 +8312,19 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ); return; } - yield* finalizeTurn(context, status); + context.promptSettled = true; + if ( + status === "completed" && + !context.interrupted && + context.compactionStatus === "running" + ) + return; + yield* finalizeTurn( + context, + status === "completed" && context.compactionStatus !== "running" + ? (context.compactionStatus ?? status) + : status, + ); }), ).pipe(Effect.asVoid), ), @@ -8533,7 +8630,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* ( ), ), startTurn, - ...(flavor.supportsCompaction === true || flavor.compactSession !== undefined + ...(flavor.compaction !== undefined || flavor.compactSession !== undefined ? { compactThread: (turnInput: ProviderAdapter.ProviderAdapterV2TurnInput) => startTurn({ diff --git a/packages/provider-grok/src/server/adapter.ts b/packages/provider-grok/src/server/adapter.ts index d826315b69fa6..6c7fdf9e90ea7 100644 --- a/packages/provider-grok/src/server/adapter.ts +++ b/packages/provider-grok/src/server/adapter.ts @@ -235,7 +235,7 @@ export function makeGrokAcpAdapterFlavor( preserveRuntimeOnSettledInterrupt: true, // Grok accepts image blocks despite advertising promptCapabilities.image:false. supportsImagePrompts: true, - supportsCompaction: true, + compaction: "marker", // Grok expands only a leading `/name`, while the `$` picker inserts `$name` from the // `grok inspect` catalog. Ask that catalog again so other `$` text is left alone. rewriteUserText: (text, turnInput) => {