Skip to content

Commit b8a2eea

Browse files
murayymurayy
committed
fix(server): a promoted follow-up follows its turn's restart continuation
A restart can cut the turn a follow-up was steered into and resume it in a continuation run. The promoted task now resolves to that continuation, so it stays open while the turn resumes and settles with the continuation's result.
1 parent 16ca007 commit b8a2eea

3 files changed

Lines changed: 223 additions & 133 deletions

File tree

‎apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts‎

Lines changed: 195 additions & 130 deletions
Original file line numberDiff line numberDiff line change
@@ -1838,123 +1838,182 @@ it.layer(layerTest)("delegated tasks across a server restart", (it) => {
18381838
);
18391839
});
18401840

1841-
it.layer(layerTest)("delegated follow-ups promoted to Steer", (it) => {
1842-
it.effect("settles a promoted follow-up with the turn it steered, not as cancelled", () =>
1843-
Effect.gen(function* () {
1844-
const orchestrator = yield* Orchestrator.OrchestratorV2;
1845-
const eventSink = yield* EventSink.EventSinkV2;
1846-
const now = yield* DateTime.now;
1847-
const threadId = ThreadId.make("thread:steer-parent");
1848-
const projectId = ProjectId.make("project:steer-parent");
1849-
const runId = RunId.make("run:steer-parent");
1850-
const rootNodeId = NodeId.make("node:steer-parent-root");
1851-
yield* seedParentWithTerminalTask({
1852-
threadId,
1853-
projectId,
1854-
runId,
1855-
rootNodeId,
1856-
taskId: NodeId.make("node:steer-parent-settled"),
1857-
deliveryState: "delivered",
1858-
now,
1859-
});
1860-
const child = yield* seedRestartCancelledChild({
1861-
parentThreadId: threadId,
1862-
projectId,
1863-
parentRunId: runId,
1864-
rootNodeId,
1865-
name: "steer-child",
1866-
completionWake: "always",
1867-
continuationPending: false,
1868-
runStatus: "completed",
1869-
now,
1870-
});
1871-
const followUpTask = (taskId: NodeId, prompt: string) => ({
1872-
id: EventId.make(`event:${taskId}`),
1873-
type: "subagent.updated" as const,
1841+
/**
1842+
* A child whose second follow-up was queued as its own task, then promoted to
1843+
* Steer into the first follow-up's running turn.
1844+
*/
1845+
const seedPromotedFollowUp = (name: string) =>
1846+
Effect.gen(function* () {
1847+
const eventSink = yield* EventSink.EventSinkV2;
1848+
const now = yield* DateTime.now;
1849+
const threadId = ThreadId.make(`thread:${name}-parent`);
1850+
const runId = RunId.make(`run:${name}-parent`);
1851+
const rootNodeId = NodeId.make(`node:${name}-parent-root`);
1852+
const projectId = ProjectId.make(`project:${name}-parent`);
1853+
yield* seedParentWithTerminalTask({
1854+
threadId,
1855+
projectId,
1856+
runId,
1857+
rootNodeId,
1858+
taskId: NodeId.make(`node:${name}-parent-settled`),
1859+
deliveryState: "delivered",
1860+
now,
1861+
});
1862+
const child = yield* seedRestartCancelledChild({
1863+
parentThreadId: threadId,
1864+
projectId,
1865+
parentRunId: runId,
1866+
rootNodeId,
1867+
name,
1868+
completionWake: "always",
1869+
continuationPending: false,
1870+
runStatus: "completed",
1871+
now,
1872+
});
1873+
const followUpTask = (taskId: NodeId, prompt: string) => ({
1874+
id: EventId.make(`event:${taskId}`),
1875+
type: "subagent.updated" as const,
1876+
threadId,
1877+
runId,
1878+
nodeId: taskId,
1879+
driver,
1880+
providerInstanceId: modelSelection.instanceId,
1881+
occurredAt: now,
1882+
payload: {
1883+
id: taskId,
18741884
threadId,
18751885
runId,
1876-
nodeId: taskId,
1886+
parentNodeId: rootNodeId,
1887+
origin: "app_owned" as const,
1888+
createdBy: "agent" as const,
18771889
driver,
18781890
providerInstanceId: modelSelection.instanceId,
1879-
occurredAt: now,
1891+
providerThreadId: null,
1892+
childThreadId: child.childThreadId,
1893+
nativeTaskRef: null,
1894+
prompt,
1895+
title: null,
1896+
model: null,
1897+
completionWake: "always" as const,
1898+
status: "pending" as const,
1899+
result: null,
1900+
startedAt: now,
1901+
completedAt: null,
1902+
updatedAt: now,
1903+
},
1904+
});
1905+
const steeredRunId = RunId.make(`run:${name}:2`);
1906+
const steeredTaskId = NodeId.make(`node:${name}-steered`);
1907+
const promotedTaskId = NodeId.make(`node:${name}-promoted`);
1908+
const promotedMessageId = MessageId.make(`message:${name}-promoted`);
1909+
const promotedRun = runEvent({
1910+
threadId: child.childThreadId,
1911+
runId: RunId.make(`run:${name}:3`),
1912+
ordinal: 3,
1913+
status: "cancelled",
1914+
userMessageId: promotedMessageId,
1915+
now,
1916+
});
1917+
const steeredRun = (
1918+
status: OrchestrationV2Run["status"],
1919+
id = steeredRunId,
1920+
ordinal = 2,
1921+
restartContinuationOfRunId?: RunId,
1922+
) => {
1923+
const event = runEvent({ threadId: child.childThreadId, runId: id, ordinal, status, now });
1924+
return {
1925+
...event,
18801926
payload: {
1881-
id: taskId,
1882-
threadId,
1883-
runId,
1884-
parentNodeId: rootNodeId,
1885-
origin: "app_owned" as const,
1886-
createdBy: "agent" as const,
1887-
driver,
1888-
providerInstanceId: modelSelection.instanceId,
1889-
providerThreadId: null,
1890-
childThreadId: child.childThreadId,
1891-
nativeTaskRef: null,
1892-
prompt,
1893-
title: null,
1894-
model: null,
1895-
completionWake: "always" as const,
1896-
status: "pending" as const,
1897-
result: null,
1898-
startedAt: now,
1899-
completedAt: null,
1900-
updatedAt: now,
1927+
...event.payload,
1928+
delegatedTaskId: steeredTaskId,
1929+
...(restartContinuationOfRunId === undefined ? {} : { restartContinuationOfRunId }),
19011930
},
1902-
});
1903-
const steeredRunId = RunId.make("run:steer-child:2");
1904-
const steeredTaskId = NodeId.make("node:steer-child-steered");
1905-
const promotedTaskId = NodeId.make("node:steer-child-promoted");
1906-
const promotedMessageId = MessageId.make("message:steer-child-promoted");
1907-
const steeredRun = runEvent({
1908-
threadId: child.childThreadId,
1909-
runId: steeredRunId,
1910-
ordinal: 2,
1911-
status: "running",
1912-
now,
1913-
});
1914-
const promotedRun = runEvent({
1915-
threadId: child.childThreadId,
1916-
runId: RunId.make("run:steer-child:3"),
1917-
ordinal: 3,
1918-
status: "cancelled",
1919-
userMessageId: promotedMessageId,
1920-
now,
1921-
});
1922-
// Both follow-ups were queued as their own tasks. The second was promoted into the first's turn.
1923-
yield* eventSink.write({
1924-
commandId: CommandId.make("command:steer-child:promoted"),
1925-
events: [
1926-
followUpTask(steeredTaskId, "Merge the fixture branch."),
1927-
followUpTask(promotedTaskId, "Also merge the pricing branch."),
1928-
{ ...steeredRun, payload: { ...steeredRun.payload, delegatedTaskId: steeredTaskId } },
1929-
{
1930-
...promotedRun,
1931-
payload: { ...promotedRun.payload, startedAt: null, delegatedTaskId: promotedTaskId },
1931+
};
1932+
};
1933+
yield* eventSink.write({
1934+
commandId: CommandId.make(`command:${name}:promoted`),
1935+
events: [
1936+
followUpTask(steeredTaskId, "Merge the fixture branch."),
1937+
followUpTask(promotedTaskId, "Also merge the pricing branch."),
1938+
steeredRun("running"),
1939+
{
1940+
...promotedRun,
1941+
payload: { ...promotedRun.payload, startedAt: null, delegatedTaskId: promotedTaskId },
1942+
},
1943+
{
1944+
id: EventId.make(`event:${name}:promoted-message`),
1945+
type: "message.updated",
1946+
threadId: child.childThreadId,
1947+
runId: steeredRunId,
1948+
occurredAt: now,
1949+
payload: {
1950+
id: promotedMessageId,
1951+
threadId: child.childThreadId,
1952+
runId: steeredRunId,
1953+
nodeId: null,
1954+
role: "user",
1955+
text: "Also merge the pricing branch.",
1956+
attachments: [],
1957+
streaming: false,
1958+
createdBy: "agent",
1959+
creationSource: "mcp",
1960+
createdAt: now,
1961+
updatedAt: now,
19321962
},
1963+
},
1964+
],
1965+
});
1966+
/** Ends `resultRunId` with an assistant reply; `run` is that run's terminal event. */
1967+
const finish = (resultRunId: RunId, run: ReturnType<typeof steeredRun>) =>
1968+
eventSink.write({
1969+
commandId: CommandId.make(`command:${resultRunId}:completed`),
1970+
events: [
19331971
{
1934-
id: EventId.make("event:steer-child:promoted-message"),
1972+
id: EventId.make(`event:${resultRunId}:result`),
19351973
type: "message.updated",
19361974
threadId: child.childThreadId,
1937-
runId: steeredRunId,
1975+
runId: resultRunId,
19381976
occurredAt: now,
19391977
payload: {
1940-
id: promotedMessageId,
1978+
id: MessageId.make(`message:${resultRunId}:result`),
19411979
threadId: child.childThreadId,
1942-
runId: steeredRunId,
1980+
runId: resultRunId,
19431981
nodeId: null,
1944-
role: "user",
1945-
text: "Also merge the pricing branch.",
1982+
role: "assistant",
1983+
text: "Merged both branches.",
19461984
attachments: [],
19471985
streaming: false,
19481986
createdBy: "agent",
1949-
creationSource: "mcp",
1987+
creationSource: "server",
19501988
createdAt: now,
19511989
updatedAt: now,
19521990
},
19531991
},
1992+
{ ...run, id: EventId.make(`event:${resultRunId}:finished`) },
19541993
],
19551994
});
1995+
return {
1996+
threadId,
1997+
child,
1998+
steeredRunId,
1999+
steeredRun,
2000+
steeredTaskId,
2001+
promotedTaskId,
2002+
promotedRunId: promotedRun.payload.id,
2003+
finish,
2004+
};
2005+
});
2006+
2007+
it.layer(layerTest)("delegated follow-ups promoted to Steer", (it) => {
2008+
it.effect("settles a promoted follow-up with the turn it steered, not as cancelled", () =>
2009+
Effect.gen(function* () {
2010+
const orchestrator = yield* Orchestrator.OrchestratorV2;
2011+
const eventSink = yield* EventSink.EventSinkV2;
2012+
const now = yield* DateTime.now;
2013+
const seeded = yield* seedPromotedFollowUp("steer-child");
2014+
const { threadId, child, promotedTaskId, steeredTaskId } = seeded;
19562015

1957-
yield* orchestrator.recoverDelegatedTask(child.childThreadId, promotedRun.payload.id);
2016+
yield* orchestrator.recoverDelegatedTask(child.childThreadId, seeded.promotedRunId);
19582017

19592018
const steering = yield* orchestrator.getThreadProjection(threadId);
19602019
assert.equal(steering.subagents.find((row) => row.id === promotedTaskId)?.status, "pending");
@@ -1991,43 +2050,8 @@ it.layer(layerTest)("delegated follow-ups promoted to Steer", (it) => {
19912050
},
19922051
],
19932052
});
1994-
yield* eventSink.write({
1995-
commandId: CommandId.make("command:steer-child:completed"),
1996-
events: [
1997-
{
1998-
id: EventId.make("event:steer-child:result"),
1999-
type: "message.updated",
2000-
threadId: child.childThreadId,
2001-
runId: steeredRunId,
2002-
occurredAt: now,
2003-
payload: {
2004-
id: MessageId.make("message:steer-child:result"),
2005-
threadId: child.childThreadId,
2006-
runId: steeredRunId,
2007-
nodeId: null,
2008-
role: "assistant",
2009-
text: "Merged both branches.",
2010-
attachments: [],
2011-
streaming: false,
2012-
createdBy: "agent",
2013-
creationSource: "server",
2014-
createdAt: now,
2015-
updatedAt: now,
2016-
},
2017-
},
2018-
{
2019-
...steeredRun,
2020-
id: EventId.make("event:steer-child:steered-completed"),
2021-
payload: {
2022-
...steeredRun.payload,
2023-
delegatedTaskId: steeredTaskId,
2024-
status: "completed" as const,
2025-
completedAt: now,
2026-
},
2027-
},
2028-
],
2029-
});
2030-
yield* orchestrator.recoverDelegatedTask(child.childThreadId, steeredRunId);
2053+
yield* seeded.finish(seeded.steeredRunId, seeded.steeredRun("completed"));
2054+
yield* orchestrator.recoverDelegatedTask(child.childThreadId, seeded.steeredRunId);
20312055

20322056
const settled = yield* orchestrator.getThreadProjection(threadId);
20332057
const task = (id: NodeId) => settled.subagents.find((row) => row.id === id);
@@ -2039,4 +2063,45 @@ it.layer(layerTest)("delegated follow-ups promoted to Steer", (it) => {
20392063
);
20402064
}),
20412065
);
2066+
2067+
it.effect("keeps a promoted follow-up open while a restart continues its turn", () =>
2068+
Effect.gen(function* () {
2069+
const orchestrator = yield* Orchestrator.OrchestratorV2;
2070+
const eventSink = yield* EventSink.EventSinkV2;
2071+
const seeded = yield* seedPromotedFollowUp("steer-restart-child");
2072+
const { threadId, child, promotedTaskId, steeredRunId } = seeded;
2073+
const continuationRunId = RunId.make("run:steer-restart-child:4");
2074+
// A restart cut the steered turn, and its continuation resumed it.
2075+
yield* eventSink.write({
2076+
commandId: reconcileCommandId("steer-restart-child-cut"),
2077+
events: [seeded.steeredRun("cancelled")],
2078+
});
2079+
yield* eventSink.write({
2080+
commandId: CommandId.make("command:steer-restart-child:continued"),
2081+
events: [seeded.steeredRun("running", continuationRunId, 4, steeredRunId)],
2082+
});
2083+
2084+
yield* orchestrator.recoverDelegatedTask(child.childThreadId, steeredRunId);
2085+
2086+
const continuing = yield* orchestrator.getThreadProjection(threadId);
2087+
assert.equal(
2088+
continuing.subagents.find((row) => row.id === promotedTaskId)?.status,
2089+
"pending",
2090+
);
2091+
assert.isTrue(
2092+
yield* orchestrator.delegatedTaskResultPending(child.childThreadId, promotedTaskId),
2093+
);
2094+
2095+
yield* seeded.finish(
2096+
continuationRunId,
2097+
seeded.steeredRun("completed", continuationRunId, 4, steeredRunId),
2098+
);
2099+
yield* orchestrator.recoverDelegatedTask(child.childThreadId, continuationRunId);
2100+
2101+
const settled = yield* orchestrator.getThreadProjection(threadId);
2102+
const promoted = settled.subagents.find((row) => row.id === promotedTaskId);
2103+
assert.equal(promoted?.status, "completed");
2104+
assert.equal(promoted?.result, "Merged both branches.");
2105+
}),
2106+
);
20422107
});

‎apps/server/src/orchestration-v2/SubagentProjection.test.ts‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -219,6 +219,23 @@ it("settles a follow-up promoted to Steer with the run its message joined", () =
219219
)?.id,
220220
steered.id,
221221
);
222+
// A restart cut the steered turn; its continuation now carries the result.
223+
const continuation: OrchestrationV2Run = {
224+
...run,
225+
id: RunId.make("run:continuation"),
226+
ordinal: 3,
227+
restartContinuationOfRunId: steered.id,
228+
};
229+
assert.equal(
230+
delegatedTaskResultRun(
231+
{
232+
runs: [{ ...steered, status: "cancelled" }, queued, continuation],
233+
messages: [{ ...message, runId: steered.id }],
234+
},
235+
taskId,
236+
)?.id,
237+
continuation.id,
238+
);
222239
});
223240

224241
it("waits for nested work and retains the report across monitor acknowledgements", () => {

0 commit comments

Comments
 (0)