diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts index 3f771d6abee08..58f9718e533cf 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts @@ -17902,6 +17902,79 @@ describe("ClaudeAdapterV2 background wake turns", () => { ), ); + it.effect("ends a goal check without re-prompting while T3 work will wake the thread", () => + Effect.gen(function* () { + const harness = yield* makeWakeHarness; + const mcpSessions = yield* McpProviderSessions.McpProviderSessions; + yield* mcpSessions.set({ + environmentId: EnvironmentId.make("environment-goal-wake"), + threadId: harness.threadId, + providerSessionId: "mcp-session-goal-wake", + providerInstanceId: ProviderInstanceId.make("claudeAgent"), + endpoint: "http://127.0.0.1:43123/mcp", + authorizationHeader: "Bearer secret-claude-token", + browserToolsAvailable: true, + hasPendingWake: Effect.succeed(true), + }); + const stop = Effect.promise(() => + Promise.all( + (harness.getOpenedOptions()?.hooks?.Stop ?? []).flatMap((matcher) => + matcher.hooks.map((hook) => + hook( + { + hook_event_name: "Stop", + stop_hook_active: true, + session_id: WAKE_NATIVE_SESSION, + transcript_path: "/transcript.jsonl", + cwd: "/workspace", + }, + undefined, + { signal: new AbortController().signal }, + ), + ), + ), + ), + ); + const condition = "the delegated review is addressed"; + yield* harness.runtime.startTurn( + makeClaudeTestTurnInput({ + threadId: harness.threadId, + providerThread: harness.providerThread, + now: yield* DateTime.now, + attemptId: RunAttemptId.make("goal-wake-attempt"), + text: `/goal ${condition}`, + attachments: [], + }), + ); + // Without a goal, Stop keeps Claude Code's own behavior. + assert.deepEqual(yield* stop, [{}]); + yield* harness.offerAndWait(syntheticFrame("goal-wake-set", `Goal set: ${condition}`)); + assert.deepEqual(yield* stop, [{ continue: false }]); + // The turn ends without an error and the goal waits for the wake. + yield* Queue.offer( + harness.sdkMessages, + makeResultFrame({ + uuid: "goal-wake-result", + result: "Waiting for the review.", + terminalReason: "stop_hook_prevented", + }), + ); + const terminal = yield* Queue.take(harness.terminalReceipts); + assert.equal(terminal.status, "completed"); + assert.equal(goalStatuses(harness.events).at(-1)?.status, "active"); + }).pipe( + Effect.scoped, + Effect.provide( + Layer.mergeAll( + IdAllocator.layer, + McpProviderSessions.layer, + NodeServices.layer, + NodeCrypto.layer, + ), + ), + ), + ); + it.effect( "defers goal completion until an ambient subagent finishes and Claude checks again", () => diff --git a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts index 98c920f5053db..d526c848bcfc1 100644 --- a/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts @@ -1539,6 +1539,7 @@ export function makeClaudeQueryOptions(input: { readonly canUseTool?: CanUseTool; readonly onUserDialog?: ClaudeQueryOptions["onUserDialog"]; readonly onElicitation?: ClaudeQueryOptions["onElicitation"]; + readonly onStop?: HookCallback; readonly supportedDialogKinds?: ClaudeQueryOptions["supportedDialogKinds"]; readonly allowDangerouslySkipPermissions?: boolean; readonly appendSystemPromptFileText?: string; @@ -1663,8 +1664,15 @@ export function makeClaudeQueryOptions(input: { ...launchAppendSystemPrompt, ].join("\n\n"), }, - ...(deviceInstructions - ? { hooks: { SubagentStart: [{ hooks: [subagentDeviceInstructions] }] } } + ...(deviceInstructions || input.onStop !== undefined + ? { + hooks: { + ...(deviceInstructions + ? { SubagentStart: [{ hooks: [subagentDeviceInstructions] }] } + : {}), + ...(input.onStop === undefined ? {} : { Stop: [{ hooks: [input.onStop] }] }), + }, + } : {}), ...(Object.keys(extraArgs).length === 0 ? {} : { extraArgs }), }; @@ -10383,6 +10391,16 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( ) { const queryPolicy = yield* queryPolicyForTurn(turnInput); const mcpSession = yield* mcpSessions.read(turnInput.threadId); + // Claude Code defers its goal Stop hook only for its own background tasks, so a goal + // re-prompts in a loop while T3 work that will wake this thread is still running. + const onStop: HookCallback = () => + runPromise( + Effect.gen(function* () { + if (goalsByNativeThread.get(nativeThreadId)?.status !== "active") return {}; + const pendingWake = (yield* mcpSessions.read(turnInput.threadId))?.hasPendingWake; + return pendingWake !== undefined && (yield* pendingWake) ? { continue: false } : {}; + }), + ); const mcpOverrides = claudeMcpQueryOverrides({ mcpSession, readOnlySandbox: @@ -10541,6 +10559,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* ( canUseTool, onUserDialog, onElicitation, + onStop, supportedDialogKinds: ["resume_return"], ...(appendSystemPromptFileText === undefined ? {} : { appendSystemPromptFileText }), ...(toolSearchOverride === undefined ? {} : { toolSearchOverride }), diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index d8c046c018651..8f41b71a1f6f6 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -16,6 +16,7 @@ import { ProviderDriverKind, ProviderInstanceId, type ProviderSessionId, + RunId, ThreadId, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; @@ -1769,6 +1770,43 @@ it.effect("ProviderSessionManagerV2 drains subscribers when the provider stops", }), ); +it("ProviderSessionManagerV2 counts a delegated task as a wake only while its result can be delivered", () => { + const runId = RunId.make("run-delegating"); + const thread = { archivedAt: null, deletedAt: null }; + const task = { + origin: "app_owned" as const, + status: "running" as const, + runId, + completionDelivery: undefined, + }; + const run = (disposition: "open" | "disposed") => ({ + id: runId, + delegatedCompletion: { disposition, nextGeneration: 1, delivery: null }, + }); + assert.isTrue( + ProviderSessionManager.delegatedTaskWillWake({ + thread, + runs: [run("open")], + subagents: [task], + }), + ); + // An interrupted completion wake disposes the cohort, so the result never wakes the thread. + assert.isFalse( + ProviderSessionManager.delegatedTaskWillWake({ + thread, + runs: [run("disposed")], + subagents: [task], + }), + ); + assert.isFalse( + ProviderSessionManager.delegatedTaskWillWake({ + thread, + runs: [run("open")], + subagents: [{ ...task, origin: "provider_native" }], + }), + ); +}); + it.effect( "ProviderSessionManagerV2 issues MCP credentials before opening and revokes them on close", () => @@ -1814,6 +1852,8 @@ it.effect( new Set(["preview", "orchestration", "worktree", "pull-requests"]), ); + assert.isFalse(yield* captured!.hasPendingWake!); + yield* manager.close(providerSessionId); assert.isUndefined(yield* mcpSessions.read(threadId)); assert.isUndefined(yield* registry.resolve(token!)); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index f06011676c667..1259c0047d93f 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -4,10 +4,13 @@ import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; import { isOrchestrationV2WorkActive, ModelSelection, + type OrchestrationV2AppThread, OrchestrationV2DomainEvent, OrchestrationV2ProviderSession, type OrchestrationV2ProviderThread, + type OrchestrationV2Run, OrchestrationV2RuntimeRequest, + type OrchestrationV2Subagent, ProviderInstanceId, ProviderSessionId, ThreadId, @@ -71,6 +74,30 @@ const WORKING_RUN_STATUSES: ReadonlySet = new Set([ "waiting", ]); +/** Matches planDelegatedCompletionDelivery: true while a delegated task's result still wakes the thread. */ +export const delegatedTaskWillWake = (records: { + readonly thread: Pick; + readonly runs: ReadonlyArray>; + readonly subagents: ReadonlyArray< + Pick + >; +}): boolean => + records.thread.archivedAt === null && + records.thread.deletedAt === null && + records.subagents.some((task) => { + const run = records.runs.find((candidate) => candidate.id === task.runId); + const delivery = task.completionDelivery?.state; + return ( + task.origin === "app_owned" && + isOrchestrationV2WorkActive(task.status) && + run !== undefined && + (run.delegatedCompletion?.disposition ?? "open") === "open" && + delivery !== "acknowledged" && + delivery !== "delivered" && + delivery !== "disposed" + ); + }); + export const ProviderSessionReleaseReason = Schema.Literals([ "idle_timeout", "runtime_error", @@ -497,7 +524,15 @@ export const layerWithOptions = ( capabilities, subagent, }); - yield* mcpSessions.set(credential.config); + yield* mcpSessions.set({ + ...credential.config, + hasPendingWake: projectionStore + .getThreadRecords(threadId, ["runs", "subagents"]) + .pipe( + Effect.map(delegatedTaskWillWake), + Effect.orElseSucceed(() => false), + ), + }); reserveMcpCredential(threadId, credential.config.providerSessionId); return { mcpCredentialId: credential.config.providerSessionId, issued: true }; }), diff --git a/packages/provider-core/src/server/mcpSession.ts b/packages/provider-core/src/server/mcpSession.ts index f2152dde0ccf1..f69d105702786 100644 --- a/packages/provider-core/src/server/mcpSession.ts +++ b/packages/provider-core/src/server/mcpSession.ts @@ -1,4 +1,5 @@ import type { EnvironmentId, ProviderInstanceId, ThreadId } from "@t3tools/contracts"; +import type * as Effect from "effect/Effect"; export interface McpProviderSessionConfig { readonly environmentId: EnvironmentId; @@ -11,6 +12,8 @@ export interface McpProviderSessionConfig { readonly browserToolsAvailable: boolean; readonly capabilities?: ReadonlySet; readonly agentDeviceEnvironment?: Readonly>; + /** True while a delegated task this thread started will wake it with its result. */ + readonly hasPendingWake?: Effect.Effect; } export function withAgentDeviceEnvironment(