From cd39a86b65375b0ffaaea0b72dbc37181dda8776 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Mon, 19 Oct 2026 21:09:10 -0700 Subject: [PATCH 1/3] fix(server): deliver subagent results without history handoffs --- .../DelegatedCompletionDelivery.test.ts | 93 +++++++++++++++++++ .../src/orchestration-v2/Orchestrator.ts | 88 +++++------------- .../ProviderTurnStartService.memory.test.ts | 2 +- .../ProviderTurnStartService.ts | 8 ++ .../ProviderTurnStartMemory.fixture.mjs | 13 ++- 5 files changed, 132 insertions(+), 72 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index c188d60d4b7d..d0aff5d573d4 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -670,6 +670,7 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { assert.include(message?.text ?? "", String(firstTaskId)); assert.include(message?.text ?? "", String(secondTaskId)); assert.include(message?.text ?? "", "task_status"); + assert.include(message?.text ?? "", "child finished"); }), ); @@ -1503,6 +1504,98 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( ); it.layer(layerTest)("delegated tasks across a server restart", (it) => { + it.effect("queues a child result without creating a provider history handoff", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:result-message-parent"); + const projectId = ProjectId.make("project:result-message"); + const runId = RunId.make("run:result-message-parent"); + const rootNodeId = NodeId.make("node:result-message-parent"); + yield* seedParentWithTerminalTask({ + threadId, + projectId, + runId, + rootNodeId, + taskId: NodeId.make("node:result-message-seed"), + deliveryState: "acknowledged", + now, + }); + const child = yield* seedRestartCancelledChild({ + parentThreadId: threadId, + projectId, + parentRunId: runId, + rootNodeId, + name: "result-message-child", + completionWake: "always", + continuationPending: false, + runStatus: "completed", + now, + }); + const parent = yield* orchestrator.getThreadProjection(threadId); + const childProjection = yield* orchestrator.getThreadProjection(child.childThreadId); + const providerThreadId = ProviderThreadId.make("provider-thread:result-message-child"); + yield* eventSink.write({ + commandId: CommandId.make("command:result-message-child-provider"), + events: [ + { + id: EventId.make("event:result-message-child-provider"), + type: "provider-thread.updated", + threadId: child.childThreadId, + occurredAt: now, + payload: { + ...parent.providerThreads[0]!, + id: providerThreadId, + appThreadId: child.childThreadId, + }, + }, + { + id: EventId.make("event:result-message-child-run"), + type: "run.updated", + threadId: child.childThreadId, + runId: child.childRunId, + occurredAt: now, + payload: { ...childProjection.runs[0]!, providerThreadId }, + }, + ], + }); + + yield* orchestrator.recoverDelegatedTasks; + const recovered = yield* orchestrator.getThreadProjection(threadId); + const task = recovered.subagents.find((candidate) => candidate.id === child.taskId)!; + assert.equal(task.status, "completed"); + assert.isNotNull(task.result); + assert.isEmpty(recovered.contextHandoffs); + assert.equal(recovered.contextTransfers.length, 1); + + const delivery = recovered.runs.find((candidate) => candidate.id === runId)! + .delegatedCompletion!.delivery!; + const messageId = delivery.messageId; + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make("command:result-message-delivery"), + threadId, + messageId, + text: "Delegated task finished.", + attachments: [], + dispatchMode: { type: "queue_after_active" }, + createdBy: "agent", + creationSource: "server", + delegatedCompletion: { + parentRunId: runId, + generation: delivery.generation, + taskIds: [child.taskId], + }, + }); + const queued = yield* orchestrator.getThreadProjection(threadId); + assert.include( + queued.messages.find((message) => message.id === messageId)?.text ?? "", + task.result!, + ); + }), + ); + it.effect("holds a restart-cancelled child for its continuation's result", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index dbe9f9b9cd52..34a84cb1e056 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -590,11 +590,18 @@ function legacyPullRequestLink( ); } -function delegatedCompletionWakeDetail(taskIds: ReadonlyArray): string { - const taskList = taskIds.join(", "); - return taskIds.length === 1 - ? `Delegated task ${taskList} reached a terminal state. Use task_status with taskId ${taskList} to read the result.` - : `Delegated tasks ${taskList} reached terminal states. Use task_status with each taskId to read the results.`; +function delegatedCompletionWakeDetail( + taskIds: ReadonlyArray, + tasks: ReadonlyArray, +): string { + return taskIds + .map((taskId) => { + const task = tasks.find((candidate) => candidate.id === taskId); + return task?.result == null + ? `Delegated task ${taskId} reached a terminal state. Use task_status with taskId ${taskId} to read the result.` + : `Delegated task ${taskId}${task.title === null ? "" : ` (${task.title})`} [${task.status}]:\n${task.result}`; + }) + .join("\n\n"); } function isTerminalDelegatedTaskStatus(status: OrchestrationV2Subagent["status"]): boolean { @@ -2136,7 +2143,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio occurredAt: now, payload: { ...message, - text: delegatedCompletionWakeDetail(remainingTaskIds), + text: delegatedCompletionWakeDetail(remainingTaskIds, projection.subagents), delegatedCompletion: { parentRunId: parentRun.id, generation: delivery.generation, @@ -4981,7 +4988,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const dispatchText = delegatedCompletion === undefined ? command.text - : delegatedCompletionWakeDetail(delegatedCompletion.taskIds); + : delegatedCompletionWakeDetail(delegatedCompletion.taskIds, projection.subagents); const sourcePlanProjection = command.sourcePlanRef === undefined ? null @@ -10192,7 +10199,12 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ? undefined : { ...message, - text: delegatedCompletionWakeDetail(taskIds), + text: delegatedCompletionWakeDetail( + taskIds, + input.parentProjection.subagents.map((task) => + task.id === input.updatedTask.id ? input.updatedTask : task, + ), + ), delegatedCompletion: { parentRunId: input.parentRun.id, generation: delivery.generation, @@ -10482,46 +10494,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio targetThreadId: parentThreadId, type: "subagent_result", }); - const childProviderThread = - childRun.providerThreadId === null - ? undefined - : childProjection.providerThreads.find( - (candidate) => candidate.id === childRun.providerThreadId, - ); - const parentProviderThread = - parentRun?.providerThreadId === null || parentRun?.providerThreadId === undefined - ? undefined - : parentProjection.providerThreads.find( - (candidate) => candidate.id === parentRun.providerThreadId, - ); - const resultHandoff: OrchestrationV2ContextHandoff | null = - parentRun === undefined || - childProviderThread === undefined || - parentProviderThread === undefined - ? null - : { - id: yield* idAllocator.allocate.contextHandoff({ - threadId: parentThreadId, - fromProviderInstanceId: childRun.providerInstanceId, - toProviderInstanceId: parentRun.providerInstanceId, - }), - transferId: resultTransferId, - threadId: parentThreadId, - targetRunId: parentRun.id, - fromProviderThreadIds: [childProviderThread.id], - toProviderThreadId: parentProviderThread.id, - coveredRunOrdinals: { - from: childRun.ordinal, - to: childRun.ordinal, - }, - strategy: "manual_context", - status: "ready", - summaryMessageId: result.messageId, - summaryText: result.text, - createdByProviderInstanceId: childRun.providerInstanceId, - createdAt: now, - updatedAt: now, - }; const resultTransfer: OrchestrationV2ContextTransfer = { id: resultTransferId, type: "subagent_result", @@ -10537,13 +10509,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio parentRun?.providerInstanceId ?? parentProjection.thread.providerInstanceId, targetRunId: parentRun?.id ?? null, status: "consumed", - resolution: - resultHandoff === null - ? null - : { - strategy: "portable_context", - contextHandoffId: resultHandoff.id, - }, + resolution: null, createdBy: "system", error: null, createdAt: now, @@ -10632,18 +10598,6 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }, }, ]), - ...(resultHandoff === null - ? [] - : [ - { - type: "context-handoff.updated" as const, - threadId: parentThreadId, - ...(parentRun === undefined ? {} : { runId: parentRun.id }), - providerInstanceId: childRun.providerInstanceId, - occurredAt: now, - payload: resultHandoff, - }, - ]), { type: "context-transfer.created", threadId: parentThreadId, diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.memory.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.memory.test.ts index 566e35bf3dc5..91f36959e76b 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.memory.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.memory.test.ts @@ -6,7 +6,7 @@ import { expect, it } from "vite-plus/test"; const execFile = NodeUtil.promisify(NodeChildProcess.execFile); -it.each(["regular", "handoff", "handoff-compact", "handoff-failure"])( +it.each(["regular", "handoff", "handoff-compact", "handoff-failure", "subagent-result"])( "releases startup history while %s run workers remain alive", async (mode) => { const root = NodeURL.fileURLToPath(new URL("../../../../", import.meta.url)); diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 1f57b2a8a40a..d611fd5967b8 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -240,8 +240,16 @@ export const layer: Layer.Layer< const checkpointScope = projection.checkpointScopes.find( (candidate) => candidate.id === rootNode?.checkpointScopeId, ); + // Older versions also stored subagent replies as history handoffs. Their + // queued completion messages carry the results without a history budget. + const subagentResultTransfers = new Set( + projection.contextTransfers + .filter((transfer) => transfer.type === "subagent_result") + .map((transfer) => transfer.id), + ); const handoffs = projection.contextHandoffs.filter( (handoff) => + (handoff.transferId == null || !subagentResultTransfers.has(handoff.transferId)) && handoff.status === "ready" && (handoff.targetRunId === run.id || (handoff.toProviderThreadId === run.providerThreadId && diff --git a/apps/server/src/orchestration-v2/testkit/ProviderTurnStartMemory.fixture.mjs b/apps/server/src/orchestration-v2/testkit/ProviderTurnStartMemory.fixture.mjs index 840da2742323..6ac1f2b172bd 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderTurnStartMemory.fixture.mjs +++ b/apps/server/src/orchestration-v2/testkit/ProviderTurnStartMemory.fixture.mjs @@ -4,7 +4,7 @@ import * as NodeCrypto from "node:crypto"; import * as NodeAssert from "node:assert/strict"; const root = process.argv[2]; const mode = process.argv[3]; -const withHandoff = mode.startsWith("handoff"); +const withHandoff = mode.startsWith("handoff") || mode === "subagent-result"; const require = NodeModule.createRequire(root + "/apps/server/package.json"); const load = (name) => import(NodeURL.pathToFileURL(require.resolve("effect/" + name))); const [Effect, Layer, FileSystem, Path] = await Promise.all([ @@ -160,7 +160,10 @@ function fixture(index) { driver: "codex", providerSessionId: "session", nativeThreadRef: { nativeId: "native", driver: "codex", strength: "strong" }, - contextUsage: { usedTokens: 0, maxTokens: 128000 }, + contextUsage: + mode === "subagent-result" + ? { usedTokens: 390458, maxTokens: 475000 } + : { usedTokens: 0, maxTokens: 128000 }, handoffIds: [], }, ], @@ -170,19 +173,21 @@ function fixture(index) { ? [ { id: "handoff", + transferId: mode === "subagent-result" ? "subagent-transfer" : null, threadId: "thread", targetRunId: runId, toProviderThreadId: "provider", fromProviderThreadIds: ["provider"], coveredRunOrdinals: { from: 1, to: 1 }, status: "ready", - strategy: "full_thread_summary", + strategy: mode === "subagent-result" ? "manual_context" : "full_thread_summary", summaryText: "Synthetic summary", handoffIds: [], }, ] : [], - contextTransfers: [], + contextTransfers: + mode === "subagent-result" ? [{ id: "subagent-transfer", type: "subagent_result" }] : [], providerSessions: [], providerTurns: [], subagents: [], From 26cf033ab76e7ae408c2107881dc4d0ec912c710 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Mon, 19 Oct 2026 21:19:31 -0700 Subject: [PATCH 2/3] fix(server): bound queued subagent result payloads --- .../DelegatedCompletionDelivery.test.ts | 82 +++++++++++++++++++ .../src/orchestration-v2/Orchestrator.ts | 15 +++- 2 files changed, 94 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index d0aff5d573d4..1f07b9b24f38 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -674,6 +674,88 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { }), ); + it.effect.each([ + { + name: "oversized UTF-8 result", + first: "界".repeat(6_000), + second: "small result", + inlineFirst: false, + inlineSecond: true, + }, + { + name: "combined results", + first: "a".repeat(10_000), + second: "b".repeat(10_000), + inlineFirst: true, + inlineSecond: false, + }, + ])("uses task_status for $name beyond the inline budget", (testCase) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make(`thread:inline-budget:${testCase.name}`); + const runId = RunId.make(`run:inline-budget:${testCase.name}`); + const firstTaskId = NodeId.make(`node:inline-first:${testCase.name}`); + const secondTaskId = NodeId.make(`node:inline-second:${testCase.name}`); + const messageId = MessageId.make(`message:delegated-delivery:${threadId}`); + yield* seedParentWithTerminalTask({ + threadId, + projectId: ProjectId.make(`project:inline-budget:${testCase.name}`), + runId, + rootNodeId: NodeId.make(`node:inline-budget:${testCase.name}`), + taskId: firstTaskId, + deliveryState: "claimed", + completionWake: "always", + deliveryTaskIds: [firstTaskId, secondTaskId], + now, + }); + const parent = yield* orchestrator.getThreadProjection(threadId); + const task = parent.subagents[0]!; + yield* sink.write({ + events: [ + { id: firstTaskId, result: testCase.first }, + { id: secondTaskId, result: testCase.second }, + ].map(({ id, result }) => ({ + id: EventId.make(`event:inline-budget:${id}`), + type: "subagent.updated" as const, + threadId, + runId, + nodeId: id, + occurredAt: now, + payload: { ...task, id, result }, + })), + }); + yield* orchestrator.dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`command:inline-budget:${testCase.name}`), + threadId, + messageId, + text: "Delegated tasks completed.", + attachments: [], + dispatchMode: { type: "queue_after_active" }, + createdBy: "agent", + creationSource: "server", + delegatedCompletion: { + parentRunId: runId, + generation: 1, + taskIds: [firstTaskId, secondTaskId], + }, + }); + const projection = yield* orchestrator.getThreadProjection(threadId); + const message = projection.messages.find((candidate) => candidate.id === messageId)!; + assert.equal(message.text.includes(testCase.first), testCase.inlineFirst); + assert.equal(message.text.includes(testCase.second), testCase.inlineSecond); + const deferredTaskId = testCase.inlineFirst ? secondTaskId : firstTaskId; + assert.include(message.text, `Use task_status with taskId ${deferredTaskId}`); + assert.deepEqual(message.delegatedCompletion?.taskIds, [firstTaskId, secondTaskId]); + assert.equal( + projection.subagents.find((candidate) => candidate.id === deferredTaskId)?.result, + testCase.inlineFirst ? testCase.second : testCase.first, + ); + }), + ); + it.effect("does not re-offer when wake-policy upgrades after delivered ownership settled", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 34a84cb1e056..f056c5057aef 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -590,16 +590,25 @@ function legacyPullRequestLink( ); } +const DELEGATED_RESULT_INLINE_BYTES = 16_000; + function delegatedCompletionWakeDetail( taskIds: ReadonlyArray, tasks: ReadonlyArray, ): string { + let remainingBytes = DELEGATED_RESULT_INLINE_BYTES; return taskIds .map((taskId) => { const task = tasks.find((candidate) => candidate.id === taskId); - return task?.result == null - ? `Delegated task ${taskId} reached a terminal state. Use task_status with taskId ${taskId} to read the result.` - : `Delegated task ${taskId}${task.title === null ? "" : ` (${task.title})`} [${task.status}]:\n${task.result}`; + if (task?.result != null) { + const detail = `Delegated task ${taskId}${task.title === null ? "" : ` (${task.title})`} [${task.status}]:\n${task.result}`; + const bytes = Buffer.byteLength(detail); + if (bytes <= remainingBytes) { + remainingBytes -= bytes; + return detail; + } + } + return `Delegated task ${taskId} reached a terminal state. Use task_status with taskId ${taskId} to read the result.`; }) .join("\n\n"); } From 4a5d56a9b65a2af852e2fe8f524ec5d32895a03d Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Mon, 19 Oct 2026 22:08:32 -0700 Subject: [PATCH 3/3] fix(server): retain task status in deferred result messages --- .../DelegatedCompletionDelivery.test.ts | 12 ++++++++---- apps/server/src/orchestration-v2/Orchestrator.ts | 5 +++-- 2 files changed, 11 insertions(+), 6 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 1f07b9b24f38..18b80136e9d9 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -714,16 +714,16 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { const task = parent.subagents[0]!; yield* sink.write({ events: [ - { id: firstTaskId, result: testCase.first }, - { id: secondTaskId, result: testCase.second }, - ].map(({ id, result }) => ({ + { id: firstTaskId, result: testCase.first, status: "failed" as const }, + { id: secondTaskId, result: testCase.second, status: "cancelled" as const }, + ].map(({ id, result, status }) => ({ id: EventId.make(`event:inline-budget:${id}`), type: "subagent.updated" as const, threadId, runId, nodeId: id, occurredAt: now, - payload: { ...task, id, result }, + payload: { ...task, id, result, status, title: "Inspect regression" }, })), }); yield* orchestrator.dispatch({ @@ -747,6 +747,10 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { assert.equal(message.text.includes(testCase.first), testCase.inlineFirst); assert.equal(message.text.includes(testCase.second), testCase.inlineSecond); const deferredTaskId = testCase.inlineFirst ? secondTaskId : firstTaskId; + assert.include( + message.text, + `Delegated task ${deferredTaskId} (Inspect regression) [${testCase.inlineFirst ? "cancelled" : "failed"}]`, + ); assert.include(message.text, `Use task_status with taskId ${deferredTaskId}`); assert.deepEqual(message.delegatedCompletion?.taskIds, [firstTaskId, secondTaskId]); assert.equal( diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index f056c5057aef..582baf89a848 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -600,15 +600,16 @@ function delegatedCompletionWakeDetail( return taskIds .map((taskId) => { const task = tasks.find((candidate) => candidate.id === taskId); + const label = `Delegated task ${taskId}${task?.title == null ? "" : ` (${task.title})`}${task === undefined ? " reached a terminal state" : ` [${task.status}]`}`; if (task?.result != null) { - const detail = `Delegated task ${taskId}${task.title === null ? "" : ` (${task.title})`} [${task.status}]:\n${task.result}`; + const detail = `${label}:\n${task.result}`; const bytes = Buffer.byteLength(detail); if (bytes <= remainingBytes) { remainingBytes -= bytes; return detail; } } - return `Delegated task ${taskId} reached a terminal state. Use task_status with taskId ${taskId} to read the result.`; + return `${label}. Use task_status with taskId ${taskId} to read the result.`; }) .join("\n\n"); }