From e44ca4ff8c18838f7866a0aee4c7753eccfcf869 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Mon, 20 Jul 2026 00:25:24 +0000 Subject: [PATCH 1/9] fix: report live workspace-turn state from task_await instead of stale settlements A workspace turn that settled terminally (provider stream error whose auto-retry was not detected, or 'interrupted after restart') could self-heal: the child auto-retries the same turn and completes it, but settleWorkspaceTurn refused to touch already-terminal records, so a parent orchestrator's task_await kept reporting the stale interrupted/error status forever. Two complementary fixes: - Event-time: the strictly turn-correlated stream-end path may now resettle a stale interrupted/error record (never completed) with the turn's real outcome, re-arming the notify_on_terminal wake-up so the parent learns the corrected result. Duplicate replays stay idempotent (no change when status+messageId already match). - Read-time: normalizeWorkspaceTurnRecord reconciles settled interrupted/error records against the child's live state. Snapshot reads (task_await) also repair from durable history when a correlated final assistant message proves the turn finished; and when the child is actively retrying the same turn (streaming or pending auto-retry, with the turn's prompt still the newest user message) the handle is revived to 'running' and re-registered so normal settlement resumes. List paths skip the per-record history read for perf and only use the cheap runtime-activity gate. --- src/node/services/taskService.test.ts | 292 ++++++++++++++++++++++++++ src/node/services/taskService.ts | 219 ++++++++++++++++++- 2 files changed, 502 insertions(+), 9 deletions(-) diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index 3b34aadd8e..aa77db28e1 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -3724,6 +3724,298 @@ describe("TaskService", () => { }); }); + test("correlated stream-end corrects a stale error settlement after self-healed retry", async () => { + const { config, parentId, taskService } = await startWorkspaceTurnForTest(); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "error", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + error: "Stream error: provider overloaded", + terminalAttentionNotifiedAt: "2026-06-19T00:00:02.000Z", + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + const internal = taskService as unknown as { + handleStreamEnd: (event: StreamEndEvent) => Promise; + }; + + await internal.handleStreamEnd({ + type: "stream-end", + workspaceId: "childworkspace", + messageId: "msg_retry_final", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + muxMetadata, + }, + parts: [{ type: "text", text: "Recovered after retry" }], + }); + + const snapshot = await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle"); + expect(snapshot).toMatchObject({ + status: "completed", + messageId: "msg_retry_final", + reportMarkdown: "Recovered after retry", + }); + expect(snapshot?.error).toBeUndefined(); + expect(snapshot?.terminalAttentionNotifiedAt).toBeUndefined(); + }); + + test("duplicate correlated stream-end replay keeps a settled error handle unchanged", async () => { + const { config, parentId, taskService } = await startWorkspaceTurnForTest(); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "error", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + messageId: "msg_truncated_replay", + error: "Workspace turn ended before completion (finishReason: length)", + terminalAttentionNotifiedAt: "2026-06-19T00:00:02.000Z", + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + const internal = taskService as unknown as { + handleStreamEnd: (event: StreamEndEvent) => Promise; + }; + + await internal.handleStreamEnd({ + type: "stream-end", + workspaceId: "childworkspace", + messageId: "msg_truncated_replay", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "length", + muxMetadata, + }, + parts: [{ type: "text", text: "Partial text" }], + }); + + const snapshot = await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle"); + expect(snapshot).toMatchObject({ + status: "error", + messageId: "msg_truncated_replay", + updatedAt: "2026-06-19T00:00:01.000Z", + terminalAttentionNotifiedAt: "2026-06-19T00:00:02.000Z", + }); + }); + + test("correlated stream-end never overwrites a completed workspace turn", async () => { + const { config, parentId, taskService } = await startWorkspaceTurnForTest(); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "completed", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + messageId: "msg_first", + reportMarkdown: "First result", + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + const internal = taskService as unknown as { + handleStreamEnd: (event: StreamEndEvent) => Promise; + }; + + await internal.handleStreamEnd({ + type: "stream-end", + workspaceId: "childworkspace", + messageId: "msg_second", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + muxMetadata, + }, + parts: [{ type: "text", text: "Second result" }], + }); + + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "completed", + messageId: "msg_first", + reportMarkdown: "First result", + }); + }); + + test("getWorkspaceTurnSnapshot repairs a stale error handle from self-healed history", async () => { + const { config, parentId, taskService, historyService } = await startWorkspaceTurnForTest(); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + const appendResult = await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_selfhealed", "assistant", "Self-healed final text", { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + muxMetadata, + }) + ); + expect(appendResult.success).toBe(true); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "error", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + error: "Stream error: provider overloaded", + }); + + // List paths skip history repair for settled handles (no runtime activity), so the + // stale record stays visible there until a snapshot read reconciles it. + const listed = await taskService.listWorkspaceTurnTasks(parentId, { statuses: ["error"] }); + expect(listed.map((record) => record.handleId)).toContain("wst_handle"); + + const snapshot = await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle"); + expect(snapshot).toMatchObject({ + status: "completed", + messageId: "msg_selfhealed", + reportMarkdown: "Self-healed final text", + }); + expect(snapshot?.error).toBeUndefined(); + }); + + test("getWorkspaceTurnSnapshot revives an interrupted handle while the child retries the same turn", async () => { + const hasPendingAutoRetry = mock((workspaceId: string) => workspaceId === "childworkspace"); + const { config, parentId, taskService, historyService } = await startWorkspaceTurnForTest({ + hasPendingAutoRetry, + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + const appendResult = await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_prompt", "user", "Summarize", { muxMetadata }) + ); + expect(appendResult.success).toBe(true); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "interrupted", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + error: "Workspace turn interrupted after restart", + terminalAttentionNotifiedAt: "2026-06-19T00:00:02.000Z", + }); + const internal = taskService as unknown as { + activeWorkspaceTurnHandleByWorkspaceId: Map< + string, + { handleId: string; ownerWorkspaceId: string } + >; + }; + internal.activeWorkspaceTurnHandleByWorkspaceId.clear(); + + const snapshot = await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle"); + expect(snapshot).toMatchObject({ status: "running", workspaceId: "childworkspace" }); + expect(snapshot?.error).toBeUndefined(); + expect(snapshot?.terminalAttentionNotifiedAt).toBeUndefined(); + expect(internal.activeWorkspaceTurnHandleByWorkspaceId.get("childworkspace")).toEqual({ + handleId: "wst_handle", + ownerWorkspaceId: parentId, + }); + }); + + test("a newer unrelated child prompt does not revive a settled workspace turn", async () => { + const isStreaming = mock((workspaceId: string) => workspaceId === "childworkspace"); + const { config, parentId, taskService, historyService } = await startWorkspaceTurnForTest({ + isStreaming, + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + expect( + ( + await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_prompt", "user", "Summarize", { muxMetadata }) + ) + ).success + ).toBe(true); + expect( + ( + await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_manual", "user", "Manual follow-up", {}) + ) + ).success + ).toBe(true); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "error", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + error: "Stream error: provider overloaded", + }); + const internal = taskService as unknown as { + activeWorkspaceTurnHandleByWorkspaceId: Map< + string, + { handleId: string; ownerWorkspaceId: string } + >; + }; + internal.activeWorkspaceTurnHandleByWorkspaceId.clear(); + + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "error", + error: "Stream error: provider overloaded", + }); + }); + test("workspace-turn stale recovery uses deferred history after archived descendants stop blocking", async () => { const { config, parentId, projectPath, taskService, historyService } = await startWorkspaceTurnForTest(); diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index a202597ece..699523330e 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -5227,6 +5227,13 @@ export class TaskService { waiterSettlement: | { status: "completed"; result: WorkspaceTurnWaitResult } | { status: "error"; error: Error }; + /** + * Allow replacing an already-settled interrupted/error record (never completed). + * Only the strictly turn-correlated stream-end path may set this: a handle that + * settled from a transient failure can self-heal when the child auto-retries the + * same turn, and the correlated stream-end proves the turn's real outcome. + */ + allowTerminalResettle?: boolean; }): Promise { assert( params.next.handleId === params.record.handleId, @@ -5258,7 +5265,15 @@ export class TaskService { "settleWorkspaceTurn requires current record to match workspaceId" ); - if (this.isTerminalWorkspaceTurnStatus(current.status)) { + // A completed record is immutable; a stale interrupted/error record may be corrected + // once by an explicitly allowed resettle, but only when the new settlement actually + // changes the outcome (duplicate stream-end replays must stay idempotent). + const resettleStaleTerminal = + params.allowTerminalResettle === true && + this.isTerminalWorkspaceTurnStatus(current.status) && + current.status !== "completed" && + (params.next.status !== current.status || params.next.messageId !== current.messageId); + if (this.isTerminalWorkspaceTurnStatus(current.status) && !resettleStaleTerminal) { const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(params.record.workspaceId); if ( active?.handleId === params.record.handleId && @@ -5285,11 +5300,24 @@ export class TaskService { } // Decide the terminal wake-up using persisted policy + the restart-safe dedupe marker. + // A resettle corrects a previously reported outcome, so it re-arms the wake-up even if + // the stale settlement was already notified/consumed. const policy = resolveBackgroundWorkAttentionPolicy(current.attentionPolicy); const shouldNotify = - policy === "notify_on_terminal" && current.terminalAttentionNotifiedAt == null; - - await this.taskHandleStore.upsertWorkspaceTurn(params.next); + policy === "notify_on_terminal" && + (current.terminalAttentionNotifiedAt == null || resettleStaleTerminal); + + const nextRecord = { ...params.next }; + if (resettleStaleTerminal) { + log.debug("Workspace turn resettled from stale terminal status", { + handleId: current.handleId, + workspaceId: current.workspaceId, + staleStatus: current.status, + nextStatus: nextRecord.status, + }); + delete nextRecord.terminalAttentionNotifiedAt; + } + await this.taskHandleStore.upsertWorkspaceTurn(nextRecord); const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(params.record.workspaceId); if ( active?.handleId === params.record.handleId && @@ -5302,7 +5330,7 @@ export class TaskService { params.waiterSettlement ); this.markTaskForegroundRelevant(params.record.handleId); - await this.cleanupDisposableWorkspaceTurn(params.next); + await this.cleanupDisposableWorkspaceTurn(nextRecord); this.scheduleMaybeStartQueuedTasks(); // A foreground waiter that received this terminal result already integrates it, so suppress @@ -5316,8 +5344,8 @@ export class TaskService { } return { kind: "notify", - outcome: workspaceTurnTerminalOutcome(params.next.status), - ...(params.next.title != null ? { title: params.next.title } : {}), + outcome: workspaceTurnTerminalOutcome(nextRecord.status), + ...(nextRecord.title != null ? { title: nextRecord.title } : {}), }; } ); @@ -6161,7 +6189,16 @@ export class TaskService { } private async normalizeWorkspaceTurnRecord( - record: WorkspaceTurnTaskHandleRecord + record: WorkspaceTurnTaskHandleRecord, + options: { + /** + * Also reconcile settled interrupted/error records against the child's durable + * history (one history read per record). Enabled for single-handle snapshot reads + * (task_await) but not for list paths, which would pay that read for every + * historical terminal handle on each call. + */ + repairSettledTurnsFromHistory?: boolean; + } = {} ): Promise { assert(record.ownerWorkspaceId.length > 0, "normalizeWorkspaceTurnRecord requires owner id"); assert(record.handleId.length > 0, "normalizeWorkspaceTurnRecord requires handle id"); @@ -6200,9 +6237,164 @@ export class TaskService { return await this.taskHandleStore.getWorkspaceTurn(record.ownerWorkspaceId, record.handleId); } + if (record.status === "interrupted" || record.status === "error") { + return await this.reconcileSettledWorkspaceTurn(record, { + repairFromHistory: options.repairSettledTurnsFromHistory === true, + }); + } + + return record; + } + + /** + * Settled interrupted/error handles can go stale: a stream error or restart can settle + * the handle even though the child workspace self-heals by auto-retrying the same turn + * (retries replay the turn's synthetic prompt, so their output still correlates via + * muxMetadata). Reconcile against the child's live state at read time so task_await and + * task_list report reality instead of the stale settlement. + */ + private async reconcileSettledWorkspaceTurn( + record: WorkspaceTurnTaskHandleRecord, + options: { repairFromHistory: boolean } + ): Promise { + assert( + record.status === "interrupted" || record.status === "error", + "reconcileSettledWorkspaceTurn requires a settled interrupted/error record" + ); + // Cheap in-memory evidence of a same-turn retry, checked first so list paths do not + // pay a history read for every historical terminal handle. Queued/preparing turns are + // intentionally excluded: unrelated queued input must not revive a settled handle. + const retryLive = + this.aiService.isStreaming(record.workspaceId) || + this.workspaceService.hasPendingAutoRetry(record.workspaceId); + if (!retryLive && !options.repairFromHistory) { + return record; + } + + const historyResult = await this.historyService.getHistoryFromLatestBoundary( + record.workspaceId + ); + if (!historyResult.success) { + return record; + } + const allowDeferredMessages = + (record.deferredMessageIds?.length ?? 0) === 0 || + !(await this.hasActiveWorkspaceTurnDeferredBlockers(record)); + for (const message of historyResult.data.toReversed()) { + if (this.isDeferredWorkspaceTurnMessage(record, message.id) && !allowDeferredMessages) { + continue; + } + const event = this.buildWorkspaceTurnStreamEndEventFromHistory(record, message); + if (event != null) { + // The turn produced a correlated final assistant message after settlement, so the + // child self-healed and finished. Persist the true outcome when repair is allowed. + const recovered = this.buildTerminalWorkspaceTurnRecordFromEvent(record, event); + if ( + !options.repairFromHistory || + (recovered.status === record.status && recovered.messageId === record.messageId) + ) { + return record; + } + return await this.persistRepairedSettledWorkspaceTurn(record, recovered); + } + if (message.role !== "user") { + continue; + } + // Auto-retry replays the newest accepted user message, so live child activity belongs + // to this turn only while its prompt is still that newest message. Any newer user + // message means the activity is a different turn. + const metadata = this.getWorkspaceTurnMetadataFromValue(message.metadata?.muxMetadata); + const correlatedPrompt = + metadata != null && + metadata.taskHandleId === record.handleId && + metadata.ownerWorkspaceId === record.ownerWorkspaceId && + metadata.turnId === record.turnId; + if (correlatedPrompt && retryLive) { + return await this.reviveRetryingWorkspaceTurn(record); + } + return record; + } return record; } + private async persistRepairedSettledWorkspaceTurn( + record: WorkspaceTurnTaskHandleRecord, + recovered: WorkspaceTurnTaskHandleRecord + ): Promise { + const next = await this.workspaceTurnSettlementLocks.withLock(record.handleId, async () => { + const current = await this.taskHandleStore.getWorkspaceTurn( + record.ownerWorkspaceId, + record.handleId + ); + // A concurrent settlement/repair wins; only replace the stale status we reconciled. + if (current == null || current.status !== record.status) { + return current; + } + log.debug("Workspace turn repaired from self-healed child history", { + handleId: record.handleId, + workspaceId: record.workspaceId, + staleStatus: record.status, + nextStatus: recovered.status, + }); + await this.taskHandleStore.upsertWorkspaceTurn(recovered); + return recovered; + }); + if (next === recovered) { + await this.cleanupDisposableWorkspaceTurn(recovered); + const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(record.workspaceId); + if ( + active?.handleId === record.handleId && + active.ownerWorkspaceId === record.ownerWorkspaceId + ) { + this.activeWorkspaceTurnHandleByWorkspaceId.delete(record.workspaceId); + } + } + return next; + } + + private async reviveRetryingWorkspaceTurn( + record: WorkspaceTurnTaskHandleRecord + ): Promise { + return await this.workspaceTurnSettlementLocks.withLock(record.handleId, async () => { + const current = await this.taskHandleStore.getWorkspaceTurn( + record.ownerWorkspaceId, + record.handleId + ); + // A concurrent transition wins; only revive the stale status we reconciled against. + if (current == null || current.status !== record.status) { + return current; + } + // Another turn already owns the child workspace; the activity is not this turn's retry. + const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(record.workspaceId); + if ( + active != null && + (active.handleId !== record.handleId || active.ownerWorkspaceId !== record.ownerWorkspaceId) + ) { + return current; + } + const next: WorkspaceTurnTaskHandleRecord = { + ...current, + status: "running", + updatedAt: getIsoNow(), + }; + delete next.error; + // The revived turn's next terminal transition is a new outcome; re-arm its wake-up. + delete next.terminalAttentionNotifiedAt; + await this.taskHandleStore.upsertWorkspaceTurn(next); + // Re-register so stream-end/abort/error settlement paths own the handle again. + this.activeWorkspaceTurnHandleByWorkspaceId.set(record.workspaceId, { + handleId: record.handleId, + ownerWorkspaceId: record.ownerWorkspaceId, + }); + log.debug("Workspace turn revived: child is retrying the same turn", { + handleId: record.handleId, + workspaceId: record.workspaceId, + staleStatus: record.status, + }); + return next; + }); + } + async getWorkspaceTurnSnapshot( ownerWorkspaceId: string, handleId: string @@ -6214,7 +6406,11 @@ export class TaskService { if (record == null) { return null; } - return await this.normalizeWorkspaceTurnRecord(record); + // Snapshot reads back task_await, which must report the child's live state even when + // a stale settlement (interrupted/error) was later corrected by a self-healed retry. + return await this.normalizeWorkspaceTurnRecord(record, { + repairSettledTurnsFromHistory: true, + }); } async listWorkspaceTurnTasks( @@ -8405,6 +8601,11 @@ export class TaskService { next.status === "completed" ? { status: "completed", result: this.buildWorkspaceTurnWaitResult(next) } : { status: "error", error: new Error(next.error ?? "Workspace turn failed") }, + // This stream-end is strictly correlated (workspaceId + turnId), so it proves the + // delegated turn's real outcome. A handle that settled interrupted/error from a + // transient failure (provider error, restart) may have self-healed via auto-retry + // of the same turn; let this settlement correct that stale record. + allowTerminalResettle: true, }); return true; } From cb50f704ceb80852d206c65011976eaf3bb2d54b Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Mon, 20 Jul 2026 02:10:01 +0000 Subject: [PATCH 2/9] fix: reset terminal-attention tombstone when resettling/reviving a stale workspace turn Codex P2: TerminalAttentionStore.enqueueIfAbsent leaves delivered tombstones untouched (keyed workspace_turn:), so a corrected terminal outcome after a resettle would never wake the owner. Delete the existing notification record before enqueueing the corrected outcome, and also clear it when reviving a retrying turn so the eventual real settlement can notify. --- src/node/services/taskService.test.ts | 71 +++++++++++++++++++++++++++ src/node/services/taskService.ts | 17 ++++++- 2 files changed, 87 insertions(+), 1 deletion(-) diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index aa77db28e1..ab5cc6c18d 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -3773,6 +3773,65 @@ describe("TaskService", () => { expect(snapshot?.terminalAttentionNotifiedAt).toBeUndefined(); }); + test("resettled workspace turn re-arms a consumed notify_on_terminal wake-up", async () => { + const { config, parentId, taskService } = await startWorkspaceTurnForTest(); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "error", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + error: "Stream error: provider overloaded", + attentionPolicy: "notify_on_terminal", + terminalAttentionNotifiedAt: "2026-06-19T00:00:02.000Z", + }); + // The stale error's wake-up was already delivered; without the tombstone reset, + // enqueueIfAbsent would swallow the corrected outcome's notification. + const terminalAttentionStore = new TerminalAttentionStore(config); + await terminalAttentionStore.enqueueIfAbsent({ + ownerWorkspaceId: parentId, + sourceKind: "workspace_turn", + sourceId: "wst_handle", + outputDelivery: "requires_task_await", + terminalOutcome: "error", + }); + await terminalAttentionStore.markDelivered(parentId, "workspace_turn:wst_handle"); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + const internal = taskService as unknown as { + handleStreamEnd: (event: StreamEndEvent) => Promise; + }; + + await internal.handleStreamEnd({ + type: "stream-end", + workspaceId: "childworkspace", + messageId: "msg_retry_final", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + muxMetadata, + }, + parts: [{ type: "text", text: "Recovered after retry" }], + }); + + const notification = await terminalAttentionStore.get(parentId, "workspace_turn:wst_handle"); + expect(notification).toMatchObject({ terminalOutcome: "completed" }); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "completed", + reportMarkdown: "Recovered after retry", + }); + }); + test("duplicate correlated stream-end replay keeps a settled error handle unchanged", async () => { const { config, parentId, taskService } = await startWorkspaceTurnForTest(); await new TaskHandleStore(config).upsertWorkspaceTurn({ @@ -3944,6 +4003,17 @@ describe("TaskService", () => { error: "Workspace turn interrupted after restart", terminalAttentionNotifiedAt: "2026-06-19T00:00:02.000Z", }); + // Delivered tombstone from the stale settlement; revive must clear it so the revived + // turn's eventual real settlement can enqueue a fresh wake-up. + const terminalAttentionStore = new TerminalAttentionStore(config); + await terminalAttentionStore.enqueueIfAbsent({ + ownerWorkspaceId: parentId, + sourceKind: "workspace_turn", + sourceId: "wst_handle", + outputDelivery: "requires_task_await", + terminalOutcome: "interrupted", + }); + await terminalAttentionStore.markDelivered(parentId, "workspace_turn:wst_handle"); const internal = taskService as unknown as { activeWorkspaceTurnHandleByWorkspaceId: Map< string, @@ -3956,6 +4026,7 @@ describe("TaskService", () => { expect(snapshot).toMatchObject({ status: "running", workspaceId: "childworkspace" }); expect(snapshot?.error).toBeUndefined(); expect(snapshot?.terminalAttentionNotifiedAt).toBeUndefined(); + expect(await terminalAttentionStore.get(parentId, "workspace_turn:wst_handle")).toBeNull(); expect(internal.activeWorkspaceTurnHandleByWorkspaceId.get("childworkspace")).toEqual({ handleId: "wst_handle", ownerWorkspaceId: parentId, diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index 699523330e..5eeeafd187 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -5249,7 +5249,7 @@ export class TaskService { const pendingNotify = await this.workspaceTurnSettlementLocks.withLock( params.record.handleId, async (): Promise< - | { kind: "notify"; outcome: TerminalAttentionOutcome; title?: string } + | { kind: "notify"; outcome: TerminalAttentionOutcome; title?: string; resettled: boolean } | { kind: "drain_pending" } | null > => { @@ -5345,6 +5345,7 @@ export class TaskService { return { kind: "notify", outcome: workspaceTurnTerminalOutcome(nextRecord.status), + resettled: resettleStaleTerminal, ...(nextRecord.title != null ? { title: nextRecord.title } : {}), }; } @@ -5361,6 +5362,14 @@ export class TaskService { // Enqueue the terminal wake-up outside the lock. The persisted notification is the restart-safe // record of intent; only after it is accepted do we set terminalAttentionNotifiedAt on the // handle so a duplicate settlement / stale recovery cannot double-wake. + if (pendingNotify.resettled) { + // The stale settlement's wake-up may already be delivered/consumed; enqueueIfAbsent + // treats that tombstone as "already notified" and would swallow the corrected outcome. + await this.terminalAttentionStore.delete( + params.record.ownerWorkspaceId, + TerminalAttentionStore.notificationId("workspace_turn", params.record.handleId) + ); + } await this.enqueueTerminalAttention({ ownerWorkspaceId: params.record.ownerWorkspaceId, sourceKind: "workspace_turn", @@ -6379,7 +6388,13 @@ export class TaskService { }; delete next.error; // The revived turn's next terminal transition is a new outcome; re-arm its wake-up. + // The notification tombstone must go too: enqueueIfAbsent would otherwise treat the + // stale settlement's delivered wake-up as "already notified" and swallow the new one. delete next.terminalAttentionNotifiedAt; + await this.terminalAttentionStore.delete( + record.ownerWorkspaceId, + TerminalAttentionStore.notificationId("workspace_turn", record.handleId) + ); await this.taskHandleStore.upsertWorkspaceTurn(next); // Re-register so stream-end/abort/error settlement paths own the handle again. this.activeWorkspaceTurnHandleByWorkspaceId.set(record.workspaceId, { From de8ad3ba0dc1ac5655ed987f877fe06f74c286c1 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Mon, 20 Jul 2026 02:17:26 +0000 Subject: [PATCH 3/9] fix: continue history repair scan past unrelated newer prompts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex P2: an unrelated user prompt newer than the turn's correlated final assistant message must only disable the live-retry revive, not abort the durable-history repair scan — otherwise a turn that self-healed and completed before the child received manual input would keep reporting the stale failure. --- src/node/services/taskService.test.ts | 54 +++++++++++++++++++++++++++ src/node/services/taskService.ts | 22 ++++++++--- 2 files changed, 70 insertions(+), 6 deletions(-) diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index ab5cc6c18d..ae38eec8a5 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -4087,6 +4087,60 @@ describe("TaskService", () => { }); }); + test("history repair scans past newer unrelated prompts to a correlated final message", async () => { + const { config, parentId, taskService, historyService } = await startWorkspaceTurnForTest(); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + // The turn self-healed and finished, THEN the child received an unrelated manual + // prompt before the parent ever called task_await. + expect( + ( + await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_selfhealed_final", "assistant", "Self-healed final text", { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + muxMetadata, + }) + ) + ).success + ).toBe(true); + expect( + ( + await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_manual_later", "user", "Manual follow-up", {}) + ) + ).success + ).toBe(true); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "error", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + error: "Stream error: provider overloaded", + }); + + const snapshot = await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle"); + expect(snapshot).toMatchObject({ + status: "completed", + messageId: "msg_selfhealed_final", + reportMarkdown: "Self-healed final text", + }); + expect(snapshot?.error).toBeUndefined(); + }); + test("workspace-turn stale recovery uses deferred history after archived descendants stop blocking", async () => { const { config, parentId, projectPath, taskService, historyService } = await startWorkspaceTurnForTest(); diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index 5eeeafd187..75057fbfe9 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -6289,6 +6289,12 @@ export class TaskService { const allowDeferredMessages = (record.deferredMessageIds?.length ?? 0) === 0 || !(await this.hasActiveWorkspaceTurnDeferredBlockers(record)); + // Auto-retry replays the newest accepted user message, so live child activity belongs + // to this turn only while its prompt is still that newest message. Any newer user + // message disables the revive — but only the revive: a correlated final assistant + // message older than that unrelated prompt still proves the turn completed, so the + // durable-history repair scan must continue past it. + let reviveAllowed = retryLive; for (const message of historyResult.data.toReversed()) { if (this.isDeferredWorkspaceTurnMessage(record, message.id) && !allowDeferredMessages) { continue; @@ -6309,19 +6315,23 @@ export class TaskService { if (message.role !== "user") { continue; } - // Auto-retry replays the newest accepted user message, so live child activity belongs - // to this turn only while its prompt is still that newest message. Any newer user - // message means the activity is a different turn. const metadata = this.getWorkspaceTurnMetadataFromValue(message.metadata?.muxMetadata); const correlatedPrompt = metadata != null && metadata.taskHandleId === record.handleId && metadata.ownerWorkspaceId === record.ownerWorkspaceId && metadata.turnId === record.turnId; - if (correlatedPrompt && retryLive) { - return await this.reviveRetryingWorkspaceTurn(record); + if (correlatedPrompt) { + if (reviveAllowed) { + return await this.reviveRetryingWorkspaceTurn(record); + } + // A correlated final must be newer than the prompt; nothing older can repair. + return record; + } + reviveAllowed = false; + if (!options.repairFromHistory) { + return record; } - return record; } return record; } From 74691ae6c52ef1111ed74e71840fdb20f8728d4c Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Mon, 20 Jul 2026 02:27:23 +0000 Subject: [PATCH 4/9] fix: compare updatedAt before reviving/repairing a reconciled workspace turn Codex P2: between the reconcile read and the settlement lock, the live retry itself can fail and settle a NEWER record with the same status (error -> error). Guarding on status alone would revive that fresh terminal failure back to running, stranding task_await. Compare updatedAt as record identity in both revive and repair paths. --- src/node/services/taskService.test.ts | 42 +++++++++++++++++++++++++++ src/node/services/taskService.ts | 21 +++++++++++--- 2 files changed, 59 insertions(+), 4 deletions(-) diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index ae38eec8a5..1603b96084 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -4087,6 +4087,48 @@ describe("TaskService", () => { }); }); + test("revive does not clobber a newer same-status settlement written after the stale read", async () => { + const { config, parentId, taskService } = await startWorkspaceTurnForTest(); + const taskHandleStore = new TaskHandleStore(config); + // The record currently on disk: a FRESH error settled by the live retry itself. + const freshError = { + kind: "workspace_turn" as const, + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "error" as const, + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:05:00.000Z", + createdWorkspace: true, + disposableWorkspace: false, + error: "Stream error: retry also failed", + }; + await taskHandleStore.upsertWorkspaceTurn(freshError); + const internal = taskService as unknown as { + reviveRetryingWorkspaceTurn: (record: typeof freshError) => Promise; + }; + + // Reconcile observed an OLDER error record (same status, earlier updatedAt) before the + // retry failed; the revive must notice the newer settlement and leave it untouched. + const revived = await internal.reviveRetryingWorkspaceTurn({ + ...freshError, + updatedAt: "2026-06-19T00:00:01.000Z", + error: "Stream error: provider overloaded", + }); + + expect(revived).toMatchObject({ + status: "error", + updatedAt: "2026-06-19T00:05:00.000Z", + error: "Stream error: retry also failed", + }); + expect(await taskHandleStore.getWorkspaceTurn(parentId, "wst_handle")).toMatchObject({ + status: "error", + updatedAt: "2026-06-19T00:05:00.000Z", + error: "Stream error: retry also failed", + }); + }); + test("history repair scans past newer unrelated prompts to a correlated final message", async () => { const { config, parentId, taskService, historyService } = await startWorkspaceTurnForTest(); const muxMetadata = { diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index 75057fbfe9..17a742b64c 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -6345,8 +6345,14 @@ export class TaskService { record.ownerWorkspaceId, record.handleId ); - // A concurrent settlement/repair wins; only replace the stale status we reconciled. - if (current == null || current.status !== record.status) { + // A concurrent settlement/repair wins; only replace the exact record we reconciled. + // Comparing updatedAt (not just status) matters: a concurrent settlement can produce + // a NEWER record with the same status that must not be clobbered by our stale read. + if ( + current == null || + current.status !== record.status || + current.updatedAt !== record.updatedAt + ) { return current; } log.debug("Workspace turn repaired from self-healed child history", { @@ -6379,8 +6385,15 @@ export class TaskService { record.ownerWorkspaceId, record.handleId ); - // A concurrent transition wins; only revive the stale status we reconciled against. - if (current == null || current.status !== record.status) { + // A concurrent transition wins; only revive the exact record we reconciled against. + // Comparing updatedAt (not just status) matters: the live retry itself can fail and + // settle a NEWER record with the same status (e.g. error → error) between our read + // and this lock — reviving that fresh terminal failure would strand task_await. + if ( + current == null || + current.status !== record.status || + current.updatedAt !== record.updatedAt + ) { return current; } // Another turn already owns the child workspace; the activity is not this turn's retry. From d434df2484ba56f5517d5bc8013297b72c95586d Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Mon, 20 Jul 2026 02:35:39 +0000 Subject: [PATCH 5/9] fix: restrict workspace-turn self-heal correction to stale settlements Codex P2 x2: explicit interrupts (user Esc, task_terminate, cancel reasons) persist status interrupted without the stale-restart marker; they must stay terminal even if a late correlated stream-end lands or same-turn retry evidence appears, so canceled work never resurfaces as completed/running. Gate both the stream-end resettle and the read-time reconcile on isSelfHealEligibleSettledWorkspaceTurn: status error (transient stream failures) or interrupted with the restart marker. --- src/node/services/taskService.test.ts | 91 +++++++++++++++++++++++++++ src/node/services/taskService.ts | 38 ++++++++--- 2 files changed, 122 insertions(+), 7 deletions(-) diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index 1603b96084..3b2a93749a 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -3881,6 +3881,97 @@ describe("TaskService", () => { }); }); + test("late correlated stream-end does not resettle an explicitly interrupted workspace turn", async () => { + const { config, parentId, taskService } = await startWorkspaceTurnForTest(); + // Explicit interrupt (user Esc / task_terminate): status interrupted WITHOUT the + // stale-restart marker. An in-flight stream-end completing after the cancel must not + // make the canceled turn appear completed. + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "interrupted", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + const internal = taskService as unknown as { + handleStreamEnd: (event: StreamEndEvent) => Promise; + }; + + await internal.handleStreamEnd({ + type: "stream-end", + workspaceId: "childworkspace", + messageId: "msg_late_final", + metadata: { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + muxMetadata, + }, + parts: [{ type: "text", text: "Late final text" }], + }); + + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "interrupted", + updatedAt: "2026-06-19T00:00:01.000Z", + }); + }); + + test("explicitly interrupted workspace turns are not revived by same-turn retry evidence", async () => { + const isStreaming = mock((workspaceId: string) => workspaceId === "childworkspace"); + const { config, parentId, taskService, historyService } = await startWorkspaceTurnForTest({ + isStreaming, + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + expect( + ( + await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_prompt", "user", "Summarize", { muxMetadata }) + ) + ).success + ).toBe(true); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "interrupted", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + }); + const internal = taskService as unknown as { + activeWorkspaceTurnHandleByWorkspaceId: Map< + string, + { handleId: string; ownerWorkspaceId: string } + >; + }; + internal.activeWorkspaceTurnHandleByWorkspaceId.clear(); + + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "interrupted", + updatedAt: "2026-06-19T00:00:01.000Z", + }); + }); + test("correlated stream-end never overwrites a completed workspace turn", async () => { const { config, parentId, taskService } = await startWorkspaceTurnForTest(); await new TaskHandleStore(config).upsertWorkspaceTurn({ diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index 17a742b64c..feaee37141 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -689,6 +689,25 @@ const WORKSPACE_TURN_RECOVERABLE_STREAM_ERRORS: ReadonlySet = n "context_exceeded", ]); +/** Marker persisted by settleStaleWorkspaceTurn when restart recovery interrupts a handle. */ +const WORKSPACE_TURN_STALE_RESTART_ERROR = "Workspace turn interrupted after restart"; + +/** + * Settled workspace-turn records eligible for self-heal correction (resettle from a + * correlated stream-end, or read-time repair/revive): transient stream-error settlements + * (status "error") and stale restart-recovery interrupts. Explicit interrupts — user Esc, + * task_terminate, cancel reasons — must stay terminal even if a late correlated stream-end + * or same-turn retry evidence arrives, so canceled work never resurfaces as completed. + */ +function isSelfHealEligibleSettledWorkspaceTurn( + record: Pick +): boolean { + if (record.status === "error") { + return true; + } + return record.status === "interrupted" && record.error === WORKSPACE_TURN_STALE_RESTART_ERROR; +} + /** * A workspace-turn stream error may resolve without parent intervention when * the child can still make progress on its own. The caller must still confirm @@ -5265,13 +5284,15 @@ export class TaskService { "settleWorkspaceTurn requires current record to match workspaceId" ); - // A completed record is immutable; a stale interrupted/error record may be corrected - // once by an explicitly allowed resettle, but only when the new settlement actually - // changes the outcome (duplicate stream-end replays must stay idempotent). + // A completed record is immutable; a self-heal-eligible settled record (transient + // error / stale restart interrupt — never an explicit user interrupt) may be + // corrected once by an explicitly allowed resettle, but only when the new settlement + // actually changes the outcome (duplicate stream-end replays must stay idempotent). const resettleStaleTerminal = params.allowTerminalResettle === true && this.isTerminalWorkspaceTurnStatus(current.status) && current.status !== "completed" && + isSelfHealEligibleSettledWorkspaceTurn(current) && (params.next.status !== current.status || params.next.messageId !== current.messageId); if (this.isTerminalWorkspaceTurnStatus(current.status) && !resettleStaleTerminal) { const active = this.activeWorkspaceTurnHandleByWorkspaceId.get(params.record.workspaceId); @@ -6217,7 +6238,7 @@ export class TaskService { // task_await agree on the self-healed terminal status. if ( record.status === "interrupted" && - record.error === "Workspace turn interrupted after restart" && + record.error === WORKSPACE_TURN_STALE_RESTART_ERROR && (record.deferredMessageIds?.length ?? 0) > 0 ) { const recovered = await this.recoverTerminalWorkspaceTurnFromHistory(record); @@ -6246,7 +6267,10 @@ export class TaskService { return await this.taskHandleStore.getWorkspaceTurn(record.ownerWorkspaceId, record.handleId); } - if (record.status === "interrupted" || record.status === "error") { + if ( + (record.status === "interrupted" || record.status === "error") && + isSelfHealEligibleSettledWorkspaceTurn(record) + ) { return await this.reconcileSettledWorkspaceTurn(record, { repairFromHistory: options.repairSettledTurnsFromHistory === true, }); @@ -7347,14 +7371,14 @@ export class TaskService { ...record, status: "interrupted", updatedAt: getIsoNow(), - error: "Workspace turn interrupted after restart", + error: WORKSPACE_TURN_STALE_RESTART_ERROR, }; await this.settleWorkspaceTurn({ record, next, waiterSettlement: { status: "error", - error: new Error("Workspace turn interrupted after restart"), + error: new Error(WORKSPACE_TURN_STALE_RESTART_ERROR), }, }); } From 50bd17154200c590f5559ac4df913061796246b2 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Mon, 20 Jul 2026 02:44:50 +0000 Subject: [PATCH 6/9] tests: prove active-only task_list revives stale retrying workspace turns Codex round-5 finding assumed listWorkspaceTurnTasks applies the status filter at the store level before normalization; it actually reads all records, normalizes (reconcile/revive) each, then filters on the normalized status. Add a regression test demonstrating a stale error handle with a live auto-retry is revived and included in the active-only listing. --- src/node/services/taskService.test.ts | 50 +++++++++++++++++++++++++++ 1 file changed, 50 insertions(+) diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index 3b2a93749a..c92e9f5e90 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -4124,6 +4124,56 @@ describe("TaskService", () => { }); }); + test("active-only listWorkspaceTurnTasks revives and includes a stale retrying handle", async () => { + const hasPendingAutoRetry = mock((workspaceId: string) => workspaceId === "childworkspace"); + const { config, parentId, taskService, historyService } = await startWorkspaceTurnForTest({ + hasPendingAutoRetry, + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + expect( + ( + await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_prompt", "user", "Summarize", { muxMetadata }) + ) + ).success + ).toBe(true); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "error", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + error: "Stream error: provider overloaded", + }); + const internal = taskService as unknown as { + activeWorkspaceTurnHandleByWorkspaceId: Map< + string, + { handleId: string; ownerWorkspaceId: string } + >; + }; + internal.activeWorkspaceTurnHandleByWorkspaceId.clear(); + + // task_list defaults to active statuses; the status filter applies AFTER + // normalization, so the stale-but-retrying handle is revived and reported as + // running instead of silently disappearing from the active view. + const listed = await taskService.listWorkspaceTurnTasks(parentId, { + statuses: ["queued", "starting", "running"], + }); + expect(listed.map((record) => record.handleId)).toContain("wst_handle"); + expect(listed.find((record) => record.handleId === "wst_handle")?.status).toBe("running"); + }); + test("a newer unrelated child prompt does not revive a settled workspace turn", async () => { const isStreaming = mock((workspaceId: string) => workspaceId === "childworkspace"); const { config, parentId, taskService, historyService } = await startWorkspaceTurnForTest({ From 756859870fc2189ee3cbef7f8bc9a28166ea4306 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Mon, 20 Jul 2026 02:58:14 +0000 Subject: [PATCH 7/9] fix: gate stale repair on child background work; revive in turn-end blocker scan Codex P1: settled records never accumulate deferredMessageIds, so the history repair could report completed while the child still had active descendant/workflow/nested-turn work. Gate the repair on hasActiveWorkspaceTurnDeferredBlockers (lazily computed) to match the deferred stream-end semantics of active handles. Codex P2: listActiveWorkspaceTurnTaskIdsForOwner (parent turn-end blocker scan) previously asked the store for active statuses only, so a stale error handle with a live same-turn retry was invisible and the parent could end its turn. Reconcile self-heal-eligible settled records there too (runtime checks short-circuit before any history read). --- src/node/services/taskService.test.ts | 124 ++++++++++++++++++++++++++ src/node/services/taskService.ts | 59 +++++++++--- 2 files changed, 169 insertions(+), 14 deletions(-) diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index c92e9f5e90..0f644047ab 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -4124,6 +4124,130 @@ describe("TaskService", () => { }); }); + test("history repair of a stale error handle waits for active child background work", async () => { + const { config, parentId, projectPath, taskService, historyService } = + await startWorkspaceTurnForTest(); + // The retried turn emitted its correlated final while a descendant task was still + // running; repairing to completed before it finishes would hand the parent an + // incomplete result that active handles avoid via deferred stream-ends. + await config.editConfig((cfg) => { + const project = Array.from(cfg.projects.values())[0]; + assert(project, "test project must exist"); + project.workspaces.push({ + path: path.join(projectPath, "descendant-task"), + id: "descendant-task", + name: "descendant-task", + createdAt: "2026-06-19T00:00:00.000Z", + runtimeConfig: { type: "local" }, + parentWorkspaceId: "childworkspace", + taskStatus: "running", + }); + return cfg; + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + expect( + ( + await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_blocked_final", "assistant", "Blocked final text", { + model: "anthropic:claude-opus-4-6", + agentId: "exec", + finishReason: "stop", + muxMetadata, + }) + ) + ).success + ).toBe(true); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "error", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + error: "Stream error: provider overloaded", + }); + + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "error", + error: "Stream error: provider overloaded", + }); + + await config.editConfig((cfg) => { + const descendant = Array.from(cfg.projects.values()) + .flatMap((project) => project.workspaces) + .find((workspace) => workspace.id === "descendant-task"); + assert(descendant, "descendant task must exist"); + descendant.archivedAt = "2026-06-19T00:01:00.000Z"; + return cfg; + }); + + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "completed", + messageId: "msg_blocked_final", + reportMarkdown: "Blocked final text", + }); + }); + + test("turn-end blocker scan revives and includes a stale retrying handle", async () => { + const hasPendingAutoRetry = mock((workspaceId: string) => workspaceId === "childworkspace"); + const { config, parentId, taskService, historyService } = await startWorkspaceTurnForTest({ + hasPendingAutoRetry, + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + expect( + ( + await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_prompt", "user", "Summarize", { muxMetadata }) + ) + ).success + ).toBe(true); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "error", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + error: "Stream error: provider overloaded", + }); + const internal = taskService as unknown as { + activeWorkspaceTurnHandleByWorkspaceId: Map< + string, + { handleId: string; ownerWorkspaceId: string } + >; + listActiveWorkspaceTurnTaskIdsForOwner: (ownerWorkspaceId: string) => Promise; + }; + internal.activeWorkspaceTurnHandleByWorkspaceId.clear(); + + // The parent turn-end path must treat the stale-but-retrying handle as live work so + // the parent cannot end its turn while the child is still running. + expect(await internal.listActiveWorkspaceTurnTaskIdsForOwner(parentId)).toContain("wst_handle"); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "running", + workspaceId: "childworkspace", + }); + }); + test("active-only listWorkspaceTurnTasks revives and includes a stale retrying handle", async () => { const hasPendingAutoRetry = mock((workspaceId: string) => workspaceId === "childworkspace"); const { config, parentId, taskService, historyService } = await startWorkspaceTurnForTest({ diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index feaee37141..5cb870dd87 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -6310,9 +6310,16 @@ export class TaskService { if (!historyResult.success) { return record; } - const allowDeferredMessages = - (record.deferredMessageIds?.length ?? 0) === 0 || - !(await this.hasActiveWorkspaceTurnDeferredBlockers(record)); + // Lazily computed: whether the child still has active descendant/workflow/nested-turn + // work. A settled record cannot accumulate deferredMessageIds (the deferral path skips + // inactive handles), so the repair itself must be gated on these blockers too — + // otherwise a retried turn could be reported completed while its background work is + // still running, which active handles avoid via deferred stream-ends. + let blockersActive: boolean | null = null; + const deferredBlockersActive = async (): Promise => { + blockersActive ??= await this.hasActiveWorkspaceTurnDeferredBlockers(record); + return blockersActive; + }; // Auto-retry replays the newest accepted user message, so live child activity belongs // to this turn only while its prompt is still that newest message. Any newer user // message disables the revive — but only the revive: a correlated final assistant @@ -6320,7 +6327,10 @@ export class TaskService { // durable-history repair scan must continue past it. let reviveAllowed = retryLive; for (const message of historyResult.data.toReversed()) { - if (this.isDeferredWorkspaceTurnMessage(record, message.id) && !allowDeferredMessages) { + if ( + this.isDeferredWorkspaceTurnMessage(record, message.id) && + (await deferredBlockersActive()) + ) { continue; } const event = this.buildWorkspaceTurnStreamEndEventFromHistory(record, message); @@ -6334,6 +6344,9 @@ export class TaskService { ) { return record; } + if (await deferredBlockersActive()) { + return record; + } return await this.persistRepairedSettledWorkspaceTurn(record, recovered); } if (message.role !== "user") { @@ -7422,23 +7435,41 @@ export class TaskService { private async listActiveWorkspaceTurnTaskIdsForOwner( ownerWorkspaceId: string ): Promise { - const records = await this.taskHandleStore.listWorkspaceTurns(ownerWorkspaceId, { - statuses: ["queued", "starting", "running"], - }); + const records = await this.taskHandleStore.listWorkspaceTurns(ownerWorkspaceId); const taskIds: string[] = []; for (const record of records) { if ( - record.status !== "queued" && - record.status !== "starting" && - record.status !== "running" + record.status === "queued" || + record.status === "starting" || + record.status === "running" ) { + if (!(await this.isLiveWorkspaceTurn(record))) { + await this.settleStaleWorkspaceTurn(record); + continue; + } + taskIds.push(record.handleId); continue; } - if (!(await this.isLiveWorkspaceTurn(record))) { - await this.settleStaleWorkspaceTurn(record); - continue; + // A stale settled handle whose child is actively retrying the same turn is live + // delegated work: revive it here too, so the parent's turn-end blocker scan cannot + // finish while the child is still running. Cheap for historical handles — reconcile + // short-circuits on in-memory runtime checks before touching history. + if ( + (record.status === "interrupted" || record.status === "error") && + isSelfHealEligibleSettledWorkspaceTurn(record) + ) { + const latest = await this.reconcileSettledWorkspaceTurn(record, { + repairFromHistory: false, + }); + if ( + latest != null && + (latest.status === "queued" || + latest.status === "starting" || + latest.status === "running") + ) { + taskIds.push(latest.handleId); + } } - taskIds.push(record.handleId); } return taskIds; } From a91068e12985366229e78faad30c759a4a60e4bf Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Mon, 20 Jul 2026 03:13:26 +0000 Subject: [PATCH 8/9] fix: keep stale workspace turns live through child handoff windows MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex P2 (round 7): between retry streams the child can be momentarily quiet — stream ended, synthetic background-await queued/preparing, or descendant/workflow/nested work still running — and the previous cheap gate (isStreaming || hasPendingAutoRetry) returned the stale terminal record, letting the parent turn-end blocker scan finish early. Extend the liveness gate (cheap-first, blockers cached) and, when a correlated final exists but blockers are active, revive the handle with the final recorded as deferred so standard deferred recovery settles the true outcome once blockers finish. --- src/node/services/taskService.test.ts | 78 +++++++++++++++++++++++++-- src/node/services/taskService.ts | 54 ++++++++++++------- 2 files changed, 108 insertions(+), 24 deletions(-) diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index 0f644047ab..c7c024183f 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -4128,8 +4128,10 @@ describe("TaskService", () => { const { config, parentId, projectPath, taskService, historyService } = await startWorkspaceTurnForTest(); // The retried turn emitted its correlated final while a descendant task was still - // running; repairing to completed before it finishes would hand the parent an - // incomplete result that active handles avoid via deferred stream-ends. + // running; reporting completed before it finishes would hand the parent an + // incomplete result that active handles avoid via deferred stream-ends. Instead the + // handle is revived with the final recorded as deferred, then settles once the + // blocker is gone. await config.editConfig((cfg) => { const project = Array.from(cfg.projects.values())[0]; assert(project, "test project must exist"); @@ -4177,10 +4179,13 @@ describe("TaskService", () => { error: "Stream error: provider overloaded", }); - expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ - status: "error", - error: "Stream error: provider overloaded", + const blocked = await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle"); + expect(blocked).toMatchObject({ + status: "running", + deferredMessageIds: ["msg_blocked_final"], }); + expect(blocked?.error).toBeUndefined(); + expect(blocked?.reportMarkdown).toBeUndefined(); await config.editConfig((cfg) => { const descendant = Array.from(cfg.projects.values()) @@ -4198,6 +4203,69 @@ describe("TaskService", () => { }); }); + test("turn-end blocker scan keeps a stale handle live between retry streams via child blockers", async () => { + const { config, parentId, projectPath, taskService, historyService } = + await startWorkspaceTurnForTest(); + // Codex handoff gap: the retried child's stream ended, no auto-retry is pending, but + // its descendant work is still running. The blocker scan must still treat the turn as + // live so the parent cannot end its turn during that window. + await config.editConfig((cfg) => { + const project = Array.from(cfg.projects.values())[0]; + assert(project, "test project must exist"); + project.workspaces.push({ + path: path.join(projectPath, "descendant-task"), + id: "descendant-task", + name: "descendant-task", + createdAt: "2026-06-19T00:00:00.000Z", + runtimeConfig: { type: "local" }, + parentWorkspaceId: "childworkspace", + taskStatus: "running", + }); + return cfg; + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + expect( + ( + await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_prompt", "user", "Summarize", { muxMetadata }) + ) + ).success + ).toBe(true); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "error", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + error: "Stream error: provider overloaded", + }); + const internal = taskService as unknown as { + activeWorkspaceTurnHandleByWorkspaceId: Map< + string, + { handleId: string; ownerWorkspaceId: string } + >; + listActiveWorkspaceTurnTaskIdsForOwner: (ownerWorkspaceId: string) => Promise; + }; + internal.activeWorkspaceTurnHandleByWorkspaceId.clear(); + + expect(await internal.listActiveWorkspaceTurnTaskIdsForOwner(parentId)).toContain("wst_handle"); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "running", + workspaceId: "childworkspace", + }); + }); + test("turn-end blocker scan revives and includes a stale retrying handle", async () => { const hasPendingAutoRetry = mock((workspaceId: string) => workspaceId === "childworkspace"); const { config, parentId, taskService, historyService } = await startWorkspaceTurnForTest({ diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index 5cb870dd87..a3ca1c3e0e 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -6294,13 +6294,26 @@ export class TaskService { record.status === "interrupted" || record.status === "error", "reconcileSettledWorkspaceTurn requires a settled interrupted/error record" ); - // Cheap in-memory evidence of a same-turn retry, checked first so list paths do not - // pay a history read for every historical terminal handle. Queued/preparing turns are - // intentionally excluded: unrelated queued input must not revive a settled handle. - const retryLive = + // Lazily computed and cached: whether the child still has active descendant/workflow/ + // nested-turn work. A settled record cannot accumulate deferredMessageIds (the deferral + // path skips inactive handles), so both the liveness gate and the repair must consult + // these blockers — otherwise a retried turn could be reported completed (or dropped + // from active scans) while its background work is still running. + let blockersActive: boolean | null = null; + const deferredBlockersActive = async (): Promise => { + blockersActive ??= await this.hasActiveWorkspaceTurnDeferredBlockers(record); + return blockersActive; + }; + // Evidence the turn is still live, ordered cheap-first so list paths rarely pay a + // history read for historical terminal handles: an active stream or pending auto-retry + // of the same turn; a queued/preparing continuation (e.g. the synthetic background-await + // resume between retry streams); or still-running child background work. + const turnLive = this.aiService.isStreaming(record.workspaceId) || - this.workspaceService.hasPendingAutoRetry(record.workspaceId); - if (!retryLive && !options.repairFromHistory) { + this.workspaceService.hasPendingAutoRetry(record.workspaceId) || + this.workspaceService.hasPendingQueuedOrPreparingTurn(record.workspaceId) || + (await deferredBlockersActive()); + if (!turnLive && !options.repairFromHistory) { return record; } @@ -6310,22 +6323,12 @@ export class TaskService { if (!historyResult.success) { return record; } - // Lazily computed: whether the child still has active descendant/workflow/nested-turn - // work. A settled record cannot accumulate deferredMessageIds (the deferral path skips - // inactive handles), so the repair itself must be gated on these blockers too — - // otherwise a retried turn could be reported completed while its background work is - // still running, which active handles avoid via deferred stream-ends. - let blockersActive: boolean | null = null; - const deferredBlockersActive = async (): Promise => { - blockersActive ??= await this.hasActiveWorkspaceTurnDeferredBlockers(record); - return blockersActive; - }; // Auto-retry replays the newest accepted user message, so live child activity belongs // to this turn only while its prompt is still that newest message. Any newer user // message disables the revive — but only the revive: a correlated final assistant // message older than that unrelated prompt still proves the turn completed, so the // durable-history repair scan must continue past it. - let reviveAllowed = retryLive; + let reviveAllowed = turnLive; for (const message of historyResult.data.toReversed()) { if ( this.isDeferredWorkspaceTurnMessage(record, message.id) && @@ -6336,7 +6339,14 @@ export class TaskService { const event = this.buildWorkspaceTurnStreamEndEventFromHistory(record, message); if (event != null) { // The turn produced a correlated final assistant message after settlement, so the - // child self-healed and finished. Persist the true outcome when repair is allowed. + // child self-healed. While its background work is still running, revive the handle + // with the final recorded as deferred: the parent stays blocked, and the standard + // deferred-recovery machinery settles the true outcome once blockers finish. + if (reviveAllowed && (await deferredBlockersActive())) { + return await this.reviveRetryingWorkspaceTurn(record, { + deferredMessageIds: [event.messageId], + }); + } const recovered = this.buildTerminalWorkspaceTurnRecordFromEvent(record, event); if ( !options.repairFromHistory || @@ -6415,7 +6425,8 @@ export class TaskService { } private async reviveRetryingWorkspaceTurn( - record: WorkspaceTurnTaskHandleRecord + record: WorkspaceTurnTaskHandleRecord, + options: { deferredMessageIds?: string[] } = {} ): Promise { return await this.workspaceTurnSettlementLocks.withLock(record.handleId, async () => { const current = await this.taskHandleStore.getWorkspaceTurn( @@ -6445,6 +6456,11 @@ export class TaskService { ...current, status: "running", updatedAt: getIsoNow(), + // An already-observed correlated final (blocked on child background work) rides + // along as deferred so standard deferred recovery settles it once blockers finish. + ...(options.deferredMessageIds != null + ? { deferredMessageIds: options.deferredMessageIds } + : {}), }; delete next.error; // The revived turn's next terminal transition is a new outcome; re-arm its wake-up. From 6091f5c44362fb9cc08e48560b9af34d0bdec65c Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Mon, 20 Jul 2026 03:21:43 +0000 Subject: [PATCH 9/9] fix: make workspace-turn liveness gate fully in-memory MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Codex P2 x2 (round 8): drop hasPendingQueuedOrPreparingTurn from the gate — queued manual input is not yet in history, so it defeated the newest-correlated-prompt guard and could revive unrelated work. The synthetic background-await continuation window it covered is a subset of blockers-active, now represented by the cheap in-memory hasActiveDescendantAgentTasks hint instead of the full hasActiveWorkspaceTurnDeferredBlockers (which reads child history and would run for every historical failed handle on parent stream-ends). Full blocker checks remain lazily evaluated in the deferred/repair branches only. --- src/node/services/taskService.test.ts | 56 +++++++++++++++++++++++++++ src/node/services/taskService.ts | 19 ++++----- 2 files changed, 66 insertions(+), 9 deletions(-) diff --git a/src/node/services/taskService.test.ts b/src/node/services/taskService.test.ts index c7c024183f..9cfd3ecb83 100644 --- a/src/node/services/taskService.test.ts +++ b/src/node/services/taskService.test.ts @@ -4366,6 +4366,62 @@ describe("TaskService", () => { expect(listed.find((record) => record.handleId === "wst_handle")?.status).toBe("running"); }); + test("queued manual input does not revive a settled workspace turn", async () => { + // Ordinary queued input is not yet in history, so the newest-correlated-prompt guard + // cannot see it; the liveness gate must not treat it as a same-turn continuation. + const hasQueuedMessages = mock((workspaceId: string) => workspaceId === "childworkspace"); + const hasPendingQueuedOrPreparingTurn = mock( + (workspaceId: string) => workspaceId === "childworkspace" + ); + const { config, parentId, taskService, historyService } = await startWorkspaceTurnForTest({ + hasQueuedMessages, + hasPendingQueuedOrPreparingTurn, + }); + const muxMetadata = { + type: "workspace-turn-task" as const, + taskHandleId: "wst_handle", + ownerWorkspaceId: parentId, + turnId: "turn", + }; + expect( + ( + await historyService.appendToHistory( + "childworkspace", + createMuxMessage("msg_prompt", "user", "Summarize", { muxMetadata }) + ) + ).success + ).toBe(true); + await new TaskHandleStore(config).upsertWorkspaceTurn({ + kind: "workspace_turn", + handleId: "wst_handle", + ownerWorkspaceId: parentId, + workspaceId: "childworkspace", + turnId: "turn", + status: "error", + createdAt: "2026-06-19T00:00:00.000Z", + updatedAt: "2026-06-19T00:00:01.000Z", + createdWorkspace: true, + disposableWorkspace: false, + error: "Stream error: provider overloaded", + }); + const internal = taskService as unknown as { + activeWorkspaceTurnHandleByWorkspaceId: Map< + string, + { handleId: string; ownerWorkspaceId: string } + >; + listActiveWorkspaceTurnTaskIdsForOwner: (ownerWorkspaceId: string) => Promise; + }; + internal.activeWorkspaceTurnHandleByWorkspaceId.clear(); + + expect(await internal.listActiveWorkspaceTurnTaskIdsForOwner(parentId)).not.toContain( + "wst_handle" + ); + expect(await taskService.getWorkspaceTurnSnapshot(parentId, "wst_handle")).toMatchObject({ + status: "error", + error: "Stream error: provider overloaded", + }); + }); + test("a newer unrelated child prompt does not revive a settled workspace turn", async () => { const isStreaming = mock((workspaceId: string) => workspaceId === "childworkspace"); const { config, parentId, taskService, historyService } = await startWorkspaceTurnForTest({ diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts index a3ca1c3e0e..52f84e69c1 100644 --- a/src/node/services/taskService.ts +++ b/src/node/services/taskService.ts @@ -6296,23 +6296,24 @@ export class TaskService { ); // Lazily computed and cached: whether the child still has active descendant/workflow/ // nested-turn work. A settled record cannot accumulate deferredMessageIds (the deferral - // path skips inactive handles), so both the liveness gate and the repair must consult - // these blockers — otherwise a retried turn could be reported completed (or dropped - // from active scans) while its background work is still running. + // path skips inactive handles), so the repair/deferred decisions below must consult + // these blockers — otherwise a retried turn could be reported completed while its + // background work is still running. let blockersActive: boolean | null = null; const deferredBlockersActive = async (): Promise => { blockersActive ??= await this.hasActiveWorkspaceTurnDeferredBlockers(record); return blockersActive; }; - // Evidence the turn is still live, ordered cheap-first so list paths rarely pay a - // history read for historical terminal handles: an active stream or pending auto-retry - // of the same turn; a queued/preparing continuation (e.g. the synthetic background-await - // resume between retry streams); or still-running child background work. + // Evidence the turn is still live. All checks are in-memory so historical terminal + // handles stay cheap on list/blocker-scan paths (no history reads). Queued/preparing + // turns are deliberately excluded: ordinary queued manual input is not yet in history, + // so it would defeat the newest-correlated-prompt guard below and revive unrelated + // work. The synthetic background-await continuation between retry streams is instead + // covered by the descendant hint — it is only queued while such blockers are active. const turnLive = this.aiService.isStreaming(record.workspaceId) || this.workspaceService.hasPendingAutoRetry(record.workspaceId) || - this.workspaceService.hasPendingQueuedOrPreparingTurn(record.workspaceId) || - (await deferredBlockersActive()); + this.hasActiveDescendantAgentTasks(this.config.loadConfigOrDefault(), record.workspaceId); if (!turnLive && !options.repairFromHistory) { return record; }