From f45888d13acc47fa29427a2915a0c3c64fee5e87 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 26 Oct 2026 21:35:09 +0000 Subject: [PATCH 1/4] test(server): acknowledging a child result can revert a checkpointed run to waiting Red test for pingdotgg/t3code#24322. A delegated completion acknowledge plans run.updated from the parent run it read while waiting. When checkpoint capture commits the run as completed before the acknowledge commits, the full stale payload overwrites it, leaving the run waiting with a checkpoint. Co-Authored-By: Claude Opus 5.5 --- .../DelegatedCompletionDelivery.test.ts | 96 +++++++++++++++++++ 1 file changed, 96 insertions(+) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 3bf563c1f794a..4b5b1538f008a 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -1140,6 +1140,102 @@ const seedRestartCancelledChild = (input: { it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( "subagent follow-ups", (it) => { + it.effect( + "keeps a checkpoint completion committed between an acknowledge's read and write", + () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:ack-capture-race"); + const runId = RunId.make("run:ack-capture-race"); + const taskId = NodeId.make("node:ack-capture-race-task"); + const checkpointId = CheckpointId.make("checkpoint:ack-capture-race"); + yield* seedParentWithTerminalTask({ + threadId, + projectId: ProjectId.make("project:ack-capture-race"), + runId, + rootNodeId: NodeId.make("node:ack-capture-race-root"), + taskId, + deliveryState: "delivered", + completionWake: "always", + deliveryTaskIds: [taskId], + now, + }); + const parent = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!; + const waiting = { ...parent, status: "waiting" as const }; + yield* sink.write({ + events: [ + { + id: EventId.make("event:ack-capture-waiting"), + type: "run.updated", + threadId, + runId, + occurredAt: now, + payload: waiting, + }, + ], + }); + const commit = sink.commitCommand; + let captured = false; + yield* Effect.acquireUseRelease( + Effect.sync(() => + vi.spyOn(sink, "commitCommand").mockImplementation((input) => + Effect.gen(function* () { + if ( + !captured && + input.commandType === "delegated_task.completion-delivery.acknowledge" + ) { + captured = true; + // Commit capture after the acknowledge planned its stale waiting snapshot. + const { delegatedCompletion: _cohort, ...captureRun } = waiting; + yield* commit({ + commandId: CommandId.make("command:ack-capture-race"), + threadId, + commandType: "checkpoint.capture", + acceptedAt: now, + effects: [], + events: [ + { + id: EventId.make("event:ack-capture-completed"), + type: "run.updated", + threadId, + runId, + occurredAt: now, + payload: { + ...captureRun, + status: "completed", + completedAt: now, + checkpointId, + }, + }, + ], + }); + } + return yield* commit(input); + }), + ), + ), + () => + orchestrator.dispatch({ + type: "delegated_task.completion-delivery.acknowledge", + commandId: CommandId.make("command:ack-capture-race:ack"), + parentThreadId: threadId, + taskId, + observedByRunId: runId, + }), + (spy) => Effect.sync(() => spy.mockRestore()), + ); + assert.isTrue(captured); + const finished = yield* orchestrator.getThreadProjection(threadId); + const completed = finished.runs.find((run) => run.id === runId)!; + assert.equal(completed.status, "completed"); + assert.deepEqual(completed.completedAt, now); + assert.equal(completed.checkpointId, checkpointId); + assert.isNull(completed.delegatedCompletion?.delivery); + }), + ); + it.effect("keeps a checkpoint completion committed between finalize's read and write", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; From ec6c96f0c323aa08917f3ea2529afd828aee045d Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 26 Oct 2026 21:40:55 +0000 Subject: [PATCH 2/4] fix(server): acknowledging a child result no longer reverts a checkpointed run to waiting Delegated completion acknowledge, dispose, and wake-policy commands plan the parent run's cohort update from a run snapshot read before commit. When checkpoint capture committed the run as completed in between, the stale full payload overwrote it, leaving the run waiting and its queue stuck. commitCommand now takes the run ids whose run.updated only changes delegatedCompletion and rebuilds those from the stored row inside the commit transaction, sharing the merge writeWithEffects already uses. It is scoped by run id because an acknowledge can also cancel a queued delivery run in the same batch. Fixes pingdotgg/t3code#24322 Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/EventSink.ts | 59 +++++++++++-------- .../src/orchestration-v2/Orchestrator.ts | 19 +++++- 2 files changed, 52 insertions(+), 26 deletions(-) diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index af8fa5a95af7b..d882859916140 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -142,6 +142,8 @@ export interface EventSinkV2Shape { readonly acceptedAt: DateTime.Utc; readonly events: ReadonlyArray; readonly effects: ReadonlyArray; + /** Runs whose run.updated is rebuilt as in writeWithEffects' updateDelegatedCompletionOnly. */ + readonly delegatedCompletionOnlyRunIds?: ReadonlyArray; readonly cancelUnsettledEffects?: { readonly effectTypes: ReadonlyArray; readonly reason: string; @@ -383,6 +385,31 @@ const layerBase: Layer.Layer< return yield* runOutcomes.prepare(storedEvents); }); + // Checkpoint capture can finish after a cohort writer reads its parent run. + // Merge in this transaction so the event log and projection keep that completion. + const keepStoredRunLifecycle = ( + events: ReadonlyArray, + isCohortOnly: (runId: RunId) => boolean, + ) => + Effect.forEach(events, (event) => + Effect.gen(function* () { + if (event.type !== "run.updated" || !isCohortOnly(event.payload.id)) return event; + const { runs } = yield* projectionStore.getThreadRecords(event.threadId, ["runs"], { + runIds: [event.payload.id], + }); + const current = runs[0]; + return current === undefined + ? event + : { + ...event, + payload: { + ...current, + delegatedCompletion: event.payload.delegatedCompletion, + }, + }; + }), + ); + const writeEffect = Effect.fn("orchestrationV2.EventSink.write")(function* ( input: Parameters[0], ) { @@ -408,29 +435,8 @@ const layerBase: Layer.Layer< `; if (owners.length > 0) return { committed: [], outcomes: [] }; } - // Checkpoint capture can finish after a cohort writer reads its parent run. - // Merge in this transaction so the event log and projection keep that completion. const events = input.updateDelegatedCompletionOnly - ? yield* Effect.forEach(input.events, (event) => - Effect.gen(function* () { - if (event.type !== "run.updated") return event; - const { runs } = yield* projectionStore.getThreadRecords( - event.threadId, - ["runs"], - { runIds: [event.payload.id] }, - ); - const current = runs[0]; - return current === undefined - ? event - : { - ...event, - payload: { - ...current, - delegatedCompletion: event.payload.delegatedCompletion, - }, - }; - }), - ) + ? yield* keepStoredRunLifecycle(input.events, () => true) : input.events; const normalized = yield* normalizeEvents( input.guardPendingUserInputCancellations === true @@ -646,7 +652,14 @@ const layerBase: Layer.Layer< return { ...existing, committed: false as const, cancelledEffectIds: [] }; } - const normalized = yield* normalizeEvents(input.events); + const cohortOnlyRunIds = input.delegatedCompletionOnlyRunIds ?? []; + const normalized = yield* normalizeEvents( + cohortOnlyRunIds.length === 0 + ? input.events + : yield* keepStoredRunLifecycle(input.events, (runId) => + cohortOnlyRunIds.includes(runId), + ), + ); const storedEvents = yield* eventStore.append({ commandId: input.commandId, events: normalized, diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 0c6ca268f1e04..4fec0bfd832f6 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -2358,7 +2358,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio run: deliveryRun, now, }); - return; + return parentRun.id; } const message = completionDeliveryMessage(projection, delivery); if (message !== undefined) { @@ -2382,6 +2382,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); } } + return parentRun.id; }); const disposeDelegatedCompletionCohort = Effect.fn( @@ -7659,6 +7660,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio payload: completionPlan.message, }); } + return completionPlan.parentRun?.id; }); const dispatchCreatedThreadRecord = Effect.fn("orchestrationV2.dispatch.createdThreadRecord")( @@ -11752,6 +11754,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio readonly effectTypes: ReadonlyArray; readonly reason: string; }; + readonly delegatedCompletionOnlyRunIds?: ReadonlyArray; }, OrchestratorV2Error > { @@ -11769,6 +11772,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio readonly reason: string; } | undefined; + let delegatedCompletionOnlyRunId: RunId | undefined; switch (command.type) { case "thread.create": yield* dispatchThreadCreate(command, events); @@ -12076,11 +12080,14 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio yield* dispatchDelegatedTaskRequest(command, events, effects); break; case "delegated_task.wake-policy": - yield* dispatchDelegatedTaskWakePolicy(command, events); + delegatedCompletionOnlyRunId = yield* dispatchDelegatedTaskWakePolicy(command, events); break; case "delegated_task.completion-delivery.acknowledge": case "delegated_task.completion-delivery.dispose": - yield* dispatchDelegatedTaskCompletionDeliveryResolution(command, events); + delegatedCompletionOnlyRunId = yield* dispatchDelegatedTaskCompletionDeliveryResolution( + command, + events, + ); break; case "thread.created.record": yield* dispatchCreatedThreadRecord(command, events); @@ -12095,6 +12102,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio events: yield* Ref.get(events), effects: yield* Ref.get(effects), ...(cancelUnsettledEffects === undefined ? {} : { cancelUnsettledEffects }), + ...(delegatedCompletionOnlyRunId === undefined + ? {} + : { delegatedCompletionOnlyRunIds: [delegatedCompletionOnlyRunId] }), }; }); @@ -12271,6 +12281,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...(plan.cancelUnsettledEffects === undefined ? {} : { cancelUnsettledEffects: plan.cancelUnsettledEffects }), + ...(plan.delegatedCompletionOnlyRunIds === undefined + ? {} + : { delegatedCompletionOnlyRunIds: plan.delegatedCompletionOnlyRunIds }), }) .pipe( Effect.mapError( From 2df519ecef4f77d7e0e85408bd3c9112ccde8f88 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 27 Oct 2026 06:05:36 +0000 Subject: [PATCH 3/4] test(server): takeover tests: wake-policy can revert a checkpointed run to running Drop the PR's acknowledge race test, which duplicates the coverage #25672 landed on main, and add a wake-policy case to main's parametrised settle race test. The case upgrades a terminal task to completionWake "always" while its parent run is live and commits the checkpoint completion between the command's read and commit; the run must keep "completed", completedAt and checkpointId while taking the planned delegatedCompletion. Each case now also asserts the planned delegatedCompletion is what lands. Co-Authored-By: Claude Opus 5.5 --- .../DelegatedCompletionDelivery.test.ts | 188 ++++++------------ 1 file changed, 59 insertions(+), 129 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index f5473f76b0c73..463077e35f319 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -147,7 +147,7 @@ const seedParentWithTerminalTask = (input: { readonly runId: RunId; readonly rootNodeId: NodeId; readonly taskId: NodeId; - readonly deliveryState: "delivered" | "claimed" | "acknowledged" | "disposed"; + readonly deliveryState: "pending" | "delivered" | "claimed" | "acknowledged" | "disposed"; readonly completionWake?: "always" | "settled_only"; readonly deliveryTaskIds?: ReadonlyArray; readonly now: DateTime.Utc; @@ -976,34 +976,38 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { ); it.effect.each([ - { before: "running", after: "failed", capture: false }, - { before: "waiting", after: "completed", capture: true }, + { name: "ack-failed", before: "running", after: "failed", capture: false, ack: true }, + { name: "ack-completed", before: "waiting", after: "completed", capture: true, ack: true }, + { name: "wake-completed", before: "running", after: "completed", capture: true, ack: false }, ] as const)( - "keeps a $after run that settled between task_status's acknowledge read and commit", - ({ before, after, capture }) => + "keeps a run that settled between a delegated completion update's read and commit ($name)", + ({ name, before, after, capture, ack }) => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; const sink = yield* EventSink.EventSinkV2; const now = yield* DateTime.now; - const threadId = ThreadId.make(`thread:ack-settle-race-${after}`); - const runId = RunId.make(`run:ack-settle-race-${after}`); - const taskId = NodeId.make(`node:ack-settle-race-${after}-task`); - const checkpointId = CheckpointId.make(`checkpoint:ack-settle-race-${after}`); + const threadId = ThreadId.make(`thread:settle-race-${name}`); + const runId = RunId.make(`run:settle-race-${name}`); + const taskId = NodeId.make(`node:settle-race-${name}-task`); + const checkpointId = CheckpointId.make(`checkpoint:settle-race-${name}`); + const commandType = ack + ? "delegated_task.completion-delivery.acknowledge" + : "delegated_task.wake-policy"; yield* seedParentWithTerminalTask({ threadId, - projectId: ProjectId.make(`project:ack-settle-race-${after}`), + projectId: ProjectId.make(`project:settle-race-${name}`), runId, - rootNodeId: NodeId.make(`node:ack-settle-race-${after}-root`), + rootNodeId: NodeId.make(`node:settle-race-${name}-root`), taskId, - deliveryState: "claimed", - deliveryTaskIds: [taskId], + deliveryState: ack ? "claimed" : "pending", + ...(ack ? { deliveryTaskIds: [taskId] } : {}), now, }); const parent = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!; yield* sink.write({ events: [ { - id: EventId.make(`event:ack-settle-race-${after}-before`), + id: EventId.make(`event:settle-race-${name}-before`), type: "run.updated", threadId, runId, @@ -1014,26 +1018,27 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { }); const { delegatedCompletion: _cohort, ...settledRun } = parent; const commit = sink.commitCommand; - let settled = false; + let planned: OrchestrationV2Run["delegatedCompletion"]; yield* Effect.acquireUseRelease( Effect.sync(() => vi.spyOn(sink, "commitCommand").mockImplementation((input) => Effect.gen(function* () { - if ( - !settled && - input.commandType === "delegated_task.completion-delivery.acknowledge" - ) { - settled = true; - // The provider turn ends (or its checkpoint lands) after acknowledge read the run. + if (planned === undefined && input.commandType === commandType) { + const plannedRun = input.events.find((event) => event.type === "run.updated"); + planned = + plannedRun?.type === "run.updated" + ? plannedRun.payload.delegatedCompletion + : undefined; + // The provider turn ends (or its checkpoint lands) after the command read the run. yield* commit({ - commandId: CommandId.make(`command:ack-settle-race-${after}`), + commandId: CommandId.make(`command:settle-race-${name}`), threadId, commandType: capture ? "checkpoint.capture" : "provider.event", acceptedAt: now, effects: [], events: [ { - id: EventId.make(`event:ack-settle-race-${after}-settled`), + id: EventId.make(`event:settle-race-${name}-settled`), type: "run.updated", threadId, runId, @@ -1053,22 +1058,43 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { ), ), () => - orchestrator.dispatch({ - type: "delegated_task.completion-delivery.acknowledge", - commandId: CommandId.make(`command:ack-settle-race-${after}:task-status`), - parentThreadId: threadId, - taskId, - observedByRunId: runId, - }), + orchestrator.dispatch( + ack + ? { + type: "delegated_task.completion-delivery.acknowledge", + commandId: CommandId.make(`command:settle-race-${name}:update`), + parentThreadId: threadId, + taskId, + observedByRunId: runId, + } + : { + type: "delegated_task.wake-policy", + commandId: CommandId.make(`command:settle-race-${name}:update`), + parentThreadId: threadId, + taskId, + completionWake: "always", + }, + ), (spy) => Effect.sync(() => spy.mockRestore()), ); - assert.isTrue(settled); + assert.isDefined(planned); + assert.notDeepEqual(planned, parent.delegatedCompletion); const run = (yield* orchestrator.getThreadProjection(threadId)).runs.find( (candidate) => candidate.id === runId, )!; assert.deepEqual( - { status: run.status, completedAt: run.completedAt, checkpointId: run.checkpointId }, - { status: after, completedAt: now, checkpointId: capture ? checkpointId : null }, + { + status: run.status, + completedAt: run.completedAt, + checkpointId: run.checkpointId, + delegatedCompletion: run.delegatedCompletion, + }, + { + status: after, + completedAt: now, + checkpointId: capture ? checkpointId : null, + delegatedCompletion: planned, + }, ); }), ); @@ -1238,102 +1264,6 @@ const seedRestartCancelledChild = (input: { it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))( "subagent follow-ups", (it) => { - it.effect( - "keeps a checkpoint completion committed between an acknowledge's read and write", - () => - Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const sink = yield* EventSink.EventSinkV2; - const now = yield* DateTime.now; - const threadId = ThreadId.make("thread:ack-capture-race"); - const runId = RunId.make("run:ack-capture-race"); - const taskId = NodeId.make("node:ack-capture-race-task"); - const checkpointId = CheckpointId.make("checkpoint:ack-capture-race"); - yield* seedParentWithTerminalTask({ - threadId, - projectId: ProjectId.make("project:ack-capture-race"), - runId, - rootNodeId: NodeId.make("node:ack-capture-race-root"), - taskId, - deliveryState: "delivered", - completionWake: "always", - deliveryTaskIds: [taskId], - now, - }); - const parent = (yield* orchestrator.getThreadProjection(threadId)).runs[0]!; - const waiting = { ...parent, status: "waiting" as const }; - yield* sink.write({ - events: [ - { - id: EventId.make("event:ack-capture-waiting"), - type: "run.updated", - threadId, - runId, - occurredAt: now, - payload: waiting, - }, - ], - }); - const commit = sink.commitCommand; - let captured = false; - yield* Effect.acquireUseRelease( - Effect.sync(() => - vi.spyOn(sink, "commitCommand").mockImplementation((input) => - Effect.gen(function* () { - if ( - !captured && - input.commandType === "delegated_task.completion-delivery.acknowledge" - ) { - captured = true; - // Commit capture after the acknowledge planned its stale waiting snapshot. - const { delegatedCompletion: _cohort, ...captureRun } = waiting; - yield* commit({ - commandId: CommandId.make("command:ack-capture-race"), - threadId, - commandType: "checkpoint.capture", - acceptedAt: now, - effects: [], - events: [ - { - id: EventId.make("event:ack-capture-completed"), - type: "run.updated", - threadId, - runId, - occurredAt: now, - payload: { - ...captureRun, - status: "completed", - completedAt: now, - checkpointId, - }, - }, - ], - }); - } - return yield* commit(input); - }), - ), - ), - () => - orchestrator.dispatch({ - type: "delegated_task.completion-delivery.acknowledge", - commandId: CommandId.make("command:ack-capture-race:ack"), - parentThreadId: threadId, - taskId, - observedByRunId: runId, - }), - (spy) => Effect.sync(() => spy.mockRestore()), - ); - assert.isTrue(captured); - const finished = yield* orchestrator.getThreadProjection(threadId); - const completed = finished.runs.find((run) => run.id === runId)!; - assert.equal(completed.status, "completed"); - assert.deepEqual(completed.completedAt, now); - assert.equal(completed.checkpointId, checkpointId); - assert.isNull(completed.delegatedCompletion?.delivery); - }), - ); - it.effect("keeps a checkpoint completion committed between finalize's read and write", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; From 5abcaf34fe13cf513b7335b5cedccc5ccd908a30 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 27 Oct 2026 06:08:37 +0000 Subject: [PATCH 4/4] fix(server): changing a task's wake policy no longer reverts a checkpointed parent run When a wake-policy change offers a finished child's result to a live parent run, it plans the run's delegated completion cohort from a snapshot read before commit. If checkpoint capture committed the run as completed in between, the stale full payload set it back to running. Commit only the cohort update, as acknowledge and dispose already do. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/Orchestrator.ts | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index ee8ab83b952aa..41fca70960985 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -7660,6 +7660,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio payload: completionPlan.message, }); } + return completionPlan.parentRun?.id; }); const dispatchCreatedThreadRecord = Effect.fn("orchestrationV2.dispatch.createdThreadRecord")( @@ -12079,7 +12080,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio yield* dispatchDelegatedTaskRequest(command, events, effects); break; case "delegated_task.wake-policy": - yield* dispatchDelegatedTaskWakePolicy(command, events); + updateDelegatedCompletionOnlyRunId = yield* dispatchDelegatedTaskWakePolicy( + command, + events, + ); break; case "delegated_task.completion-delivery.acknowledge": case "delegated_task.completion-delivery.dispose":