Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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");
}),
);

Expand Down Expand Up @@ -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;
Expand Down
88 changes: 21 additions & 67 deletions apps/server/src/orchestration-v2/Orchestrator.ts
Original file line number Diff line number Diff line change
Expand Up @@ -590,11 +590,18 @@ function legacyPullRequestLink(
);
}

function delegatedCompletionWakeDetail(taskIds: ReadonlyArray<string>): 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<string>,
tasks: ReadonlyArray<OrchestrationV2Subagent>,
): 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");
Comment thread
coderabbitai[bot] marked this conversation as resolved.
}

function isTerminalDelegatedTaskStatus(status: OrchestrationV2Subagent["status"]): boolean {
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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",
Expand All @@ -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,
Expand Down Expand Up @@ -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,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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));
Expand Down
8 changes: 8 additions & 0 deletions apps/server/src/orchestration-v2/ProviderTurnStartService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 &&
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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([
Expand Down Expand Up @@ -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: [],
},
],
Expand All @@ -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: [],
Expand Down
Loading