From 715946889c0a3559c07c976bb8b7dbe9c64b86fd Mon Sep 17 00:00:00 2001 From: Jreevo Date: Tue, 6 Oct 2026 15:13:31 -0700 Subject: [PATCH 1/4] fix(server): a Claude /goal no longer reads as complete while a background subagent runs Claude Code skips a goal's Stop hook check while background work runs and checks again on a later turn. The adapter inferred completion from a normal turn end plus an empty roster, but the roster leaves subagents out, so a turn that ended while a background Agent ran marked the goal complete. The completion check now also requires that the live process runs no background work of the types Claude Code 2.1.292 waits on: local_agent, remote_agent, in_process_teammate and local_workflow on the subagent path (local_bash is already on the roster). Tasks of other types, such as mcp_task and monitor_ws, are exempt, as they are in Claude. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../Adapters/ClaudeAdapterV2.test.ts | 216 ++++++++++++++++++ .../Adapters/ClaudeAdapterV2.ts | 35 ++- 2 files changed, 249 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index ec65e9a5ac5c..9aaec15d9123 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -14018,6 +14018,222 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + // Level entries from Claude Code 2.1.292 carry no ambient flag. + const backgroundTasksLevel = ( + uuid: string, + tasks: ReadonlyArray<{ readonly taskId: string; readonly taskType: string }>, + ) => + claudeSdkFrame({ + type: "system", + subtype: "background_tasks_changed", + tasks: tasks.map((task) => ({ + task_id: task.taskId, + task_type: task.taskType, + description: "Background task", + })), + uuid, + session_id: WAKE_NATIVE_SESSION, + } as unknown as SDKMessage); + // Frame order captured from Claude Code 2.1.292: the Agent call, the + // level, the backgrounded task_started, its async ack, the model's reply, + // then the result while the subagent still runs. + const backgroundAgentLaunchFrames = (input: { + readonly prefix: string; + readonly taskId: string; + }) => { + const toolUseId = `toolu-${input.prefix}`; + return [ + claudeSdkFrame({ + type: "assistant", + message: { + model: "claude-sonnet-4-6", + id: `msg-${input.prefix}-launch`, + type: "message", + role: "assistant", + content: [ + { + type: "tool_use", + id: toolUseId, + name: "Agent", + input: { + description: "Implement the fix", + subagent_type: "general-purpose", + run_in_background: true, + prompt: "Implement the fix.", + }, + }, + ], + }, + parent_tool_use_id: null, + uuid: `${input.prefix}-launch`, + session_id: WAKE_NATIVE_SESSION, + }), + backgroundTasksLevel(`${input.prefix}-level`, [ + { taskId: input.taskId, taskType: "local_agent" }, + ]), + claudeSdkFrame({ + type: "system", + subtype: "task_started", + task_id: input.taskId, + tool_use_id: toolUseId, + description: "Implement the fix", + subagent_type: "general-purpose", + is_backgrounded: true, + task_type: "local_agent", + prompt: "Implement the fix.", + uuid: `${input.prefix}-task-started`, + session_id: WAKE_NATIVE_SESSION, + }), + claudeSdkFrame({ + type: "user", + message: { + role: "user", + content: [ + { + type: "tool_result", + tool_use_id: toolUseId, + content: [{ type: "text", text: "Async agent launched successfully." }], + }, + ], + }, + parent_tool_use_id: null, + uuid: `${input.prefix}-ack`, + session_id: WAKE_NATIVE_SESSION, + }), + ]; + }; + + it.effect("defers goal completion while a background subagent runs", () => + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const condition = "both reviewers find no bugs"; + const taskId = "task-goal-background-subagent"; + const now = yield* DateTime.now; + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("goal-background-attempt"), + text: `/goal ${condition}`, + attachments: [], + }), + ); + for (const frame of [ + syntheticFrame("goal-background-set", `Goal set: ${condition}`), + ...backgroundAgentLaunchFrames({ prefix: "goal-background", taskId }), + makeAssistantTextFrame({ + uuid: "goal-background-wait", + text: "Waiting for the background subagent before round 1.", + }), + makeResultFrame({ + uuid: "goal-background-result", + result: "Waiting for the background subagent before round 1.", + }), + ]) { + yield* Queue.offer(harness.sdkMessages, frame); + } + yield* Queue.take(harness.terminalReceipts); + // Claude skips the goal check while the subagent runs. + assert.equal(goalStatuses(harness.events).at(-1)?.status, "active"); + + // The subagent ends; Claude opens a turn on its own, and the goal + // check at its end passes. + yield* harness.offerAndWait( + claudeSdkFrame({ + type: "system", + subtype: "task_notification", + task_id: taskId, + tool_use_id: "toolu-goal-background", + status: "completed", + output_file: `/tmp/${taskId}.output`, + summary: "Fix implemented.", + uuid: "goal-background-notification", + session_id: WAKE_NATIVE_SESSION, + }), + ); + yield* harness.offerAndWait(backgroundTasksLevel("goal-background-level-empty", [])); + yield* harness.offerAndWait(wakeTurnInit); + assert.lengthOf(harness.continuationRequests, 1); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("goal-background-wake"), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 2, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + yield* harness.offerAndWait( + makeAssistantTextFrame({ uuid: "goal-background-wake-work", text: "No bugs found." }), + ); + yield* harness.offerAndWait( + makeResultFrame({ + uuid: "goal-background-wake-result", + result: "No bugs found.", + origin: { kind: "task-notification" }, + }), + ); + yield* Queue.take(harness.terminalReceipts); + assert.deepEqual(goalStatuses(harness.events).at(-1), { + objective: condition, + status: "complete", + checks: 0, + }); + }).pipe(Effect.scoped, Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + ); + + // Claude's goal check waits on agents, workflows and shells by task type, + // ambient or not, and never on mcp_task or monitor_mcp/monitor_ws tasks. + // A task with no type is treated as one Claude waits on. + it.effect.each([ + { taskType: "mcp_task", ambient: false, status: "complete" }, + { taskType: "local_agent", ambient: true, status: "active" }, + { taskType: undefined, ambient: false, status: "active" }, + ] as const)("treats a running $taskType (ambient $ambient) as goal $status", (input) => + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const condition = "the docs build"; + const taskId = `task-goal-${input.taskType}`; + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make(`goal-${input.taskType}-attempt`), + text: `/goal ${condition}`, + attachments: [], + }), + ); + for (const frame of [ + syntheticFrame(`goal-${input.taskType}-set`, `Goal set: ${condition}`), + backgroundTasksLevel(`goal-${input.taskType}-level`, [ + { taskId, taskType: input.taskType ?? "local_agent" }, + ]), + claudeSdkFrame({ + type: "system", + subtype: "task_started", + task_id: taskId, + description: "Watch the docs", + ...(input.taskType === undefined ? {} : { task_type: input.taskType }), + ...(input.ambient ? { skip_transcript: true, ambient: true } : {}), + uuid: `goal-${input.taskType}-task-started`, + session_id: WAKE_NATIVE_SESSION, + }), + makeAssistantTextFrame({ uuid: `goal-${input.taskType}-work`, text: "The docs build." }), + makeResultFrame({ uuid: `goal-${input.taskType}-result`, result: "The docs build." }), + ]) { + yield* Queue.offer(harness.sdkMessages, frame); + } + yield* Queue.take(harness.terminalReceipts); + assert.equal(goalStatuses(harness.events).at(-1)?.status, input.status); + }).pipe(Effect.scoped, Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + ); + it.effect("keeps a goal active after a command turn with no model output", () => Effect.gen(function* () { const harness = yield* makeWakeHarness; diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 9a871c49d400..73d97cdd7d1f 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -2043,6 +2043,17 @@ const CLAUDE_OPAQUE_BACKGROUND_TASK_KINDS: ReadonlyMap< Exclude > = new Map([["local_bash", "command"]]); +// Task types whose running tasks make Claude skip a goal's Stop hook check, +// as Claude Code 2.1.292 decides it. local_bash (Monitor tool runs included) +// is the roster's; the rest run on the subagent path. mcp_task, monitor_mcp +// and monitor_ws tasks do not defer the check. +const CLAUDE_GOAL_DEFERRING_TASK_TYPES: ReadonlySet = new Set([ + "local_agent", + "remote_agent", + "in_process_teammate", + "local_workflow", +]); + function isClaudeOpaqueBackgroundTaskType(taskType: string | null | undefined): boolean { return typeof taskType === "string" && CLAUDE_OPAQUE_BACKGROUND_TASK_KINDS.has(taskType); } @@ -3633,6 +3644,8 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }>({ userTurns: new Map(), reports: new Map() }); // Subagents Claude started in the background. Only their ends wake the root. const backgroundedSubagentTaskIds = yield* Ref.make>(new Set()); + // Subagent-path tasks whose type Claude's goal check does not wait for. + const goalExemptTaskIds = yield* Ref.make>(new Set()); // Subagents another subagent started (spawn_depth above 1). Claude // reports their end to the owning subagent, so it never wakes the root. const nestedSubagentTaskIds = yield* Ref.make>(new Set()); @@ -6072,20 +6085,27 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( input.context.input.providerThread.nativeConversationHeadRef !== null; // While a goal is set, Claude ends a turn on its own only after the // goal's Stop hook passes. It defers that check while background - // work runs, and a hook that stops the turn reports another reason. + // work of the types it waits on runs (subagents stay off the roster; + // see CLAUDE_GOAL_DEFERRING_TASK_TYPES), and a hook that stops the + // turn reports another reason. // SDK mode does not report an evaluator timeout or an impossible // verdict, so those still read as complete. const goal = nativeThreadId === null ? undefined : goalsByNativeThread.get(nativeThreadId); const goalChecked = goalCheckedTurns.delete(input.context.providerTurnId); const terminalReason = input.result?.terminal_reason; + const liveQuery = yield* Ref.get(queryContext); if ( nativeThreadId !== null && goal?.status === "active" && goalChecked && input.status === "completed" && (terminalReason === undefined || terminalReason === "completed") && - roster.size === 0 + roster.size === 0 && + !( + liveQuery?.nativeThreadId === nativeThreadId && + (yield* liveProcessRunsBackgroundWork(liveQuery, { forGoalCheck: true })) + ) ) { goalsByNativeThread.set(nativeThreadId, { objective: goal.objective, @@ -7253,6 +7273,14 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( activeContext: context, }); } else { + if ( + message.task_type !== undefined && + !CLAUDE_GOAL_DEFERRING_TASK_TYPES.has(message.task_type) + ) { + yield* Ref.update(goalExemptTaskIds, (current) => + new Set(current).add(message.task_id), + ); + } const launch = message.tool_use_id === undefined ? undefined @@ -8361,6 +8389,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( // already buffered is done: the buffer outlives the process. const liveProcessRunsBackgroundWork = Effect.fnUntraced(function* ( live: ClaudeLiveQueryContext, + options?: { readonly forGoalCheck?: boolean }, ) { if ( rosterForNativeThread( @@ -8371,10 +8400,12 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( return true; } const buffered = (yield* Ref.get(wakeBuffers)).get(live.nativeThreadId)?.messages ?? []; + const ignored = options?.forGoalCheck === true ? yield* Ref.get(goalExemptTaskIds) : null; for (const [taskId, subagent] of yield* Ref.get(sessionSubagentsByTaskId)) { if ( subagent.task.status === "running" && !live.subagentsFromEarlierProcesses.has(subagent) && + !(ignored?.has(taskId) ?? false) && !buffered.some( (message) => message.type === "system" && From 83ae424c6a92ee62646b1f45efd5629f39aeec93 Mon Sep 17 00:00:00 2001 From: Chrysler Ade Chandra <10206122+AdeChrysler@users.noreply.github.com> Date: Wed, 14 Oct 2026 22:15:06 +0800 Subject: [PATCH 2/4] test(server): provide MCP provider sessions to the background goal tests The adapter now needs McpProviderSessions, so the two goal-deferral tests from the cherry-picked fix use the same test layer as the rest of the file. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/ClaudeAdapterV2.test.ts | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 9aaec15d9123..158761eff43d 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -14184,7 +14184,12 @@ describe("ClaudeAdapterV2 background wake turns", () => { status: "complete", checks: 0, }); - }).pipe(Effect.scoped, Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), ); // Claude's goal check waits on agents, workflows and shells by task type, @@ -14231,7 +14236,12 @@ describe("ClaudeAdapterV2 background wake turns", () => { } yield* Queue.take(harness.terminalReceipts); assert.equal(goalStatuses(harness.events).at(-1)?.status, input.status); - }).pipe(Effect.scoped, Effect.provide(Layer.merge(IdAllocator.layer, NodeServices.layer))), + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll(IdAllocator.layer, McpProviderSessions.layer, NodeServices.layer), + ), + ), ); it.effect("keeps a goal active after a command turn with no model output", () => From 6297edd000ecdd4d53b314703dc0f819232d2d59 Mon Sep 17 00:00:00 2001 From: Chrysler Ade Chandra <10206122+AdeChrysler@users.noreply.github.com> Date: Wed, 14 Oct 2026 22:15:06 +0800 Subject: [PATCH 3/4] fix(server): a Claude /goal reads as complete only when Claude says it is met T3 marked a Claude goal complete whenever a turn ended normally with no background work left, because in SDK mode Claude reports a goal's verdict only in its session transcript. A check that Claude skipped, one whose evaluator failed or timed out, or a goal Claude judged impossible all read as complete, and stayed that way. The query runner now gives the SDK a session store, so the CLI mirrors its transcript on stdout (--session-mirror), and keeps only the root session's goal_status records. They are applied when they arrive, including after the turn ended: a met record completes the goal, an unmet one counts a check and reopens a complete goal, a set or cleared record starts or drops it, and an impossible verdict drops it (Claude ends the goal, and Resume would send `/goal resume`, which Claude reads as a new goal). While a process delivers these records, a turn end no longer implies a met goal and Stop hook feedback no longer counts checks. A CLI whose mirror delivers nothing keeps the inferred completion. The store loads nothing, so a resumed CLI still reads its own transcript and config directory. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/ClaudeAdapterV2.test.ts | 674 ++++++++++++++++-- .../Adapters/ClaudeAdapterV2.ts | 273 ++++++- docs/internals/providers.md | 9 +- docs/user/composer.md | 1 + 4 files changed, 870 insertions(+), 87 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 158761eff43d..164d42ac569b 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -4,6 +4,7 @@ import * as NodeFS from "node:fs"; import * as NodeOS from "node:os"; import * as NodePath from "node:path"; import * as NodeStream from "node:stream"; +import * as NodeV8 from "node:v8"; import type { Query as ClaudeQuery, @@ -1937,15 +1938,25 @@ describe("ClaudeAdapterV2 native fork", () => { // Stands in for the Claude CLI behind the SDK's spawn hook: records how it was // started and answers stdin control requests, except mcp_set_servers if asked. -function makeFakeClaudeCli(answerSetServers = true) { +// `onPrompt` answers each user message with the frames it emits. +function makeFakeClaudeCli( + answerSetServers = true, + onPrompt?: (emit: (frame: object) => void) => void, +) { const spawns: Array = []; const controlRequests: Array = []; const stdins: Array = []; + const stdouts: Array = []; + // Writes one stdout frame of the latest process. + const emit = (frame: object) => { + stdouts.at(-1)?.write(`${JSON.stringify(frame)}\n`); + }; const spawn = (options: SpawnOptions): SpawnedProcess => { spawns.push(options); const stdin = new NodeStream.PassThrough(); const stdout = new NodeStream.PassThrough(); stdins.push(stdin); + stdouts.push(stdout); const child = Object.assign(new NodeEvents.EventEmitter(), { stdin, stdout, @@ -1963,6 +1974,7 @@ function makeFakeClaudeCli(answerSetServers = true) { for (let end = pending.indexOf("\n"); end >= 0; end = pending.indexOf("\n")) { const frame = JSON.parse(pending.slice(0, end)); pending = pending.slice(end + 1); + if (frame.type === "user") onPrompt?.(emit); if (frame.type !== "control_request") continue; controlRequests.push(frame.request); const setServers = frame.request.subtype === "mcp_set_servers"; @@ -1983,7 +1995,7 @@ function makeFakeClaudeCli(answerSetServers = true) { }; // The SDK closes the CLI by ending its stdin. const closed = () => stdins.length > 0 && stdins.every((stdin) => stdin.writableEnded); - return { spawn, spawns, controlRequests, closed }; + return { spawn, spawns, controlRequests, closed, emit }; } describe("ClaudeAdapterV2 MCP credential channel", () => { @@ -2049,6 +2061,198 @@ describe("ClaudeAdapterV2 MCP credential channel", () => { ); }); +// The SDK's transcript mirror, end to end through its real query(): Claude Code +// 2.1.299 writes `/goal` verdicts only to the session transcript, and emits +// that transcript as `transcript_mirror` frames when a session store is set. +describe("ClaudeAdapterV2 goal status mirror", () => { + // Never read or written: the fake CLI only names files under it. + const configDir = "/tmp/t3-claude-goal-mirror-config"; + const nativeThreadId = "a1b2c3d4-0000-4000-8000-000000000001"; + const projectDir = `${configDir}/projects/-tmp-goal-mirror`; + const transcript = `${projectDir}/${nativeThreadId}.jsonl`; + const condition = "the assistant has said pineapple in two separate turns"; + const entry = (uuid: string, attachment: object, isSidechain = false) => ({ + parentUuid: null, + isSidechain, + type: "attachment", + uuid, + timestamp: "2026-10-14T12:26:45.582Z", + entrypoint: "sdk-ts", + sessionId: nativeThreadId, + version: "2.1.299", + attachment, + }); + const unmet = { + type: "goal_status", + met: false, + condition, + reason: "Only one assistant turn says pineapple.", + } as const; + const met = { + type: "goal_status", + met: true, + condition, + reason: "Two separate assistant turns say pineapple.", + iterations: 3, + durationMs: 14_310, + tokens: 2_048, + } as const; + const cleared = { type: "goal_status", met: true, sentinel: true, condition } as const; + const t3McpServer = { type: "http" as const, url: "http://127.0.0.1:43123/mcp" }; + const openWith = (cli: ReturnType, resume = false) => + Effect.gen(function* () { + const runner = yield* ClaudeAdapterV2.ClaudeAgentSdkQueryRunner; + return yield* runner.open({ + threadId: ThreadId.make("thread-claude-goal-mirror"), + providerSessionId: ProviderSessionId.make("provider-session-claude-goal-mirror"), + options: { + ...ClaudeAdapterV2.makeClaudeQueryOptions({ + modelSelection: CLAUDE_TEST_MODEL_SELECTION, + nativeThreadId, + resume, + cwd: null, + environment: { PATH: "/usr/bin", CLAUDE_CONFIG_DIR: configDir }, + mcpServers: { "t3-code": t3McpServer }, + }), + pathToClaudeCodeExecutable: "/opt/claude/cli.js", + spawnClaudeCodeProcess: cli.spawn, + }, + }); + }).pipe( + Effect.provide(ClaudeAdapterV2.layerQueryRunner), + Effect.provideService( + ProviderEventLoggers.ProviderEventLoggers, + ProviderEventLoggers.NoOpProviderEventLoggers, + ), + Effect.provide(NodeServices.layer), + ); + const prompt: SDKUserMessage = { + type: "user", + message: { role: "user", content: "Say pineapple." }, + parent_tool_use_id: null, + }; + + const resultFrame = { + type: "result", + subtype: "success", + is_error: false, + duration_ms: 1, + duration_api_ms: 1, + num_turns: 1, + result: "Pineapple.", + stop_reason: "end_turn", + total_cost_usd: 0, + usage: {}, + modelUsage: {}, + permission_denials: [], + uuid: "result-pineapple", + session_id: nativeThreadId, + }; + + it.effect("streams the root transcript's goal records ahead of the result they precede", () => + Effect.gen(function* () { + const cli = makeFakeClaudeCli(true, (emit) => { + emit({ + type: "assistant", + message: { + model: "claude-haiku-5-5", + id: "msg-pineapple", + type: "message", + role: "assistant", + content: [{ type: "text", text: "Pineapple." }], + }, + parent_tool_use_id: null, + uuid: "assistant-pineapple", + session_id: nativeThreadId, + }); + emit({ + type: "transcript_mirror", + filePath: transcript, + entries: [ + entry("hook-success", { type: "hook_success", hookName: "Stop" }), + entry("goal-unmet", unmet), + ], + }); + // A subagent's transcript, and a sidechain entry in the root one. + emit({ + type: "transcript_mirror", + filePath: `${projectDir}/${nativeThreadId}/subagents/agent-a1.jsonl`, + entries: [entry("subagent-goal", met)], + }); + emit({ + type: "transcript_mirror", + filePath: transcript, + entries: [entry("sidechain-goal", met, true), entry("goal-met", met)], + }); + emit(resultFrame); + }); + const session = yield* openWith(cli); + yield* session.offer(prompt); + const messages = yield* session.messages.pipe( + // Claude clears the goal while it is idle: nothing else follows. + Stream.tap((message) => + Effect.sync(() => { + if (message.type !== "result") return; + cli.emit({ + type: "transcript_mirror", + filePath: transcript, + entries: [entry("goal-cleared", cleared)], + }); + }), + ), + Stream.take(5), + Stream.runCollect, + ); + yield* session.close; + + assert.include(cli.spawns[0]?.args ?? [], "--session-mirror"); + const order = messages.map((message) => + message.type === "goal_status_mirror" + ? { sessionId: message.sessionId, records: message.records } + : message.type, + ); + // A record can overtake an SDK message read in the same tick, never + // the result it preceded. + assert.include(order, "assistant"); + assert.deepEqual( + order.filter((message) => message !== "assistant"), + [ + { sessionId: nativeThreadId, records: [unmet] }, + { sessionId: nativeThreadId, records: [met] }, + "result", + { sessionId: nativeThreadId, records: [cleared] }, + ], + ); + }), + ); + + // With a session store, the SDK asks it for the session before it spawns a + // resumed CLI. An empty answer leaves the CLI on its own transcript and + // config directory, and the MCP registration waits for the spawn. + it.effect("resumes from the CLI's own transcript and settings", () => + Effect.gen(function* () { + const cli = makeFakeClaudeCli(true, (emit) => emit(resultFrame)); + const session = yield* openWith(cli, true); + yield* session.offer(prompt); + const messages = yield* session.messages.pipe(Stream.take(1), Stream.runCollect); + yield* session.close; + + assert.deepEqual( + messages.map((message) => message.type), + ["result"], + ); + const spawned = cli.spawns[0]; + assert.include(spawned?.args ?? [], `--resume=${nativeThreadId}`); + assert.include(spawned?.args ?? [], "--session-mirror"); + assert.equal(spawned?.env.CLAUDE_CONFIG_DIR, configDir); + assert.deepInclude(cli.controlRequests, { + subtype: "mcp_set_servers", + servers: { "t3-code": t3McpServer }, + }); + }), + ); +}); + describe("ClaudeAdapterV2 native session identity", () => { const openTurnWithOrdinal = (providerTurnOrdinal: number, nativeThreadHasTurns?: boolean) => Effect.scoped( @@ -2387,6 +2591,10 @@ describe("ClaudeAdapterV2 background wake turns", () => { yield* Queue.unbounded< Extract >(); + const providerThreadReceipts = + yield* Queue.unbounded< + Extract + >(); let openedOptions: ClaudeAdapterV2.ClaudeAgentSdkQueryOptions | undefined; const adapter = yield* ClaudeAdapterV2.makeClaudeAdapterV2({ instanceId: ClaudeAdapterV2.CLAUDE_DEFAULT_INSTANCE_ID, @@ -2496,6 +2704,9 @@ describe("ClaudeAdapterV2 background wake turns", () => { if (event.type === "turn_item.updated" && event.turnItem.type === "system_notice") { yield* Queue.offer(systemNoticeReceipts, event); } + if (event.type === "provider_thread.updated") { + yield* Queue.offer(providerThreadReceipts, event); + } }), ), Effect.forkScoped, @@ -2533,6 +2744,7 @@ describe("ClaudeAdapterV2 background wake turns", () => { terminalReceipts, subagentReceipts, systemNoticeReceipts, + providerThreadReceipts, getOpenedOptions: () => openedOptions, terminalEvents, hasPendingBackgroundWork, @@ -14244,6 +14456,300 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + // Claude's own `goal_status` records, as the transcript mirror delivers + // them (see "ClaudeAdapterV2 goal status mirror"). Sequences follow + // Claude Code 2.1.299 sessions: each unmet check is Stop hook feedback + // followed by its record; a skipped or failed check writes nothing. + const goalRecords = ( + records: ReadonlyArray, + sessionId = WAKE_NATIVE_SESSION, + ) => claudeSdkFrame({ type: "goal_status_mirror", sessionId, records }); + const goalSetRecord = (condition: string) => + ({ type: "goal_status", met: false, sentinel: true, condition }) as const; + const goalClearedRecord = (condition: string) => + ({ type: "goal_status", met: true, sentinel: true, condition }) as const; + const goalUnmetRecord = (condition: string, reason: string) => + ({ type: "goal_status", met: false, condition, reason }) as const; + const goalMetRecord = (condition: string) => + ({ type: "goal_status", met: true, condition, reason: "The condition holds." }) as const; + const goalTestLayer = Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + ); + const startGoalTurn = ( + harness: Effect.Success, + input: { + readonly attempt: string; + readonly text: string; + readonly goal?: OrchestrationV2ProviderThread["goal"]; + }, + ) => + Effect.gen(function* () { + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: + input.goal === undefined + ? harness.providerThread + : { ...harness.providerThread, goal: input.goal }, + now: yield* DateTime.now, + attemptId: RunAttemptId.make(input.attempt), + text: input.text, + attachments: [], + }), + ); + }); + const offerAll = ( + harness: Effect.Success, + frames: ReadonlyArray, + ) => Effect.forEach(frames, (frame) => Queue.offer(harness.sdkMessages, frame)); + + it.effect("keeps a goal pursued when Claude's check writes no verdict", () => + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const condition = "all tests pass"; + yield* startGoalTurn(harness, { + attempt: "goal-mirror-error", + text: `/goal ${condition}`, + }); + yield* offerAll(harness, [ + syntheticFrame("goal-mirror-error-set", `Goal set: ${condition}`), + goalRecords([goalSetRecord(condition)]), + makeAssistantTextFrame({ uuid: "goal-mirror-error-work-1", text: "Fixing a test." }), + stopHookFeedback("goal-mirror-error-check", condition, "One test still fails."), + goalRecords([goalUnmetRecord(condition, "One test still fails.")]), + makeAssistantTextFrame({ uuid: "goal-mirror-error-work-2", text: "All tests pass." }), + // Another session's verdict, and then a turn end whose check + // errored: Claude writes no record for it. + goalRecords([goalMetRecord(condition)], "native-thread-other"), + makeResultFrame({ uuid: "goal-mirror-error-result", result: "All tests pass." }), + ]); + const terminal = yield* Queue.take(harness.terminalReceipts); + assert.equal(terminal.status, "completed"); + // One check, counted once although both its feedback and its record arrived. + assert.deepEqual(goalStatuses(harness.events), [ + { objective: condition, status: "active", checks: 0 }, + { + objective: condition, + status: "active", + checks: 1, + lastCheck: "One test still fails.", + }, + { + objective: condition, + status: "active", + checks: 1, + lastCheck: "One test still fails.", + }, + ]); + }).pipe(Effect.scoped, Effect.provide(goalTestLayer)), + ); + + it.effect("completes a goal when Claude's met verdict arrives before the result", () => + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const condition = "the word pineapple appears twice"; + yield* startGoalTurn(harness, { attempt: "goal-mirror-met", text: `/goal ${condition}` }); + yield* offerAll(harness, [ + syntheticFrame("goal-mirror-met-set", `Goal set: ${condition}`), + goalRecords([goalSetRecord(condition)]), + makeAssistantTextFrame({ uuid: "goal-mirror-met-work-1", text: "Pineapple." }), + stopHookFeedback("goal-mirror-met-check", condition, "Only one turn says it."), + goalRecords([goalUnmetRecord(condition, "Only one turn says it.")]), + makeAssistantTextFrame({ uuid: "goal-mirror-met-work-2", text: "Pineapple again." }), + goalRecords([goalMetRecord(condition)]), + makeResultFrame({ uuid: "goal-mirror-met-result", result: "Pineapple again." }), + ]); + yield* Queue.take(harness.terminalReceipts); + // Complete when the verdict arrives, and still at the turn's end. + assert.deepEqual(goalStatuses(harness.events).slice(1), [ + { + objective: condition, + status: "active", + checks: 1, + lastCheck: "Only one turn says it.", + }, + { objective: condition, status: "complete", checks: 1 }, + { objective: condition, status: "complete", checks: 1 }, + ]); + }).pipe(Effect.scoped, Effect.provide(goalTestLayer)), + ); + + it.effect("completes a goal only on Claude's verdict after a background subagent", () => + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const condition = "both reviewers find no bugs"; + const taskId = "task-goal-mirror-subagent"; + const now = yield* DateTime.now; + yield* startGoalTurn(harness, { + attempt: "goal-mirror-subagent", + text: `/goal ${condition}`, + }); + yield* offerAll(harness, [ + syntheticFrame("goal-mirror-subagent-set", `Goal set: ${condition}`), + goalRecords([goalSetRecord(condition)]), + ...backgroundAgentLaunchFrames({ prefix: "goal-mirror-subagent", taskId }), + makeAssistantTextFrame({ + uuid: "goal-mirror-subagent-wait", + text: "Waiting for the reviewers.", + }), + makeResultFrame({ + uuid: "goal-mirror-subagent-result", + result: "Waiting for the reviewers.", + }), + ]); + yield* Queue.take(harness.terminalReceipts); + assert.equal(goalStatuses(harness.events).at(-1)?.status, "active"); + + yield* harness.offerAndWait( + claudeSdkFrame({ + type: "system", + subtype: "task_notification", + task_id: taskId, + tool_use_id: "toolu-goal-mirror-subagent", + status: "completed", + output_file: `/tmp/${taskId}.output`, + summary: "No bugs found.", + uuid: "goal-mirror-subagent-notification", + session_id: WAKE_NATIVE_SESSION, + }), + ); + yield* harness.offerAndWait(backgroundTasksLevel("goal-mirror-subagent-level", [])); + yield* harness.offerAndWait(wakeTurnInit); + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now, + attemptId: RunAttemptId.make("goal-mirror-subagent-wake"), + text: "Background task completed.", + attachments: [], + providerTurnOrdinal: 2, + messageCreatedBy: "agent", + messageCreationSource: "provider", + }), + ); + yield* harness.offerAndWait( + makeAssistantTextFrame({ uuid: "goal-mirror-subagent-wake-work", text: "No bugs." }), + ); + yield* harness.offerAndWait( + makeResultFrame({ + uuid: "goal-mirror-subagent-wake-result", + result: "No bugs.", + origin: { kind: "task-notification" }, + }), + ); + yield* Queue.take(harness.terminalReceipts); + // Nothing running and the turn ended, but Claude has not said so yet. + assert.equal(goalStatuses(harness.events).at(-1)?.status, "active"); + + // The verdict lands while Claude is idle. + yield* Queue.clear(harness.providerThreadReceipts); + yield* harness.offerAndWait(goalRecords([goalMetRecord(condition)])); + const update = yield* Queue.take(harness.providerThreadReceipts); + assert.deepEqual(update.providerThread.goal, { + objective: condition, + status: "complete", + checks: 0, + }); + }).pipe(Effect.scoped, Effect.provide(goalTestLayer)), + ); + + // Claude drops a goal it judges impossible. Resume would send + // `/goal resume`, which Claude reads as a new goal, so T3 shows no goal. + it.effect("drops a goal Claude judges impossible", () => + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const condition = "the moon is made of cheese"; + yield* startGoalTurn(harness, { + attempt: "goal-mirror-impossible", + text: `/goal ${condition}`, + }); + yield* offerAll(harness, [ + syntheticFrame("goal-mirror-impossible-set", `Goal set: ${condition}`), + goalRecords([goalSetRecord(condition)]), + makeAssistantTextFrame({ uuid: "goal-mirror-impossible-work", text: "It is rock." }), + goalRecords([ + { + type: "goal_status", + met: false, + failed: true, + condition, + reason: "The moon is rock; no work can change that.", + }, + ]), + makeResultFrame({ uuid: "goal-mirror-impossible-result", result: "It is rock." }), + ]); + yield* Queue.take(harness.terminalReceipts); + const updates = harness.events.flatMap((event) => + event.type === "provider_thread.updated" ? [event.providerThread.goal] : [], + ); + assert.isNull(updates.at(-1)); + assert.notInclude( + goalStatuses(harness.events).map((goal) => goal.status), + "complete", + ); + }).pipe(Effect.scoped, Effect.provide(goalTestLayer)), + ); + + // A complete goal can come from an earlier inference, or from a version + // that inferred more. Claude's next unmet check puts it back. + it.effect("returns a complete goal to active when Claude's next check finds it unmet", () => + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const condition = "the docs build"; + yield* startGoalTurn(harness, { + attempt: "goal-mirror-reopen", + text: "Keep going.", + goal: { objective: condition, status: "complete", checks: 1 }, + }); + yield* offerAll(harness, [ + // This process's first mirror batch holds no goal record. + goalRecords([]), + makeAssistantTextFrame({ uuid: "goal-mirror-reopen-work-1", text: "Building." }), + stopHookFeedback("goal-mirror-reopen-check", condition, "The build fails."), + goalRecords([goalUnmetRecord(condition, "The build fails.")]), + makeAssistantTextFrame({ uuid: "goal-mirror-reopen-work-2", text: "Fixed it." }), + makeResultFrame({ uuid: "goal-mirror-reopen-result", result: "Fixed it." }), + ]); + yield* Queue.take(harness.terminalReceipts); + assert.deepEqual(goalStatuses(harness.events).at(-1), { + objective: condition, + status: "active", + checks: 2, + lastCheck: "The build fails.", + }); + }).pipe(Effect.scoped, Effect.provide(goalTestLayer)), + ); + + it.effect("resets on a new goal and drops a goal Claude clears", () => + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const previous = "the old objective"; + const condition = "the release notes are written"; + yield* startGoalTurn(harness, { + attempt: "goal-mirror-clear", + text: `/goal ${condition}`, + goal: { objective: previous, status: "active", checks: 3, lastCheck: "Not yet." }, + }); + yield* offerAll(harness, [ + syntheticFrame("goal-mirror-clear-set", `Goal set: ${condition}`), + goalRecords([goalSetRecord(condition)]), + makeAssistantTextFrame({ uuid: "goal-mirror-clear-work", text: "Writing notes." }), + // Claude clears a goal itself on auth, credit or context errors. + goalRecords([goalClearedRecord(condition)]), + makeResultFrame({ uuid: "goal-mirror-clear-result", result: "Writing notes." }), + ]); + yield* Queue.take(harness.terminalReceipts); + const goals = harness.events.flatMap((event) => + event.type === "provider_thread.updated" ? [event.providerThread.goal] : [], + ); + assert.deepInclude(goals, { objective: condition, status: "active", checks: 0 }); + assert.isNull(goals.at(-1)); + }).pipe(Effect.scoped, Effect.provide(goalTestLayer)), + ); + it.effect("keeps a goal active after a command turn with no model output", () => Effect.gen(function* () { const harness = yield* makeWakeHarness; @@ -14286,54 +14792,122 @@ describe("ClaudeAdapterV2 background wake turns", () => { }); describe("ClaudeAdapterV2 query message stream", () => { - it.effect("closes the query when the message stream is interrupted mid-read", () => - Effect.gen(function* () { - let closed = false; - let releaseRead = () => {}; - const readStarted = Promise.withResolvers(); - // Never yields a message — the read stays pending until close() flips - // `closed` and releases the in-flight await. - // oxlint-disable-next-line require-yield - async function* sdkMessages(): AsyncGenerator { - for (;;) { - if (closed) return; - await new Promise((resolve) => { - releaseRead = resolve; - readStarted.resolve(); - }); + it.effect.each([{ mirror: "without" }, { mirror: "with" }])( + "closes the query when the message stream is interrupted mid-read $mirror a goal mirror", + ({ mirror }) => + Effect.gen(function* () { + const goalStatusMirror = + mirror === "with" ? ClaudeAdapterV2.makeClaudeGoalStatusMirror() : undefined; + let closed = false; + let releaseRead = () => {}; + const readStarted = Promise.withResolvers(); + // Never yields a message — the read stays pending until close() flips + // `closed` and releases the in-flight await. + // oxlint-disable-next-line require-yield + async function* sdkMessages(): AsyncGenerator { + for (;;) { + if (closed) return; + await new Promise((resolve) => { + releaseRead = resolve; + readStarted.resolve(); + }); + } } - } - const generator = sdkMessages(); - const close = () => { - closed = true; - releaseRead(); - }; - const query = { - next: () => generator.next(), - return: async (value?: void) => { - close(); - return generator.return(value); - }, - throw: (error?: unknown) => generator.throw(error), - [Symbol.asyncIterator]: () => generator, - close, - } as unknown as ClaudeQuery; - - const scope = yield* Scope.make(); - yield* Stream.fromAsyncIterable( - ClaudeAdapterV2.claudeQueryMessages(query), - (cause) => cause, - ).pipe( - Stream.runForEach(() => Effect.void), - Effect.forkIn(scope), - ); - yield* Effect.promise(() => readStarted.promise); - - // Iterating query[Symbol.asyncIterator]() directly deadlocks here: - // the raw generator's return() queues behind the in-flight read and - // scope close never completes. - yield* Scope.close(scope, Exit.void); - assert.isTrue(closed); - }), + const generator = sdkMessages(); + const close = () => { + closed = true; + releaseRead(); + }; + const query = { + next: () => generator.next(), + return: async (value?: void) => { + close(); + return generator.return(value); + }, + throw: (error?: unknown) => generator.throw(error), + [Symbol.asyncIterator]: () => generator, + close, + } as unknown as ClaudeQuery; + + const scope = yield* Scope.make(); + yield* Stream.fromAsyncIterable( + ClaudeAdapterV2.claudeQueryMessages(query, goalStatusMirror), + (cause) => cause, + ).pipe( + Stream.runForEach(() => Effect.void), + Effect.forkIn(scope), + ); + yield* Effect.promise(() => readStarted.promise); + + // Iterating query[Symbol.asyncIterator]() directly deadlocks here: + // the raw generator's return() queues behind the in-flight read and + // scope close never completes. + yield* Scope.close(scope, Exit.void); + assert.isTrue(closed); + }), ); + + // The SDK flushes the mirror before it ends the stream, so a record that + // lands with the end still reaches the adapter. + it("delivers a goal record that arrives with the end of the stream", async () => { + const mirror = ClaudeAdapterV2.makeClaudeGoalStatusMirror(); + const met = { type: "goal_status", met: true, condition: "the docs build" } as const; + const query = { + next: async () => { + await mirror.sessionStore.append({ projectKey: "p", sessionId: "s" }, [ + { type: "attachment", attachment: met }, + ]); + return { done: true, value: undefined }; + }, + return: async () => ({ done: true, value: undefined }), + } as unknown as ClaudeQuery; + const messages: Array = []; + for await (const message of ClaudeAdapterV2.claudeQueryMessages(query, mirror)) { + messages.push(message); + } + + assert.deepEqual(messages, [{ type: "goal_status_mirror", sessionId: "s", records: [met] }]); + }); + + // A session without a goal never gets a goal record, and T3 reads one SDK + // message per streamed token, so what the stream keeps must not grow with + // the messages it reads. Counts live promises after a full collection. + it("keeps nothing per message while the mirror delivers no goal record", async () => { + let sent = 0; + const query = { + next: () => + new Promise>((resolve) => + setImmediate(() => resolve({ done: false, value: claudeStreamEventFrame(sent++) })), + ), + return: async () => ({ done: true, value: undefined }), + } as unknown as ClaudeQuery; + const mirror = ClaudeAdapterV2.makeClaudeGoalStatusMirror(); + // The process's first root batch confirms the mirror; no record follows. + await mirror.sessionStore.append({ projectKey: "p", sessionId: "s" }, [{ type: "user" }]); + const messages = ClaudeAdapterV2.claudeQueryMessages(query, mirror)[Symbol.asyncIterator](); + const read = async (count: number) => { + for (let index = 0; index < count; index += 1) await messages.next(); + }; + + await read(1_000); + const before = NodeV8.queryObjects(Promise, { format: "count" }); + await read(10_000); + const after = NodeV8.queryObjects(Promise, { format: "count" }); + + assert.isBelow(after - before, 1_000); + }); }); + +function claudeStreamEventFrame(index: number): SDKMessage { + return { + type: "stream_event", + event: { + type: "content_block_delta", + index: 0, + delta: { type: "text_delta", text: `${index}` }, + }, + parent_tool_use_id: null, + uuid: `00000000-0000-4000-8000-${index.toString().padStart(12, "0")}`, + session_id: "s", + } as SDKMessage; +} diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 73d97cdd7d1f..c0d4fedbb86e 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -28,6 +28,7 @@ import { type PermissionResult, type PermissionUpdate, type Query as ClaudeQuery, + type SessionStore, type Settings as ClaudeSdkSettings, type SpawnOptions as ClaudeSpawnOptions, type SpawnedProcess as ClaudeSpawnedProcess, @@ -414,8 +415,39 @@ export interface ClaudeAgentSdkQueryOpenInput { readonly providerSessionId: OrchestrationV2ProviderSession["id"]; } +// Claude Code's transcript record for `/goal`. The CLI writes it to the session +// transcript only; the SDK types do not declare it. +const ClaudeGoalStatusEntry = Schema.Struct({ + type: Schema.Literal("attachment"), + isSidechain: Schema.optional(Schema.Boolean), + attachment: Schema.Struct({ + type: Schema.Literal("goal_status"), + condition: Schema.String, + met: Schema.Boolean, + sentinel: Schema.optional(Schema.Boolean), + failed: Schema.optional(Schema.Boolean), + reason: Schema.optional(Schema.String), + }), +}); +const isClaudeGoalStatusEntry = Schema.is(ClaudeGoalStatusEntry); +export type ClaudeGoalStatusRecord = (typeof ClaudeGoalStatusEntry.Type)["attachment"]; + +/** + * The `/goal` records Claude wrote to the root session transcript, which reach + * T3 only through the SDK's transcript mirror. A process's first one may hold + * no record: it shows that the mirror works for that process. + */ +export interface ClaudeGoalStatusMirrorMessage { + readonly type: "goal_status_mirror"; + readonly sessionId: string; + readonly records: ReadonlyArray; +} + +/** What a query session streams: the SDK's messages and mirrored goal records. */ +export type ClaudeQueryMessage = SDKMessage | ClaudeGoalStatusMirrorMessage; + export interface ClaudeAgentSdkQuerySession { - readonly messages: Stream.Stream; + readonly messages: Stream.Stream; readonly offer: (message: SDKUserMessage) => Effect.Effect; readonly setModel: (model: string) => Effect.Effect; readonly applyFlagSettings: ( @@ -508,14 +540,98 @@ function closeClaudeQuery(queryRuntime: ClaudeQuery) { }); } +export interface ClaudeGoalStatusMirror { + /** The SDK `sessionStore` that receives one CLI process's transcript mirror. */ + readonly sessionStore: SessionStore; + readonly take: () => ClaudeGoalStatusMirrorMessage | undefined; + /** + * Settles when the next message arrives. Only the latest call is kept, so a + * reader that waits again drops its earlier, unsettled wait. + */ + readonly arrival: () => Promise; +} + +/** + * Keeps the `/goal` records of one CLI process's transcript mirror. Claude + * reports a goal's verdict nowhere else in SDK mode. Subagent transcripts + * arrive under a subpath and are ignored. + */ +export function makeClaudeGoalStatusMirror(): ClaudeGoalStatusMirror { + const pending: Array = []; + // One waiter at a time: most sessions never get a goal record, and every + // reaction parked on a promise that never settles would stay for the + // process's lifetime. + let waiter: PromiseWithResolvers | undefined; + let confirmed = false; + return { + sessionStore: { + append: async (key, entries) => { + if (key.subpath !== undefined) return; + const records = entries.flatMap((entry) => + isClaudeGoalStatusEntry(entry) && entry.isSidechain !== true ? [entry.attachment] : [], + ); + if (records.length === 0 && confirmed) return; + confirmed = true; + pending.push({ type: "goal_status_mirror", sessionId: key.sessionId, records }); + waiter?.resolve(); + waiter = undefined; + }, + // Nothing is stored here, so a resume reads the CLI's own transcript. + load: async () => null, + }, + take: () => pending.shift(), + arrival: () => { + waiter = Promise.withResolvers(); + return waiter.promise; + }, + }; +} + // Iterate the Query itself, not query[Symbol.asyncIterator]() (the raw // sdkMessages generator). The raw generator's return() queues behind the // in-flight read of the next CLI message and never settles while the CLI is // idle, deadlocking stream interruption (and with it, session scope close). // Query.return() runs cleanup() first, which closes the transport and // unblocks that read. -export function claudeQueryMessages(queryRuntime: ClaudeQuery): AsyncIterable { - return { [Symbol.asyncIterator]: () => queryRuntime }; +// +// Mirrored goal records join the stream as soon as the store receives them, +// so one that lands while the CLI is idle is read at once. The SDK hands a +// turn's mirror batches to the store before it yields that turn's result, so +// a record never follows the result it preceded. It can overtake an SDK +// message read in the same tick. +export function claudeQueryMessages( + queryRuntime: ClaudeQuery, + goalStatusMirror?: ClaudeGoalStatusMirror, +): AsyncIterable { + if (goalStatusMirror === undefined) { + return { [Symbol.asyncIterator]: () => queryRuntime }; + } + return { + [Symbol.asyncIterator]: () => { + let read: Promise> | undefined; + return { + next: async () => { + for (;;) { + const mirrored = goalStatusMirror.take(); + if (mirrored !== undefined) return { done: false, value: mirrored }; + const arrival = goalStatusMirror.arrival(); + const current = (read ??= queryRuntime.next()); + const sdkFirst = await Promise.race([ + current.then(() => true), + arrival.then(() => false), + ]); + if (sdkFirst) { + const next = goalStatusMirror.take(); + if (next !== undefined) return { done: false, value: next }; + read = undefined; + return current; + } + } + }, + return: (value) => queryRuntime.return(value), + }; + }, + }; } export interface ClaudeAgentSdkLoggedQueryOptions { @@ -639,7 +755,7 @@ export type ClaudeAgentSdkProtocolLogEvent = | { readonly direction: "incoming"; readonly stage: "decoded"; - readonly payload: SDKMessage; + readonly payload: ClaudeQueryMessage; }; export type ClaudeAgentSdkProtocolLogger = ( @@ -768,6 +884,12 @@ export const layerQueryRunner: Layer.Layer< const spawnClaudeCodeProcess = options.spawnClaudeCodeProcess ?? (platform === "win32" ? undefined : spawnClaudeProcessGroup); + // The SDK refuses a mirror without local transcript writes or with + // file checkpointing; both are off unless a caller opts in. + const goalStatusMirror = + options.persistSession === false || options.enableFileCheckpointing === true + ? undefined + : makeClaudeGoalStatusMirror(); // The CLI is running once query() returns, so it is closed if anything // before the session is handed out fails or is interrupted. const queryRuntime = yield* Effect.uninterruptibleMask((restore) => @@ -775,10 +897,13 @@ export const layerQueryRunner: Layer.Layer< try: () => query({ prompt, - options: - spawnClaudeCodeProcess === undefined - ? options - : { ...options, spawnClaudeCodeProcess }, + options: { + ...options, + ...(spawnClaudeCodeProcess === undefined ? {} : { spawnClaudeCodeProcess }), + ...(goalStatusMirror === undefined + ? {} + : { sessionStore: goalStatusMirror.sessionStore, sessionStoreFlush: "eager" }), + }, }), catch: (cause) => queryRunnerError(cause, "query"), }).pipe( @@ -835,8 +960,9 @@ export const layerQueryRunner: Layer.Layer< ); return { - messages: Stream.fromAsyncIterable(claudeQueryMessages(queryRuntime), (cause) => - queryRunnerError(cause, "fromAsyncIterable"), + messages: Stream.fromAsyncIterable( + claudeQueryMessages(queryRuntime, goalStatusMirror), + (cause) => queryRunnerError(cause, "fromAsyncIterable"), ).pipe( Stream.tap((message) => logProtocolEvent({ @@ -1277,17 +1403,22 @@ function textFromClaudeContent(content: SDKAssistantMessage["message"]["content" // In SDK mode Claude reports `/goal` only through the transcript (its // `active_goal` event is remote-only): synthetic command output names the -// goal, and each unmet evaluator check returns as Stop hook feedback. +// goal, and each unmet evaluator check returns as Stop hook feedback. The +// verdicts themselves are `goal_status` records, which only the transcript +// mirror delivers (see nextClaudeGoalFromRecord). const CLAUDE_GOAL_SET_PREFIX = "Goal set: "; const CLAUDE_GOAL_ACTIVE = /^Goal active: ([\s\S]+?) \((?:not yet evaluated|(\d+) turns?)\)/u; /** * The goal after one root SDK frame, or undefined when the frame says nothing - * about it. Completion has no frame of its own; see finalizeActiveTurn. + * about it. Completion has no frame of its own; see finalizeActiveTurn. Once + * the mirror delivers goal records, they count the checks instead of Stop + * hook feedback, which reports the same checks. */ function nextClaudeGoal( current: OrchestrationV2ProviderGoal | null, message: SDKMessage, + goalStatusMirrored: boolean, ): OrchestrationV2ProviderGoal | null | undefined { if (message.type === "assistant") { if (message.parent_tool_use_id !== null || message.message.model !== "") { @@ -1309,6 +1440,7 @@ function nextClaudeGoal( }; } if ( + goalStatusMirrored || message.type !== "user" || message.parent_tool_use_id !== null || message.isSynthetic !== true || @@ -1331,6 +1463,34 @@ function nextClaudeGoal( }; } +/** + * The goal after one `goal_status` record from the root session transcript. + * Claude writes one when a goal is set or cleared (a `sentinel` record), after + * each check that finds it unmet, and when it judges the goal met or + * impossible. A check it skips (background work still running) or that fails + * to run writes nothing, so the goal stays as it was. + */ +function nextClaudeGoalFromRecord( + current: OrchestrationV2ProviderGoal | null, + record: ClaudeGoalStatusRecord, +): OrchestrationV2ProviderGoal | null { + const objective = record.condition.trim(); + if (objective.length === 0) return current; + // Cleared, including the clears Claude makes itself on auth, credit or + // context errors. + if (record.sentinel === true) { + return record.met ? null : { objective, status: "active", checks: 0 }; + } + // Impossible: Claude drops the goal, as on a clear. T3 has no Claude status + // for it, and the statuses that offer Resume would send `/goal resume`, + // which Claude reads as a new goal. + if (record.failed === true) return null; + const checks = current?.objective === objective ? (current.checks ?? 0) : 0; + if (record.met) return { objective, status: "complete", checks }; + if (record.reason === undefined) return current; + return { objective, status: "active", checks: checks + 1, lastCheck: record.reason.trim() }; +} + const providerGoalsEqual = Schema.toEquivalence(Schema.NullOr(OrchestrationV2ProviderGoal)); /** Compares a subagent result with its routed text regardless of block joins. */ @@ -3118,6 +3278,9 @@ interface ClaudeLiveQueryContext { permissionMode: PermissionMode; // Stop, rollback or fork is closing this process; its work is ending. stopping: boolean; + // This process's transcript mirror delivers Claude's own `/goal` records, + // so they alone move the goal (see applyMirroredClaudeGoal). + goalStatusMirrored: boolean; // Registry entries still running when this process opened. Their process // is gone and never reports their end; any later task_started replaces the // entry, so an entry still in this set runs nowhere. @@ -4024,13 +4187,33 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( }); }); + /** Records a native thread's goal and writes any change onto its provider thread. */ + const writeClaudeGoal = Effect.fnUntraced(function* ( + nativeThreadId: string, + next: OrchestrationV2ProviderGoal | null, + ) { + if (providerGoalsEqual(goalsByNativeThread.get(nativeThreadId) ?? null, next)) return; + goalsByNativeThread.set(nativeThreadId, next); + const providerThread = (yield* Ref.get(lastProviderThreadByNativeThread)).get( + nativeThreadId, + ); + if (providerThread === undefined) return; + const updated = { ...providerThread, goal: next, updatedAt: yield* DateTime.now }; + yield* rememberProviderThread(updated); + yield* emitProviderEvent({ + type: "provider_thread.updated", + driver: CLAUDE_PROVIDER, + providerThread: updated, + }); + }); + /** Applies one root frame's goal signal and writes any change onto the provider thread. */ const trackClaudeGoal = Effect.fnUntraced(function* (input: { - readonly nativeThreadId: string; + readonly live: ClaudeLiveQueryContext; readonly context: ActiveClaudeTurnContext; readonly message: SDKMessage; }) { - const current = goalsByNativeThread.get(input.nativeThreadId) ?? null; + const current = goalsByNativeThread.get(input.live.nativeThreadId) ?? null; if ( current?.status === "active" && input.message.type === "assistant" && @@ -4039,20 +4222,35 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ) { goalCheckedTurns.add(input.context.providerTurnId); } - const next = nextClaudeGoal(current, input.message); - if (next === undefined || providerGoalsEqual(current, next)) return; - goalsByNativeThread.set(input.nativeThreadId, next); - const providerThread = (yield* Ref.get(lastProviderThreadByNativeThread)).get( - input.nativeThreadId, + const next = nextClaudeGoal(current, input.message, input.live.goalStatusMirrored); + if (next !== undefined) yield* writeClaudeGoal(input.live.nativeThreadId, next); + }); + + /** + * Applies the `/goal` records a process's transcript mirror delivered, + * whenever they arrive, including after the turn they belong to ended. + * From then on they alone decide the goal for that process: Claude + * writes a record for every verdict, so a turn end without one is a + * check that was skipped or failed, not a met goal. + */ + const applyMirroredClaudeGoal = Effect.fnUntraced(function* ( + live: ClaudeLiveQueryContext, + message: ClaudeGoalStatusMirrorMessage, + ) { + if ( + (yield* Ref.get(queryContext))?.query !== live.query || + message.sessionId !== live.nativeThreadId + ) { + return; + } + live.goalStatusMirrored = true; + yield* writeClaudeGoal( + live.nativeThreadId, + message.records.reduce( + nextClaudeGoalFromRecord, + goalsByNativeThread.get(live.nativeThreadId) ?? null, + ), ); - if (providerThread === undefined) return; - const updated = { ...providerThread, goal: next, updatedAt: yield* DateTime.now }; - yield* rememberProviderThread(updated); - yield* emitProviderEvent({ - type: "provider_thread.updated", - driver: CLAUDE_PROVIDER, - providerThread: updated, - }); }); const rememberOpaqueTasks = (tasks: ReadonlyArray) => @@ -6083,13 +6281,13 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( const clearConversationHead = input.status === "completed" && input.context.input.providerThread.nativeConversationHeadRef !== null; - // While a goal is set, Claude ends a turn on its own only after the - // goal's Stop hook passes. It defers that check while background - // work of the types it waits on runs (subagents stay off the roster; - // see CLAUDE_GOAL_DEFERRING_TASK_TYPES), and a hook that stops the - // turn reports another reason. - // SDK mode does not report an evaluator timeout or an impossible - // verdict, so those still read as complete. + // Without a transcript mirror (see applyMirroredClaudeGoal), a met + // goal is inferred: while a goal is set, Claude ends a turn on its + // own only after the goal's Stop hook passes. It defers that check + // while background work of the types it waits on runs (subagents + // stay off the roster; see CLAUDE_GOAL_DEFERRING_TASK_TYPES), and + // a hook that stops the turn reports another reason. An evaluator + // timeout or an impossible verdict still reads as complete there. const goal = nativeThreadId === null ? undefined : goalsByNativeThread.get(nativeThreadId); const goalChecked = goalCheckedTurns.delete(input.context.providerTurnId); @@ -6101,6 +6299,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( goalChecked && input.status === "completed" && (terminalReason === undefined || terminalReason === "completed") && + !(liveQuery?.nativeThreadId === nativeThreadId && liveQuery.goalStatusMirrored) && roster.size === 0 && !( liveQuery?.nativeThreadId === nativeThreadId && @@ -6924,7 +7123,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( } return; } - yield* trackClaudeGoal({ nativeThreadId: liveQuery.nativeThreadId, context, message }); + yield* trackClaudeGoal({ live: liveQuery, context, message }); // Subagent narration belongs to its child thread, never the parent log. if (message.type === "stream_event" && !message.parent_tool_use_id) { @@ -8608,6 +8807,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( openedPermissionMode: queryOptions.permissionMode, permissionMode: queryOptions.permissionMode, stopping: false, + goalStatusMirrored: false, subagentsFromEarlierProcesses: new Set( [...(yield* Ref.get(sessionSubagentsByTaskId)).values()].filter( (subagent) => subagent.task.status === "running", @@ -8619,6 +8819,9 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( Stream.runForEach((message) => sdkMessageLock.withPermits(1)( Effect.suspend(() => { + if (message.type === "goal_status_mirror") { + return applyMirroredClaudeGoal(context, message); + } if ( message.type === "system" && (message.subtype === "init" || message.subtype === "status") && diff --git a/docs/internals/providers.md b/docs/internals/providers.md index 5f9642c2db27..ee7942362369 100644 --- a/docs/internals/providers.md +++ b/docs/internals/providers.md @@ -121,8 +121,13 @@ the next goal turn on its own milliseconds after the last one completes, so the [adapter](../../apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts) keeps the run open and adds that turn to it. One run can therefore own several native turns. `/goal` commands that start no Codex turn settle on a provider turn without a native ref, which native rollback must -not count. Claude's SDK mode emits no goal events; its adapter reads goal state from the -synthetic command output and Stop hook feedback in the transcript. +not count. Claude's SDK mode emits no goal events, and only its session transcript records a +verdict. The [adapter](../../apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts) gives +the SDK a session store so the CLI mirrors that transcript on stdout, and takes Claude's +`goal_status` records from it. The store loads nothing, so a resume still reads the CLI's own +transcript. A launcher that moves `CLAUDE_CONFIG_DIR` behind the SDK's back gets no mirror; the +adapter then reads the synthetic command output and Stop hook feedback, and infers a met goal +when a turn ends with no background work left. Capabilities must describe what the provider can actually do. Antigravity can capture workspace checkpoints but cannot roll back its conversation. The [checkpoint boundary](./overview.md#turn-completion-and-checkpoints) diff --git a/docs/user/composer.md b/docs/user/composer.md index 7e1be24eda36..3db3c5842147 100644 --- a/docs/user/composer.md +++ b/docs/user/composer.md @@ -244,6 +244,7 @@ row above the composer shows the goal and its progress. pauses it. - Stopping Claude ends the current turn, but the goal stays set. Claude checks it again at the end of your next message. +- Claude ends a goal it judges impossible, and the goal row goes away. ## Context in your message From a35a1024ba5a23a921c59d59e3924f105c023384 Mon Sep 17 00:00:00 2001 From: Chrysler Ade Chandra <10206122+AdeChrysler@users.noreply.github.com> Date: Thu, 15 Oct 2026 00:08:30 +0800 Subject: [PATCH 4/4] fix(server): a Claude goal record for a new objective replaces the old goal Claude Code accepts a goal evaluator answer without a reason, and then writes an unmet goal_status record with no reason. The adapter ignored such a record, so a stale goal for another objective stayed on screen, and a goal shown as complete was not reopened by the check that found it unmet. It now counts the record as an unmet check for its own objective, without a new last-check reason. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/ClaudeAdapterV2.test.ts | 30 +++++++++++++++++++ .../Adapters/ClaudeAdapterV2.ts | 10 +++++-- 2 files changed, 38 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 164d42ac569b..d5171317c83c 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -14723,6 +14723,36 @@ describe("ClaudeAdapterV2 background wake turns", () => { }).pipe(Effect.scoped, Effect.provide(goalTestLayer)), ); + // Claude Code checks the evaluator's answer with `reason` optional, and an + // answer without one writes an unmet record without `reason`. It is still a + // check that found the record's goal unmet. + it.effect.each([ + { shown: "a different objective", objective: "the old objective" }, + { shown: "the same objective, complete", objective: "the docs build" }, + ])("pursues the record's goal after an unmet check without a reason over $shown", (input) => + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const condition = "the docs build"; + yield* startGoalTurn(harness, { + attempt: "goal-mirror-no-reason", + text: "Keep going.", + goal: { objective: input.objective, status: "complete", checks: 1, lastCheck: "Old." }, + }); + yield* offerAll(harness, [ + goalRecords([]), + makeAssistantTextFrame({ uuid: "goal-mirror-no-reason-work", text: "Building." }), + goalRecords([{ type: "goal_status", met: false, condition }]), + makeResultFrame({ uuid: "goal-mirror-no-reason-result", result: "Building." }), + ]); + yield* Queue.take(harness.terminalReceipts); + assert.deepEqual(goalStatuses(harness.events).at(-1), { + objective: condition, + status: "active", + checks: input.objective === condition ? 2 : 1, + }); + }).pipe(Effect.scoped, Effect.provide(goalTestLayer)), + ); + it.effect("resets on a new goal and drops a goal Claude clears", () => Effect.gen(function* () { const harness = yield* makeWakeHarness; diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index c0d4fedbb86e..acf04e69a4af 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -1487,8 +1487,14 @@ function nextClaudeGoalFromRecord( if (record.failed === true) return null; const checks = current?.objective === objective ? (current.checks ?? 0) : 0; if (record.met) return { objective, status: "complete", checks }; - if (record.reason === undefined) return current; - return { objective, status: "active", checks: checks + 1, lastCheck: record.reason.trim() }; + // An unmet check. Claude accepts an evaluator answer without a reason, and + // then the record has none. + return { + objective, + status: "active", + checks: checks + 1, + ...(record.reason === undefined ? {} : { lastCheck: record.reason.trim() }), + }; } const providerGoalsEqual = Schema.toEquivalence(Schema.NullOr(OrchestrationV2ProviderGoal));