From 8cc978398161a27a7bf8ca13e6f3eb00f1f23d57 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 26 Oct 2026 21:34:55 +0000 Subject: [PATCH 1/3] test(server): reproduce issue 25000 run stuck waiting after delivery accept Reproduces pingdotgg/t3code#25000: notification.delivery.accept plans a run.updated from the parent's waiting snapshot; when checkpoint capture commits completed in between, the accept overwrites it back to waiting with no checkpoint, so the run blocks its queue forever. Co-Authored-By: Claude Opus 5.5 --- .../DelegatedCompletionDelivery.test.ts | 116 ++++++++++++++++++ 1 file changed, 116 insertions(+) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index 3bf563c1f794a..75699b3384377 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -623,6 +623,122 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { }), ); + it.effect("keeps a checkpoint completion committed while accepting a delivery", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:accept-capture-race"); + const runId = RunId.make("run:accept-capture-race"); + const taskId = NodeId.make("node:accept-capture-race-task"); + const checkpointId = CheckpointId.make("checkpoint:accept-capture-race"); + const messageId = MessageId.make(`message:delegated-delivery:${threadId}`); + yield* seedParentWithTerminalTask({ + threadId, + runId, + projectId: ProjectId.make("project:accept-capture-race"), + rootNodeId: NodeId.make("node:accept-capture-race-root"), + taskId, + deliveryState: "claimed", + completionWake: "always", + deliveryTaskIds: [taskId], + now, + }); + const projection = yield* orchestrator.getThreadProjection(threadId); + const waiting = { ...projection.runs[0]!, status: "waiting" as const }; + const task = projection.subagents[0]!; + yield* sink.write({ + events: [ + { + id: EventId.make("event:accept-capture-race-waiting"), + type: "run.updated", + threadId, + runId, + occurredAt: now, + payload: waiting, + }, + { + id: EventId.make("event:accept-capture-race-message"), + type: "message.updated", + threadId, + runId, + occurredAt: now, + payload: { + id: messageId, + threadId, + runId, + nodeId: task.parentNodeId, + role: "user", + text: "Background task finished", + attachments: [], + streaming: false, + createdBy: "agent", + creationSource: "server", + createdAt: now, + updatedAt: now, + delegatedCompletion: { parentRunId: runId, generation: 1, taskIds: [taskId] }, + }, + }, + ], + }); + 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 === "notification.delivery.accept") { + captured = true; + // Checkpoint capture holds no thread lock; it commits after accept planned from the waiting run. + const { delegatedCompletion: _cohort, ...captureRun } = waiting; + yield* commit({ + commandId: CommandId.make("command:effect:checkpoint.capture:accept-race"), + threadId, + commandType: "checkpoint.capture", + acceptedAt: now, + effects: [], + events: [ + { + id: EventId.make("event:accept-capture-race-completed"), + type: "run.updated", + threadId, + runId, + occurredAt: now, + payload: { + ...captureRun, + status: "completed", + completedAt: now, + checkpointId, + }, + }, + ], + }); + } + return yield* commit(input); + }), + ), + ), + () => + orchestrator.dispatch({ + type: "notification.delivery.accept", + commandId: CommandId.make("command:accept-capture-race"), + threadId, + messageId, + }), + (spy) => Effect.sync(() => spy.mockRestore()), + ); + assert.isTrue(captured); + const finished = yield* orchestrator.getThreadProjection(threadId); + const parent = finished.runs.find((run) => run.id === runId)!; + assert.equal( + finished.subagents.find((row) => row.id === taskId)?.completionDelivery?.state, + "delivered", + ); + assert.equal(parent.status, "completed"); + assert.equal(parent.checkpointId, checkpointId); + }), + ); + it.effect("builds completion text and metadata from the same live cohort", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; From 8cfa0bbc8ca0c0403edb3498f2c15e62acae42b2 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Mon, 26 Oct 2026 21:38:58 +0000 Subject: [PATCH 2/3] fix(server): delivery acceptance no longer reverts a completed run to waiting Fixes pingdotgg/t3code#25000 Co-Authored-By: Claude Opus 5.5 --- apps/server/src/orchestration-v2/EventSink.ts | 50 ++++++++++--------- .../src/orchestration-v2/Orchestrator.ts | 4 ++ 2 files changed, 31 insertions(+), 23 deletions(-) diff --git a/apps/server/src/orchestration-v2/EventSink.ts b/apps/server/src/orchestration-v2/EventSink.ts index af8fa5a95af7b..89d760789dd82 100644 --- a/apps/server/src/orchestration-v2/EventSink.ts +++ b/apps/server/src/orchestration-v2/EventSink.ts @@ -136,6 +136,8 @@ export interface EventSinkV2Shape { EventSinkV2Error >; readonly commitCommand: (input: { + /** See writeWithEffects. */ + readonly updateDelegatedCompletionOnly?: boolean; readonly commandId: CommandId; readonly threadId: ThreadId; readonly commandType: string; @@ -383,6 +385,25 @@ const layerBase: Layer.Layer< return yield* runOutcomes.prepare(storedEvents); }); + // Checkpoint capture can finish after a cohort writer reads its parent run. + // Merge in the writer's transaction so the event log and projection keep that completion. + const mergeDelegatedCompletionOnly = (events: ReadonlyArray) => + Effect.forEach(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 }, + }; + }), + ); + const writeEffect = Effect.fn("orchestrationV2.EventSink.write")(function* ( input: Parameters[0], ) { @@ -408,29 +429,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* mergeDelegatedCompletionOnly(input.events) : input.events; const normalized = yield* normalizeEvents( input.guardPendingUserInputCancellations === true @@ -646,7 +646,11 @@ const layerBase: Layer.Layer< return { ...existing, committed: false as const, cancelledEffectIds: [] }; } - const normalized = yield* normalizeEvents(input.events); + const normalized = yield* normalizeEvents( + input.updateDelegatedCompletionOnly + ? yield* mergeDelegatedCompletionOnly(input.events) + : input.events, + ); 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..d7476aff0872a 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -12268,6 +12268,10 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio acceptedAt, events: plan.events, effects: plan.effects, + // Accept only advances delivery bookkeeping; checkpoint capture may end the run meanwhile. + ...(command.type === "notification.delivery.accept" + ? { updateDelegatedCompletionOnly: true } + : {}), ...(plan.cancelUnsettledEffects === undefined ? {} : { cancelUnsettledEffects: plan.cancelUnsettledEffects }), From 34668f0e468d534622a3b3a907c7a68372d26ffd Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 27 Oct 2026 06:00:17 +0000 Subject: [PATCH 3/3] test(server): fold the takeover's accept race into the acknowledge table Takeover tests for #25666: the standalone accept test duplicated the seeding and commit spy of the acknowledge/dispose race test from #25672, so accept is now a row in that table. The row fails without the Orchestrator fix for the same reason (run stays waiting). Co-Authored-By: Claude Opus 5.5 --- .../DelegatedCompletionDelivery.test.ts | 206 ++++++------------ 1 file changed, 65 insertions(+), 141 deletions(-) diff --git a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts index aadf994c17fe6..78bf5dcce18d7 100644 --- a/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts +++ b/apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts @@ -623,122 +623,6 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { }), ); - it.effect("keeps a checkpoint completion committed while accepting a delivery", () => - Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const sink = yield* EventSink.EventSinkV2; - const now = yield* DateTime.now; - const threadId = ThreadId.make("thread:accept-capture-race"); - const runId = RunId.make("run:accept-capture-race"); - const taskId = NodeId.make("node:accept-capture-race-task"); - const checkpointId = CheckpointId.make("checkpoint:accept-capture-race"); - const messageId = MessageId.make(`message:delegated-delivery:${threadId}`); - yield* seedParentWithTerminalTask({ - threadId, - runId, - projectId: ProjectId.make("project:accept-capture-race"), - rootNodeId: NodeId.make("node:accept-capture-race-root"), - taskId, - deliveryState: "claimed", - completionWake: "always", - deliveryTaskIds: [taskId], - now, - }); - const projection = yield* orchestrator.getThreadProjection(threadId); - const waiting = { ...projection.runs[0]!, status: "waiting" as const }; - const task = projection.subagents[0]!; - yield* sink.write({ - events: [ - { - id: EventId.make("event:accept-capture-race-waiting"), - type: "run.updated", - threadId, - runId, - occurredAt: now, - payload: waiting, - }, - { - id: EventId.make("event:accept-capture-race-message"), - type: "message.updated", - threadId, - runId, - occurredAt: now, - payload: { - id: messageId, - threadId, - runId, - nodeId: task.parentNodeId, - role: "user", - text: "Background task finished", - attachments: [], - streaming: false, - createdBy: "agent", - creationSource: "server", - createdAt: now, - updatedAt: now, - delegatedCompletion: { parentRunId: runId, generation: 1, taskIds: [taskId] }, - }, - }, - ], - }); - 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 === "notification.delivery.accept") { - captured = true; - // Checkpoint capture holds no thread lock; it commits after accept planned from the waiting run. - const { delegatedCompletion: _cohort, ...captureRun } = waiting; - yield* commit({ - commandId: CommandId.make("command:effect:checkpoint.capture:accept-race"), - threadId, - commandType: "checkpoint.capture", - acceptedAt: now, - effects: [], - events: [ - { - id: EventId.make("event:accept-capture-race-completed"), - type: "run.updated", - threadId, - runId, - occurredAt: now, - payload: { - ...captureRun, - status: "completed", - completedAt: now, - checkpointId, - }, - }, - ], - }); - } - return yield* commit(input); - }), - ), - ), - () => - orchestrator.dispatch({ - type: "notification.delivery.accept", - commandId: CommandId.make("command:accept-capture-race"), - threadId, - messageId, - }), - (spy) => Effect.sync(() => spy.mockRestore()), - ); - assert.isTrue(captured); - const finished = yield* orchestrator.getThreadProjection(threadId); - const parent = finished.runs.find((run) => run.id === runId)!; - assert.equal( - finished.subagents.find((row) => row.id === taskId)?.completionDelivery?.state, - "delivered", - ); - assert.equal(parent.status, "completed"); - assert.equal(parent.checkpointId, checkpointId); - }), - ); - it.effect("builds completion text and metadata from the same live cohort", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; @@ -1092,24 +976,27 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { ); it.effect.each([ - { before: "running", after: "failed", capture: false }, - { before: "waiting", after: "completed", capture: true }, + { command: "acknowledge", before: "running", after: "failed", capture: false }, + { command: "acknowledge", before: "waiting", after: "completed", capture: true }, + { command: "accept", before: "waiting", after: "completed", capture: true }, ] as const)( - "keeps a $after run that settled between task_status's acknowledge read and commit", - ({ before, after, capture }) => + "keeps a $after run that settled between $command's read and commit", + ({ command, before, after, capture }) => 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 race = `${command}-settle-race-${after}`; + const threadId = ThreadId.make(`thread:${race}`); + const runId = RunId.make(`run:${race}`); + const taskId = NodeId.make(`node:${race}-task`); + const checkpointId = CheckpointId.make(`checkpoint:${race}`); + const messageId = MessageId.make(`message:delegated-delivery:${threadId}`); yield* seedParentWithTerminalTask({ threadId, - projectId: ProjectId.make(`project:ack-settle-race-${after}`), + projectId: ProjectId.make(`project:${race}`), runId, - rootNodeId: NodeId.make(`node:ack-settle-race-${after}-root`), + rootNodeId: NodeId.make(`node:${race}-root`), taskId, deliveryState: "claimed", deliveryTaskIds: [taskId], @@ -1119,13 +1006,35 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { yield* sink.write({ events: [ { - id: EventId.make(`event:ack-settle-race-${after}-before`), + id: EventId.make(`event:${race}-before`), type: "run.updated", threadId, runId, occurredAt: now, payload: { ...parent, status: before }, }, + { + id: EventId.make(`event:${race}-message`), + type: "message.updated", + threadId, + runId, + occurredAt: now, + payload: { + id: messageId, + threadId, + runId, + nodeId: parent.rootNodeId, + role: "user", + text: "Background task finished", + attachments: [], + streaming: false, + createdBy: "agent", + creationSource: "server", + createdAt: now, + updatedAt: now, + delegatedCompletion: { parentRunId: runId, generation: 1, taskIds: [taskId] }, + }, + }, ], }); const { delegatedCompletion: _cohort, ...settledRun } = parent; @@ -1137,19 +1046,22 @@ it.layer(layerTest)("delegated completion delivery repairs", (it) => { Effect.gen(function* () { if ( !settled && - input.commandType === "delegated_task.completion-delivery.acknowledge" + input.commandType === + (command === "accept" + ? "notification.delivery.accept" + : "delegated_task.completion-delivery.acknowledge") ) { settled = true; - // The provider turn ends (or its checkpoint lands) after acknowledge read the run. + // 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:${race}`), threadId, commandType: capture ? "checkpoint.capture" : "provider.event", acceptedAt: now, effects: [], events: [ { - id: EventId.make(`event:ack-settle-race-${after}-settled`), + id: EventId.make(`event:${race}-settled`), type: "run.updated", threadId, runId, @@ -1169,19 +1081,31 @@ 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( + command === "accept" + ? { + type: "notification.delivery.accept", + commandId: CommandId.make(`command:${race}:accept`), + threadId, + messageId, + } + : { + type: "delegated_task.completion-delivery.acknowledge", + commandId: CommandId.make(`command:${race}:task-status`), + parentThreadId: threadId, + taskId, + observedByRunId: runId, + }, + ), (spy) => Effect.sync(() => spy.mockRestore()), ); assert.isTrue(settled); - const run = (yield* orchestrator.getThreadProjection(threadId)).runs.find( - (candidate) => candidate.id === runId, - )!; + const finished = yield* orchestrator.getThreadProjection(threadId); + assert.equal( + finished.subagents.find((row) => row.id === taskId)?.completionDelivery?.state, + command === "accept" ? "delivered" : "acknowledged", + ); + const run = finished.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 },