From ae14239f2b83818bb78bf092cb15929ae94f421c Mon Sep 17 00:00:00 2001 From: Emi Griffith Date: Thu, 12 Nov 2026 12:54:08 -0800 Subject: [PATCH 1/4] fix(server): handle steering while waiting for child work --- .../src/mcp/OrchestratorMcpService.test.ts | 49 +++++-- apps/server/src/mcp/OrchestratorMcpService.ts | 45 +++++- .../src/mcp/toolkits/orchestrator/tools.ts | 2 +- .../Adapters/OpenCode2AdapterV2.test.ts | 128 ++++++++++++++++++ .../Adapters/OpenCode2AdapterV2.testkit.ts | 17 ++- .../ThreadManagementService.test.ts | 64 +++++++++ .../ThreadManagementService.ts | 51 ++++++- apps/server/src/peer/PeerForwarding.test.ts | 79 ++++++++++- apps/server/src/peer/PeerForwarding.ts | 96 ++++++++++--- .../src/provider/agentContextBudget.json | 2 +- evals/mcp-tool-use/cases/orchestration.ndjson | 1 + packages/contracts/src/orchestratorMcp.ts | 8 +- .../src/server/v2/adapter.ts | 19 +++ 13 files changed, 510 insertions(+), 51 deletions(-) diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index 8f349f77b71aa..904940f584acc 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -1273,6 +1273,7 @@ describe("OrchestratorMcpService provider resolution", () => { it.effect.each([ "parent completion", + "caller steering", "nested child completion", "child run completion", "child question", @@ -1285,15 +1286,19 @@ describe("OrchestratorMcpService provider resolution", () => { const nestedId = NodeId.make("node:mcp-wait-nested"); const childRunId = RunId.make("run:mcp-wait-child"); const childRootId = NodeId.make("node:mcp-wait-child-root"); - const eventThreadId = wake === "parent completion" ? parentThreadId : childThreadId; + const eventThreadId = + wake === "parent completion" || wake === "caller steering" ? parentThreadId : childThreadId; const eventType = - wake === "child run completion" - ? "run.updated" - : wake === "child question" || wake === "child approval" - ? "runtime-request.updated" - : "subagent.updated"; + wake === "caller steering" + ? "turn-item.updated" + : wake === "child run completion" + ? "run.updated" + : wake === "child question" || wake === "child approval" + ? "runtime-request.updated" + : "subagent.updated"; const terminal = wake.endsWith("completion"); let ready = false; + const commands: Array = []; const task = () => ({ id: taskId, threadId: parentThreadId, @@ -1326,7 +1331,7 @@ describe("OrchestratorMcpService provider resolution", () => { : [], nodes: [{ id: childRootId, runId: childRunId }], runtimeRequests: - ready && !terminal + ready && !terminal && wake !== "caller steering" ? [ { id: "request:mcp-wait-answer", @@ -1357,11 +1362,13 @@ describe("OrchestratorMcpService provider resolution", () => { Stream.concat(Stream.fromQueue(events)), ) : Stream.never, - dispatch: () => - Effect.succeed({ + dispatch: (command) => { + commands.push(command); + return Effect.succeed({ sequence: 1, storedEvents: [{ event: { type: "subagent.updated", payload: task() } }], - } as never), + } as never); + }, }), ServerSettings.layerTest(), providerRegistryLayer([ @@ -1417,7 +1424,15 @@ describe("OrchestratorMcpService provider resolution", () => { event: { type: eventType, threadId: eventThreadId, - payload: { id: wake === "parent completion" ? taskId : nestedId }, + payload: + wake === "caller steering" + ? { + type: "user_message", + runId: parentRunId, + messageId: MessageId.make("message:mcp-wait-steer"), + inputIntent: "steer", + } + : { id: wake === "parent completion" ? taskId : nestedId }, }, } as unknown as OrchestrationV2StoredEvent); const result = yield* Fiber.join(waiting); @@ -1427,7 +1442,17 @@ describe("OrchestratorMcpService provider resolution", () => { if (wake === "parent completion") assert.equal(result.summary, "done"); } else { assert.equal(result.status, "running"); - if (wake === "child question") assert.isTrue(result.hasPendingQuestion); + if (wake === "caller steering") { + assert.equal(result.interruptedByMessageId, "message:mcp-wait-steer"); + assert.notProperty(result, "next"); + assert.isTrue( + commands.some( + (command) => + command.type === "delegated_task.wake-policy" && + command.completionWake === "always", + ), + ); + } else if (wake === "child question") assert.isTrue(result.hasPendingQuestion); else { assert.deepEqual(result.pendingApproval, { requestId: "request:mcp-wait-answer", diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index 3a43e2a1751e9..82303d38725c2 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -136,6 +136,7 @@ const DEFAULT_TASK_WAIT_TIMEOUT_MS = 4 * 60 * 1_000; const MAX_WAIT_TIMEOUT_MS = 60 * 60 * 1_000; const TASK_WAKE_EVENTS = [ { thread: "parent", eventType: "subagent.updated" }, + { thread: "parent", eventType: "turn-item.updated" }, { thread: "child", eventType: "run.updated" }, { thread: "child", eventType: "subagent.updated" }, { thread: "child", eventType: "provider-thread.updated" }, @@ -1768,7 +1769,7 @@ const make = Effect.gen(function* () { Exit.isSuccess(exit) && delivered(exit.value) ? Effect.void : upgradeWake, ), ); - if (Option.isSome(waited) && delivered(waited)) return waited.value; + if (Option.isSome(waited)) return waited.value; return yield* readTask(scope, taskId, Option.isNone(waited), true); }); @@ -1892,18 +1893,30 @@ const make = Effect.gen(function* () { }) .pipe( Stream.filter( - ({ event }) => + (stored) => thread === "child" || - event.type !== "subagent.updated" || - event.payload.id === taskId, + (stored.event.type === "turn-item.updated" + ? ThreadManagementService.steeringInputMessage(stored) !== undefined + : stored.event.type !== "subagent.updated" || + stored.event.payload.id === taskId), ), ), ), { concurrency: "unbounded" }, ).pipe( Stream.mapError(streamError), - Stream.mapEffect(() => readTask(scope, taskId, false, true)), - Stream.filter(endsTaskWait), + Stream.mapEffect((stored) => + Effect.gen(function* () { + const result = yield* readTask(scope, taskId, false, true); + const messageId = ThreadManagementService.steeringInputMessage(stored); + if (messageId === undefined) return result; + const { next: _next, ...current } = result; + return { ...current, interruptedByMessageId: messageId }; + }), + ), + Stream.filter( + (result) => result.interruptedByMessageId !== undefined || endsTaskWait(result), + ), Stream.runHead, ); }, @@ -3220,7 +3233,15 @@ const make = Effect.gen(function* () { } satisfies OrchestratorMcpThreadSendResult; }), waitForThread: Effect.fn("OrchestratorMcpService.waitForThread")(function* (scope, input) { - const { parent, target } = yield* loadScopedThread(scope, input.threadId, []); + const { parent, target } = yield* loadScopedThread(scope, input.threadId, ["runs"]); + const callerRun = + parent === undefined + ? undefined + : ThreadManagementService.latestActiveRun({ + runs: parent.runs.filter( + (run) => run.providerInstanceId === scope.thread?.providerInstanceId, + ), + }); const result = yield* threadManagement .waitForThread({ projectId: target.thread.projectId, @@ -3231,6 +3252,11 @@ const make = Effect.gen(function* () { Math.max(1, input.timeoutMs ?? DEFAULT_WAIT_TIMEOUT_MS), ), ...(input.untilDelegatedWorkSettled === true ? { untilDelegatedWorkSettled: true } : {}), + ...(callerRun === undefined || parent === undefined + ? {} + : { + caller: { threadId: parent.thread.id, runId: callerRun.id }, + }), }) .pipe(Effect.mapError(threadManagementFailure)); const status = result.run === null ? "idle" : runOutcome(result.run); @@ -3261,6 +3287,11 @@ const make = Effect.gen(function* () { runId: result.run?.id ?? null, status, timedOut: result.timedOut, + ...(result.interruptedByMessageId === undefined + ? {} + : { + interruptedByMessageId: result.interruptedByMessageId, + }), ...(wakesCaller ? { next: ORCHESTRATOR_MCP_RUNNING_TASK_NEXT } : {}), ...(result.resultRun === undefined ? {} diff --git a/apps/server/src/mcp/toolkits/orchestrator/tools.ts b/apps/server/src/mcp/toolkits/orchestrator/tools.ts index 38d8fb4e0ada5..8d891d7ac8096 100644 --- a/apps/server/src/mcp/toolkits/orchestrator/tools.ts +++ b/apps/server/src/mcp/toolkits/orchestrator/tools.ts @@ -290,7 +290,7 @@ const ThreadSendTool = Tool.make("t3_thread_send", { const ThreadWaitTool = Tool.make("t3_thread_wait", { description: - "Wait for a thread's run to finish. No `runId`: latest. Idle: at once. On timeout, with `next` end your turn; without, call again.", + "Wait for a thread's run to finish. No `runId`: latest. Idle: at once. `interruptedByMessageId`: handle incoming input first; target keeps running. On timeout, with `next` end your turn; without, call again.", parameters: OrchestratorMcpThreadWaitInput, success: OrchestratorMcpThreadWaitResult, failure: OrchestratorMcpFailure, diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index acfbdad41ea8e..18ac1c07e609e 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -333,6 +333,7 @@ const resumed = ( readonly resolveExtractedTextPath?: (path: string) => Effect.Effect; readonly fileSystem?: FileSystem.FileSystem; readonly onPrompt?: (input: unknown) => void; + readonly onRequest?: (operation: string) => void; readonly borrowers?: { current: number }; }, ) => @@ -371,6 +372,7 @@ const resumed = ( ...(options?.borrowers === undefined ? {} : { borrowers: options.borrowers }), ...(options?.fileSystem === undefined ? {} : { fileSystem: options.fileSystem }), ...(options?.onPrompt === undefined ? {} : { onPrompt: options.onPrompt }), + ...(options?.onRequest === undefined ? {} : { onRequest: options.onRequest }), ...(options?.external === undefined ? {} : { external: options.external }), ...(options?.replayGate === undefined ? {} : { replayGate: options.replayGate }), ...(options?.attachmentsDir === undefined @@ -1648,6 +1650,132 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { event("session.execution.started", { sessionID: CHILD }), ]; + it.effect.each([false, true])( + "answers steering while foreground subagents keep running, active goal=%s", + (goal) => + Effect.gen(function* () { + const gate = makeProviderReplayGate(["release-child"]); + const requests: Array = []; + const steerId = `msg_t3_steer_${SESSION}:message:opencode2-adapter:steer`; + const foreground = backgroundLaunch(CHILD).map((entry) => { + if (entry.type !== "emit_inbound" || !entry.frame || typeof entry.frame !== "object") + return entry; + const frame = entry.frame as { event?: { type: string; data: Record } }; + if (frame.event?.type !== "session.tool.called") return entry; + return event("session.tool.called", { + ...frame.event.data, + input: { description: "Sleep", prompt: "sleep", background: false }, + }); + }); + const { runtime, thread } = yield* resumed( + [ + ...foreground, + ...(goal + ? [ + event("rpc.opencode.goal.changed", { + sessionID: SESSION, + goal: nativeGoal("active"), + }), + ] + : []), + event("session.execution.started", { sessionID: CHILD }), + out("session.prompt", { sessionID: SESSION, id: steerId, delivery: "steer" }), + replyData("session.prompt", { + id: steerId, + sessionID: SESSION, + time: { created: 1790656601500 }, + type: "user", + payload: { text: "Also say STEERED." }, + delivery: "steer", + }), + out("session.background", { sessionID: SESSION }), + reply("session.background", null), + returnedRunning("call-background"), + event("session.inbox.delivered", { sessionID: SESSION, inboxID: steerId }), + event("session.text.ended", { + sessionID: SESSION, + assistantMessageID: "msg_steer_reply", + ordinal: 0, + text: "STEERED", + }), + event("session.execution.succeeded", { sessionID: SESSION }), + event("session.execution.succeeded", { sessionID: CHILD }, "release-child"), + childReport("msg_child_report", "CHILD_DONE"), + ...(goal + ? [ + event("rpc.opencode.goal.changed", { + sessionID: SESSION, + goal: nativeGoal("complete"), + }), + ] + : []), + ], + { + version: "2.2.5", + replayGate: gate, + onRequest: (operation) => requests.push(operation), + }, + ); + const childStarted = yield* Deferred.make(); + const childDone = yield* Deferred.make(); + const answered = yield* Deferred.make(); + const parentEnded = yield* Deferred.make(); + const terminals: Array = []; + let childStatus: string | undefined; + let callStatus: string | undefined; + let result: string | null = null; + yield* runtime.events.pipe( + Stream.runForEach((e) => { + if (e.type === "turn.terminal") { + terminals.push(e); + return Deferred.succeed(parentEnded, undefined); + } + if ( + e.type === "provider_turn.updated" && + e.providerTurn.nativeTurnRef?.nativeId?.startsWith(`${CHILD}:turn:`) + ) { + childStatus = e.providerTurn.status; + if (childStatus === "running") return Deferred.succeed(childStarted, undefined); + } + if (e.type === "subagent.updated" && e.subagent.title === "Sleep") { + callStatus = e.subagent.status; + result = e.subagent.result; + if (callStatus === "completed") return Deferred.succeed(childDone, undefined); + } + return e.type === "message.updated" && e.message.text === "STEERED" + ? Deferred.succeed(answered, undefined) + : Effect.void; + }), + Effect.forkScoped, + ); + yield* runtime.startTurn(withLineage(thread)); + yield* Deferred.await(childStarted); + yield* runtime.steerTurn({ + threadId, + runId: RunId.make("run:opencode2-adapter"), + providerThread: thread, + providerTurnId: yield* providerTurnId, + message: { + ...turnInput(thread).message, + messageId: MessageId.make("message:opencode2-adapter:steer"), + text: "Also say STEERED.", + }, + }); + assert.include(requests, "session.background"); + assert.isTrue(yield* runtime.hasPendingBackgroundWorkForThread!(thread)); + yield* Deferred.await(answered); + yield* Effect.promise(() => gate.waitForReached("release-child")); + if (!goal) yield* Deferred.await(parentEnded); + assert.equal(childStatus, "running"); + assert.equal(callStatus, "running"); + assert.lengthOf(terminals, goal ? 0 : 1); + gate.release("release-child"); + yield* Deferred.await(childDone); + assert.equal(result, "CHILD_DONE"); + yield* Deferred.await(parentEnded); + }).pipe(Effect.scoped), + ); + it.effect.each([false, true])( "settles a failed background subagent with recovered child=%s", (recoveredChild) => diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts index c031e8d529132..e311774459730 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts @@ -112,6 +112,7 @@ const operationOf = ( if (method === "POST" && rest === "/command") return { type: "session.command", input }; if (method === "POST" && rest === "/compact") return { type: "session.compact", input }; if (method === "POST" && rest === "/interrupt") return { type: "session.interrupt", input }; + if (method === "POST" && rest === "/background") return { type: "session.background", input }; if (method === "POST" && rest === "/model") return { type: "session.switchModel", input }; if (method === "POST" && rest === "/agent") return { type: "session.switchAgent", input }; if (method === "POST" && rest === "/move") return { type: "session.move", input }; @@ -152,6 +153,7 @@ const replayHttpClient = ( controller: OpenCodeReplayController, replayGate: ProviderReplayGate | undefined, onPrompt: ((input: unknown) => void) | undefined, + onRequest: ((operation: string) => void) | undefined, ) => HttpClient.make((request, url) => Effect.tryPromise({ @@ -170,6 +172,7 @@ const replayHttpClient = ( query, raw === undefined ? undefined : decodeJson(raw), ); + onRequest?.(operation.type); // HTTP streams and requests use separate connections; deliver prior recorded inbound data before matching requests. if (operation.type !== "event.subscribe") await controller.untilInboundDelivered(); if (operation.type === "session.prompt") onPrompt?.(operation.input); @@ -177,9 +180,16 @@ const replayHttpClient = ( if (operation.type === "event.subscribe") { controller.exited = false; const encoder = new TextEncoder(); - const events = controller.events(undefined, replayGate?.beforeEmit); + const abort = new AbortController(); + const events = controller.events( + abort.signal, + (label) => replayGate?.beforeEmit(label, abort.signal) ?? Promise.resolve(), + ); const frames = events[Symbol.asyncIterator](); const body = new ReadableStream({ + cancel() { + abort.abort(); + }, async pull(stream) { const next = await frames.next(); if (next.done === true) stream.close(); @@ -218,6 +228,7 @@ export const replayServer = ( options?: { readonly external?: boolean; readonly onPrompt?: (input: unknown) => void; + readonly onRequest?: (operation: string) => void; readonly replayGate?: ProviderReplayGate; readonly borrowers?: { current: number }; }, @@ -233,7 +244,7 @@ export const replayServer = ( const opencode = yield* OpenCode2Client.make.pipe( Effect.provideService( HttpClient.HttpClient, - replayHttpClient(controller, options?.replayGate, options?.onPrompt), + replayHttpClient(controller, options?.replayGate, options?.onPrompt, options?.onRequest), ), ); const connection = { @@ -266,6 +277,7 @@ const makeReplayAdapter = ( options?: { readonly external?: boolean; readonly onPrompt?: (input: unknown) => void; + readonly onRequest?: (operation: string) => void; readonly replayGate?: ProviderReplayGate; readonly borrowers?: { current: number }; }, @@ -320,6 +332,7 @@ export const openCode2ReplayRuntime = ( readonly external?: boolean; readonly version?: string; readonly onPrompt?: (input: unknown) => void; + readonly onRequest?: (operation: string) => void; readonly borrowers?: { current: number }; readonly replayGate?: ProviderReplayGate; readonly attachmentsDir?: string; diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index ab846252c9e58..d5f1f82dfcedb 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -270,6 +270,70 @@ it.effect("waitForThread follows a restart continuation instead of the cut run", }), ); +it.effect("waitForThread yields to caller input without ending the child's run", () => + Effect.gen(function* () { + const projectId = ProjectId.make("project:responsive-wait"); + const threadId = ThreadId.make("thread:responsive-wait-child"); + const callerThreadId = ThreadId.make("thread:responsive-wait-parent"); + const callerRunId = RunId.make("run:responsive-wait-parent"); + const messageId = MessageId.make("message:responsive-wait"); + const subscribed = yield* Deferred.make(); + const inputs = yield* Queue.unbounded(); + const run = { id: RunId.make("run:responsive-wait-child"), status: "running" as const }; + const layerTest = ThreadManagementService.layer.pipe( + Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), + Layer.provide( + Layer.mock(Orchestrator.OrchestratorV2)({ + getThreadEventSequence: () => Effect.succeed(0), + streamStoredEventsFrom: (input) => + input?.threadId === callerThreadId + ? Stream.fromQueue(inputs) + : Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe( + Stream.drain, + Stream.concat(Stream.never), + ), + getThreadRecords: () => + Effect.succeed({ + thread: { id: threadId, projectId, deletedAt: null }, + runs: [run], + } as unknown as OrchestrationV2ThreadProjection), + }), + ), + Layer.provide(ServerSettings.layerTest()), + Layer.provide(layerReceiptsTest), + ); + const service = yield* ThreadManagementService.ThreadManagementService.pipe( + Effect.provide(layerTest), + ); + const waiting = yield* service + .waitForThread({ + projectId, + threadId, + timeoutMs: 1000, + caller: { threadId: callerThreadId, runId: callerRunId }, + }) + .pipe(Effect.forkChild); + yield* Deferred.await(subscribed); + for (const inputIntent of ["queued_turn", "steer"] as const) { + yield* Queue.offer(inputs, { + sequence: 1, + event: { + type: "turn-item.updated", + threadId: callerThreadId, + payload: { type: "user_message", runId: callerRunId, messageId, inputIntent }, + }, + } as unknown as OrchestrationV2StoredEvent); + } + yield* TestClock.adjust("1 second"); + expect(yield* Fiber.join(waiting)).toMatchObject({ + timedOut: false, + interruptedByMessageId: messageId, + run: { status: "running" }, + }); + expect(run.status).toBe("running"); + }), +); + it.effect("waitForThread can wait past an ended run for the thread's delegated work", () => Effect.gen(function* () { const projectId = ProjectId.make("project:thread-management:wait-delegated"); diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index e1a1abdc1819b..c6b2a449d679e 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -27,6 +27,7 @@ import { type OrchestrationV2CreationSource, type OrchestrationV2LinkedSender, type OrchestrationV2Run, + type OrchestrationV2StoredEvent, type OrchestrationV2ThreadShellSnapshot, type OrchestrationV2ThreadProjection, type OrchestrationV2ThreadTranscript, @@ -173,6 +174,7 @@ export interface ThreadManagementWaitInput { readonly threadId: ThreadId; readonly runId?: RunId; readonly timeoutMs: number; + readonly caller?: { readonly threadId: ThreadId; readonly runId: RunId }; /** * Also wait until the thread's own delegated work settles: an ended run * still owes the thread's answer while a child, a background task or a @@ -185,6 +187,7 @@ export interface ThreadManagementWaitResult { readonly threadId: ThreadId; readonly run: OrchestrationV2Run | null; readonly timedOut: boolean; + readonly interruptedByMessageId?: MessageId; /** With `untilDelegatedWorkSettled`, once settled: the run with the thread's final answer. */ readonly resultRun?: OrchestrationV2Run; } @@ -575,6 +578,17 @@ function latestSteerableRun( .toSorted((left, right) => right.ordinal - left.ordinal)[0]; } +export function steeringInputMessage(stored: OrchestrationV2StoredEvent, runId?: RunId) { + const event = stored.event; + return event.type === "turn-item.updated" && + event.payload.type === "user_message" && + (runId === undefined || event.payload.runId === runId) && + (event.payload.inputIntent === "steer" || + event.payload.inputIntent === "promoted_queued_to_steer") + ? event.payload.messageId + : undefined; +} + const SETTLE_AFTER_RUN_WAIT_MS = 24 * 60 * 60 * 1_000; const FORCE_STOP_REASON = "Force stopped by user"; const FORCE_STOP_RELEASE_WAIT = "10 seconds"; @@ -949,6 +963,12 @@ const make = Effect.gen(function* () { const afterSequence = yield* orchestrator .getThreadEventSequence(input.threadId) .pipe(Effect.mapError(loadError)); + const callerAfterSequence = + input.caller === undefined + ? undefined + : yield* orchestrator + .getThreadEventSequence(input.caller.threadId) + .pipe(Effect.mapError(loadError)); const target = yield* getProjectThreadRecords(input, ["runs"]); const selectedRun = input.runId === undefined @@ -1064,7 +1084,36 @@ const make = Effect.gen(function* () { Stream.runHead, Effect.timeoutOption(Duration.millis(Math.max(1, input.timeoutMs))), ); - const waited = Option.flatten(yield* wait); + const caller = input.caller; + const incoming = + caller === undefined + ? Effect.never + : orchestrator + .streamStoredEventsFrom({ + threadId: caller.threadId, + afterSequence: callerAfterSequence!, + eventType: "turn-item.updated", + }) + .pipe( + Stream.mapError(loadError), + Stream.map((stored) => steeringInputMessage(stored, caller.runId)), + Stream.filter((id): id is MessageId => id !== undefined), + Stream.runHead, + Effect.flatMap((id) => (Option.isSome(id) ? Effect.succeed(id.value) : Effect.never)), + ); + const waitedOutcome = yield* Effect.raceFirst( + wait.pipe(Effect.map((result) => ({ type: "settled" as const, result }))), + incoming.pipe(Effect.map((messageId) => ({ type: "input" as const, messageId }))), + ); + if (waitedOutcome.type === "input") { + const current = yield* evaluate((yield* getProjectThreadRecords(input, ["runs"])).runs); + return { + ...settledResult(current), + timedOut: false, + interruptedByMessageId: waitedOutcome.messageId, + }; + } + const waited = Option.flatten(waitedOutcome.result); if (Option.isSome(waited)) return settledResult(waited.value); // The run may have reached a terminal status while the timeout was // winning the race; the final projection read decides what actually diff --git a/apps/server/src/peer/PeerForwarding.test.ts b/apps/server/src/peer/PeerForwarding.test.ts index 3c0e2652ea48b..e21578718012c 100644 --- a/apps/server/src/peer/PeerForwarding.test.ts +++ b/apps/server/src/peer/PeerForwarding.test.ts @@ -5,6 +5,7 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { EnvironmentId, type OrchestrationV2Run, + type OrchestrationV2StoredEvent, type OrchestrationV2ThreadShell, type SourceControlCloneRepositoryInput, ProjectId, @@ -14,10 +15,14 @@ import { } from "@t3tools/contracts"; import { expect, it } from "@effect/vitest"; import * as Context from "effect/Context"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; +import * as Queue from "effect/Queue"; +import * as Stream from "effect/Stream"; import { McpSchema, McpServer } from "effect/ai"; import * as ServerConfig from "../config.ts"; @@ -76,7 +81,11 @@ const usersThread = ThreadId.make("thread:box-users-own"); * started, whose run finishes on the third wait, and the box user's own. * Every send and wait is recorded. */ -const boxThreads = (seen: Ref.Ref, linkSession: Ref.Ref) => { +const boxThreads = ( + seen: Ref.Ref, + linkSession: Ref.Ref, + blockedWait?: Deferred.Deferred, +) => { const base = { ...liveThreadShell(boxThread, { runtimeMode: "auto" }), projectId: boxProject }; const linked = Ref.get(linkSession).pipe( Effect.map((sessionId) => ({ ...base, linkOrigin: { sessionId, label: "T3 Code · Laptop" } })), @@ -97,7 +106,7 @@ const boxThreads = (seen: Ref.Ref, linkSession: Ref.Ref) => { listProjectThreads: () => linked.pipe(Effect.map((shell) => [shell] as never)), sendToThread: (input) => Ref.update(seen, (current) => ({ ...current, sends: [...current.sends, input] })).pipe( - Effect.as({ run: run("running"), delivery: "started" } as never), + Effect.as({ run: run("running"), message: {}, delivery: "started" } as never), ), dispatch: (command) => Ref.update(seen, (current) => ({ @@ -133,6 +142,11 @@ const boxThreads = (seen: Ref.Ref, linkSession: Ref.Ref) => { })), waitForThread: (input) => Ref.updateAndGet(seen, (current) => ({ ...current, waits: [...current.waits, input] })).pipe( + Effect.tap(() => + blockedWait !== undefined && input.timeoutMs > 1 + ? Deferred.succeed(blockedWait, undefined).pipe(Effect.andThen(Effect.never)) + : Effect.void, + ), Effect.map(({ waits }) => ({ threadId: input.threadId, run: run(waits.length < 3 ? "running" : "completed"), @@ -216,7 +230,11 @@ const boxProjects = (seen: Ref.Ref) => * OAuth. `linkSession` is filled in once A links, so B's fixture thread reads * as one that link started. */ -const serveBox = (seen: Ref.Ref, linkSession: Ref.Ref) => +const serveBox = ( + seen: Ref.Ref, + linkSession: Ref.Ref, + blockedWait?: Deferred.Deferred, +) => servePeer( box, Layer.mergeAll( @@ -226,7 +244,7 @@ const serveBox = (seen: Ref.Ref, linkSession: Ref.Ref) => ).pipe( Layer.provide(DocumentText.layer.pipe(Layer.provideMerge(NodeServices.layer))), Layer.provide(NodeCrypto.layer), - Layer.provide(boxThreads(seen, linkSession)), + Layer.provide(boxThreads(seen, linkSession, blockedWait)), Layer.provide(Layer.mock(ThreadSearch.ThreadSearch)({})), Layer.provide(boxLaunches(seen)), Layer.provide( @@ -273,6 +291,8 @@ const mcpClient = McpSchema.McpServerClient.of({ * caller whose modes the test picks. Its local threads never matter here. */ const makeLaptop = Effect.gen(function* () { + const inputs = yield* Queue.unbounded(); + const inputSubscribed = yield* Deferred.make(); const linking = yield* layerLinkingEnvironment(laptop).pipe(Layer.build); const layerHere = Layer.mergeAll( McpHttpServer.layerOrchestratorToolkit, @@ -299,6 +319,12 @@ const makeLaptop = Effect.gen(function* () { Layer.provide(NodeCrypto.layer), Layer.provideMerge( Layer.mock(ThreadManagement.ThreadManagementService)({ + getThreadEventSequence: () => Effect.succeed(0), + streamStoredEventsFrom: () => + Stream.fromEffect(Deferred.succeed(inputSubscribed, undefined)).pipe( + Stream.drain, + Stream.concat(Stream.fromQueue(inputs)), + ), getThreadShell: (threadId) => Effect.succeed( threadId === ThreadId.make("thread:laptop-from-desk") @@ -370,7 +396,7 @@ const makeLaptop = Effect.gen(function* () { ), Effect.provideService(McpSchema.McpServerClient, mcpClient), ); - return { links: Context.get(linking, PeerLinks.PeerLinks), call }; + return { links: Context.get(linking, PeerLinks.PeerLinks), call, inputs, inputSubscribed }; }); const failureOf = (result: McpSchema.CallToolResult) => { @@ -378,6 +404,49 @@ const failureOf = (result: McpSchema.CallToolResult) => { return result.isError === true && text?.type === "text" ? JSON.parse(text.text) : undefined; }; +it.effect("incoming local steering releases a forwarded wait without stopping remote work", () => + Effect.scoped( + Effect.gen(function* () { + const seen = yield* Ref.make(nothingSeen); + const linkSession = yield* Ref.make(""); + const blocked = yield* Deferred.make(); + const b = yield* serveBox(seen, linkSession, blocked); + const a = yield* makeLaptop; + yield* linkTo(a.links, b, "auto"); + yield* Ref.set(linkSession, (yield* b.linkedSessions)[0]!.sessionId); + const waiting = yield* a + .call("thread:laptop-full", "t3_thread_wait", { + environmentId: box.environmentId, + threadId: boxThread, + timeoutMs: 600_000, + }) + .pipe(Effect.forkChild); + yield* Deferred.await(a.inputSubscribed); + yield* Deferred.await(blocked); + yield* Queue.offer(a.inputs, { + sequence: 1, + event: { + type: "turn-item.updated", + threadId: ThreadId.make("thread:laptop-full"), + payload: { + type: "user_message", + inputIntent: "steer", + messageId: "message:remote-wait-input", + }, + }, + } as unknown as OrchestrationV2StoredEvent); + expect((yield* Fiber.join(waiting)).structuredContent).toMatchObject({ + status: "running", + timedOut: false, + interruptedByMessageId: "message:remote-wait-input", + }); + const observed = yield* Ref.get(seen); + expect(observed.stops).toEqual([]); + expect(observed.waits.map((wait) => wait.timeoutMs)).toEqual([50_000, 1]); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + it.effect("an agent here works on a linked environment's threads within its own modes", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/peer/PeerForwarding.ts b/apps/server/src/peer/PeerForwarding.ts index e10d851f76913..1c3ddf70abf66 100644 --- a/apps/server/src/peer/PeerForwarding.ts +++ b/apps/server/src/peer/PeerForwarding.ts @@ -1,5 +1,6 @@ import { type EnvironmentId, + type MessageId, OrchestratorMcpFailure, type OrchestratorMcpEnvironmentLinksResult, OrchestratorMcpThreadInterruptInput, @@ -15,6 +16,8 @@ import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; import * as Hex from "effect/encoding/Hex"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Stream from "effect/Stream"; import * as Schema from "effect/Schema"; import type * as Tool from "effect/ai/Tool"; @@ -191,33 +194,84 @@ const make = Effect.gen(function* () { const waitForThread: PeerForwarding["Service"]["waitForThread"] = Effect.fn( "PeerForwarding.waitForThread", )(function* (scope, environmentId, input) { + const callerThreadId = scope.thread?.threadId; + const afterSequence = + callerThreadId === undefined + ? 0 + : yield* threads.getThreadEventSequence(callerThreadId).pipe( + Effect.mapError( + () => + new OrchestratorMcpFailure({ + code: "orchestration_error", + message: "Unable to watch incoming messages while waiting.", + }), + ), + ); let remaining = Math.min( MAX_WAIT_TIMEOUT_MS, Math.max(1, input.timeoutMs ?? DEFAULT_WAIT_TIMEOUT_MS), ); let runId = input.runId; - while (true) { - const chunk = Math.min(remaining, FORWARDED_WAIT_CHUNK_MS); - const result = yield* forward( - scope, - "t3_thread_wait", - OrchestratorMcpThreadWaitInput, - OrchestratorMcpThreadWaitResult, - environmentId, - { - threadId: input.threadId, - ...(runId === undefined ? {} : { runId }), - timeoutMs: chunk, - ...(input.untilDelegatedWorkSettled === undefined - ? {} - : { untilDelegatedWorkSettled: input.untilDelegatedWorkSettled }), - }, + const wait = Effect.gen(function* () { + while (true) { + const chunk = Math.min(remaining, FORWARDED_WAIT_CHUNK_MS); + const result = yield* forward( + scope, + "t3_thread_wait", + OrchestratorMcpThreadWaitInput, + OrchestratorMcpThreadWaitResult, + environmentId, + { + threadId: input.threadId, + ...(runId === undefined ? {} : { runId }), + timeoutMs: chunk, + ...(input.untilDelegatedWorkSettled === undefined + ? {} + : { untilDelegatedWorkSettled: input.untilDelegatedWorkSettled }), + }, + ); + remaining -= chunk; + if (!result.timedOut || remaining <= 0) return result; + // Later pieces wait for the run the first one picked, not a newer one. + runId = result.runId ?? undefined; + } + }); + if (callerThreadId === undefined) return yield* wait; + const incoming = threads + .streamStoredEventsFrom({ + threadId: callerThreadId, + afterSequence, + eventType: "turn-item.updated", + }) + .pipe( + Stream.map((stored) => ThreadManagement.steeringInputMessage(stored)), + Stream.filter((id): id is MessageId => id !== undefined), + Stream.runHead, + Effect.mapError( + () => + new OrchestratorMcpFailure({ + code: "orchestration_error", + message: "Unable to watch incoming messages while waiting.", + }), + ), + Effect.flatMap((id) => (Option.isSome(id) ? Effect.succeed(id.value) : Effect.never)), ); - remaining -= chunk; - if (!result.timedOut || remaining <= 0) return result; - // Later pieces wait for the run the first one picked, not a newer one. - runId = result.runId ?? undefined; - } + const outcome = yield* Effect.raceFirst( + wait.pipe(Effect.map((result) => ({ type: "result" as const, result }))), + incoming.pipe(Effect.map((messageId) => ({ type: "input" as const, messageId }))), + ); + if (outcome.type === "result") return outcome.result; + // Read the target's current outcome without continuing the blocking wait. + const result = yield* forward( + scope, + "t3_thread_wait", + OrchestratorMcpThreadWaitInput, + OrchestratorMcpThreadWaitResult, + environmentId, + { ...input, runId, timeoutMs: 1 }, + ); + const { next: _next, ...current } = result; + return { ...current, timedOut: false, interruptedByMessageId: outcome.messageId }; }); // While the beta is off, agents start nothing over links. Tasks already running there keep diff --git a/apps/server/src/provider/agentContextBudget.json b/apps/server/src/provider/agentContextBudget.json index e166af3f78340..9ff8cc5f20b1d 100644 --- a/apps/server/src/provider/agentContextBudget.json +++ b/apps/server/src/provider/agentContextBudget.json @@ -1,5 +1,5 @@ { "claudeSystemAppendBytes": 5597, "codexAdditionalContextBytes": 7327, - "mcpToolListJsonBytes": 48126 + "mcpToolListJsonBytes": 48248 } diff --git a/evals/mcp-tool-use/cases/orchestration.ndjson b/evals/mcp-tool-use/cases/orchestration.ndjson index 520bd2abf10d3..a3fec9a89bc4a 100644 --- a/evals/mcp-tool-use/cases/orchestration.ndjson +++ b/evals/mcp-tool-use/cases/orchestration.ndjson @@ -40,3 +40,4 @@ {"id": "settle-after-delayed-merge-waits", "area": "orchestration", "kind": "guard", "pattern": "settle before delayed work finishes", "experience": "Asked to merge a PR once its checks pass and then settle this thread, an agent following the #26749 settle bullet can settle in the first turn; a settled thread refuses watch_pull_request and drops PR-watch wakes, so the merge never happens.", "rule": "Asked to settle ... a thread, including this one: call t3_thread_organize. (the request is to settle after the merge, not now)", "privacy": "commit-safe", "notes": "From the CodeRabbit finding on #26765. Settling this thread applies at turn end, and Orchestrator.ts rejects watch wakes on a settled thread. Good: watch #123 and end the turn without settling. gh pr merge --auto plus a settle would still lose the wake that confirms the merge.", "setup": {"role": "top", "priorTurns": [{"role": "user", "text": "Okay this looks good to me. Any risk merging it?"}, {"role": "assistant", "text": "Low risk. [#123](https://github.com/acme/app/pull/123) only adds an accessible title to the HTML preview frame, and the new test covers it. Two checks are still running: Check and Test Web.", "toolCalls": [{"tool": "Bash", "input": {"command": "gh pr view 123 --json title,files --jq '.title, .files[].path'"}, "result": "fix(web): HTML preview frame gets an accessible title\napps/web/src/HtmlPreview.tsx\napps/web/src/HtmlPreview.test.tsx"}, {"tool": "Bash", "input": {"command": "gh pr checks 123"}, "result": "Lint\tpass\t41s\nTypecheck\tpass\t1m2s\nCheck\tpending\t0\nTest Web\tpending\t0"}, {"tool": "link_pull_request", "input": {"url": "https://github.com/acme/app/pull/123"}, "result": {"host": "github.com", "repository": "acme/app", "number": 123, "url": "https://github.com/acme/app/pull/123", "alreadyLinked": false}}]}], "mocks": {"watch_pull_request": "default", "link_pull_request": "default", "t3_thread_organize": {"settlesWhenTurnEnds": true}}, "world": {"branch": "fix/html-preview-title", "commits": [{"sha": "abc1234", "subject": "fix(web): HTML preview frame gets an accessible title"}], "diff": "diff --git a/apps/web/src/HtmlPreview.tsx b/apps/web/src/HtmlPreview.tsx\n--- a/apps/web/src/HtmlPreview.tsx\n+++ b/apps/web/src/HtmlPreview.tsx\n@@ -1,3 +1,3 @@\n export function HtmlPreview({ html }: { html: string }) {\n- return