diff --git a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts index 2ff2125a51800..104e2ec438992 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.activity.test.ts @@ -693,6 +693,7 @@ it("taskStatus returns task.providerInstanceId rather than the driver kind", asy visibleTurnItems: [], providerTurns: [], runtimeRequests: [], + turnItems: [], messages: [], contextTransfers: [ { diff --git a/apps/server/src/mcp/OrchestratorMcpService.test.ts b/apps/server/src/mcp/OrchestratorMcpService.test.ts index a017590b1f30a..2539bd96446f3 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.test.ts @@ -44,6 +44,7 @@ import * as TestClock from "effect/testing/TestClock"; import { CodexProviderCapabilitiesV2 } from "../orchestration-v2/Adapters/CodexAdapterV2.ts"; import * as CommandReceiptStore from "../orchestration-v2/CommandReceiptStore.ts"; +import * as ProviderSessionManager from "../orchestration-v2/ProviderSessionManager.ts"; import * as EffectWorker from "../orchestration-v2/EffectWorker.ts"; import * as LegacyV1ThreadImporter from "../orchestration-v2/legacy/LegacyV1ThreadImporter.ts"; import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; @@ -356,7 +357,12 @@ describe("OrchestratorMcpService", () => { const childThreadId = ThreadId.make("thread:mcp-next-child"); const taskId = NodeId.make("node:mcp-next-task"); let finished = false; + let queued = false; let pendingRequest = false; + let requestOwner: "task" | "finished" | "unrelated" | "blocker" = "task"; + let pendingSecret = false; + let secretCompleted = false; + let secretOwner: "task" | "finished" | "unrelated" = "task"; const parentProjection = () => ({ thread: { id: parentThreadId }, @@ -378,22 +384,55 @@ describe("OrchestratorMcpService", () => { }) as unknown as OrchestrationV2ThreadProjection; const childRun = () => ({ id: RunId.make("run:mcp-next-child"), - ordinal: 1, - status: finished ? ("completed" as const) : ("running" as const), + ordinal: 2, + status: finished + ? ("completed" as const) + : queued + ? ("queued" as const) + : ("running" as const), delegatedTaskId: taskId, }); const childProjection = () => ({ thread: { id: childThreadId, title: "Child task" }, - runs: [childRun()], + runs: [ + childRun(), + { id: "run:finished", ordinal: 0, status: "completed" }, + { id: "run:unrelated", ordinal: 3, status: "running" }, + ...(queued ? [{ id: "run:blocker", ordinal: 1, status: "running" }] : []), + ], contextTransfers: [], messages: [], subagents: [], providerThreads: [], providerTurns: [], - runtimeRequests: pendingRequest ? [{ status: "pending" }] : [], - nodes: [], - turnItems: [], + runtimeRequests: pendingRequest + ? [ + { + nodeId: "node:question", + kind: "user_input", + status: "pending", + responseCapability: { type: "live" }, + }, + ] + : [], + nodes: [ + { + id: "node:question", + runId: requestOwner === "task" ? childRun().id : `run:${requestOwner}`, + }, + ], + turnItems: pendingSecret + ? [ + { + type: "secret_request", + runId: secretOwner === "task" ? childRun().id : `run:${secretOwner}`, + nodeId: "node:secret", + status: secretCompleted ? "completed" : "waiting", + label: "Private credential request", + }, + ] + : [], }) as unknown as OrchestrationV2ThreadProjection; const layerDependencies = Layer.mergeAll( NodeServices.layer, @@ -408,6 +447,8 @@ describe("OrchestratorMcpService", () => { threadId: childThreadId, run: childRun() as unknown as OrchestrationV2ThreadProjection["runs"][number], timedOut: !finished, + hasPendingQuestion: false, + pendingApproval: null, }), dispatch: () => Effect.succeed({} as never), }), @@ -448,11 +489,49 @@ describe("OrchestratorMcpService", () => { const asking = yield* service.waitForThread(scope, { threadId: childThreadId }); assert.isTrue(asking.timedOut); assert.notProperty(asking, "next"); + const question = yield* service.taskStatus(scope, taskId); + assert.isTrue(question.hasPendingQuestion); + assert.notProperty(question, "next"); + for (const owner of ["finished", "unrelated"] as const) { + requestOwner = owner; + const unrelated = yield* service.taskStatus(scope, taskId); + assert.isFalse(unrelated.hasPendingQuestion); + assert.equal(unrelated.next, ORCHESTRATOR_MCP_RUNNING_TASK_NEXT); + } + queued = true; + requestOwner = "blocker"; + const blocked = yield* service.taskStatus(scope, taskId); + assert.isTrue(blocked.hasPendingQuestion); + assert.notProperty(blocked, "next"); + requestOwner = "unrelated"; + assert.isFalse((yield* service.taskStatus(scope, taskId)).hasPendingQuestion); + queued = false; + requestOwner = "task"; pendingRequest = false; + pendingSecret = true; + const secret = yield* service.taskStatus(scope, taskId); + assert.isTrue(secret.hasPendingQuestion); + assert.notProperty(secret, "next"); + assert.notInclude(JSON.stringify(secret), "Private credential request"); + secretCompleted = true; + assert.isFalse((yield* service.taskStatus(scope, taskId)).hasPendingQuestion); + secretCompleted = false; + for (const owner of ["finished", "unrelated"] as const) { + secretOwner = owner; + const unrelated = yield* service.taskStatus(scope, taskId); + assert.isFalse(unrelated.hasPendingQuestion); + assert.equal(unrelated.next, ORCHESTRATOR_MCP_RUNNING_TASK_NEXT); + } + secretOwner = "task"; + pendingSecret = false; + assert.isFalse((yield* service.taskStatus(scope, taskId)).hasPendingQuestion); + finished = true; + pendingSecret = true; const completed = yield* service.taskStatus(scope, taskId); assert.equal(completed.status, "completed"); + assert.isFalse(completed.hasPendingQuestion); assert.notProperty(completed, "next"); const waited = yield* service.waitForThread(scope, { threadId: childThreadId }); assert.equal(waited.status, "completed"); @@ -493,6 +572,7 @@ describe("OrchestratorMcpService", () => { providerThreads: [], providerTurns: [], runtimeRequests: [], + turnItems: [], } as unknown as OrchestrationV2ThreadProjection; const layerDependencies = Layer.mergeAll( NodeServices.layer, @@ -690,6 +770,7 @@ describe("OrchestratorMcpService", () => { providerThreads: [], providerTurns: [], runtimeRequests: [], + turnItems: [], } as unknown as OrchestrationV2ThreadProjection; const layerDependencies = Layer.mergeAll( NodeServices.layer, @@ -780,6 +861,7 @@ describe("OrchestratorMcpService", () => { providerThreads: [], providerTurns: [], runtimeRequests: [], + turnItems: [], } as unknown as OrchestrationV2ThreadProjection; const layerDependencies = Layer.mergeAll( NodeServices.layer, @@ -874,6 +956,7 @@ describe("OrchestratorMcpService", () => { providerThreads: [], providerTurns: [], runtimeRequests: [], + turnItems: [], } as unknown as OrchestrationV2ThreadProjection; const layerDependencies = Layer.mergeAll( NodeServices.layer, @@ -1002,6 +1085,7 @@ describe("OrchestratorMcpService", () => { providerThreads: [], providerTurns: [], runtimeRequests: [], + turnItems: [], }, ], [ @@ -1027,6 +1111,7 @@ describe("OrchestratorMcpService", () => { providerThreads: [], providerTurns: [], runtimeRequests: [], + turnItems: [], }, ], ]) as unknown as ReadonlyMap; @@ -1271,6 +1356,132 @@ describe("OrchestratorMcpService provider resolution", () => { turnItems: [], } as unknown as OrchestrationV2ThreadProjection; + it.effect("wakes a task wait for a newly posted secret card, not transcript updates", () => + Effect.gen(function* () { + let pendingSecret = false; + let childReads = 0; + const task = { + id: taskId, + threadId: parentThreadId, + origin: "app_owned", + childThreadId, + driver: "codex", + model: "gpt-5.4", + status: "running", + result: null, + completionDelivery: { state: "pending" }, + }; + const child = () => + ({ + ...childProjection, + runs: [ + { + id: RunId.make("run:secret-child"), + ordinal: 1, + status: "running", + delegatedTaskId: taskId, + }, + ], + nodes: [], + turnItems: pendingSecret + ? [ + { + type: "secret_request", + runId: RunId.make("run:secret-child"), + nodeId: "node:secret", + status: "waiting", + label: "Private credential request", + }, + ] + : [], + }) as unknown as OrchestrationV2ThreadProjection; + const layerDependencies = Layer.mergeAll( + NodeServices.layer, + NodeCrypto.layer, + Layer.mock(ThreadManagementService.ThreadManagementService)({ + hasReceipt: () => Effect.succeed(false), + assertProviderInstanceEnabled: () => Effect.void, + delegatedTaskResultPending: () => Effect.succeed(false), + getThreadRecords: (threadId) => + Effect.sync(() => { + if (threadId === parentThreadId) return parentProjection([task]); + childReads++; + return child(); + }), + getThreadShell: (threadId) => Effect.succeed(liveThreadShell(threadId)), + getThreadEventSequence: () => Effect.succeed(1), + isProviderInstanceEnabledForProject: () => Effect.succeed(true), + dispatch: () => + Effect.succeed({ + sequence: 1, + storedEvents: [ + { + sequence: 1, + commandId: null, + event: { type: "subagent.updated", payload: task }, + }, + ], + } as never), + streamStoredEventsFrom: (input) => { + if (input?.threadId !== childThreadId || input.eventType !== "turn-item.updated") + return Stream.empty; + let beforeTranscript = 0; + return Stream.fromEffect( + Effect.sync(() => { + beforeTranscript = childReads; + return { + event: { type: "turn-item.updated", payload: { type: "assistant_message" } }, + } as never; + }), + ).pipe( + Stream.concat( + Stream.fromEffect( + Effect.sync(() => { + assert.equal( + childReads, + beforeTranscript, + "transcript updates must not reload the projection", + ); + pendingSecret = true; + return { + event: { type: "turn-item.updated", payload: { type: "secret_request" } }, + } as never; + }), + ), + ), + ); + }, + }), + providerRegistryLayer([ + providerSnapshot({ + instanceId: codexInstanceId, + driver: ProviderDriverKind.make("codex"), + model: "gpt-5.4", + }), + ]), + adapterRegistryLayer([codexInstanceId]), + ServerSettings.layerTest(), + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(SecretRequests.SecretRequests)({}), + Layer.mock(RemoteDelegation.RemoteDelegation)({}), + Layer.mock(ScheduledTaskService.ScheduledTaskService)({}), + ); + yield* Effect.gen(function* () { + const service = yield* OrchestratorMcpService.OrchestratorMcpService; + const waited = yield* service.delegateTask(scope, { + task: "Wait for credentials", + mode: "wait", + clientRequestId: "secret-wait", + }); + assert.isTrue(waited.hasPendingQuestion); + assert.isFalse(waited.waitTimedOut); + assert.equal(waited.status, "running"); + assert.notProperty(waited, "next"); + assert.notInclude(JSON.stringify(waited), "Private credential request"); + }).pipe(Effect.provide(layerOrchestratorService.pipe(Layer.provide(layerDependencies)))); + }), + ); + it.effect.each([ "parent completion", "nested child completion", @@ -3898,6 +4109,100 @@ describe("OrchestratorMcpService thread lookup", () => { }), ); + it.effect.each([ + { name: "client", agent: false, pending: false }, + { name: "linked peer", linked: true, pending: false }, + { name: "thread caller", agent: true, pending: true }, + ])("preserves wait semantics for $name", (scenario) => + Effect.gen(function* () { + const runId = RunId.make("run:compat-wait"); + const projection = { + ...idleThreadProjection(launched), + runs: [{ id: runId, ordinal: 1, status: "running" }], + nodes: [{ id: "node:compat-wait", runId }], + runtimeRequests: [ + { + id: "request:compat-wait", + nodeId: "node:compat-wait", + kind: "command", + status: "pending", + responseCapability: { type: "live" }, + createdAt: DateTime.makeUnsafe(0), + }, + ], + } as unknown as OrchestrationV2ThreadProjection; + const management = yield* ThreadManagementService.ThreadManagementService.pipe( + Effect.provide( + ThreadManagementService.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({}), + Layer.mock(CommandReceiptStore.CommandReceiptStoreV2)({}), + ServerSettings.layerTest(), + Layer.mock(Orchestrator.OrchestratorV2)({ + getThreadEventSequence: () => Effect.succeed(0), + getThreadRecords: () => Effect.succeed(projection), + streamStoredEventsFrom: () => Stream.never, + }), + ), + ), + ), + ), + ); + const sent = yield* Ref.make(0); + yield* Effect.gen(function* () { + const mcp = yield* OrchestratorMcpService.OrchestratorMcpService; + const waiting = yield* mcp + .waitForThread( + { + ...client, + ...(scenario.linked + ? { + client: { + sessionId: "link-compat", + label: "A", + access: "full-access", + linked: true, + }, + } + : {}), + ...(scenario.agent + ? { + thread: { + threadId: launched.id, + providerSessionId: "session:compat-wait", + providerInstanceId: ProviderInstanceId.make("codex"), + }, + } + : {}), + }, + { + threadId: launched.id, + runId, + timeoutMs: 60_000, + }, + ) + .pipe(Effect.forkChild); + yield* TestClock.adjust("1 minute"); + const result = yield* Fiber.join(waiting); + assert.equal(result.timedOut, !scenario.pending); + assert.equal(result.status, "running"); + assert.equal( + result.pendingApproval?.requestId ?? null, + scenario.pending ? "request:compat-wait" : null, + ); + }).pipe( + Effect.provide( + service([launched], sent, { + getThreadRecords: () => Effect.succeed(projection), + getProjectThreadRecords: () => Effect.succeed(projection), + waitForThread: management.waitForThread, + }), + ), + ); + }), + ); + it.effect("waits on the caller's own thread without a separate shell lookup", () => Effect.gen(function* () { const sent = yield* Ref.make(0); @@ -3919,6 +4224,8 @@ describe("OrchestratorMcpService thread lookup", () => { runId: null, status: "idle", timedOut: false, + hasPendingQuestion: false, + pendingApproval: null, }); }).pipe( Effect.provide( @@ -3926,7 +4233,13 @@ describe("OrchestratorMcpService thread lookup", () => { getThreadShell: () => Effect.die("Unexpected shell lookup for the caller"), getThreadRecords: () => Effect.succeed(idleThreadProjection(launched)), waitForThread: () => - Effect.succeed({ threadId: launched.id, run: null, timedOut: false }), + Effect.succeed({ + threadId: launched.id, + run: null, + timedOut: false, + hasPendingQuestion: false, + pendingApproval: null, + }), }), ), ); diff --git a/apps/server/src/mcp/OrchestratorMcpService.ts b/apps/server/src/mcp/OrchestratorMcpService.ts index fc741bcac6ab5..0679fb4d203ab 100644 --- a/apps/server/src/mcp/OrchestratorMcpService.ts +++ b/apps/server/src/mcp/OrchestratorMcpService.ts @@ -529,22 +529,38 @@ function endsTaskWait(result: OrchestratorMcpDelegateTaskResult): boolean { const isApprovalRequestKind = Schema.is(ProviderRequestKind); -function pendingTaskApproval( - child: Pick, +function activeTaskRunIds( + child: Pick, taskId: NodeId, originalRun: OrchestrationV2Run | undefined, ) { - const taskRunIds = new Set( + const queuedOrdinal = child.runs.reduce( + (last, run) => + run.status === "queued" && + (run.delegatedTaskId === taskId || + (run.delegatedTaskId === undefined && run.id === originalRun?.id)) + ? Math.max(last, run.ordinal) + : last, + -1, + ); + return new Set( child.runs .filter( (run) => !ThreadManagementService.isTerminalRunStatus(run.status) && (run.delegatedTaskId === taskId || // Legacy tasks own only their original run, not later turns in the child thread. - (run.delegatedTaskId === undefined && run.id === originalRun?.id)), + (run.delegatedTaskId === undefined && run.id === originalRun?.id) || + (run.status !== "queued" && run.ordinal < queuedOrdinal)), ) .map((run) => run.id), ); +} + +function pendingTaskApproval( + child: Pick, + taskRunIds: ReadonlySet, +) { let approval: OrchestratorMcpDelegateTaskResult["pendingApproval"] = null; let oldestCreatedAt = Infinity; for (const request of child.runtimeRequests) { @@ -1623,9 +1639,14 @@ const make = Effect.gen(function* () { "providerTurns", "runtimeRequests", "nodes", + "turnItems", ], - // Progress reads only active or undelivered tasks under the child. - { messageRoles: ["user"], omitSettledLeafNodes: true, omitSettledSubagents: true }, + { + messageRoles: ["user"], + omitSettledLeafNodes: true, + omitSettledSubagents: true, + turnItemTypes: ["secret_request"], + }, ) .pipe(Effect.mapError(threadManagementFailure)); const childRun = delegatedTaskRun(childControls, task); @@ -1727,6 +1748,7 @@ const make = Effect.gen(function* () { instanceId: task.providerInstanceId, model: task.model, }; + const taskRunIds = activeTaskRunIds(childControls, task.id, childRun); const result = { taskId: task.id, childThreadId: task.childThreadId, @@ -1755,16 +1777,30 @@ const make = Effect.gen(function* () { : latestTerminalResultContextTransferId, hasPendingQuestion: !isTerminalTaskStatus(status) && - childControls.runtimeRequests.some( - (request) => - request.kind === "user_input" && - request.status === "pending" && - request.responseCapability.type === "live", - ), + (childControls.turnItems.some( + (item) => + item.type === "secret_request" && + item.status === "waiting" && + item.nodeId !== null && + item.runId !== null && + taskRunIds.has(item.runId), + ) || + childControls.runtimeRequests.some((request) => { + const requestRunId = childControls.nodes.find( + (node) => node.id === request.nodeId, + )?.runId; + return ( + request.kind === "user_input" && + request.status === "pending" && + request.responseCapability.type === "live" && + requestRunId != null && + taskRunIds.has(requestRunId) + ); + })), pendingApproval: isTerminalTaskStatus(status) || task.completionDelivery?.state === "disposed" ? null - : pendingTaskApproval(childControls, task.id, childRun), + : pendingTaskApproval(childControls, taskRunIds), waitTimedOut, } satisfies OrchestratorMcpDelegateTaskResult; const response: OrchestratorMcpDelegateTaskResult = @@ -1972,28 +2008,41 @@ const make = Effect.gen(function* () { childThreadId === null ? TASK_WAKE_EVENTS.filter(({ thread }) => thread === "parent") : TASK_WAKE_EVENTS; - return yield* Stream.mergeAll( - wakeEvents.map(({ thread, eventType }) => + const streams = wakeEvents.map(({ thread, eventType }) => + threadManagement + .streamStoredEventsFrom({ + threadId: + thread === "parent" || childThreadId === null ? scope.thread.threadId : childThreadId, + afterSequence, + eventType, + }) + .pipe( + Stream.filter( + ({ event }) => + thread === "child" || + event.type !== "subagent.updated" || + event.payload.id === taskId, + ), + ), + ); + if (childThreadId !== null) { + streams.push( threadManagement .streamStoredEventsFrom({ - threadId: - thread === "parent" || childThreadId === null - ? scope.thread.threadId - : childThreadId, + threadId: childThreadId, afterSequence, - eventType, + eventType: "turn-item.updated", }) .pipe( Stream.filter( - ({ event }) => - thread === "child" || - event.type !== "subagent.updated" || - event.payload.id === taskId, + (stored) => + stored.event.type === "turn-item.updated" && + stored.event.payload.type === "secret_request", ), ), - ), - { concurrency: "unbounded" }, - ).pipe( + ); + } + return yield* Stream.mergeAll(streams, { concurrency: "unbounded" }).pipe( Stream.mapError(streamError), Stream.mapEffect(() => readTask(scope, taskId, false, true)), Stream.filter(endsTaskWait), @@ -3356,6 +3405,8 @@ const make = Effect.gen(function* () { projectId: target.thread.projectId, threadId: input.threadId, ...(input.runId === undefined ? {} : { runId: input.runId }), + // CLIs and linked peers' followers read every early return as a finished run. + returnOnInput: scope.thread !== undefined, timeoutMs: Math.min( MAX_WAIT_TIMEOUT_MS, Math.max(1, input.timeoutMs ?? DEFAULT_WAIT_TIMEOUT_MS), @@ -3391,6 +3442,8 @@ const make = Effect.gen(function* () { runId: result.run?.id ?? null, status, timedOut: result.timedOut, + hasPendingQuestion: result.hasPendingQuestion, + pendingApproval: result.pendingApproval, ...(wakesCaller ? { next: ORCHESTRATOR_MCP_RUNNING_TASK_NEXT } : {}), ...(result.resultRun === undefined ? {} diff --git a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts index 2ec6e939927e4..84085c2f48b88 100644 --- a/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts +++ b/apps/server/src/mcp/OrchestratorMcpToolkit.integration.test.ts @@ -674,6 +674,8 @@ describe("orchestrator MCP toolkit", () => { runId: queuedFollowup.runId, status: "completed", timedOut: false, + hasPendingQuestion: false, + pendingApproval: null, }; expect(wait.structuredContent).toEqual(expectedWait); expect(messageReads).toEqual([]); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 8b967f648165f..56b592c180e44 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -2045,6 +2045,8 @@ const commandReadsSkipSettledLeafNodes = Effect.gen(function* () { ["settled-run-live-tool", settledRun, "tool_call", "running"], ["settled-run-todo", settledRun, "todo_list", "running"], ["active-run-done-tool", activeRun, "tool_call", "completed"], + ["active-approval", activeRun, "approval_request", "waiting"], + ["active-question", activeRun, "user_input_request", "waiting"], ["active-run-todo", activeRun, "todo_list", "running"], ] as const; for (const [name, run, kind, status] of nodes) { @@ -2127,6 +2129,8 @@ const commandReadsSkipSettledLeafNodes = Effect.gen(function* () { assert.deepEqual( nodeNames(yield* store.getThreadRecords(threadId, ["nodes"], { omitSettledLeafNodes: true })), [ + "active-approval", + "active-question", "active-run-done-tool", "active-run-todo", "settled-root", @@ -2142,7 +2146,12 @@ const commandReadsSkipSettledLeafNodes = Effect.gen(function* () { // A todo list stays running after its run ends. Counted as live work, every old run of a // long thread would ride along with each window, with its nodes and subagents. const liveWindow = (yield* store.getThreadSnapshotWindow(threadId, { rowLimit: 1 })).projection; - assert.deepEqual(nodeNames(liveWindow), ["active-run-todo", "settled-run-live-tool"]); + assert.deepEqual(nodeNames(liveWindow), [ + "active-approval", + "active-question", + "active-run-todo", + "settled-run-live-tool", + ]); assert.deepEqual( liveWindow.plans.map((plan) => plan.id), [PlanId.make("plan:settled-leaf-nodes:active-run-todo")], diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts index c2502902ef8e3..eae86106781b7 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.test.ts @@ -204,6 +204,264 @@ it.effect( }, ); +it.effect.each([ + { + kind: "user_input", + capability: "live", + requestOrdinal: 1, + selectedStatus: "running", + blocks: true, + }, + { + kind: "command", + capability: "live", + requestOrdinal: 1, + selectedStatus: "running", + blocks: true, + }, + { + kind: "user_input", + capability: "live", + requestOrdinal: 0, + selectedStatus: "queued", + blocks: true, + }, + { + kind: "user_input", + capability: "live", + requestOrdinal: 2, + selectedStatus: "queued", + blocks: true, + }, + { + kind: "user_input", + capability: "message", + requestOrdinal: 1, + selectedStatus: "running", + blocks: false, + }, + { + kind: "user_input", + capability: "live", + requestOrdinal: 2, + selectedStatus: "running", + blocks: false, + }, + { + kind: "user_input", + capability: "live", + requestOrdinal: 1, + selectedStatus: "completed", + blocks: false, + }, +] as const)("waitForThread scopes pending input to unfinished work: %j", (scenario) => + Effect.gen(function* () { + const projectId = ProjectId.make("project:wait-input"); + const threadId = ThreadId.make("thread:wait-input"); + const runId = RunId.make("run:wait-input"); + const requestRunId = scenario.requestOrdinal === 1 ? runId : RunId.make("run:other"); + const projection = { + thread: { id: threadId, projectId, deletedAt: null }, + runs: [ + { id: runId, ordinal: 1, status: scenario.selectedStatus }, + ...(requestRunId === runId + ? [] + : [{ id: requestRunId, ordinal: scenario.requestOrdinal, status: "running" }]), + ], + nodes: [{ id: "node:request", runId: requestRunId }], + turnItems: [], + runtimeRequests: [ + { + id: "request:input", + nodeId: "node:request", + kind: scenario.kind, + status: "pending", + responseCapability: { type: scenario.capability }, + createdAt: yield* DateTime.now, + }, + ], + } as unknown as OrchestrationV2ThreadProjection; + const layerTest = ThreadManagementService.layer.pipe( + Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), + Layer.provide( + Layer.mock(Orchestrator.OrchestratorV2)({ + getThreadEventSequence: () => Effect.succeed(0), + getThreadRecords: () => Effect.succeed(projection), + restartContinuationPending: () => Effect.succeed(false), + streamStoredEventsFrom: () => Stream.never, + }), + ), + Layer.provide(ServerSettings.layerTest()), + Layer.provide(layerReceiptsTest), + ); + const service = yield* ThreadManagementService.ThreadManagementService.pipe( + Effect.provide(layerTest), + ); + const fiber = yield* service + .waitForThread({ projectId, threadId, runId, timeoutMs: 60_000, returnOnInput: true }) + .pipe(Effect.forkChild); + yield* TestClock.adjust("1 minute"); + const result = yield* Fiber.join(fiber); + expect(result).toMatchObject({ + run: { id: runId, status: scenario.selectedStatus }, + timedOut: !scenario.blocks && scenario.selectedStatus !== "completed", + hasPendingQuestion: scenario.blocks && scenario.kind === "user_input", + pendingApproval: + scenario.blocks && scenario.kind === "command" + ? { requestId: "request:input", kind: "command", runId: requestRunId } + : null, + }); + }), +); + +it.effect.each([ + { selected: "running", owner: "running", ordinal: 1, status: "waiting", blocks: true }, + { selected: "queued", owner: "running", ordinal: 0, status: "waiting", blocks: true }, + { selected: "queued", owner: "running", ordinal: 2, status: "waiting", blocks: true }, + { selected: "running", owner: "running", ordinal: 2, status: "waiting", blocks: false }, + { selected: "running", owner: "completed", ordinal: 0, status: "waiting", blocks: false }, + { selected: "running", owner: "running", ordinal: 1, status: "completed", blocks: false }, + { selected: "completed", owner: "completed", ordinal: 1, status: "waiting", blocks: false }, +] as const)("waitForThread scopes secret input to unfinished work: %j", (scenario) => + Effect.gen(function* () { + const projectId = ProjectId.make("project:wait-secret"); + const threadId = ThreadId.make("thread:wait-secret"); + const runId = RunId.make("run:wait-secret"); + const ownerId = scenario.ordinal === 1 ? runId : RunId.make("run:secret-owner"); + const projection = { + thread: { id: threadId, projectId, deletedAt: null }, + runs: [ + { id: runId, ordinal: 1, status: scenario.selected }, + ...(ownerId === runId + ? [] + : [{ id: ownerId, ordinal: scenario.ordinal, status: scenario.owner }]), + ], + nodes: [], + runtimeRequests: [], + turnItems: [ + { type: "secret_request", runId: ownerId, nodeId: "node:secret", status: scenario.status }, + ], + } as unknown as OrchestrationV2ThreadProjection; + const layerTest = ThreadManagementService.layer.pipe( + Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), + Layer.provide( + Layer.mock(Orchestrator.OrchestratorV2)({ + getThreadEventSequence: () => Effect.succeed(0), + getThreadRecords: () => Effect.succeed(projection), + restartContinuationPending: () => Effect.succeed(false), + streamStoredEventsFrom: () => Stream.never, + }), + ), + Layer.provide(ServerSettings.layerTest()), + Layer.provide(layerReceiptsTest), + ); + const service = yield* ThreadManagementService.ThreadManagementService.pipe( + Effect.provide(layerTest), + ); + const fiber = yield* service + .waitForThread({ projectId, threadId, runId, timeoutMs: 60_000, returnOnInput: true }) + .pipe(Effect.forkChild); + yield* TestClock.adjust("1 minute"); + expect(yield* Fiber.join(fiber)).toMatchObject({ + run: { id: runId, status: scenario.selected }, + timedOut: !scenario.blocks && scenario.selected !== "completed", + hasPendingQuestion: scenario.blocks, + pendingApproval: null, + }); + }), +); + +it.effect.each(["command", "user_input", "secret_request"] as const)( + "waitForThread wakes on a new %s while terminal-only callers keep waiting", + (kind) => + Effect.gen(function* () { + const projectId = ProjectId.make("project:wait-event"); + const threadId = ThreadId.make("thread:wait-event"); + const runId = RunId.make("run:wait-event"); + const subscribed = yield* Deferred.make(); + const events = yield* Queue.unbounded(); + const now = yield* DateTime.now; + let requests: OrchestrationV2ThreadProjection["runtimeRequests"] = []; + let turnItems: OrchestrationV2ThreadProjection["turnItems"] = []; + let reads = 0; + const eventType = kind === "secret_request" ? "turn-item.updated" : "runtime-request.updated"; + const layerTest = ThreadManagementService.layer.pipe( + Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), + Layer.provide( + Layer.mock(Orchestrator.OrchestratorV2)({ + getThreadEventSequence: () => Effect.succeed(7), + getThreadRecords: () => + Effect.sync(() => { + reads += 1; + return { + thread: { id: threadId, projectId, deletedAt: null }, + runs: [{ id: runId, ordinal: 1, status: "running" }], + nodes: [{ id: "node:request", runId }], + runtimeRequests: requests, + turnItems, + } as unknown as OrchestrationV2ThreadProjection; + }), + streamStoredEventsFrom: (input) => + input?.eventType === eventType + ? Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe( + Stream.drain, + Stream.concat(Stream.fromQueue(events)), + ) + : Stream.never, + }), + ), + Layer.provide(ServerSettings.layerTest()), + Layer.provide(layerReceiptsTest), + ); + const service = yield* ThreadManagementService.ThreadManagementService.pipe( + Effect.provide(layerTest), + ); + const fiber = yield* service + .waitForThread({ projectId, threadId, runId, timeoutMs: 600_000, returnOnInput: true }) + .pipe(Effect.forkChild); + yield* Deferred.await(subscribed); + if (kind === "secret_request") { + turnItems = [ + { type: kind, runId, nodeId: "node:secret", status: "waiting" }, + ] as unknown as OrchestrationV2ThreadProjection["turnItems"]; + yield* Queue.offer(events, { + sequence: 8, + event: { type: eventType, threadId, payload: { type: "assistant_message" } }, + } as unknown as OrchestrationV2StoredEvent); + } else { + requests = [ + { + id: "request:approval", + nodeId: "node:request", + kind, + status: "pending", + responseCapability: { type: "live" }, + createdAt: now, + }, + ] as unknown as OrchestrationV2ThreadProjection["runtimeRequests"]; + } + yield* Queue.offer(events, { + sequence: 9, + event: { type: eventType, threadId, payload: turnItems[0] ?? requests[0] }, + } as unknown as OrchestrationV2StoredEvent); + expect(yield* Fiber.join(fiber)).toMatchObject({ + timedOut: false, + hasPendingQuestion: kind !== "command", + pendingApproval: kind === "command" ? { requestId: "request:approval", kind, runId } : null, + }); + expect(reads).toBe(2); + const terminalWait = yield* service + .waitForThread({ projectId, threadId, runId, timeoutMs: 60_000 }) + .pipe(Effect.forkChild); + yield* TestClock.adjust("1 minute"); + expect(yield* Fiber.join(terminalWait)).toMatchObject({ + timedOut: true, + hasPendingQuestion: false, + pendingApproval: null, + }); + }), +); + it.effect.each([ { label: "fails a send that starts a turn", running: false }, { label: "still steers a running turn", running: true }, @@ -335,6 +593,9 @@ it.effect("waitForThread follows a restart continuation instead of the cut run", ({ thread: { id: threadId, projectId, deletedAt: null }, runs, + nodes: [], + runtimeRequests: [], + turnItems: [], }) as unknown as OrchestrationV2ThreadProjection, ), }), @@ -370,92 +631,114 @@ it.effect("waitForThread follows a restart continuation instead of the cut run", }), ); -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"); - const threadId = ThreadId.make("thread:thread-management:wait-delegated"); - const runId = RunId.make("run:thread-management:wait-delegated"); - const wakeRunId = RunId.make("run:thread-management:wait-delegated:wake"); - const subscribed = yield* Deferred.make(); - const events = yield* Queue.unbounded(); - const at = (seconds: number) => DateTime.makeUnsafe(seconds * 1_000); - const run = (id: RunId, ordinal: number, status: OrchestrationV2Run["status"], end: number) => - ({ - id, - ordinal, - status, - startedAt: at(end - 1), - completedAt: status === "running" ? null : at(end), - }) as unknown as OrchestrationV2Run; - // The run ended, but a task it delegated still runs and will wake the thread. - let runs = [run(runId, 1, "completed", 1)]; - let subagents = [{ status: "running" }]; - const stored = (sequence: number, event: object) => - ({ sequence, event: { threadId, ...event } }) as unknown as OrchestrationV2StoredEvent; - const layerTest = ThreadManagementService.layer.pipe( - Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), - Layer.provide( - Layer.mock(Orchestrator.OrchestratorV2)({ - getThreadEventSequence: () => Effect.succeed(0), - restartContinuationPending: () => Effect.succeed(false), - delegatedTaskResultPending: () => Effect.succeed(false), - streamStoredEventsFrom: (input) => - input?.eventType === "subagent.updated" - ? Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe( - Stream.drain, - Stream.concat(Stream.fromQueue(events)), - ) - : Stream.never, - getThreadRecords: () => - Effect.sync( - () => - ({ - thread: { id: threadId, projectId, deletedAt: null }, - runs, - messages: [], - subagents, - providerThreads: [], - }) as unknown as OrchestrationV2ThreadProjection, - ), - }), - ), - Layer.provide(ServerSettings.layerTest()), - Layer.provide(layerReceiptsTest), - ); - const service = yield* ThreadManagementService.ThreadManagementService.pipe( - Effect.provide(layerTest), - ); - // Without the option, the ended run is the answer, as before. - expect( - yield* service.waitForThread({ projectId, threadId, runId, timeoutMs: 60 * 60 * 1_000 }), - ).toMatchObject({ timedOut: false, run: { id: runId } }); - expect( - yield* service.waitForThread({ projectId, threadId, runId, timeoutMs: 60 * 60 * 1_000 }), - ).not.toHaveProperty("resultRun"); +it.effect.each(["completion", "secret_request"] as const)( + "waitForThread follows delegated work until %s", + (ending) => + Effect.gen(function* () { + const projectId = ProjectId.make("project:thread-management:wait-delegated"); + const threadId = ThreadId.make("thread:thread-management:wait-delegated"); + const runId = RunId.make("run:thread-management:wait-delegated"); + const wakeRunId = RunId.make("run:thread-management:wait-delegated:wake"); + const subscribed = yield* Deferred.make(); + const events = yield* Queue.unbounded(); + const at = (seconds: number) => DateTime.makeUnsafe(seconds * 1_000); + const run = (id: RunId, ordinal: number, status: OrchestrationV2Run["status"], end: number) => + ({ + id, + ordinal, + status, + startedAt: at(end - 1), + completedAt: status === "running" ? null : at(end), + }) as unknown as OrchestrationV2Run; + // The run ended, but a task it delegated still runs and will wake the thread. + let runs = [run(runId, 1, "completed", 1)]; + let subagents = [{ status: "running" }]; + let turnItems: OrchestrationV2ThreadProjection["turnItems"] = []; + const eventType = ending === "completion" ? "subagent.updated" : "turn-item.updated"; + const stored = (sequence: number, event: object) => + ({ sequence, event: { threadId, ...event } }) as unknown as OrchestrationV2StoredEvent; + const layerTest = ThreadManagementService.layer.pipe( + Layer.provide(Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({})), + Layer.provide( + Layer.mock(Orchestrator.OrchestratorV2)({ + getThreadEventSequence: () => Effect.succeed(0), + restartContinuationPending: () => Effect.succeed(false), + delegatedTaskResultPending: () => Effect.succeed(false), + streamStoredEventsFrom: (input) => + input?.eventType === eventType + ? Stream.fromEffect(Deferred.succeed(subscribed, undefined)).pipe( + Stream.drain, + Stream.concat(Stream.fromQueue(events)), + ) + : Stream.never, + getThreadRecords: () => + Effect.sync( + () => + ({ + thread: { id: threadId, projectId, deletedAt: null }, + runs, + messages: [], + subagents, + providerThreads: [], + nodes: [], + runtimeRequests: [], + turnItems, + }) as unknown as OrchestrationV2ThreadProjection, + ), + }), + ), + Layer.provide(ServerSettings.layerTest()), + Layer.provide(layerReceiptsTest), + ); + const service = yield* ThreadManagementService.ThreadManagementService.pipe( + Effect.provide(layerTest), + ); + // Without the option, the ended run is the answer, as before. + expect( + yield* service.waitForThread({ projectId, threadId, runId, timeoutMs: 60 * 60 * 1_000 }), + ).toMatchObject({ timedOut: false, run: { id: runId } }); + expect( + yield* service.waitForThread({ projectId, threadId, runId, timeoutMs: 60 * 60 * 1_000 }), + ).not.toHaveProperty("resultRun"); - const fiber = yield* service - .waitForThread({ - projectId, - threadId, - runId, - timeoutMs: 60 * 60 * 1_000, - untilDelegatedWorkSettled: true, - }) - .pipe(Effect.forkChild); - // A wait that returns early fails below instead of hanging here. - yield* Effect.raceFirst(Deferred.await(subscribed), Fiber.await(fiber)); - // The task finishes; its delivery wakes the thread, whose next run answers. - subagents = [{ status: "completed" }]; - runs = [...runs, run(wakeRunId, 2, "completed", 3)]; - yield* Queue.offer(events, stored(1, { type: "subagent.updated", payload: {} })); - const result = yield* Fiber.join(fiber); + const fiber = yield* service + .waitForThread({ + projectId, + threadId, + runId, + timeoutMs: 60 * 60 * 1_000, + untilDelegatedWorkSettled: true, + returnOnInput: true, + }) + .pipe(Effect.forkChild); + // A wait that returns early fails below instead of hanging here. + yield* Effect.raceFirst(Deferred.await(subscribed), Fiber.await(fiber)); + // The task finishes; its delivery wakes the thread, whose next run answers. + subagents = [{ status: "completed" }]; + runs = [...runs, run(wakeRunId, 2, ending === "completion" ? "completed" : "running", 3)]; + turnItems = + ending === "completion" + ? [] + : ([ + { + type: "secret_request", + runId: wakeRunId, + nodeId: "node:wake-secret", + status: "waiting", + }, + ] as unknown as OrchestrationV2ThreadProjection["turnItems"]); + yield* Queue.offer(events, stored(1, { type: eventType, payload: turnItems[0] ?? {} })); + const result = yield* Fiber.join(fiber); - expect(result).toMatchObject({ - timedOut: false, - run: { id: runId, status: "completed" }, - resultRun: { id: wakeRunId, status: "completed" }, - }); - }), + expect(result).toMatchObject({ + timedOut: false, + run: { id: runId, status: "completed" }, + hasPendingQuestion: ending === "secret_request", + }); + if (ending === "completion") + expect(result.resultRun).toMatchObject({ id: wakeRunId, status: "completed" }); + else expect(result).not.toHaveProperty("resultRun"); + }), ); it.effect("waitForThread settles delegated work on a thread rolled back past its only run", () => @@ -482,6 +765,9 @@ it.effect("waitForThread settles delegated work on a thread rolled back past its Effect.succeed({ thread: { id: threadId, projectId, deletedAt: null }, runs: [rolledBack], + nodes: [], + runtimeRequests: [], + turnItems: [], messages: [], subagents: [], providerThreads: [], diff --git a/apps/server/src/orchestration-v2/ThreadManagementService.ts b/apps/server/src/orchestration-v2/ThreadManagementService.ts index 482674b772d43..9458bbb55dff1 100644 --- a/apps/server/src/orchestration-v2/ThreadManagementService.ts +++ b/apps/server/src/orchestration-v2/ThreadManagementService.ts @@ -35,6 +35,8 @@ import { ProjectId, PROVIDER_SEND_TURN_MAX_FILE_BYTES, ProviderInstanceId, + ProviderRequestKind, + type OrchestratorMcpThreadWaitResult, type ProviderSessionId, resolveProjectProviderInstanceEnabled, RunId, @@ -72,6 +74,7 @@ import * as ServerSettings from "../serverSettings.ts"; import { ThreadWorktreeRestore } from "../workspace/restoreMissingWorktree.ts"; export type ThreadManagementSendMode = "auto" | "queue" | "steer" | "restart"; +const isApprovalRequestKind = Schema.is(ProviderRequestKind); export interface ThreadManagementProvenance { readonly createdBy: OrchestrationV2Actor; @@ -174,6 +177,8 @@ export interface ThreadManagementWaitInput { readonly threadId: ThreadId; readonly runId?: RunId; readonly timeoutMs: number; + /** Also return when live input blocks the run, leaving its status as is. */ + readonly returnOnInput?: boolean; /** * 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 @@ -186,6 +191,8 @@ export interface ThreadManagementWaitResult { readonly threadId: ThreadId; readonly run: OrchestrationV2Run | null; readonly timedOut: boolean; + readonly hasPendingQuestion: boolean; + readonly pendingApproval: OrchestratorMcpThreadWaitResult["pendingApproval"]; /** With `untilDelegatedWorkSettled`, once settled: the run with the thread's final answer. */ readonly resultRun?: OrchestrationV2Run; } @@ -995,7 +1002,15 @@ const make = Effect.gen(function* () { const afterSequence = yield* orchestrator .getThreadEventSequence(input.threadId) .pipe(Effect.mapError(loadError)); - const target = yield* getProjectThreadRecords(input, ["runs"]); + const fields = input.returnOnInput + ? (["runs", "nodes", "runtimeRequests", "turnItems"] as const) + : (["runs"] as const); + const filter = { + turnItemTypes: ["secret_request"], + omitSettledLeafNodes: true, + runtimeRequestStatuses: ["pending"], + } as const; + const target = yield* getProjectThreadRecords(input, fields, filter); const selectedRun = input.runId === undefined ? latestRun(target) @@ -1007,12 +1022,25 @@ const make = Effect.gen(function* () { }); } if (selectedRun === undefined) { - return { threadId: input.threadId, run: null, timedOut: false }; + return { + threadId: input.threadId, + run: null, + timedOut: false, + hasPendingQuestion: false, + pendingApproval: null, + }; } + const untilDelegatedWorkSettled = input.untilDelegatedWorkSettled === true; // A restart cuts a run and continues it as a new run. Follow that chain, // so a cut run is not the answer while its continuation is pending. // Runs per matching event and is already covered by the waitForThread span. - const outcome = Effect.fnUntraced(function* (runs: ReadonlyArray) { + const outcome = Effect.fnUntraced(function* ( + projection: Pick< + OrchestrationV2ThreadProjection, + "runs" | "nodes" | "runtimeRequests" | "turnItems" + >, + ) { + const { runs } = projection; const current = runs.find((candidate) => candidate.id === selectedRun.id); if (current === undefined) { return yield* new ThreadManagementRunNotFoundError({ @@ -1023,9 +1051,55 @@ const make = Effect.gen(function* () { const run = latestRestartContinuation(runs, current); const pending = isTerminalRunStatus(run.status) && (yield* orchestrator.restartContinuationPending(run)); - return { run, pending, settled: isTerminalRunStatus(run.status) && !pending }; + let hasPendingQuestion = false; + let pendingApproval: ThreadManagementWaitResult["pendingApproval"] = null; + let oldestApproval = Infinity; + if (input.returnOnInput && (!isTerminalRunStatus(run.status) || untilDelegatedWorkSettled)) { + const blockingRuns = new Set( + runs + .filter( + (candidate) => + !isTerminalRunStatus(candidate.status) && + (untilDelegatedWorkSettled || + candidate.ordinal <= run.ordinal || + (run.status === "queued" && candidate.status !== "queued")), + ) + .map((candidate) => candidate.id), + ); + const nodeRuns = new Map(projection.nodes.map((node) => [node.id, node.runId])); + hasPendingQuestion = projection.turnItems.some( + (item) => + item.type === "secret_request" && + item.status === "waiting" && + item.nodeId !== null && + item.runId !== null && + blockingRuns.has(item.runId), + ); + for (const request of projection.runtimeRequests) { + if (request.status !== "pending" || request.responseCapability.type !== "live") continue; + const requestRunId = nodeRuns.get(request.nodeId); + if (requestRunId == null || !blockingRuns.has(requestRunId)) continue; + if (request.kind === "user_input") hasPendingQuestion = true; + else if (isApprovalRequestKind(request.kind)) { + const createdAt = DateTime.toEpochMillis(request.createdAt); + if (createdAt < oldestApproval) { + oldestApproval = createdAt; + pendingApproval = { requestId: request.id, kind: request.kind, runId: requestRunId }; + } + } + } + } + return { + run, + pending, + hasPendingQuestion, + pendingApproval, + settled: + (isTerminalRunStatus(run.status) && !pending) || + hasPendingQuestion || + pendingApproval !== null, + }; }); - const untilDelegatedWorkSettled = input.untilDelegatedWorkSettled === true; const delegatedWorkResult = Effect.gen(function* () { const records = yield* getProjectThreadRecords( input, @@ -1047,10 +1121,15 @@ const make = Effect.gen(function* () { return pending ? undefined : progress.resultRun; }); const evaluate = Effect.fn("orchestrationV2.threadManagement.evaluate")(function* ( - runs: ReadonlyArray, + projection: Parameters[0], ) { - const current = yield* outcome(runs); - if (!current.settled || !untilDelegatedWorkSettled) { + const current = yield* outcome(projection); + if ( + !current.settled || + !untilDelegatedWorkSettled || + current.hasPendingQuestion || + current.pendingApproval !== null + ) { return { ...current, resultRun: undefined }; } const resultRun = yield* delegatedWorkResult; @@ -1060,32 +1139,60 @@ const make = Effect.gen(function* () { threadId: input.threadId, run: current.run, timedOut: !current.settled, + hasPendingQuestion: current.hasPendingQuestion, + pendingApproval: current.pendingApproval, ...(current.resultRun === undefined ? {} : { resultRun: current.resultRun }), }); - const initial = yield* evaluate(target.runs); + const initial = yield* evaluate(target); if (initial.settled) return settledResult(initial); // A declined continuation records no event, so its effect settling wakes // the wait too. const wait = Stream.merge( Stream.mergeAll( - ( - [ - "run.updated", - "thread.deleted", - // What delegated work reads besides runs: tasks and their - // deliveries, and a provider thread's background tasks. - ...(untilDelegatedWorkSettled - ? (["subagent.updated", "provider-thread.updated"] as const) - : []), - ] as const - ).map((eventType) => + [ orchestrator.streamStoredEventsFrom({ threadId: input.threadId, afterSequence, - eventType, + eventType: "run.updated", }), - ), + orchestrator.streamStoredEventsFrom({ + threadId: input.threadId, + afterSequence, + eventType: "thread.deleted", + }), + ...(untilDelegatedWorkSettled + ? (["subagent.updated", "provider-thread.updated"] as const).map((eventType) => + orchestrator.streamStoredEventsFrom({ + threadId: input.threadId, + afterSequence, + eventType, + }), + ) + : []), + input.returnOnInput + ? orchestrator.streamStoredEventsFrom({ + threadId: input.threadId, + afterSequence, + eventType: "runtime-request.updated", + }) + : Stream.empty, + input.returnOnInput + ? orchestrator + .streamStoredEventsFrom({ + threadId: input.threadId, + afterSequence, + eventType: "turn-item.updated", + }) + .pipe( + Stream.filter( + (stored) => + stored.event.type === "turn-item.updated" && + stored.event.payload.type === "secret_request", + ), + ) + : Stream.empty, + ], { concurrency: "unbounded" }, ).pipe( Stream.mapError(loadError), @@ -1102,9 +1209,7 @@ const make = Effect.gen(function* () { : Stream.empty, ).pipe( Stream.mapEffect(() => - getProjectThreadRecords(input, ["runs"]).pipe( - Effect.flatMap((current) => evaluate(current.runs)), - ), + getProjectThreadRecords(input, fields, filter).pipe(Effect.flatMap(evaluate)), ), Stream.filter((current) => current.settled), Stream.runHead, @@ -1115,7 +1220,7 @@ const make = Effect.gen(function* () { // The run may have reached a terminal status while the timeout was // winning the race; the final projection read decides what actually // happened, so only a still-active run counts as timed out. - return settledResult(yield* evaluate((yield* getProjectThreadRecords(input, ["runs"])).runs)); + return settledResult(yield* evaluate(yield* getProjectThreadRecords(input, fields, filter))); }); const settleAfterRun: ThreadManagementServiceShape["settleAfterRun"] = Effect.fn( diff --git a/apps/server/src/peer/PeerForwarding.test.ts b/apps/server/src/peer/PeerForwarding.test.ts index 84fe35c6f21f8..0f2cb0ce91509 100644 --- a/apps/server/src/peer/PeerForwarding.test.ts +++ b/apps/server/src/peer/PeerForwarding.test.ts @@ -138,6 +138,8 @@ const boxThreads = (seen: Ref.Ref, linkSession: Ref.Ref) => { threadId: input.threadId, run: run(waits.length < 3 ? "running" : "completed"), timedOut: waits.length < 3, + hasPendingQuestion: false, + pendingApproval: null, })), ), }); @@ -475,6 +477,8 @@ it.effect("keeps a forwarded wait short and its retries scoped to the caller", ( runId: boxRun, status: "completed", timedOut: false, + hasPendingQuestion: false, + pendingApproval: null, }); // Three pieces of at most 50 s each, the later ones pinned to the run // the first one found. diff --git a/apps/server/src/peer/RemoteDelegation.test.ts b/apps/server/src/peer/RemoteDelegation.test.ts index af569e902c027..7ba4f72bfc22d 100644 --- a/apps/server/src/peer/RemoteDelegation.test.ts +++ b/apps/server/src/peer/RemoteDelegation.test.ts @@ -16,6 +16,7 @@ import { type Project, type RepositoryIdentity, RunId, + RuntimeRequestId, type ServerProvider, ThreadId, TurnItemId, @@ -114,6 +115,7 @@ interface BoxThread { readonly waitedFor: Ref.Ref>; /** The thread's run ends with a task of its own still running, whose result wakes it again. */ readonly nested: Ref.Ref; + readonly nestedInput: Ref.Ref<"question" | "approval" | null>; /** Completed once a wait there blocks on that task. */ readonly nestedWaiting: Deferred.Deferred; /** That task settled and the run it woke answered, with `laterReply`. */ @@ -162,6 +164,7 @@ const makeBoxThread = Effect.gen(function* () { executedModel: yield* Ref.make(null), waitedFor: yield* Ref.make>([]), nested: yield* Ref.make(false), + nestedInput: yield* Ref.make<"question" | "approval" | null>(null), nestedWaiting: yield* Deferred.make(), nestedSettled: yield* Deferred.make(), gone: yield* Ref.make(false), @@ -285,7 +288,13 @@ const boxToolkitLayer = (thread: BoxThread) => { yield* Ref.update(thread.waitedFor, (seen) => [...seen, input.runId]); // A thread with no run has nothing to wait for. if (yield* Ref.get(thread.runless)) { - return { threadId: input.threadId, run: null, timedOut: false }; + return { + threadId: input.threadId, + run: null, + timedOut: false, + hasPendingQuestion: false, + pendingApproval: null, + }; } if (yield* Ref.get(thread.gone)) { return yield* new ThreadManagement.ThreadManagementThreadNotFoundError({ @@ -314,6 +323,8 @@ const boxToolkitLayer = (thread: BoxThread) => { status: Option.isSome(laterDone) ? "completed" : "running", } as OrchestrationV2Run, timedOut: Option.isNone(laterDone), + hasPendingQuestion: false, + pendingApproval: null, ...(queued ? { resultRun: { @@ -333,6 +344,21 @@ const boxToolkitLayer = (thread: BoxThread) => { input.untilDelegatedWorkSettled === true && (yield* Ref.get(thread.nested)) ) { + const pendingInput = input.returnOnInput + ? yield* Ref.getAndSet(thread.nestedInput, null) + : null; + if (pendingInput !== null) { + return { + threadId: input.threadId, + run: { id: runId, status: "completed" } as OrchestrationV2Run, + timedOut: false, + hasPendingQuestion: pendingInput === "question", + pendingApproval: + pendingInput === "approval" + ? { requestId: RuntimeRequestId.make("request:box"), kind: "file-change", runId } + : null, + }; + } yield* Deferred.succeed(thread.nestedWaiting, undefined); const settled = yield* Deferred.await(thread.nestedSettled).pipe( Effect.timeoutOption(input.timeoutMs), @@ -341,6 +367,8 @@ const boxToolkitLayer = (thread: BoxThread) => { threadId: input.threadId, run: { id: runId, status: yield* Ref.get(thread.status) } as OrchestrationV2Run, timedOut: Option.isNone(settled), + hasPendingQuestion: false, + pendingApproval: null, ...(Option.isSome(settled) ? { resultRun: { id: boxLaterRunId, status: "completed" } as OrchestrationV2Run } : {}), @@ -355,6 +383,8 @@ const boxToolkitLayer = (thread: BoxThread) => { status: later ? "completed" : yield* Ref.get(thread.status), } as OrchestrationV2Run, timedOut: Option.isNone(done), + hasPendingQuestion: false, + pendingApproval: null, }; }), getTimelinePage: (threadId, options) => @@ -1440,40 +1470,45 @@ it.effect("a task's result is its own run's whole last reply, not a later run's" ).pipe(Effect.provide(NodeHttpServer.layerTest)), ); -it.effect("a task whose thread there waits on its own task ends with the answer after it", () => - Effect.scoped( - Effect.gen(function* () { - const thread = yield* makeBoxThread; - const b = yield* serveBox(thread); - const offers = yield* SubscriptionRef.make< - ReadonlyArray - >([]); - const database = yield* SqlitePersistence.layerMemory.pipe(Layer.build); - const a = yield* makeLaptop(database, offers); - yield* linkTo(a.links, b, "full-access"); - yield* seedParent(a); - const delegated = yield* delegateToBox(a); +it.effect.each([null, "question", "approval"] as const)( + "a task whose thread there waits on its own task ends with the answer after it (%s)", + (pendingInput) => + Effect.scoped( + Effect.gen(function* () { + const thread = yield* makeBoxThread; + const b = yield* serveBox(thread); + const offers = yield* SubscriptionRef.make< + ReadonlyArray + >([]); + const database = yield* SqlitePersistence.layerMemory.pipe(Layer.build); + const a = yield* makeLaptop(database, offers); + yield* linkTo(a.links, b, "full-access"); + yield* seedParent(a); + const delegated = yield* delegateToBox(a); - // The task's run there ends after delegating a review of its own. - yield* Ref.set(thread.nested, true); - yield* Ref.set(thread.reply, "Started a review; I will answer when it reports."); - yield* Ref.set(thread.status, "completed"); - yield* Deferred.succeed(thread.finished, undefined); - const first = yield* Effect.raceFirst( - Deferred.await(thread.nestedWaiting).pipe(Effect.as("waiting there" as const)), - taskResult(a, delegated.taskId).pipe(Effect.as("completed here" as const)), - ); - expect(first).toBe("waiting there"); + // The task's run there ends after delegating a review of its own. + yield* Ref.set(thread.nested, true); + yield* Ref.set(thread.nestedInput, pendingInput); + yield* Ref.set(thread.reply, "Started a review; I will answer when it reports."); + yield* Ref.set(thread.status, "completed"); + yield* Deferred.succeed(thread.finished, undefined); + const first = yield* Effect.raceFirst( + Deferred.await(thread.nestedWaiting).pipe(Effect.as("waiting there" as const)), + taskResult(a, delegated.taskId).pipe(Effect.as("completed here" as const)), + ); + expect(first).toBe("waiting there"); + // A linked follower is a client caller there, so pending input never ends its wait early. + expect(yield* Ref.get(thread.nestedInput)).toBe(pendingInput); - // The review reports and wakes the thread there, whose next run answers. - yield* Ref.set(thread.laterReply, "The review found two issues; both are fixed."); - yield* Deferred.succeed(thread.nestedSettled, undefined); - expect(yield* taskResult(a, delegated.taskId)).toMatchObject({ - status: "completed", - result: "The review found two issues; both are fixed.", - }); - }), - ).pipe(Effect.provide(NodeHttpServer.layerTest)), + // The review reports and wakes the thread there, whose next run answers. + yield* Ref.set(thread.laterReply, "The review found two issues; both are fixed."); + yield* Deferred.succeed(thread.nestedSettled, undefined); + expect(yield* taskResult(a, delegated.taskId)).toMatchObject({ + status: "completed", + result: "The review found two issues; both are fixed.", + }); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), ); it.effect("the same key from another provider session is another task", () => diff --git a/docs/orchestration-v2/orchestrator-mcp-server.md b/docs/orchestration-v2/orchestrator-mcp-server.md index 74700c2d006a2..36bf84f36bdb3 100644 --- a/docs/orchestration-v2/orchestrator-mcp-server.md +++ b/docs/orchestration-v2/orchestrator-mcp-server.md @@ -481,8 +481,22 @@ idempotent retries. Waits for a selected run to become `completed`, `failed`, `cancelled`, `interrupted`, or `rolled_back`. Without `runId`, it pins the latest run at call -time; an idle thread returns immediately. A timeout reports the latest durable -status and does not cancel work. +time. An idle thread returns immediately. + +For agents in a T3 thread, the wait also returns on blocking input. Waits from +other MCP clients, the `t3 thread` CLI and linked environments stay +terminal-only, because those callers read any wait that did not time out as a +finished run; the CLI checks for pending requests between short waits instead. + +An input-aware wait returns `timedOut: false` and `hasPendingQuestion` or +`pendingApproval` for live input on the selected run or an earlier unfinished run +blocking it. The run keeps its current status. Asynchronous message-based +questions leave the wait active. Read a fresh question with `t3_pending_request`. +Secret cards appear in thread activity. Ask the user to answer secret cards and +approvals in the thread. The result grants no approval authority. + +A timeout reports the latest durable status and leaves the work running. A `next` +hint on a timed-out wait tells the caller to end its turn. ### `t3_thread_interrupt` diff --git a/packages/contracts/src/orchestratorMcp.test.ts b/packages/contracts/src/orchestratorMcp.test.ts index d2908fa42db8c..cc42cb8d839a9 100644 --- a/packages/contracts/src/orchestratorMcp.test.ts +++ b/packages/contracts/src/orchestratorMcp.test.ts @@ -12,6 +12,23 @@ const decodeDelegateTaskResult = Schema.decodeUnknownSync(OrchestratorMcpDelegat const decodeThreadWaitResult = Schema.decodeUnknownSync(OrchestratorMcpThreadWaitResult); describe("orchestrator MCP contracts", () => { + it("decodes older peer waits while preserving explicit pending input", () => { + const decode = Schema.decodeUnknownSync(OrchestratorMcpThreadWaitResult); + const legacy = { + threadId: "thread-peer", + runId: "run-peer", + status: "running", + timedOut: true, + }; + expect(decode(legacy)).toEqual({ ...legacy, hasPendingQuestion: false, pendingApproval: null }); + const pending = { + ...legacy, + timedOut: false, + hasPendingQuestion: true, + pendingApproval: { requestId: "request-peer", kind: "command", runId: "run-peer" }, + }; + expect(decode(pending)).toEqual(pending); + }); it("decodes cross-provider delegated task requests and durable results", () => { const request = decodeDelegateTaskInput({ task: "Inspect the workspace and report the result.", diff --git a/packages/contracts/src/orchestratorMcp.ts b/packages/contracts/src/orchestratorMcp.ts index 7148eed570ef7..87d2ce19c2c7a 100644 --- a/packages/contracts/src/orchestratorMcp.ts +++ b/packages/contracts/src/orchestratorMcp.ts @@ -580,6 +580,10 @@ export const OrchestratorMcpThreadWaitResult = Schema.Struct({ runId: Schema.NullOr(RunId), status: OrchestratorMcpThreadStatus, timedOut: Schema.Boolean, + hasPendingQuestion: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(false))), + pendingApproval: Schema.NullOr( + Schema.Struct({ requestId: RuntimeRequestId, kind: ProviderRequestKind, runId: RunId }), + ).pipe(Schema.withDecodingDefault(Effect.succeed(null))), next: Schema.optional(OrchestratorMcpRunningTaskNext), /** With untilDelegatedWorkSettled, once settled: the run whose reply is the thread's result. */ resultRun: Schema.optional(Schema.Struct({ runId: RunId, status: OrchestrationV2RunStatus })),