diff --git a/apps/server/src/mcp/McpHttpServer.test.ts b/apps/server/src/mcp/McpHttpServer.test.ts index 13dba3648d0c5..0ef7335a9f11c 100644 --- a/apps/server/src/mcp/McpHttpServer.test.ts +++ b/apps/server/src/mcp/McpHttpServer.test.ts @@ -970,3 +970,119 @@ it.effect("admits provider and OAuth client credentials and points only clients }), ).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + +it.effect("keeps PR watch tools available to parents and rejects delegated child calls", () => + Effect.scoped( + Effect.gen(function* () { + const scopeFor = ( + token: string, + subagent: boolean, + ): McpInvocationContext.McpThreadInvocationScope => ({ + environmentId, + requestNamespace: token, + thread: { + threadId: ThreadId.make(`thread-${token}`), + providerSessionId: token, + providerInstanceId: ProviderInstanceId.make("codex"), + subagent, + }, + client: undefined, + capabilities: new Set(["pull-requests"]), + issuedAt: 1, + }); + const scopes = new Map([ + ["parentToken", scopeFor("parentToken", false)], + ["childToken", scopeFor("childToken", true)], + ]); + const serverLayer = McpHttpServer.layerPullRequestsToolkit.pipe( + Layer.provideMerge(McpHttpServer.layerMcpTransport), + Layer.provide( + Layer.mock(McpSessionRegistry.McpSessionRegistry)({ + resolve: (token) => Effect.succeed(scopes.get(token)), + }), + ), + Layer.provide( + Layer.mergeAll( + McpToolAccessTestkit.liveThreadsLayer, + Layer.mock(ProjectService.ProjectService)({}), + Layer.mock(PullRequestReviewGuideService.PullRequestReviewGuideService)({}), + Layer.mock(Orchestrator.OrchestratorV2)({}), + Layer.mock(ProjectionStore.ProjectionStoreV2)({}), + NodeServices.layer, + ), + ), + ); + yield* HttpRouter.serve(serverLayer, { disableListenLog: true, disableLogger: true }).pipe( + Layer.build, + ); + const httpClient = yield* HttpClient.HttpClient; + const listedTools = (token: string) => + Effect.gen(function* () { + const headers = { + accept: "application/json, text/event-stream", + authorization: `Bearer ${token}`, + }; + const initialize = yield* httpClient.post("/mcp", { + headers, + body: HttpBody.text( + `{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"probe","version":"1.0.0"}}}`, + "application/json", + ), + }); + const listed = yield* httpClient.post("/mcp", { + headers: { + ...headers, + "mcp-session-id": initialize.headers["mcp-session-id"]!, + "mcp-protocol-version": "2025-06-18", + }, + body: HttpBody.text( + `{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}`, + "application/json", + ), + }); + const body = (yield* listed.json) as { + readonly result: { readonly tools: ReadonlyArray<{ readonly name: string }> }; + }; + if (token === "childToken") { + for (const name of ["watch_pull_request", "unwatch_pull_request"]) { + const denied = yield* httpClient.post("/mcp", { + headers: { + ...headers, + "mcp-session-id": initialize.headers["mcp-session-id"]!, + "mcp-protocol-version": "2025-06-18", + }, + body: HttpBody.text( + JSON.stringify({ + jsonrpc: "2.0", + id: 3, + method: "tools/call", + params: { + name, + arguments: { + url: "https://github.com/unlockdigital/synthetic/pull/1", + threadId: "thread-parentToken", + }, + }, + }), + "application/json", + ), + }); + expect(yield* denied.json).toMatchObject({ + error: { message: `Tool '${name}' not found` }, + }); + } + } + return body.result.tools.map((tool) => tool.name); + }); + + const parentTools = yield* listedTools("parentToken"); + const childTools = yield* listedTools("childToken"); + expect(parentTools).toEqual( + expect.arrayContaining(["link_pull_request", "watch_pull_request", "unwatch_pull_request"]), + ); + expect(childTools).toContain("link_pull_request"); + expect(childTools).not.toContain("watch_pull_request"); + expect(childTools).not.toContain("unwatch_pull_request"); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); diff --git a/apps/server/src/mcp/McpInvocationContext.ts b/apps/server/src/mcp/McpInvocationContext.ts index f830da3d4a03c..57faf91819945 100644 --- a/apps/server/src/mcp/McpInvocationContext.ts +++ b/apps/server/src/mcp/McpInvocationContext.ts @@ -10,6 +10,7 @@ import { } from "@t3tools/contracts"; import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; const ALL_MCP_CAPABILITIES = [ "preview", @@ -24,6 +25,8 @@ export interface McpThreadCaller { readonly threadId: ThreadId; readonly providerSessionId: string; readonly providerInstanceId: ProviderInstanceId; + /** A delegated child task; its parent owns the work it hands back. */ + readonly subagent?: boolean; } export interface McpClientCaller { @@ -58,6 +61,16 @@ export class McpInvocationContext extends Context.Service< McpInvocationScope >()("t3/mcp/McpInvocationContext") {} +/** + * `McpSchema.EnabledWhen` for tools a delegated child must not see or call. + * EnabledWhen only receives the MCP client profile, so read the T3 caller from the request fiber. + */ +export const hiddenFromSubagents = (): boolean => { + const fiber = Fiber.getCurrent(); + const scope = fiber && Context.getOrUndefined(fiber.context, McpInvocationContext); + return scope?.thread?.subagent !== true; +}; + export type McpCapabilityError = C extends "preview" ? PreviewAutomationUnavailableError : McpCapabilityUnavailableError; diff --git a/apps/server/src/mcp/McpSessionRegistry.ts b/apps/server/src/mcp/McpSessionRegistry.ts index 04c33a1509e94..851d4e3d3d7c2 100644 --- a/apps/server/src/mcp/McpSessionRegistry.ts +++ b/apps/server/src/mcp/McpSessionRegistry.ts @@ -22,6 +22,7 @@ export interface McpCredentialRequest { */ readonly browserToolsAvailable?: boolean; readonly capabilities?: ReadonlySet; + readonly subagent?: boolean; } export interface McpIssuedCredential { @@ -109,6 +110,7 @@ const makeWithOptions = Effect.fn("McpSessionRegistry.make")(function* ( threadId: ThreadId.make(request.threadId), providerSessionId, providerInstanceId: ProviderInstanceId.make(request.providerInstanceId), + subagent: request.subagent === true, }, client: undefined, capabilities: new Set([ diff --git a/apps/server/src/mcp/toolkits/pullRequests/tools.ts b/apps/server/src/mcp/toolkits/pullRequests/tools.ts index a3a75eabaa385..073306365cec3 100644 --- a/apps/server/src/mcp/toolkits/pullRequests/tools.ts +++ b/apps/server/src/mcp/toolkits/pullRequests/tools.ts @@ -10,6 +10,7 @@ import { ThreadId, } from "@t3tools/contracts"; import * as Schema from "effect/Schema"; +import * as McpSchema from "effect/ai/McpSchema"; import * as Tool from "effect/ai/Tool"; import * as Toolkit from "effect/ai/Toolkit"; @@ -316,7 +317,7 @@ const ListThreadPullRequestsTool = Tool.make("list_thread_pull_requests", { const WatchPullRequestTool = Tool.make("watch_pull_request", { description: - "Have T3 Code watch an open pull request for this thread, linking it first if needed. T3 Code checks it every two minutes and wakes you with a message when a check fails, the required checks pass, someone else comments or reviews, or the branch starts to conflict with its base. Use this to monitor or babysit a pull request instead of polling, sleeping, or running a watcher. Only comments posted after this call wake you, so handle the existing ones first, then end your turn. A wake is news, not a merge decision: check readiness yourself before merging. While T3 Code watches, the thread stays in the user's Working list, not their inbox. When you hand the work back to the user, call unwatch_pull_request first so the thread returns to their inbox. Watching ends when the pull request merges or closes, when its thread settles or is archived, when T3 Code fails to read it 8 times in a row (a host rate limit only delays it), when the user stops this thread, or when you call unwatch_pull_request. Unsettle the thread before starting a new watch. A subagent cannot watch: its parent thread owns the pull request.", + "Have T3 Code watch an open pull request for this thread, linking it first if needed. T3 Code checks it every two minutes and wakes you with a message when a check fails, the required checks pass, someone else comments or reviews, or the branch starts to conflict with its base. Use this to monitor or babysit a pull request instead of polling, sleeping, or running a watcher. Only comments posted after this call wake you, so handle the existing ones first, then end your turn. A wake is news, not a merge decision: check readiness yourself before merging. While T3 Code watches, the thread stays in the user's Working list, not their inbox. When you hand the work back to the user, call unwatch_pull_request first so the thread returns to their inbox. Watching ends when the pull request merges or closes, when its thread settles or is archived, when T3 Code fails to read it 8 times in a row (a host rate limit only delays it), when the user stops this thread, or when you call unwatch_pull_request. Unsettle the thread before starting a new watch.", parameters: PullRequestTargetInput, success: WatchPullRequestResult, failure: PullRequestToolError, @@ -326,7 +327,8 @@ const WatchPullRequestTool = Tool.make("watch_pull_request", { .annotate(Tool.Readonly, false) .annotate(Tool.Destructive, false) .annotate(Tool.Idempotent, true) - .annotate(Tool.OpenWorld, false); + .annotate(Tool.OpenWorld, false) + .annotate(McpSchema.EnabledWhen, McpInvocationContext.hiddenFromSubagents); const UnwatchPullRequestTool = Tool.make("unwatch_pull_request", { description: @@ -340,7 +342,8 @@ const UnwatchPullRequestTool = Tool.make("unwatch_pull_request", { .annotate(Tool.Readonly, false) .annotate(Tool.Destructive, false) .annotate(Tool.Idempotent, true) - .annotate(Tool.OpenWorld, false); + .annotate(Tool.OpenWorld, false) + .annotate(McpSchema.EnabledWhen, McpInvocationContext.hiddenFromSubagents); const GuideTarget = Schema.Struct({ url: PullRequestTargetInput.fields.url, diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index f26a72365a69f..ee87e27c68e5f 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -19,7 +19,9 @@ import { TurnItemId, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Stream from "effect/Stream"; @@ -38,6 +40,8 @@ import * as WorkspacePaths from "../workspace/WorkspacePaths.ts"; import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; import * as EventSink from "./EventSink.ts"; import * as Orchestrator from "./Orchestrator.ts"; +import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts"; +import * as ThreadManagementService from "./ThreadManagementService.ts"; import { continueRestartedRun } from "./RestartContinuation.ts"; import * as RuntimeLayer from "./runtimeLayer.ts"; import * as ProviderTurnStartServiceTestkit from "./ProviderTurnStartService.testkit.ts"; @@ -105,6 +109,7 @@ const layerTestProviderInstanceRegistry = Layer.succeed( ); const layerTest = Layer.mergeAll(RuntimeLayer.layer, RuntimeLayer.layerEventSink).pipe( + Layer.provideMerge(ThreadCommandExecutor.layer), Layer.provideMerge(RuntimeLayer.layerProjectService), Layer.provide( Layer.mock(WorkspacePaths.WorkspacePaths)({ @@ -1404,6 +1409,346 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( }), ); + it.effect("delivers the run a background command's wake starts after the original result", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const requests = yield* ProviderContinuationRequests.ProviderContinuationRequests; + const now = yield* DateTime.now; + const parentThreadId = ThreadId.make("thread:background-wake-parent"); + const projectId = ProjectId.make("project:background-wake"); + const parentRunId = RunId.make("run:background-wake-parent"); + const rootNodeId = NodeId.make("node:background-wake-parent"); + yield* seedParentWithTerminalTask({ + threadId: parentThreadId, + projectId, + runId: parentRunId, + rootNodeId, + taskId: NodeId.make("node:background-wake-seed"), + deliveryState: "acknowledged", + now, + }); + const child = yield* seedRestartCancelledChild({ + parentThreadId, + projectId, + parentRunId, + rootNodeId, + name: "background-wake-child", + completionWake: "always", + continuationPending: false, + runStatus: "completed", + now, + }); + yield* orchestrator.recoverDelegatedTasks; + let firstWake = yield* requests.take; + while (firstWake.threadId !== parentThreadId) firstWake = yield* requests.take; + yield* orchestrator.dispatch({ + type: "delegated_task.completion-delivery.acknowledge", + commandId: CommandId.make("command:background-wake-ack"), + parentThreadId, + taskId: child.taskId, + observedByRunId: parentRunId, + }); + const parentRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs.find( + (run) => run.id === parentRunId, + )!; + yield* eventSink.write({ + commandId: reconcileCommandId("background-wake-parent-finished"), + events: [ + { + id: EventId.make("event:background-wake-parent-finished"), + type: "run.updated", + threadId: parentThreadId, + runId: parentRunId, + occurredAt: now, + payload: { ...parentRun, status: "completed", completedAt: now }, + }, + ], + }); + + // The shape ProviderContinuationService dispatches when the provider + // reports a background command (wire kind `background_command`) ended. + const wakeMessageId = MessageId.make("message:background-wake"); + yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "agent", + creationSource: "provider", + commandId: CommandId.make("command:background-wake"), + threadId: child.childThreadId, + messageId: wakeMessageId, + text: "Background activity updated", + notification: { + source: { kind: "command" }, + outcome: "completed", + summary: 'Command "sleep 60" finished', + }, + attachments: [], + dispatchMode: { type: "queue_after_active" }, + }); + const run = (yield* orchestrator.getThreadProjection(child.childThreadId)).runs.find( + (row) => row.userMessageId === wakeMessageId, + )!; + assert.isDefined(run.delegatedTaskId); + yield* eventSink.write({ + commandId: reconcileCommandId("background-wake-finished"), + events: [ + { + id: EventId.make("event:background-wake-result"), + type: "message.updated", + threadId: child.childThreadId, + runId: run.id, + occurredAt: now, + payload: { + id: MessageId.make("message:background-wake-result"), + threadId: child.childThreadId, + runId: run.id, + nodeId: null, + role: "assistant", + text: "The background command finished cleanly.", + attachments: [], + streaming: false, + createdBy: "agent", + creationSource: "server", + createdAt: now, + updatedAt: now, + }, + }, + { + id: EventId.make("event:background-wake-finished"), + type: "run.updated", + threadId: child.childThreadId, + runId: run.id, + occurredAt: now, + payload: { ...run, status: "completed", startedAt: now, completedAt: now }, + }, + ], + }); + yield* orchestrator.recoverDelegatedTasks; + const followup = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find( + (task) => task.id === run.delegatedTaskId, + )!; + assert.equal(followup.status, "completed"); + assert.equal(followup.result, "The background command finished cleanly."); + assert.equal(followup.completionDelivery?.state, "claimed"); + let wake = yield* requests.take; + while ( + wake.threadId !== parentThreadId || + wake.delegatedCompletion?.messageId === firstWake.delegatedCompletion?.messageId + ) { + wake = yield* requests.take; + } + assert.equal(wake.delegatedCompletion?.parentRunId, parentRunId); + }), + ); + + it.effect("preserves parent Stop racing a background wake's follow-up admission", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const executor = yield* ThreadCommandExecutor.ThreadCommandExecutor; + const threads = yield* ThreadManagementService.ThreadManagementService; + const now = yield* DateTime.now; + const parentThreadId = ThreadId.make("thread:background-stop-race-parent"); + const projectId = ProjectId.make("project:background-stop-race"); + const parentRunId = RunId.make("run:background-stop-race-parent"); + const rootNodeId = NodeId.make("node:background-stop-race-parent"); + const wakeCommandId = CommandId.make("command:background-stop-race-wake"); + const stopCommandId = CommandId.make("command:background-stop-race-stop"); + const wakeMessageId = MessageId.make("message:background-stop-race-wake"); + yield* seedParentWithTerminalTask({ + threadId: parentThreadId, + projectId, + runId: parentRunId, + rootNodeId, + taskId: NodeId.make("node:background-stop-race-seed"), + deliveryState: "acknowledged", + now, + }); + const child = yield* seedRestartCancelledChild({ + parentThreadId, + projectId, + parentRunId, + rootNodeId, + name: "background-stop-race-child", + completionWake: "always", + continuationPending: false, + runStatus: "completed", + now, + }); + yield* orchestrator.recoverDelegatedTasks; + yield* orchestrator.dispatch({ + type: "delegated_task.completion-delivery.acknowledge", + commandId: CommandId.make("command:background-stop-race-ack"), + parentThreadId, + taskId: child.taskId, + observedByRunId: parentRunId, + }); + const parent = yield* orchestrator.getThreadProjection(parentThreadId); + const parentRun = parent.runs.find((run) => run.id === parentRunId)!; + yield* eventSink.write({ + commandId: reconcileCommandId("background-stop-race-parent-finished"), + events: [ + { + id: EventId.make("event:background-stop-race-parent-finished"), + type: "run.updated", + threadId: parentThreadId, + runId: parentRunId, + occurredAt: now, + payload: { ...parentRun, status: "completed", completedAt: now }, + }, + ], + }); + const originalChildRun = (yield* orchestrator.getThreadProjection( + child.childThreadId, + )).runs.find((run) => run.id === child.childRunId)!; + const wakePlanned = yield* Deferred.make(); + const releaseWake = yield* Deferred.make(); + const stopRequestedParentLock = yield* Deferred.make(); + const wakeLocks: Array = []; + let wakeFiberId: number | undefined; + let stopFiberId: number | undefined; + const withLock = executor.withLock; + const commitCommand = eventSink.commitCommand; + yield* Effect.acquireUseRelease( + Effect.sync(() => ({ + lock: vi + .spyOn(executor, "withLock") + .mockImplementation((threadId: ThreadId, effect: Effect.Effect) => + Effect.gen(function* () { + const fiberId = yield* Effect.fiberId; + if (fiberId === wakeFiberId) wakeLocks.push(threadId); + if (fiberId === stopFiberId && threadId === parentThreadId) { + yield* Deferred.succeed(stopRequestedParentLock, undefined); + } + return yield* withLock(threadId, effect); + }), + ), + commit: vi.spyOn(eventSink, "commitCommand").mockImplementation((input) => + Effect.gen(function* () { + if (input.commandId === wakeCommandId) { + // Hold the admission snapshot while Stop requests the same parent lock. + yield* Deferred.succeed(wakePlanned, undefined); + yield* Deferred.await(releaseWake); + } + return yield* commitCommand(input); + }), + ), + })), + () => + Effect.gen(function* () { + const wake = yield* Effect.gen(function* () { + wakeFiberId = yield* Effect.fiberId; + return yield* orchestrator.dispatch({ + type: "message.dispatch", + createdBy: "agent", + creationSource: "provider", + commandId: wakeCommandId, + threadId: child.childThreadId, + messageId: wakeMessageId, + text: "Background activity updated", + notification: { + source: { kind: "command" }, + outcome: "completed", + summary: "Background command finished", + }, + attachments: [], + dispatchMode: { type: "queue_after_active" }, + }); + }).pipe(Effect.forkChild); + yield* Deferred.await(wakePlanned); + assert.deepEqual(wakeLocks, [parentThreadId, child.childThreadId]); + const stop = yield* Effect.gen(function* () { + stopFiberId = yield* Effect.fiberId; + return yield* orchestrator.dispatch({ + type: "thread.stop", + threadId: parentThreadId, + commandId: stopCommandId, + }); + }).pipe(Effect.forkChild); + yield* Deferred.await(stopRequestedParentLock); + yield* Deferred.succeed(releaseWake, undefined); + const admitted = yield* Fiber.join(wake); + const stopped = yield* Fiber.join(stop); + assert.isBelow(admitted.sequence, stopped.sequence); + }), + (spies) => + Deferred.succeed(releaseWake, undefined).pipe( + Effect.andThen( + Effect.sync(() => { + spies.commit.mockRestore(); + spies.lock.mockRestore(); + }), + ), + ), + ); + const admittedRun = (yield* orchestrator.getThreadProjection( + child.childThreadId, + )).runs.find((run) => run.userMessageId === wakeMessageId)!; + assert.isDefined(admittedRun.delegatedTaskId); + const stoppedParent = yield* orchestrator.getThreadProjection(parentThreadId); + assert.equal( + stoppedParent.subagents.find((task) => task.id === admittedRun.delegatedTaskId) + ?.completionDelivery?.state, + "disposed", + ); + // Execute the delegated Stop outbox action before any provider-start effect. + yield* threads.stopDelegatedTasks({ threadId: parentThreadId, commandId: stopCommandId }); + yield* orchestrator.recoverDelegatedTasks; + const stoppedChild = yield* orchestrator.getThreadProjection(child.childThreadId); + const interrupted = stoppedChild.runs.find((run) => run.id === admittedRun.id)!; + assert.equal(interrupted.status, "interrupted"); + const settledParent = yield* orchestrator.getThreadProjection(parentThreadId); + const followup = settledParent.subagents.find( + (task) => task.id === admittedRun.delegatedTaskId, + )!; + assert.equal(followup.status, "interrupted"); + assert.equal(followup.completionDelivery?.state, "disposed"); + assert.isNotNull(followup.result); + + yield* eventSink.write({ + commandId: reconcileCommandId("background-stop-race-stale-completion"), + events: [ + { + id: EventId.make("event:background-stop-race-stale-completion"), + type: "run.updated", + threadId: child.childThreadId, + runId: originalChildRun.id, + occurredAt: now, + payload: originalChildRun, + }, + ], + }); + yield* orchestrator.recoverDelegatedTasks; + const finished = yield* orchestrator.getThreadProjection(parentThreadId); + assert.deepEqual( + finished.subagents.filter((task) => task.id !== followup.id), + parent.subagents, + ); + assert.deepEqual( + finished.subagents.find((task) => task.id === admittedRun.delegatedTaskId), + followup, + ); + assert.deepEqual(finished.contextTransfers, settledParent.contextTransfers); + const stoppedRun = finished.runs.find((run) => run.id === parentRunId)!; + assert.equal(stoppedRun.status, "completed"); + assert.equal(stoppedRun.delegatedCompletion?.disposition, "stopped"); + assert.isNull(stoppedRun.delegatedCompletion?.delivery); + assert.equal(finished.nodes.find((node) => node.id === followup.id)?.status, "interrupted"); + assert.equal( + finished.turnItems.find( + (item) => item.type === "subagent" && item.subagentId === followup.id, + )?.status, + "interrupted", + ); + const finalChild = yield* orchestrator.getThreadProjection(child.childThreadId); + assert.deepEqual(finalChild.runs, stoppedChild.runs); + assert.equal(finalChild.runs.length, 2); + assert.isFalse( + yield* orchestrator.delegatedTaskResultPending(child.childThreadId, followup.id), + ); + }), + ); + const outcomes = [ "completed", "failed", diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 54b28cb6384d4..466bc231fd0ac 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -587,6 +587,22 @@ function cancelledAppOwnedTask( }; } +/** + * Whether a message sent into a delegated child may become a follow-up task the parent hears + * back from. Besides a person or agent writing to the child, a provider wake counts, such as a + * background command finishing: the child acts on it after its original result was delivered. + * A monitor update does not, since it reports progress rather than asking for new work. + * Dispatch takes the parent's lock for exactly these, because intake writes the parent. + */ +function mayStartSubagentFollowup( + command: Extract, +): boolean { + if (command.scheduledTaskId !== undefined || command.restartContinuationOfRunId !== undefined) + return false; + if (command.notification !== undefined) return command.notification.source.kind !== "monitor"; + return command.createdBy === "user" || command.creationSource === "mcp"; +} + function isBlockingRun(run: OrchestrationV2Run): boolean { return ( run.status === "preparing" || @@ -7427,13 +7443,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio command: Extract, events: Ref.Ref>, ) { - if ( - command.scheduledTaskId !== undefined || - command.notification !== undefined || - command.restartContinuationOfRunId !== undefined || - (command.createdBy !== "user" && command.creationSource !== "mcp") - ) - return; + if (!mayStartSubagentFollowup(command)) return; const pending = yield* Ref.get(events); const message = pending .flatMap((event) => @@ -7457,7 +7467,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio if ( child.lineage.relationshipToParent !== "subagent" || parentThreadId === null || - (command.createdBy !== "user" && command.senderThreadId !== parentThreadId) + (command.notification === undefined && + command.createdBy !== "user" && + command.senderThreadId !== parentThreadId) ) return; const run = @@ -9062,7 +9074,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio } childThreadIds.add(childThreadId); const child = yield* projectionStore - .getThreadRecords(childThreadId, ["nodes", "subagents", "turnItems"], { + .getThreadRecords(childThreadId, ["nodes", "providerTurns", "subagents", "turnItems"], { // Runless items are the subagent's; any run there is the user's own. turnItemRunIds: [null], turnItemStatuses: ["pending", "running", "waiting"], @@ -9599,7 +9611,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio yield* Ref.update(events, (current) => [...current, ...ended]); if (subagent.childThreadId !== null) { const child = yield* projectionStore - .getThreadRecords(subagent.childThreadId, ["nodes", "turnItems"], { + .getThreadRecords(subagent.childThreadId, ["nodes", "providerTurns", "turnItems"], { turnItemRunIds: [null], turnItemStatuses: ["pending", "running", "waiting"], }) @@ -12055,10 +12067,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio commandThreadId(command), dispatchWithReceiptEffect(command), ); - if ( - command.type !== "message.dispatch" || - (command.createdBy !== "user" && command.creationSource !== "mcp") - ) + if (command.type !== "message.dispatch" || !mayStartSubagentFollowup(command)) return yield* dispatch; const parentThreadId = yield* appOwnedSubagentParentThreadId(command.threadId).pipe( Effect.mapError( diff --git a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts index 6e0dd990ce388..6ed24558f01dc 100644 --- a/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionRecovery.test.ts @@ -12,6 +12,7 @@ import { ProviderInstanceId, ProviderSessionId, ProviderThreadId, + ProviderTurnId, RunId, RuntimeRequestId, ThreadId, @@ -966,3 +967,102 @@ it.effect("marks fork descendants unreadable when their source is missing or cor ); }).pipe(Effect.provide(layerTest)), ); + +it.effect("startup ends the provider turn of a native child root it cancels", () => + Effect.gen(function* () { + const projections = yield* ProjectionStore.ProjectionStoreV2; + const recovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService; + const now = yield* DateTime.now; + const parentThreadId = yield* createThread("native-child-root-turn-parent"); + const threadId = yield* createThread("native-child-root-turn"); + const rootNodeId = NodeId.make(`node:${threadId}:thread-root`); + const providerThreadId = ProviderThreadId.make(`provider-thread:${threadId}`); + const providerTurnId = ProviderTurnId.make(`provider-turn:${threadId}`); + yield* projections.apply({ + id: EventId.make(`event:${threadId}:root`), + type: "node.updated", + threadId, + nodeId: rootNodeId, + driver, + occurredAt: now, + payload: { + id: rootNodeId, + threadId, + runId: null, + parentNodeId: null, + rootNodeId, + kind: "root_turn", + status: "running", + countsForRun: false, + providerThreadId, + providerTurnId, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt: now, + completedAt: null, + }, + }); + // A native subagent's turn belongs to no run attempt. + yield* projections.apply({ + id: EventId.make(`event:${threadId}:provider-turn`), + type: "provider-turn.updated", + threadId, + nodeId: rootNodeId, + driver, + occurredAt: now, + payload: { + id: providerTurnId, + providerThreadId, + nodeId: rootNodeId, + runAttemptId: null, + nativeTurnRef: null, + ordinal: 1, + status: "running", + startedAt: now, + completedAt: null, + }, + }); + + const subagentId = NodeId.make(`node:${parentThreadId}:subagent`); + yield* projections.apply({ + id: EventId.make(`event:${parentThreadId}:subagent`), + type: "subagent.updated", + threadId: parentThreadId, + occurredAt: now, + payload: { + id: subagentId, + threadId: parentThreadId, + runId: null, + parentNodeId: NodeId.make(`node:${parentThreadId}:root`), + origin: "provider_native", + createdBy: "agent", + driver, + providerInstanceId, + providerThreadId, + childThreadId: threadId, + nativeTaskRef: null, + title: null, + model: null, + prompt: "sleep 300", + status: "running", + result: null, + startedAt: now, + completedAt: null, + updatedAt: now, + }, + }); + + yield* recovery.reconcile("startup"); + + const projection = yield* projections.getThreadProjection(threadId); + assert.deepEqual( + { + root: projection.nodes.find((node) => node.id === rootNodeId)?.status, + providerTurn: projection.providerTurns.find((turn) => turn.id === providerTurnId)?.status, + }, + { root: "cancelled", providerTurn: "cancelled" }, + "a root startup recovery cancelled left its provider turn running", + ); + }).pipe(Effect.provide(layerRecoveryTest)), +); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 4afc1d43315df..6859360f8a8a9 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -4556,14 +4556,26 @@ export const layer: Layer.Layer = FROM orchestration_v2_projection_provider_turns AS provider_turn WHERE provider_turn.thread_id = ${threadId} AND provider_turn.status IN ('pending', 'starting', 'running', 'waiting') - AND provider_turn.run_attempt_id IN ( - SELECT attempt_id FROM orchestration_v2_projection_run_attempts - WHERE thread_id = ${threadId} - AND run_id IN ( - SELECT run_id FROM orchestration_v2_projection_runs + AND ( + provider_turn.run_attempt_id IN ( + SELECT attempt_id FROM orchestration_v2_projection_run_attempts + WHERE thread_id = ${threadId} + AND run_id IN ( + SELECT run_id FROM orchestration_v2_projection_runs + WHERE thread_id = ${threadId} + AND status IN ('queued', 'preparing', 'starting', 'running', 'waiting') + ) + ) + OR ( + provider_turn.run_attempt_id IS NULL + AND provider_turn.node_id IN ( + SELECT node_id FROM orchestration_v2_projection_nodes WHERE thread_id = ${threadId} - AND status IN ('queued', 'preparing', 'starting', 'running', 'waiting') + AND run_id IS NULL + AND kind = 'root_turn' + AND status IN ('pending', 'running', 'waiting') ) + ) ) ORDER BY provider_turn.provider_thread_id ASC, provider_turn.ordinal ASC `, diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index b60758e80c728..4a9b44e12536b 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -1,6 +1,7 @@ import * as NetAddress from "effect/net/NetAddress"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; +import { vi } from "vite-plus/test"; import { AgentScope } from "@t3tools/shared/AgentScope"; import { EnvironmentId, @@ -1563,6 +1564,129 @@ it.effect( }), ); +it.effect.each([ + { relationshipToParent: null, subagent: false }, + { relationshipToParent: "subagent", subagent: true }, +] as const)( + "ProviderSessionManagerV2 marks MCP credentials of delegated children ($relationshipToParent)", + ({ relationshipToParent, subagent }) => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + const mcpConfigs = yield* Ref.make< + ReadonlyArray + >([]); + const effect = Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const registry = yield* McpSessionRegistry.McpSessionRegistry; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread-provider-session-manager-mcp-lineage"); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + const created = yield* makeThreadCreatedEvent({ idAllocator, threadId, now }); + const parentThreadId = ThreadId.make("thread-provider-session-manager-mcp-parent"); + yield* eventSink.write({ + events: [ + { + ...created, + payload: { + ...created.payload, + lineage: + relationshipToParent === null + ? created.payload.lineage + : { parentThreadId, relationshipToParent, rootThreadId: parentThreadId }, + }, + }, + ], + }); + yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy }); + + const token = (yield* Ref.get(mcpConfigs))[0]?.authorizationHeader.replace( + /^Bearer\s+/, + "", + ); + assert.isDefined(token); + assert.equal((yield* registry.resolve(token!))?.thread.subagent, subagent); + yield* manager.close(providerSessionId); + }); + + yield* effect.pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 1_000, mcpConfigs }))); + }), +); + +it.effect("ProviderSessionManagerV2 preserves child credentials when lineage lookup fails", () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + yield* Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const registry = yield* McpSessionRegistry.McpSessionRegistry; + const mcpSessions = yield* McpProviderSessions.McpProviderSessions; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread-child-lineage-retry"); + const parentThreadId = ThreadId.make("thread-child-lineage-parent"); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + const created = yield* makeThreadCreatedEvent({ idAllocator, threadId, now }); + yield* eventSink.write({ + events: [ + { + ...created, + payload: { + ...created.payload, + lineage: { + parentThreadId, + relationshipToParent: "subagent", + rootThreadId: parentThreadId, + }, + }, + }, + ], + }); + const open = manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy }); + yield* open; + const original = yield* mcpSessions.read(threadId); + assert.isDefined(original); + const token = original!.authorizationHeader.replace(/^Bearer\s+/, ""); + assert.isTrue((yield* registry.resolve(token))?.thread.subagent); + yield* manager.detach({ providerSessionId, threadId }); + + const getThread = projections.getThread; + using read = vi + .spyOn(projections, "getThread") + .mockImplementation((requestedThreadId) => + Effect.fail( + new ProjectionStore.ProjectionStoreReadError({ threadId: requestedThreadId }), + ), + ); + const failed = yield* open.pipe(Effect.exit); + assert.isTrue(Exit.isFailure(failed)); + assert.equal( + (yield* mcpSessions.read(threadId))?.providerSessionId, + original!.providerSessionId, + ); + assert.isTrue((yield* registry.resolve(token))?.thread.subagent); + assert.isTrue(yield* manager.isLive(providerSessionId)); + + read.mockImplementation(getThread); + yield* open; + assert.equal( + (yield* mcpSessions.read(threadId))?.providerSessionId, + original!.providerSessionId, + ); + assert.isTrue((yield* registry.resolve(token))?.thread.subagent); + yield* manager.close(providerSessionId); + }).pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 60_000 }))); + }), +); + it.effect( "ProviderSessionManagerV2 withholds the preview capability when agent browser access is off", () => diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 9a9238f887b12..b65166f8dfe4e 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -465,7 +465,8 @@ export const layerWithOptions = ( const prepareMcpSession = ( threadId: ThreadId, providerInstanceId: ProviderInstanceId, - ): Effect.Effect => + providerSessionId: ProviderSessionId, + ): Effect.Effect => options.configureMcp === false ? mcpSessions .clear(threadId) @@ -486,6 +487,17 @@ export const layerWithOptions = ( >(["orchestration", "worktree", "pull-requests"]); if (browserToolsAvailable) capabilities.add("preview"); if (deviceToolsAvailable) capabilities.add("device"); + const subagent = yield* projectionStore.getThread(threadId).pipe( + Effect.map((thread) => thread.lineage.relationshipToParent === "subagent"), + Effect.mapError( + (cause) => + new ProviderSessionOpenError({ + instanceId: providerInstanceId, + providerSessionId, + cause, + }), + ), + ); const existing = yield* mcpSessions.read(threadId); if (existing !== undefined) { // Reserve before the async resolve so a release cannot @@ -508,7 +520,9 @@ export const layerWithOptions = ( resolved.thread.threadId === threadId && resolved.thread.providerInstanceId === providerInstanceId && resolved.capabilities.has("preview") === browserToolsAvailable && - resolved.capabilities.has("device") === deviceToolsAvailable + resolved.capabilities.has("device") === deviceToolsAvailable && + // Refresh credentials when the thread's delegation ownership changes. + (resolved.thread.subagent === true) === subagent ) { return { mcpCredentialId: existing.providerSessionId, issued: false }; } @@ -520,6 +534,7 @@ export const layerWithOptions = ( providerInstanceId, browserToolsAvailable, capabilities, + subagent, }); yield* mcpSessions.set(credential.config); reserveMcpCredential(threadId, credential.config.providerSessionId); @@ -1403,7 +1418,14 @@ export const layerWithOptions = ( Effect.uninterruptible, ); if (attached !== undefined) { - const prepared = yield* prepareMcpSession(input.threadId, input.providerInstanceId); + const prepared = yield* withActivityError( + input.providerSessionId, + prepareMcpSession( + input.threadId, + input.providerInstanceId, + input.providerSessionId, + ), + ); preparedForCleanup = prepared; if (prepared.mcpCredentialId !== undefined) { const mcpCredentialId = prepared.mcpCredentialId; @@ -2260,6 +2282,7 @@ export const layerWithOptions = ( const prepared = yield* prepareMcpSession( input.threadId, input.modelSelection.instanceId, + input.providerSessionId, ); const mcpCredentialId = prepared.mcpCredentialId; let reservationDropped = mcpCredentialId === undefined; diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index e7937f8a1fc9f..75a746ce2feb6 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -11,6 +11,7 @@ import { type OrchestrationV2DomainEvent, type OrchestrationV2ExecutionNode, type OrchestrationV2ProviderThread, + type OrchestrationV2ProviderTurn, type OrchestrationV2Run, type OrchestrationV2RunAttempt, type OrchestrationV2Subagent, @@ -2589,6 +2590,106 @@ it.effect.each(["failure", "defect", "interrupt-failure", "ownership-lost"] as c }), ); +it.effect.each([ + { status: "completed", failure: "later-stream", interrupt: "succeeds" }, + { status: "failed", failure: "later-stream", interrupt: "succeeds" }, + { status: "completed", failure: "terminal-ingest", interrupt: "succeeds" }, + { status: "completed", failure: "terminal-ingest", interrupt: "fails" }, +] as const)( + "stores a $status root provider turn snapshot after a $failure failure only if the interrupt succeeds (it $interrupt)", + ({ status, failure, interrupt }) => + Effect.gen(function* () { + const now = yield* DateTime.now; + const terminalAt = DateTime.subtract(now, { seconds: 1 }); + const key = `terminal-snapshot:${status}:${failure}:${interrupt}`; + const durableTurns = yield* Ref.make< + ReadonlyMap + >(new Map()); + const persist = (turn: OrchestrationV2ProviderTurn) => + Ref.update(durableTurns, (current) => new Map(current).set(turn.id, turn)); + const interrupts = yield* Ref.make< + ReadonlyArray + >([]); + const ids = backgroundScenarioIds(key); + const rootTurn = ( + turnStatus: OrchestrationV2ProviderTurn["status"], + completedAt: DateTime.Utc | null, + ) => ({ + type: "provider_turn.updated" as const, + driver, + providerTurn: { + id: ids.rootProviderTurnId, + providerThreadId: ids.providerThreadId, + nodeId: ids.rootNodeId, + runAttemptId: ids.attemptId, + nativeTurnRef: null, + ordinal: 1, + status: turnStatus, + startedAt: terminalAt, + completedAt, + }, + }); + const { observed, finalEvents } = yield* captureRootRunTermination({ + key, + shouldFinalizeRun: () => Effect.succeed(true), + events: () => + Stream.make(rootTurn("running", null), rootTurn(status, terminalAt)).pipe( + Stream.concat( + failure === "later-stream" + ? Stream.fail( + new ProviderAdapter.ProviderAdapterEventStreamError({ + driver, + providerSessionId: ProviderSessionId.make(`session:${key}`), + cause: "provider process exited", + }), + ) + : Stream.empty, + ), + ), + ingestNormalized: ({ event }) => + event.type !== "provider_turn.updated" + ? Effect.succeed([]) + : failure === "terminal-ingest" && event.providerTurn.status !== "running" + ? Effect.fail( + new ProviderEventIngestor.ProviderEventPublishError({ + providerSessionId: ProviderSessionId.make(`session:${key}`), + eventCount: 1, + cause: new Error("SQLITE_FULL: database or disk is full"), + }), + ) + : persist(event.providerTurn).pipe(Effect.as([])), + interruptTurn: (input) => + Ref.update(interrupts, (current) => [...current, input]).pipe( + Effect.andThen( + interrupt === "fails" + ? Effect.fail( + new ProviderAdapter.ProviderAdapterInterruptError({ + driver, + providerThreadId: input.providerThread.id, + providerTurnId: input.providerTurnId, + cause: "provider already exited", + }), + ) + : Effect.void, + ), + ), + beforeTerminalWrite: (request) => + Effect.forEach(request.events, (event) => + event.type === "provider-turn.updated" ? persist(event.payload) : Effect.void, + ).pipe(Effect.as(true)), + }); + const calls = yield* Ref.get(interrupts); + assert.equal(calls.length, 1); + assert.isTrue(calls[0]?.requestRuntimeRestart); + assert.deepEqual(observed, ["run:failed", "pull-requests-refreshed"]); + assert.isTrue(finalEvents.some((event) => event.type === "run.updated")); + const durable = (yield* Ref.get(durableTurns)).get(ids.rootProviderTurnId); + // A failed interrupt keeps the stored turn running for the next start to retry. + assert.equal(durable?.status, interrupt === "fails" ? "running" : status); + assert.deepEqual(durable?.completedAt, interrupt === "fails" ? null : terminalAt); + }), +); + it.effect("does not interrupt an ended provider when its terminal write fails", () => Effect.gen(function* () { const interrupts = yield* Ref.make(0); diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index a2b4f1de5bcf2..39b4faa4adfc2 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -622,7 +622,8 @@ export const layer: Layer.Layer< readonly shouldFinalizeRun?: () => Effect.Effect; readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect; readonly openRunOwnedSubagents?: OpenRunOwnedSubagentProjection; - readonly interruptedProviderTurn?: OrchestrationV2ProviderTurn; + // Root turn snapshot written with the run when T3 ends the run itself. + readonly abandonedProviderTurn?: OrchestrationV2ProviderTurn; readonly terminal: ProviderTerminalEvent; readonly completedAt?: DateTime.Utc; readonly failureItemPersisted: boolean; @@ -744,7 +745,7 @@ export const layer: Layer.Layer< : [], events: [ ...cascadedSubagentEvents, - ...(input.interruptedProviderTurn === undefined + ...(input.abandonedProviderTurn === undefined ? [] : [ { @@ -755,11 +756,7 @@ export const layer: Layer.Layer< nodeId: input.rootNode.id, providerInstanceId: input.run.providerInstanceId, occurredAt: completedAt, - payload: { - ...input.interruptedProviderTurn, - status: "interrupted" as const, - completedAt, - }, + payload: input.abandonedProviderTurn, }, ]), ...(finalizedAttempt === null @@ -1464,8 +1461,9 @@ export const layer: Layer.Layer< if (yield* Ref.get(rootRunFinalized)) return; const providerThread = yield* Ref.get(latestProviderThread); const routing = yield* Ref.get(eventRouting); + const rootProviderTurn = yield* Ref.get(latestRootProviderTurn); // Stop background work too; the failed consumer cannot persist terminal events. - const interruptedProviderTurn = + const abandonedProviderTurn = routing.rootProviderTurnId !== null && !routing.rootTurnEnded ? yield* input.session .interruptTurn({ @@ -1474,14 +1472,28 @@ export const layer: Layer.Layer< requestRuntimeRestart: true, }) .pipe( - Effect.as(yield* Ref.get(latestRootProviderTurn)), + Effect.as(true), Effect.catchCauseIf( (interruptCause) => !Cause.hasInterruptsOnly(interruptCause), (interruptCause) => Effect.logWarning( "orchestration V2 failed to interrupt turn after ingestion failure", { runId: input.run.id, cause: interruptCause }, - ).pipe(Effect.as(null)), + ).pipe(Effect.as(false)), + ), + // A failed interrupt leaves the stored turn running so the next start + // retries it. The snapshot is cached before it is published, so it may + // not be stored: replay a terminal one as-is. + Effect.map((interrupted) => + !interrupted || rootProviderTurn === null + ? null + : isTerminalProviderTurnStatus(rootProviderTurn.status) + ? rootProviderTurn + : { + ...rootProviderTurn, + status: "interrupted" as const, + completedAt, + }, ), ) : null; @@ -1499,7 +1511,7 @@ export const layer: Layer.Layer< expectedStatus: "running", }, openRunOwnedSubagents: openSubagents, - ...(interruptedProviderTurn === null ? {} : { interruptedProviderTurn }), + ...(abandonedProviderTurn === null ? {} : { abandonedProviderTurn }), completedAt, terminal: makeFailedTerminalEvent( makeProviderFailure({ cause: Cause.squash(cause), class: "unknown" }), diff --git a/apps/server/src/orchestration-v2/SubagentProjection.test.ts b/apps/server/src/orchestration-v2/SubagentProjection.test.ts index e87fd7d9c6eea..d3506649f6a62 100644 --- a/apps/server/src/orchestration-v2/SubagentProjection.test.ts +++ b/apps/server/src/orchestration-v2/SubagentProjection.test.ts @@ -375,6 +375,7 @@ it.effect("ends a runless root turn's items when its node already ended", () => providerInstanceId: parentProviderInstanceId, projection: { nodes: [root], + providerTurns: [], turnItems: [item("turn-item:command"), item("turn-item:reasoning")], } as never, status: "cancelled", diff --git a/apps/server/src/orchestration-v2/ThreadStop.test.ts b/apps/server/src/orchestration-v2/ThreadStop.test.ts index 5952e5c217a3e..dbc57dec1b3a0 100644 --- a/apps/server/src/orchestration-v2/ThreadStop.test.ts +++ b/apps/server/src/orchestration-v2/ThreadStop.test.ts @@ -1336,6 +1336,25 @@ it.effect.each(["missing session", "completed", "failed", "cancelled"] as const) occurredAt: now, payload: childRoot, }); + const childTurnId = ProviderTurnId.make("turn:native-stop-child"); + yield* projections.apply({ + id: EventId.make("event:native-child-stop:child-turn"), + type: "provider-turn.updated", + threadId: childRoot.threadId, + nodeId: childRoot.id, + occurredAt: now, + payload: { + id: childTurnId, + providerThreadId: ProviderThreadId.make("provider-thread:native-stop-child"), + nodeId: childRoot.id, + runAttemptId: null, + nativeTurnRef: null, + ordinal: 1, + status: "running", + startedAt: now, + completedAt: null, + }, + }); const providerThread = before.providerThreads.find( (thread) => thread.id === run.providerThreadId, )!; @@ -1500,6 +1519,10 @@ it.effect.each(["missing session", "completed", "failed", "cancelled"] as const) ); const child = yield* orchestrator.getThreadProjection(childRoot.threadId); assert.equal(child.nodes.find((node) => node.id === childRoot.id)?.status, confirmedStatus); + assert.equal( + child.providerTurns.find((turn) => turn.id === childTurnId)?.status, + confirmedStatus, + ); assert.isNotNull(child.nodes.find((node) => node.id === childRoot.id)?.completedAt); assert.equal(settled.subagents.find((task) => task.id === sibling.id)?.status, "running"); assert.deepEqual(yield* threadState(threadId), stateBefore); diff --git a/packages/provider-core/src/server/subagentProjection.ts b/packages/provider-core/src/server/subagentProjection.ts index d5e87153ea12e..509e53584fd3c 100644 --- a/packages/provider-core/src/server/subagentProjection.ts +++ b/packages/provider-core/src/server/subagentProjection.ts @@ -271,7 +271,7 @@ export function delegatedTaskProgress(projection: { function writtenIds( emitted: ReadonlyArray, - type: "subagent.updated" | "node.updated" | "turn-item.updated", + type: "subagent.updated" | "node.updated" | "turn-item.updated" | "provider-turn.updated", ): ReadonlySet { return new Set(emitted.flatMap((event) => (event.type === type ? [event.payload.id] : []))); } @@ -359,7 +359,10 @@ export const endRunlessRootTurns = Effect.fn("subagentProjection.endRunlessRootT >(input: { readonly threadId: ThreadId; readonly providerInstanceId: ProviderInstanceId; - readonly projection: Pick; + readonly projection: Pick< + OrchestrationV2ThreadProjection, + "nodes" | "providerTurns" | "turnItems" + >; readonly status: "completed" | "failed" | "interrupted" | "cancelled"; readonly completedAt?: DateTime.Utc; readonly now: DateTime.Utc; @@ -369,6 +372,7 @@ export const endRunlessRootTurns = Effect.fn("subagentProjection.endRunlessRootT const { now, status, completedAt = now } = input; const endedNodeIds = writtenIds(input.emitted, "node.updated"); const endedItemIds = writtenIds(input.emitted, "turn-item.updated"); + const endedProviderTurnIds = writtenIds(input.emitted, "provider-turn.updated"); const scope = { threadId: input.threadId, providerInstanceId: input.providerInstanceId, @@ -421,6 +425,23 @@ export const endRunlessRootTurns = Effect.fn("subagentProjection.endRunlessRootT }, }); } + for (const providerTurn of input.projection.providerTurns) { + if ( + providerTurn.nodeId !== node.id || + providerTurn.runAttemptId !== null || + !isOrchestrationV2WorkActive(providerTurn.status) || + endedProviderTurnIds.has(providerTurn.id) + ) { + continue; + } + events.push({ + ...scope, + id: yield* input.allocateEventId(), + type: "provider-turn.updated", + nodeId: node.id, + payload: { ...providerTurn, status, completedAt }, + }); + } } return events; });