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 79038454ee737..61e4d81b5ef4e 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" }, @@ -1773,7 +1774,11 @@ 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) && + (delivered(waited) || waited.value.interruptedByMessageId !== undefined) + ) + return waited.value; return yield* readTask(scope, taskId, Option.isNone(waited), true); }); @@ -1897,18 +1902,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, ); }, @@ -3221,7 +3238,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, @@ -3232,6 +3257,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); @@ -3262,6 +3292,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..843ab6cc7e77f 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 (latest run by default). Idle returns. interruptedByMessageId: handle input; target runs. Timeout: next ends turn; else retry.", 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 ca382b39b6060..d04c265bdeba6 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -57,6 +57,8 @@ import { type ProviderReplayGate, } from "@t3tools/provider-testing/replayGate"; +import { OpenCodeReplayController } from "./OpenCodeAdapterV2.testkit.ts"; + const SESSION = "ses_f148ca2deffeJcwCnRQtb0YFNX"; const WORK = "/work/opencode2"; const instanceId = ProviderInstanceId.make("opencode"); @@ -338,6 +340,7 @@ const resumed = ( readonly resolveExtractedTextPath?: (path: string) => Effect.Effect; readonly fileSystem?: FileSystem.FileSystem; readonly onPrompt?: (input: unknown) => void; + readonly onRequest?: (operation: string) => void; }, ) => Effect.gen(function* () { @@ -375,6 +378,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?.processExit === undefined ? {} : { processExit: options.processExit }), ...(options?.beforeResponse === undefined @@ -1656,6 +1660,187 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { event("session.execution.started", { sessionID: CHILD }), ]; + it("a cancelled event subscription leaves its gated event for reconnect", async () => { + const controller = new OpenCodeReplayController({ + provider: "opencode", + protocol: "opencode2-http.sse", + version: "2", + scenario: "cancel-gated-event", + entries: [event("session.execution.started", { sessionID: CHILD }, "cancel-gate")], + }); + const entered = Promise.withResolvers(); + const abort = new AbortController(); + const stream = controller.events( + abort.signal, + () => + new Promise((resolve) => { + abort.signal.addEventListener("abort", () => resolve(), { once: true }); + entered.resolve(); + }), + ); + const events = stream[Symbol.asyncIterator](); + const next = events.next(); + await entered.promise; + abort.abort(); + assert.isTrue((await next).done); + const replacement = controller.events()[Symbol.asyncIterator](); + assert.isFalse((await replacement.next()).done); + controller.assertComplete(); + }); + + it.effect.each([ + { goal: false, late: false, fails: false }, + { goal: true, late: false, fails: false }, + { goal: false, late: true, fails: false }, + { goal: true, late: true, fails: false }, + { goal: false, late: true, fails: true }, + ])( + "answers steering while foreground subagents keep running, active goal=$goal, child starts after steer=$late, background fails=$fails", + ({ goal, late, fails }) => + Effect.gen(function* () { + const gate = makeProviderReplayGate(["release-child", "start-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( + [ + ...(late ? foreground.slice(0, 3) : foreground), + ...(goal + ? [ + event("rpc.opencode.goal.changed", { + sessionID: SESSION, + goal: nativeGoal("active"), + }), + ] + : []), + ...(!late ? [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", + }), + ...(late + ? [ + ...foreground + .slice(3) + .map((entry, index) => + entry.type === "emit_inbound" && index === 0 + ? { ...entry, label: "start-child" } + : entry, + ), + event("session.execution.started", { sessionID: CHILD }), + ] + : []), + out("session.background", { sessionID: SESSION }), + reply( + "session.background", + fails + ? { status: 500, body: { _tag: "UnknownError", message: "background failed" } } + : 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)); + if (!late) 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.", + }, + }); + if (late) { + gate.release("start-child"); + yield* Deferred.await(childStarted); + } + 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) => @@ -6563,7 +6748,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { const toolB = { sessionID: SESSION, assistantMessageID: "msg_assistant", id: "call-b" }; const spawn = { sessionID: SESSION, assistantMessageID: "msg_reply", id: "call-spawn" }; const offers: Array = []; - const gate = makeProviderReplayGate(["settled"]); + const stopped = Promise.withResolvers(); const { runtime, thread } = yield* resumed( [ ...backgroundLaunch(CHILD), @@ -6683,9 +6868,13 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { event("session.execution.started", { sessionID: SESSION }), out("session.interrupt", { sessionID: SESSION }), reply("session.interrupt", { interrupted: true }), - event("session.execution.interrupted", { sessionID: SESSION }, "settled"), ], - { replayGate: gate }, + { + beforeResponse: (path) => + Effect.sync(() => { + if (path === `/api/session/${SESSION}/interrupt`) stopped.resolve(); + }), + }, ).pipe( Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { offer: (request) => Effect.sync(() => void offers.push(request)), @@ -6728,8 +6917,7 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { requestRuntimeRestart: true, }); } - yield* Effect.promise(() => gate.waitForReached("settled")); - gate.release("settled"); + yield* Effect.promise(() => stopped.promise); assert.lengthOf(offers, 1); assert.deepEqual( collected.flatMap((event) => diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts index 000176e334696..c1e3818891a74 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, beforeResponse: ((path: string) => Effect.Effect) | undefined, ) => HttpClient.make((request, url) => @@ -171,6 +173,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); @@ -178,9 +181,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(); @@ -224,6 +234,7 @@ export const replayServer = ( readonly processExit?: Effect.Effect; readonly beforeResponse?: (path: string) => Effect.Effect; readonly onPrompt?: (input: unknown) => void; + readonly onRequest?: (operation: string) => void; readonly replayGate?: ProviderReplayGate; readonly borrowers?: { current: number }; }, @@ -243,6 +254,7 @@ export const replayServer = ( controller, options?.replayGate, options?.onPrompt, + options?.onRequest, options?.beforeResponse, ), ), @@ -280,6 +292,7 @@ const makeReplayAdapter = ( readonly processExit?: Effect.Effect; readonly beforeResponse?: (path: string) => Effect.Effect; readonly onPrompt?: (input: unknown) => void; + readonly onRequest?: (operation: string) => void; readonly replayGate?: ProviderReplayGate; readonly borrowers?: { current: number }; }, @@ -336,6 +349,7 @@ export const openCode2ReplayRuntime = ( readonly beforeResponse?: (path: string) => Effect.Effect; 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/Adapters/OpenCodeAdapterV2.testkit.ts b/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.testkit.ts index 90ce687ca591f..bcbd5b766335e 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.testkit.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCodeAdapterV2.testkit.ts @@ -256,6 +256,7 @@ export class OpenCodeReplayController { if (entry.afterMs !== undefined && entry.afterMs > 0) { await Effect.runPromise(Effect.sleep(Duration.millis(entry.afterMs))); } + if (signal?.aborted) return; const event = materializeMessageIds(frame.event, this.messageIds); this.advance(); yield event; 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..0d6fb8ed4cdc6 100644 --- a/apps/server/src/peer/PeerForwarding.test.ts +++ b/apps/server/src/peer/PeerForwarding.test.ts @@ -4,7 +4,11 @@ import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { EnvironmentId, + EventId, + MessageId, + TurnItemId, type OrchestrationV2Run, + OrchestrationV2StoredEvent, type OrchestrationV2ThreadShell, type SourceControlCloneRepositoryInput, ProjectId, @@ -14,10 +18,15 @@ import { } from "@t3tools/contracts"; import { expect, it } from "@effect/vitest"; import * as Context from "effect/Context"; +import * as Deferred from "effect/Deferred"; +import * as DateTime from "effect/DateTime"; 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 +85,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?: { started: Deferred.Deferred; release: 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 +110,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 +146,13 @@ 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.started, undefined).pipe( + Effect.andThen(Deferred.await(blockedWait.release)), + ) + : Effect.void, + ), Effect.map(({ waits }) => ({ threadId: input.threadId, run: run(waits.length < 3 ? "running" : "completed"), @@ -216,7 +236,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?: { started: Deferred.Deferred; release: Deferred.Deferred }, +) => servePeer( box, Layer.mergeAll( @@ -226,7 +250,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 +297,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, @@ -294,11 +320,17 @@ const makeLaptop = Effect.gen(function* () { ), ), Layer.provide(NodeServices.layer), - Layer.provide(PeerForwarding.layer), + Layer.provideMerge(PeerForwarding.layer), Layer.provide(Layer.succeedContext(linking)), 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 +402,14 @@ 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, + forwarding: Context.get(here, PeerForwarding.PeerForwarding), + scopeOf, + }; }); const failureOf = (result: McpSchema.CallToolResult) => { @@ -378,6 +417,95 @@ const failureOf = (result: McpSchema.CallToolResult) => { return result.isError === true && text?.type === "text" ? JSON.parse(text.text) : undefined; }; +it.effect.each([true, false])( + "incoming steering only releases an interactive forwarded wait, interactive=%s", + (interactive) => + Effect.scoped( + Effect.gen(function* () { + const seen = yield* Ref.make(nothingSeen); + const linkSession = yield* Ref.make(""); + const blocked = { + started: yield* Deferred.make(), + release: 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 input = { threadId: boxThread, timeoutMs: 600_000, untilDelegatedWorkSettled: true }; + const wait = interactive + ? a + .call("thread:laptop-full", "t3_thread_wait", { + ...input, + environmentId: box.environmentId, + }) + .pipe(Effect.map((result) => result.structuredContent)) + : a.forwarding.waitForThread(a.scopeOf("thread:laptop-full"), box.environmentId, input); + const waiting = yield* wait.pipe(Effect.forkChild); + yield* Effect.addFinalizer(() => Deferred.succeed(blocked.release, undefined)); + if (interactive) yield* Deferred.await(a.inputSubscribed); + yield* Deferred.await(blocked.started); + const now = yield* DateTime.now; + yield* Queue.offer( + a.inputs, + OrchestrationV2StoredEvent.make({ + sequence: 1, + commandId: null, + event: { + id: EventId.make("event:remote-wait-input"), + type: "turn-item.updated", + threadId: ThreadId.make("thread:laptop-full"), + occurredAt: now, + payload: { + id: TurnItemId.make("item:remote-wait-input"), + threadId: ThreadId.make("thread:laptop-full"), + runId: RunId.make("run:laptop-full"), + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "completed", + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + createdBy: "agent", + creationSource: "mcp", + freshContext: false, + type: "user_message", + inputIntent: "steer", + messageId: MessageId.make("message:remote-wait-input"), + text: "Handle this first.", + attachments: [], + }, + }, + }), + ); + if (!interactive) { + expect(yield* Deferred.isDone(a.inputSubscribed)).toBe(false); + expect(waiting.pollUnsafe()).toBeUndefined(); + yield* Deferred.succeed(blocked.release, undefined); + expect(yield* Fiber.join(waiting)).toMatchObject({ + status: "completed", + timedOut: false, + }); + return; + } + expect(yield* Fiber.join(waiting)).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([1, 50_000]); + expect(observed.waits[1]?.runId).toBe(boxRun); + }), + ).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* () { @@ -480,7 +608,7 @@ it.effect("keeps a forwarded wait short and its retries scoped to the caller", ( expect( (yield* Ref.get(seen)).waits.map(({ timeoutMs, runId }) => [timeoutMs, runId]), ).toEqual([ - [PeerForwarding.FORWARDED_WAIT_CHUNK_MS, undefined], + [1, undefined], [PeerForwarding.FORWARDED_WAIT_CHUNK_MS, boxRun], [PeerForwarding.FORWARDED_WAIT_CHUNK_MS, boxRun], ]); diff --git a/apps/server/src/peer/PeerForwarding.ts b/apps/server/src/peer/PeerForwarding.ts index e10d851f76913..17c8e1eb44bee 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"; @@ -188,36 +191,93 @@ const make = Effect.gen(function* () { input, ); - const waitForThread: PeerForwarding["Service"]["waitForThread"] = Effect.fn( - "PeerForwarding.waitForThread", - )(function* (scope, environmentId, input) { + const waitForThread = Effect.fn("PeerForwarding.waitForThread")(function* ( + scope: Scope, + environmentId: EnvironmentId, + input: OrchestratorMcpThreadWaitInput, + interruptOnSteering = false, + ) { + // Background followers must keep waiting through local input until the remote work settles. + const callerThreadId = interruptOnSteering ? scope.thread?.threadId : undefined; + 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( + const request = (timeoutMs: number) => + forward( scope, "t3_thread_wait", OrchestratorMcpThreadWaitInput, OrchestratorMcpThreadWaitResult, environmentId, - { - threadId: input.threadId, - ...(runId === undefined ? {} : { runId }), - timeoutMs: chunk, - ...(input.untilDelegatedWorkSettled === undefined - ? {} - : { untilDelegatedWorkSettled: input.untilDelegatedWorkSettled }), - }, + { ...input, runId, timeoutMs }, ); - 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; + // Pin the default run and retain its last observed state before a blocking request can stall. + let current: OrchestratorMcpThreadWaitResult | undefined; + if (callerThreadId !== undefined) { + const snapshot = yield* request(1).pipe(Effect.timeoutOption("5 seconds")); + if (Option.isNone(snapshot)) { + return yield* new OrchestratorMcpFailure({ + code: "orchestration_error", + message: "Unable to read the linked thread before waiting.", + }); + } + current = snapshot.value; + remaining -= 1; + if (!current.timedOut || remaining <= 0) return current; + runId = current.runId ?? undefined; } + const wait = Effect.gen(function* () { + while (true) { + const chunk = Math.min(remaining, FORWARDED_WAIT_CHUNK_MS); + const result = yield* request(chunk); + current = result; + 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)), + ); + 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; + const { next: _next, ...observed } = current!; + return { ...observed, timedOut: false, interruptedByMessageId: outcome.messageId }; }); // While the beta is off, agents start nothing over links. Tasks already running there keep @@ -235,6 +295,7 @@ const make = Effect.gen(function* () { scope, environmentId, params as OrchestratorMcpThreadWaitInput, + true, ) as Effect.Effect, OrchestratorMcpFailure>) : forward(scope, tool.name, tool.parametersSchema, tool.successSchema, environmentId, params); return Effect.andThen(requireEnabled, forwarded); diff --git a/apps/server/src/provider/agentContextBudget.json b/apps/server/src/provider/agentContextBudget.json index e166af3f78340..815f96e843eeb 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": 48120 } diff --git a/evals/mcp-tool-use/cases/orchestration.ndjson b/evals/mcp-tool-use/cases/orchestration.ndjson index 520bd2abf10d3..bc30cd9ddf866 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