From 94df0a913dba42f69b7eb67b570866e02e9a2783 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 23 Jul 2026 18:11:47 -0700 Subject: [PATCH 1/4] Show Waiting instead of Done for threads with open background tasks When the agent stops with background tasks (subagents, background bash, monitors) still running, the thread showed Done, then flicked back to Working when a task completion woke the agent. Track provider-reported open tasks in ingestion and park the session on the previously-unused "idle" status; sidebar v2 renders it as a grey "Waiting" with the same elapsed timer as Working. Co-Authored-By: Claude Fable 5 --- .../features/threads/thread-list-v2-items.tsx | 2 + .../src/features/threads/threadListV2.test.ts | 18 ++ .../src/features/threads/threadListV2.ts | 10 +- .../Layers/ProviderRuntimeIngestion.test.ts | 201 ++++++++++++++++++ .../Layers/ProviderRuntimeIngestion.ts | 102 ++++++++- apps/web/src/components/Sidebar.logic.test.ts | 16 ++ apps/web/src/components/Sidebar.logic.ts | 12 +- apps/web/src/components/SidebarV2.tsx | 51 +++-- 8 files changed, 381 insertions(+), 31 deletions(-) diff --git a/apps/mobile/src/features/threads/thread-list-v2-items.tsx b/apps/mobile/src/features/threads/thread-list-v2-items.tsx index 69729cb6469..99d521749e7 100644 --- a/apps/mobile/src/features/threads/thread-list-v2-items.tsx +++ b/apps/mobile/src/features/threads/thread-list-v2-items.tsx @@ -40,6 +40,8 @@ const STATUS_LABEL_BY_STATUS: Partial< approval: { label: "Approval", className: "text-amber-700 dark:text-amber-300" }, input: { label: "Input", className: "text-indigo-600 dark:text-indigo-300" }, working: { label: "Working", className: "text-sky-600 dark:text-sky-400" }, + // Colorless like the web sidebar: parked on background work, not "act now". + waiting: { label: "Waiting", className: "text-foreground-tertiary" }, failed: { label: "Failed", className: "text-red-700 dark:text-red-300" }, }; diff --git a/apps/mobile/src/features/threads/threadListV2.test.ts b/apps/mobile/src/features/threads/threadListV2.test.ts index e62b9ceda34..97708e1049f 100644 --- a/apps/mobile/src/features/threads/threadListV2.test.ts +++ b/apps/mobile/src/features/threads/threadListV2.test.ts @@ -63,6 +63,24 @@ describe("resolveThreadListV2Status", () => { "ready", ); }); + + it("resolves waiting for an idle session parked on background tasks", () => { + const thread = makeThread({ + id: ThreadId.make("t"), + title: "t", + session: { + threadId: ThreadId.make("t"), + status: "idle", + providerName: "Codex", + providerInstanceId: ProviderInstanceId.make("codex"), + runtimeMode: "full-access", + activeTurnId: null, + lastError: null, + updatedAt: NOW, + }, + }); + expect(resolveThreadListV2Status(thread)).toBe("waiting"); + }); }); describe("sortThreadsForListV2", () => { diff --git a/apps/mobile/src/features/threads/threadListV2.ts b/apps/mobile/src/features/threads/threadListV2.ts index efa68153a3f..11d688bca2a 100644 --- a/apps/mobile/src/features/threads/threadListV2.ts +++ b/apps/mobile/src/features/threads/threadListV2.ts @@ -6,11 +6,12 @@ import type { EnvironmentId, ProjectId } from "@t3tools/contracts"; * Thread List v2 model, ported from the web sidebar v2 * (apps/web/src/components/Sidebar.logic.ts + SidebarV2.tsx). * - * Four visual states, three colors: color is reserved for "act now" + * Six visual states, three colors: color is reserved for "act now" * (approval), "in motion" (working), and "broken" (failed). Ready is the - * unlabeled resting state. + * unlabeled resting state; waiting (session status "idle") is the agent + * parked on open background tasks — grey like working, not a false Done. */ -export type ThreadListV2Status = "approval" | "input" | "working" | "failed" | "ready"; +export type ThreadListV2Status = "approval" | "input" | "working" | "waiting" | "failed" | "ready"; // Settled-tail paging: recent history is the common lookup; the deep tail // stays behind an explicit Show more. Shared by the compact Home list and @@ -30,6 +31,9 @@ export function resolveThreadListV2Status( if (thread.session?.status === "running" || thread.session?.status === "starting") { return "working"; } + if (thread.session?.status === "idle") { + return "waiting"; + } if (thread.session?.status === "error") { return "failed"; } diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 74ece50cd31..44a5d2c5821 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -3174,6 +3174,207 @@ describe("ProviderRuntimeIngestion", () => { ).toBe("# Plan title"); }); + it("parks the session on idle when a turn completes with open tasks", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-waiting-turn-started"), + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: now, + turnId: asTurnId("turn-waiting-1"), + }); + harness.emit({ + type: "task.started", + eventId: asEventId("evt-waiting-task-started"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-waiting-1"), + payload: { + taskId: "bg-task-1", + description: "Babysit PR checks", + }, + }); + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-waiting-turn-completed"), + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:01.000Z", + turnId: asTurnId("turn-waiting-1"), + payload: { + state: "completed", + }, + }); + + const waiting = await waitForThread( + harness.readModel, + (entry) => entry.session?.status === "idle" && entry.session?.activeTurnId === null, + ); + expect(waiting.session?.lastError).toBeNull(); + + // The wake-up turn flips waiting straight back to running. + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-waiting-wakeup-turn-started"), + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:02.000Z", + turnId: asTurnId("turn-waiting-2"), + }); + harness.emit({ + type: "task.completed", + eventId: asEventId("evt-waiting-task-completed"), + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:03.000Z", + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-waiting-2"), + payload: { + taskId: "bg-task-1", + status: "completed", + }, + }); + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-waiting-wakeup-turn-completed"), + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:04.000Z", + turnId: asTurnId("turn-waiting-2"), + payload: { + state: "completed", + }, + }); + + // With the task closed before this turn ended, the thread is genuinely done. + await waitForThread(harness.readModel, (entry) => entry.session?.status === "ready"); + }); + + it("keeps a turn completion ready when its tasks already completed", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + + harness.emit({ + type: "task.started", + eventId: asEventId("evt-closed-task-started"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-1"), + payload: { + taskId: "fg-task-1", + }, + }); + harness.emit({ + type: "task.completed", + eventId: asEventId("evt-closed-task-completed"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-1"), + payload: { + taskId: "fg-task-1", + status: "completed", + }, + }); + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-closed-task-turn-completed"), + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:01.000Z", + turnId: asTurnId("turn-closed-task"), + payload: { + state: "completed", + }, + }); + + await waitForThread( + harness.readModel, + (entry) => entry.session?.status === "ready" && entry.session?.activeTurnId === null, + ); + }); + + it("demotes a ready session to idle when a task start arrives after turn completion", async () => { + const harness = await createHarness(); + + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-late-task-turn-completed"), + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:00.000Z", + turnId: asTurnId("turn-late-task"), + payload: { + state: "completed", + }, + }); + await waitForThread(harness.readModel, (entry) => entry.session?.status === "ready"); + + harness.emit({ + type: "task.started", + eventId: asEventId("evt-late-task-started"), + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:01.000Z", + threadId: asThreadId("thread-1"), + payload: { + taskId: "late-task-1", + description: "Monitor deploy", + }, + }); + + await waitForThread(harness.readModel, (entry) => entry.session?.status === "idle"); + }); + + it("clears open tasks on session exit so they cannot park a later session", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + + harness.emit({ + type: "task.started", + eventId: asEventId("evt-orphan-task-started"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-1"), + payload: { + taskId: "orphan-task-1", + }, + }); + harness.emit({ + type: "session.exited", + eventId: asEventId("evt-orphan-task-session-exited"), + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:01.000Z", + payload: {}, + }); + await waitForThread(harness.readModel, (entry) => entry.session?.status === "stopped"); + + // A fresh session's turn must not inherit the orphaned task. + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-orphan-task-new-turn-started"), + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:02.000Z", + turnId: asTurnId("turn-after-orphan"), + }); + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-orphan-task-new-turn-completed"), + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:03.000Z", + turnId: asTurnId("turn-after-orphan"), + payload: { + state: "completed", + }, + }); + + await waitForThread(harness.readModel, (entry) => entry.session?.status === "ready"); + }); + it("titles task activities with the task description, including on completion", async () => { const harness = await createHarness(); const now = "2026-01-01T00:00:00.000Z"; diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index a8a51b30260..7264141b021 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -746,6 +746,52 @@ const make = Effect.gen(function* () { ), ); + // Open provider-reported tasks per thread. A turn that completes while any + // remain open parks the session on "idle" (sidebar "Waiting") instead of + // "ready" ("Done"): the agent stopped, but a task completion is expected to + // wake it, and flashing Done in between is a lie. Keyed by provider + // instance so a replaced session can neither inherit nor close a + // predecessor's tasks. Plain mutable state is safe: ingestion is a single + // serial worker. Terminal session statuses sweep the entry. + const openTasksByThreadId = new Map }>(); + + const taskInstanceKey = (event: ProviderRuntimeEvent) => + event.providerInstanceId ?? event.provider; + + /** Returns true when this task is the instance's first open task. */ + const trackTaskStarted = (threadId: ThreadId, event: ProviderRuntimeEvent, taskId: string) => { + const instanceKey = taskInstanceKey(event); + const entry = openTasksByThreadId.get(threadId); + if (!entry || entry.instanceKey !== instanceKey) { + openTasksByThreadId.set(threadId, { instanceKey, taskIds: new Set([taskId]) }); + return true; + } + entry.taskIds.add(taskId); + return false; + }; + + const trackTaskCompleted = (threadId: ThreadId, event: ProviderRuntimeEvent, taskId: string) => { + const entry = openTasksByThreadId.get(threadId); + if (!entry || entry.instanceKey !== taskInstanceKey(event)) { + return; + } + entry.taskIds.delete(taskId); + if (entry.taskIds.size === 0) { + openTasksByThreadId.delete(threadId); + } + }; + + const hasOpenTasks = (threadId: ThreadId, event: ProviderRuntimeEvent) => { + const entry = openTasksByThreadId.get(threadId); + return ( + entry !== undefined && entry.instanceKey === taskInstanceKey(event) && entry.taskIds.size > 0 + ); + }; + + const clearOpenTasks = (threadId: ThreadId) => { + openTasksByThreadId.delete(threadId); + }; + const resolveThreadDetail = Effect.fn("resolveThreadDetail")(function* (threadId: ThreadId) { return yield* projectionSnapshotQuery .getThreadDetailById(threadId) @@ -1370,10 +1416,18 @@ const make = Effect.gen(function* () { event.type === "turn.completed" ) { const status = (() => { + // The agent stopping with provider-reported tasks still open is not + // "ready" (Done): a task completion is expected to wake it. Park on + // "idle" (Waiting) until real activity or session teardown moves it. + const restingStatus = hasOpenTasks(thread.id, event) ? "idle" : "ready"; switch (event.type) { case "session.state.changed": { const runtimeStatus = orchestrationSessionStatusFromRuntimeState(event.payload.state); - return hasPendingTurnStart && runtimeStatus === "ready" ? "starting" : runtimeStatus; + return runtimeStatus === "ready" + ? hasPendingTurnStart + ? "starting" + : restingStatus + : runtimeStatus; } case "turn.started": return "running"; @@ -1382,14 +1436,24 @@ const make = Effect.gen(function* () { case "turn.completed": return normalizeRuntimeTurnState(event.payload.state) === "failed" ? "error" - : "ready"; + : restingStatus; case "session.started": case "thread.started": // Provider thread/session start notifications can arrive during an // active or pending turn; preserve that lifecycle state. - return activeTurnId !== null ? "running" : hasPendingTurnStart ? "starting" : "ready"; + return activeTurnId !== null + ? "running" + : hasPendingTurnStart + ? "starting" + : restingStatus; } })(); + // A session leaving the resting/running loop can no longer be woken + // by its tasks; drop them so a stale set never parks a future + // session on Waiting. + if (status === "stopped" || status === "error" || status === "interrupted") { + clearOpenTasks(thread.id); + } const nextActiveTurnId = event.type === "turn.started" ? (eventTurnId ?? null) @@ -1407,7 +1471,7 @@ const make = Effect.gen(function* () { : event.type === "turn.completed" && normalizeRuntimeTurnState(event.payload.state) === "failed" ? (event.payload.errorMessage ?? thread.session?.lastError ?? "Turn failed") - : status === "ready" + : status === "ready" || status === "idle" ? null : (thread.session?.lastError ?? null); @@ -1671,6 +1735,7 @@ const make = Effect.gen(function* () { } if (event.type === "session.exited") { + clearOpenTasks(thread.id); yield* clearTurnStateForSession(thread.id); } @@ -1682,6 +1747,7 @@ const make = Effect.gen(function* () { : activeTurnId === null || eventTurnId === undefined || sameId(activeTurnId, eventTurnId); if (shouldApplyRuntimeError) { + clearOpenTasks(thread.id); yield* orchestrationEngine.dispatch({ type: "thread.session.set", commandId: yield* providerCommandId(event, "runtime-error-session-set"), @@ -1754,9 +1820,37 @@ const make = Effect.gen(function* () { if (description) { yield* rememberTaskDescription(thread.id, event.payload.taskId, description); } + // task.progress tracks too: it re-opens a task the tracker lost + // (server restart, missed start event) so the thread still parks on + // Waiting instead of a false Done. + const firstOpenTask = trackTaskStarted(thread.id, event, event.payload.taskId); + // A task can outlive its turn: when its start (or surviving progress) + // lands after the turn already settled the session on "ready", demote + // to "idle" now — the turn-completion path never saw the task. + if ( + firstOpenTask && + thread.session?.status === "ready" && + thread.session.activeTurnId === null && + (thread.session.providerInstanceId ?? thread.session.providerName) === + taskInstanceKey(event) + ) { + yield* orchestrationEngine.dispatch({ + type: "thread.session.set", + commandId: yield* providerCommandId(event, "open-task-session-idle"), + threadId: thread.id, + session: { + ...thread.session, + status: "idle", + lastError: null, + updatedAt: now, + }, + createdAt: now, + }); + } } let taskTitle: string | undefined; if (event.type === "task.completed") { + trackTaskCompleted(thread.id, event, event.payload.taskId); taskTitle = yield* lookupTaskDescription(thread.id, event.payload.taskId); if (!taskTitle) { const threadDetail = yield* getLoadedThreadDetail(); diff --git a/apps/web/src/components/Sidebar.logic.test.ts b/apps/web/src/components/Sidebar.logic.test.ts index 59784bf8fac..e278fab237f 100644 --- a/apps/web/src/components/Sidebar.logic.test.ts +++ b/apps/web/src/components/Sidebar.logic.test.ts @@ -628,6 +628,22 @@ describe("resolveSidebarV2Status", () => { ).toBe("working"); }); + it("reports waiting for an idle session, below approval and input", () => { + expect( + resolveSidebarV2Status({ + ...idle, + session: { ...session, status: "idle" as const, activeTurnId: null }, + }), + ).toBe("waiting"); + expect( + resolveSidebarV2Status({ + ...idle, + hasPendingApprovals: true, + session: { ...session, status: "idle" as const, activeTurnId: null }, + }), + ).toBe("approval"); + }); + it("reports failed only while the session status is error", () => { expect( resolveSidebarV2Status({ diff --git a/apps/web/src/components/Sidebar.logic.ts b/apps/web/src/components/Sidebar.logic.ts index 7aee3100d0e..2915145dcaa 100644 --- a/apps/web/src/components/Sidebar.logic.ts +++ b/apps/web/src/components/Sidebar.logic.ts @@ -395,13 +395,16 @@ export function resolveThreadRowClassName(input: { } // ── Sidebar v2 status model ───────────────────────────────────────── -// Five visual states, three colors: color is reserved for "act now" +// Six visual states, three colors: color is reserved for "act now" // (approval), "in motion" (working), and "broken" (failed). Ready is the // unlabeled resting state — the agent stopped and is waiting on the user, -// whether it finished, asked a question, or proposed a plan. +// whether it finished, asked a question, or proposed a plan. Waiting +// (session status "idle") is the agent stopped with background tasks still +// open: not the user's turn yet, so it renders grey like working, not as a +// false Done. // Unread completion is tracked separately: it describes whether a ready // thread needs attention, not what the thread is currently doing. -export type SidebarV2Status = "approval" | "input" | "working" | "failed" | "ready"; +export type SidebarV2Status = "approval" | "input" | "working" | "waiting" | "failed" | "ready"; type SidebarV2StatusInput = Pick< SidebarThreadSummary, @@ -418,6 +421,9 @@ export function resolveSidebarV2Status(thread: SidebarV2StatusInput): SidebarV2S if (thread.session?.status === "running" || thread.session?.status === "starting") { return "working"; } + if (thread.session?.status === "idle") { + return "waiting"; + } if (thread.session?.status === "error") { return "failed"; } diff --git a/apps/web/src/components/SidebarV2.tsx b/apps/web/src/components/SidebarV2.tsx index e018491348e..ec96d271c57 100644 --- a/apps/web/src/components/SidebarV2.tsx +++ b/apps/web/src/components/SidebarV2.tsx @@ -431,7 +431,8 @@ const SidebarV2Row = memo(function SidebarV2Row(props: { // findable. In-flight rows recede the same as read-ready ones (inbox-zero: // working threads aren't your problem yet) — only the colored status label // stands out. - const isInFlight = status === "working" || status === "approval" || status === "input"; + const isInFlight = + status === "working" || status === "waiting" || status === "approval" || status === "input"; const shouldRecede = (status === "ready" || isInFlight) && !isUnread && !isWoke && !props.isActive && !isSelected; // Status hues follow the system-wide convention set by sidebar v1 and the @@ -445,37 +446,45 @@ const SidebarV2Row = memo(function SidebarV2Row(props: { className: "animate-sidebar-working-text text-sky-600 motion-reduce:animate-none dark:text-sky-400", } - : status === "approval" + : status === "waiting" ? { - label: "Approval", - icon: null, - className: "text-amber-700 dark:text-amber-300", + // Same shape as Working but colorless and static: the agent is + // parked on open background work, not asking anything of you. + label: "Waiting", + icon: "working" as const, + className: "text-sidebar-muted-foreground", } - : status === "input" + : status === "approval" ? { - label: "Input", + label: "Approval", icon: null, - className: "text-indigo-600 dark:text-indigo-300", + className: "text-amber-700 dark:text-amber-300", } - : status === "failed" + : status === "input" ? { - label: "Failed", + label: "Input", icon: null, - className: "text-red-700 dark:text-red-300", + className: "text-indigo-600 dark:text-indigo-300", } - : isWoke + : status === "failed" ? { - label: "Woke", - icon: "woke" as const, - className: "text-amber-700 dark:text-amber-300", + label: "Failed", + icon: null, + className: "text-red-700 dark:text-red-300", } - : isUnread + : isWoke ? { - label: "Done", - icon: "done" as const, - className: "text-emerald-700 dark:text-emerald-300", + label: "Woke", + icon: "woke" as const, + className: "text-amber-700 dark:text-amber-300", } - : null; + : isUnread + ? { + label: "Done", + icon: "done" as const, + className: "text-emerald-700 dark:text-emerald-300", + } + : null; const gitCwd = thread.worktreePath ?? props.projectCwd; const gitStatus = useEnvironmentQuery( @@ -897,7 +906,7 @@ const SidebarV2Row = memo(function SidebarV2Row(props: { wrapper around the ticking duration would make screen readers announce every second. */} {topStatus.label} - {status === "working" ? ( + {status === "working" || status === "waiting" ? ( From 528adb2fe055000a5e7b195800607269b519590e Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 23 Jul 2026 18:19:56 -0700 Subject: [PATCH 2/4] Harden open-task tracking against stale and replayed events Review fixes: gate the terminal-status task sweep behind the lifecycle guard so rejected stale events cannot erase the active turn's tasks; ignore task starts from a provider instance that doesn't match the projected session instead of letting them displace the current set; tombstone completed task ids so late/duplicate progress cannot re-open a finished task and park the thread on Waiting. Co-Authored-By: Claude Fable 5 --- .../Layers/ProviderRuntimeIngestion.test.ts | 95 +++++++++++++++++++ .../Layers/ProviderRuntimeIngestion.ts | 73 ++++++++++---- 2 files changed, 149 insertions(+), 19 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 44a5d2c5821..ab9db9cbb96 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -3327,6 +3327,101 @@ describe("ProviderRuntimeIngestion", () => { await waitForThread(harness.readModel, (entry) => entry.session?.status === "idle"); }); + it("ignores late task progress for a completed task", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + + harness.emit({ + type: "task.started", + eventId: asEventId("evt-tombstone-task-started"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-1"), + payload: { + taskId: "tombstone-task-1", + }, + }); + harness.emit({ + type: "task.completed", + eventId: asEventId("evt-tombstone-task-completed"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-1"), + payload: { + taskId: "tombstone-task-1", + status: "completed", + }, + }); + // Duplicate/late progress must not re-open the finished task… + harness.emit({ + type: "task.progress", + eventId: asEventId("evt-tombstone-task-late-progress"), + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:01.000Z", + threadId: asThreadId("thread-1"), + payload: { + taskId: "tombstone-task-1", + description: "late duplicate progress", + }, + }); + // …so a subsequent turn completion settles on ready, not idle. + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-tombstone-turn-completed"), + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:02.000Z", + turnId: asTurnId("turn-tombstone"), + payload: { + state: "completed", + }, + }); + + await waitForThread(harness.readModel, (entry) => entry.session?.status === "ready"); + }); + + it("ignores task events from a stale provider instance", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + + // Current instance ("codex", matching the seeded session) opens a task. + harness.emit({ + type: "task.started", + eventId: asEventId("evt-current-instance-task-started"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-1"), + payload: { + taskId: "current-task-1", + }, + }); + // A delayed start from a replaced instance must not displace it. + harness.emit({ + type: "task.started", + eventId: asEventId("evt-stale-instance-task-started"), + provider: ProviderDriverKind.make("claudeAgent"), + createdAt: "2026-01-01T00:00:01.000Z", + threadId: asThreadId("thread-1"), + payload: { + taskId: "stale-task-1", + }, + }); + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-stale-instance-turn-completed"), + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:02.000Z", + turnId: asTurnId("turn-stale-instance"), + payload: { + state: "completed", + }, + }); + + // The current instance's open task still parks the thread on Waiting. + await waitForThread(harness.readModel, (entry) => entry.session?.status === "idle"); + }); + it("clears open tasks on session exit so they cannot park a later session", async () => { const harness = await createHarness(); const now = "2026-01-01T00:00:00.000Z"; diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 7264141b021..b9a5b3d63e8 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -751,23 +751,55 @@ const make = Effect.gen(function* () { // "ready" ("Done"): the agent stopped, but a task completion is expected to // wake it, and flashing Done in between is a lie. Keyed by provider // instance so a replaced session can neither inherit nor close a - // predecessor's tasks. Plain mutable state is safe: ingestion is a single - // serial worker. Terminal session statuses sweep the entry. - const openTasksByThreadId = new Map }>(); + // predecessor's tasks; closed ids are kept as tombstones so a late or + // replayed start/progress cannot resurrect a finished task. Plain mutable + // state is safe: ingestion is a single serial worker. Terminal session + // statuses sweep the entry. + const openTasksByThreadId = new Map< + ThreadId, + { instanceKey: string; taskIds: Set; closedTaskIds: Set } + >(); const taskInstanceKey = (event: ProviderRuntimeEvent) => event.providerInstanceId ?? event.provider; - /** Returns true when this task is the instance's first open task. */ - const trackTaskStarted = (threadId: ThreadId, event: ProviderRuntimeEvent, taskId: string) => { + const sessionInstanceKeyFor = (session: { + providerInstanceId?: string | undefined; + providerName: string | null; + }): string | null => session.providerInstanceId ?? session.providerName; + + /** + * Returns true when this task is the instance's first open task. The + * projected session adjudicates cross-instance races: a delayed start from + * a replaced instance must neither seed a phantom set nor displace the + * current instance's tasks, while a current-instance start may reclaim an + * entry a dead instance leaked. + */ + const trackTaskStarted = ( + threadId: ThreadId, + event: ProviderRuntimeEvent, + taskId: string, + sessionInstanceKey: string | null, + ) => { const instanceKey = taskInstanceKey(event); + if (sessionInstanceKey !== null && instanceKey !== sessionInstanceKey) { + return false; + } const entry = openTasksByThreadId.get(threadId); if (!entry || entry.instanceKey !== instanceKey) { - openTasksByThreadId.set(threadId, { instanceKey, taskIds: new Set([taskId]) }); + openTasksByThreadId.set(threadId, { + instanceKey, + taskIds: new Set([taskId]), + closedTaskIds: new Set(), + }); return true; } + if (entry.closedTaskIds.has(taskId)) { + return false; + } + const wasEmpty = entry.taskIds.size === 0; entry.taskIds.add(taskId); - return false; + return wasEmpty; }; const trackTaskCompleted = (threadId: ThreadId, event: ProviderRuntimeEvent, taskId: string) => { @@ -776,9 +808,7 @@ const make = Effect.gen(function* () { return; } entry.taskIds.delete(taskId); - if (entry.taskIds.size === 0) { - openTasksByThreadId.delete(threadId); - } + entry.closedTaskIds.add(taskId); }; const hasOpenTasks = (threadId: ThreadId, event: ProviderRuntimeEvent) => { @@ -1448,12 +1478,6 @@ const make = Effect.gen(function* () { : restingStatus; } })(); - // A session leaving the resting/running loop can no longer be woken - // by its tasks; drop them so a stale set never parks a future - // session on Waiting. - if (status === "stopped" || status === "error" || status === "interrupted") { - clearOpenTasks(thread.id); - } const nextActiveTurnId = event.type === "turn.started" ? (eventTurnId ?? null) @@ -1476,6 +1500,13 @@ const make = Effect.gen(function* () { : (thread.session?.lastError ?? null); if (shouldApplyThreadLifecycle) { + // A session leaving the resting/running loop can no longer be woken + // by its tasks; drop them so a stale set never parks a future + // session on Waiting. Gated like the session.set below: a rejected + // stale event must not erase the active turn's tasks. + if (status === "stopped" || status === "error" || status === "interrupted") { + clearOpenTasks(thread.id); + } if (event.type === "turn.started" && acceptedTurnStartedSourcePlan !== null) { yield* markSourceProposedPlanImplemented( acceptedTurnStartedSourcePlan.sourceThreadId, @@ -1823,7 +1854,12 @@ const make = Effect.gen(function* () { // task.progress tracks too: it re-opens a task the tracker lost // (server restart, missed start event) so the thread still parks on // Waiting instead of a false Done. - const firstOpenTask = trackTaskStarted(thread.id, event, event.payload.taskId); + const firstOpenTask = trackTaskStarted( + thread.id, + event, + event.payload.taskId, + thread.session ? sessionInstanceKeyFor(thread.session) : null, + ); // A task can outlive its turn: when its start (or surviving progress) // lands after the turn already settled the session on "ready", demote // to "idle" now — the turn-completion path never saw the task. @@ -1831,8 +1867,7 @@ const make = Effect.gen(function* () { firstOpenTask && thread.session?.status === "ready" && thread.session.activeTurnId === null && - (thread.session.providerInstanceId ?? thread.session.providerName) === - taskInstanceKey(event) + sessionInstanceKeyFor(thread.session) === taskInstanceKey(event) ) { yield* orchestrationEngine.dispatch({ type: "thread.session.set", From 514fccf5297ec832fdd1e5ef0a1d31de7958ec05 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Fri, 24 Jul 2026 00:21:02 -0700 Subject: [PATCH 3/4] Resolve Waiting when the last task is stopped rather than completed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A completed/failed final task wakes the agent (its notification is the wake signal), so Waiting flips straight to Working. A stopped final task is a kill — no wake follows — and previously left a live session parked on Waiting until teardown. Flip that case back to ready, and tombstone completions that outrun their start event. Co-Authored-By: Claude Fable 5 --- .../Layers/ProviderRuntimeIngestion.test.ts | 99 +++++++++++++++++++ .../Layers/ProviderRuntimeIngestion.ts | 48 ++++++++- 2 files changed, 143 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index ab9db9cbb96..5ff794cc479 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -3327,6 +3327,105 @@ describe("ProviderRuntimeIngestion", () => { await waitForThread(harness.readModel, (entry) => entry.session?.status === "idle"); }); + it("returns a waiting thread to ready when its last task is stopped", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + + harness.emit({ + type: "task.started", + eventId: asEventId("evt-stopped-task-started"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-1"), + payload: { + taskId: "stopped-task-1", + }, + }); + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-stopped-task-turn-completed"), + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:01.000Z", + turnId: asTurnId("turn-stopped-task"), + payload: { + state: "completed", + }, + }); + await waitForThread(harness.readModel, (entry) => entry.session?.status === "idle"); + + // A stopped task never wakes the agent, so Waiting must resolve here. + harness.emit({ + type: "task.completed", + eventId: asEventId("evt-stopped-task-completed"), + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:02.000Z", + threadId: asThreadId("thread-1"), + payload: { + taskId: "stopped-task-1", + status: "stopped", + }, + }); + + await waitForThread(harness.readModel, (entry) => entry.session?.status === "ready"); + }); + + it("keeps a waiting thread waiting when its last task completes (the wake turn follows)", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + + harness.emit({ + type: "task.started", + eventId: asEventId("evt-wake-task-started"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-1"), + payload: { + taskId: "wake-task-1", + }, + }); + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-wake-task-turn-completed"), + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:01.000Z", + turnId: asTurnId("turn-wake-task"), + payload: { + state: "completed", + }, + }); + await waitForThread(harness.readModel, (entry) => entry.session?.status === "idle"); + + // A successful completion wakes the agent; the status must go straight + // to running with no intermediate ready flash. + harness.emit({ + type: "task.completed", + eventId: asEventId("evt-wake-task-completed"), + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:02.000Z", + threadId: asThreadId("thread-1"), + payload: { + taskId: "wake-task-1", + status: "completed", + }, + }); + harness.emit({ + type: "turn.started", + eventId: asEventId("evt-wake-task-wake-turn"), + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:03.000Z", + turnId: asTurnId("turn-wake-follow-up"), + }); + + const thread = await waitForThread( + harness.readModel, + (entry) => entry.session?.status === "running", + ); + expect(thread.session?.activeTurnId).toBe("turn-wake-follow-up"); + }); + it("ignores late task progress for a completed task", async () => { const harness = await createHarness(); const now = "2026-01-01T00:00:00.000Z"; diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index b9a5b3d63e8..4aa6b889e2c 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -802,13 +802,27 @@ const make = Effect.gen(function* () { return wasEmpty; }; + /** Returns true when this completion closed the instance's last open task. */ const trackTaskCompleted = (threadId: ThreadId, event: ProviderRuntimeEvent, taskId: string) => { + const instanceKey = taskInstanceKey(event); const entry = openTasksByThreadId.get(threadId); - if (!entry || entry.instanceKey !== taskInstanceKey(event)) { - return; + if (!entry) { + // Completion outrunning its start: tombstone it anyway, or the late + // start would open a phantom task and park the thread on Waiting for + // a completion that already happened. + openTasksByThreadId.set(threadId, { + instanceKey, + taskIds: new Set(), + closedTaskIds: new Set([taskId]), + }); + return false; } - entry.taskIds.delete(taskId); + if (entry.instanceKey !== instanceKey) { + return false; + } + const removed = entry.taskIds.delete(taskId); entry.closedTaskIds.add(taskId); + return removed && entry.taskIds.size === 0; }; const hasOpenTasks = (threadId: ThreadId, event: ProviderRuntimeEvent) => { @@ -1885,7 +1899,33 @@ const make = Effect.gen(function* () { } let taskTitle: string | undefined; if (event.type === "task.completed") { - trackTaskCompleted(thread.id, event, event.payload.taskId); + const closedLastTask = trackTaskCompleted(thread.id, event, event.payload.taskId); + // A completed/failed final task wakes the agent — its notification is + // the wake signal — so the thread goes Waiting → Working directly and + // flipping to ready here would flash the very Done this feature + // removes. A STOPPED final task is the opposite: someone killed it, + // no wake follows, and a live session would sit on Waiting forever. + // Only that case flips back to ready. + if ( + closedLastTask && + event.payload.status === "stopped" && + thread.session?.status === "idle" && + thread.session.activeTurnId === null && + sessionInstanceKeyFor(thread.session) === taskInstanceKey(event) + ) { + yield* orchestrationEngine.dispatch({ + type: "thread.session.set", + commandId: yield* providerCommandId(event, "last-task-session-ready"), + threadId: thread.id, + session: { + ...thread.session, + status: "ready", + lastError: null, + updatedAt: now, + }, + createdAt: now, + }); + } taskTitle = yield* lookupTaskDescription(thread.id, event.payload.taskId); if (!taskTitle) { const threadDetail = yield* getLoadedThreadDetail(); From 368e0e15274419de7d6cb9d5368a5a2afc7776dd Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Fri, 24 Jul 2026 00:26:02 -0700 Subject: [PATCH 4/4] Let an explicit task start re-open a tombstoned task id MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tombstones exist to stop late/duplicate progress from resurrecting a finished task, but providers reuse task ids when an agent is resumed — a fresh task.started after completion is real work, not a replay. Progress stays blocked by the tombstone; an explicit start clears it. Co-Authored-By: Claude Fable 5 --- .../Layers/ProviderRuntimeIngestion.test.ts | 53 +++++++++++++++++++ .../Layers/ProviderRuntimeIngestion.ts | 8 ++- 2 files changed, 60 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 5ff794cc479..7f8ada41d63 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -3479,6 +3479,59 @@ describe("ProviderRuntimeIngestion", () => { await waitForThread(harness.readModel, (entry) => entry.session?.status === "ready"); }); + it("re-opens a completed task on an explicit restart (resumed agent)", async () => { + const harness = await createHarness(); + const now = "2026-01-01T00:00:00.000Z"; + + harness.emit({ + type: "task.started", + eventId: asEventId("evt-resumed-task-started"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-1"), + payload: { + taskId: "resumed-task-1", + }, + }); + harness.emit({ + type: "task.completed", + eventId: asEventId("evt-resumed-task-completed"), + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:01.000Z", + threadId: asThreadId("thread-1"), + payload: { + taskId: "resumed-task-1", + status: "completed", + }, + }); + // The agent is resumed: the same task id starts again. The tombstone + // must yield to the explicit start… + harness.emit({ + type: "task.started", + eventId: asEventId("evt-resumed-task-restarted"), + provider: ProviderDriverKind.make("codex"), + createdAt: "2026-01-01T00:00:02.000Z", + threadId: asThreadId("thread-1"), + payload: { + taskId: "resumed-task-1", + }, + }); + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-resumed-task-turn-completed"), + provider: ProviderDriverKind.make("codex"), + threadId: asThreadId("thread-1"), + createdAt: "2026-01-01T00:00:03.000Z", + turnId: asTurnId("turn-resumed-task"), + payload: { + state: "completed", + }, + }); + + // …so the turn parks on Waiting for the resumed agent. + await waitForThread(harness.readModel, (entry) => entry.session?.status === "idle"); + }); + it("ignores task events from a stale provider instance", async () => { const harness = await createHarness(); const now = "2026-01-01T00:00:00.000Z"; diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 4aa6b889e2c..6afb6e765e7 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -795,7 +795,13 @@ const make = Effect.gen(function* () { return true; } if (entry.closedTaskIds.has(taskId)) { - return false; + // task.progress for a finished task is a late duplicate and stays + // ignored. task.started is authoritative: providers reuse task ids + // when an agent is resumed, so an explicit start re-opens the task. + if (event.type !== "task.started") { + return false; + } + entry.closedTaskIds.delete(taskId); } const wasEmpty = entry.taskIds.size === 0; entry.taskIds.add(taskId);