fix(web): derive thread runtime phase from canonical turn state

apps/web/src/__tests__/canonical-runtime-reconcile.test.ts

@@ -0,0 +1,147 @@
+import { describe, expect, it, beforeEach, vi } from "vitest";
+import type { CanonicalAgentEventEnvelope } from "@mcode/contracts";
+import { useThreadStore } from "@/stores/threadStore";
+import {
+  resetThreadStoreForTests,
+  seedThreadRecord,
+  readThreadField,
+} from "@/stores/thread-store-test-utils";
+import { clearRecordCache } from "@/features/conversation/hydration/record-cache";
+import { mockTransport } from "./mocks/transport";
+
+vi.mock("@/transport", async () => ({
+  ...(await vi.importActual("@/transport")),
+  getTransport: () => mockTransport,
+}));
+
+const THREAD_ID = "thread-1";
+const TURN_ID = "turn-1";
+const EXECUTION_ID = "00000000-0000-4000-8000-000000000001";
+const OTHER_EXECUTION_ID = "00000000-0000-4000-8000-000000000099";
+const NOW = "2026-08-11T12:00:00.000Z";
+
+function envelope(
+  eventId: string,
+  acceptedSequence: number,
+  executionId: string,
+  payload: CanonicalAgentEventEnvelope["payload"],
+): CanonicalAgentEventEnvelope {
+  return {
+    eventId,
+    routing: {
+      threadId: THREAD_ID,
+      turnId: payload.type === "thread.recorded" ? undefined : TURN_ID,
+      executionId,
+    },
+    sourceProviderId: "codex",
+    sourceIdentities: [],
+    acceptedSequence,
+    durableRevision: acceptedSequence,
+    serverTimestamps: { acceptedAt: NOW, persistedAt: NOW },
+    payload,
+  };
+}
+
+function turnEvents(executionId: string, terminal?: "completed" | "interrupted"): CanonicalAgentEventEnvelope[] {
+  const events: CanonicalAgentEventEnvelope[] = [
+    envelope("e1", 1, executionId, {
+      type: "thread.recorded",
+      thread: {
+        id: THREAD_ID,
+        workspaceId: "workspace-1",
+        rootThreadId: THREAD_ID,
+        providerId: "codex",
+        providerIdentities: [],
+        activityState: "Active",
+        conversationRevision: 1,
+        rosterRevision: 0,
+        createdAt: NOW,
+        updatedAt: NOW,
+      },
+    }),
+    envelope("e2", 2, executionId, {
+      type: "turn.created",
+      turn: {
+        id: TURN_ID,
+        threadId: THREAD_ID,
+        status: "Pending",
+        trigger: { kind: "user" },
+        permissionMode: "full",
+        approvalReviewMode: "manual",
+        approvalReviewReason: "manual-requested",
+        providerIdentities: [],
+        startedAt: null,
+        endedAt: null,
+        createdAt: NOW,
+        updatedAt: NOW,
+      },
+    }),
+    envelope("e3", 3, executionId, { type: "turn.started", startedAt: NOW }),
+  ];
+  if (terminal === "completed") {
+    events.push(envelope("e4", 4, executionId, { type: "turn.completed", endedAt: NOW }));
+  } else if (terminal === "interrupted") {
+    events.push(envelope("e4", 4, executionId, { type: "turn.interrupted", endedAt: NOW, reason: "restart" }));
+  }
+  return events;
+}
+
+function seedRuntime(phase: "running" | "idle", turnExecutionId: string | null, running: boolean) {
+  useThreadStore.setState({
+    records: seedThreadRecord(THREAD_ID, { runtimePhase: phase, turnExecutionId }),
+    runningThreadIds: running ? new Set([THREAD_ID]) : new Set(),
+  });
+}
+
+describe("canonical runtime reconciliation", () => {
+  beforeEach(() => {
+    clearRecordCache();
+    resetThreadStoreForTests();
+    vi.clearAllMocks();
+  });
+
+  it("clears a stale running flag when the correlated canonical turn terminates", () => {
+    seedRuntime("running", EXECUTION_ID, true);
+
+    useThreadStore.getState().handleCanonicalAgentEvents(THREAD_ID, turnEvents(EXECUTION_ID, "completed"));
+
+    expect(readThreadField(THREAD_ID, (r) => r.runtimePhase)).toBe("completed");
+    expect(useThreadStore.getState().runningThreadIds.has(THREAD_ID)).toBe(false);
+  });
+
+  it("claims an idle record when canonical reports a running turn", () => {
+    seedRuntime("idle", null, false);
+
+    useThreadStore.getState().handleCanonicalAgentEvents(THREAD_ID, turnEvents(EXECUTION_ID));
+
+    expect(readThreadField(THREAD_ID, (r) => r.runtimePhase)).toBe("running");
+    expect(useThreadStore.getState().runningThreadIds.has(THREAD_ID)).toBe(true);
+  });
+
+  it("keeps the optimistic running window while no canonical turn correlates", () => {
+    seedRuntime("running", null, true);
+
+    useThreadStore.getState().handleCanonicalAgentEvents(THREAD_ID, turnEvents(EXECUTION_ID, "completed"));
+
+    expect(readThreadField(THREAD_ID, (r) => r.runtimePhase)).toBe("running");
+    expect(useThreadStore.getState().runningThreadIds.has(THREAD_ID)).toBe(true);
+  });
+
+  it("ignores a canonical turn carrying a different execution identity", () => {
+    seedRuntime("running", OTHER_EXECUTION_ID, true);
+
+    useThreadStore.getState().handleCanonicalAgentEvents(THREAD_ID, turnEvents(EXECUTION_ID, "completed"));
+
+    expect(readThreadField(THREAD_ID, (r) => r.runtimePhase)).toBe("running");
+    expect(useThreadStore.getState().runningThreadIds.has(THREAD_ID)).toBe(true);
+  });
+
+  it("maps an interrupted canonical turn to the interrupted phase", () => {
+    seedRuntime("running", EXECUTION_ID, true);
+
+    useThreadStore.getState().handleCanonicalAgentEvents(THREAD_ID, turnEvents(EXECUTION_ID, "interrupted"));
+
+    expect(readThreadField(THREAD_ID, (r) => r.runtimePhase)).toBe("interrupted");
+    expect(useThreadStore.getState().runningThreadIds.has(THREAD_ID)).toBe(false);
+  });
+});

apps/web/src/__tests__/connection.test.ts

@@ -234,6 +234,54 @@ describe("4001 auth failure handling", () => {
   });
 });
 
+describe("liveness heartbeat", () => {
+  it("closes a silent socket when the heartbeat goes unanswered", async () => {
+    vi.useFakeTimers();
+    const statusSpy = vi.fn();
+    const transport = createWsTransport("ws://localhost:1234", {
+      onStatusChange: statusSpy,
+    });
+    mockWsInstance.simulateOpen();
+
+    // 30s interval fires the probe; the mock never responds; the 10s
+    // heartbeat timeout closes the socket and drives the reconnect path.
+    await vi.advanceTimersByTimeAsync(30_000);
+    statusSpy.mockClear();
+    await vi.advanceTimersByTimeAsync(10_000);
+
+    expect(statusSpy).toHaveBeenCalledWith("reconnecting");
+    transport.close();
+    vi.useRealTimers();
+  });
+
+  it("keeps the socket open when the heartbeat is answered", async () => {
+    vi.useFakeTimers();
+    const statusSpy = vi.fn();
+    const transport = createWsTransport("ws://localhost:1234", {
+      onStatusChange: statusSpy,
+    });
+    mockWsInstance.simulateOpen();
+    const sendSpy = vi.spyOn(mockWsInstance, "send");
+
+    await vi.advanceTimersByTimeAsync(30_000);
+    const heartbeat = sendSpy.mock.calls
+      .map(([raw]) => JSON.parse(raw as string) as { id: string; method?: string })
+      .find((message) => message.method === "app.version");
+    expect(heartbeat).toBeDefined();
+    mockWsInstance.onmessage?.({
+      data: JSON.stringify({ id: heartbeat!.id, result: "0.0.1" }),
+    });
+
+    await vi.advanceTimersByTimeAsync(10_000);
+    expect(mockWsInstance.readyState).toBe(1);
+    statusSpy.mockClear();
+    await vi.advanceTimersByTimeAsync(10_000);
+    expect(statusSpy).not.toHaveBeenCalledWith("reconnecting");
+    transport.close();
+    vi.useRealTimers();
+  });
+});
+
 /**
  * Browser-faithful socket: send() throws while CONNECTING and silently
  * discards while CLOSING/CLOSED, matching the WebSocket spec. The silent

apps/web/src/features/conversation/messages/__tests__/turn-lifecycle.test.tsx

@@ -25,6 +25,7 @@ function canonicalTurn(status: AgentTurn["status"], trigger: AgentTurn["trigger"
   return {
     id: "canonical-turn", threadId: THREAD, status, trigger, permissionMode: "full",
     approvalReviewMode: "manual", approvalReviewReason: "manual-requested",
+    executionId: "00000000-0000-4000-8000-000000000042",
     providerIdentities: [], startedAt: NOW, endedAt: null, createdAt: NOW, updatedAt: NOW,
   };
 }

apps/web/src/stores/thread-lifecycle.ts

@@ -1,24 +1,53 @@
-import type { AgentTurn, TurnRuntimePhase } from "@mcode/contracts";
+import type { AgentTurn, AgentTurnStatus, TurnRuntimePhase } from "@mcode/contracts";
 import type { ThreadRecord } from "./thread-record";
 
-/** Selects the provider-owned child lifecycle when no local execution is active. */
-export function getCanonicalLifecycleTurn(threadId: string, record: ThreadRecord): AgentTurn | undefined {
-  if (record.runtimePhase === "running" || record.runtimePhase === "finalizing") return undefined;
-  const latest = Object.values(record.canonicalAgent.state.turns)
+function latestCanonicalTurn(threadId: string, record: ThreadRecord): AgentTurn | undefined {
+  return Object.values(record.canonicalAgent.state.turns)
     .filter((turn) => turn.threadId === threadId)
     .sort((left, right) => Date.parse(left.startedAt ?? left.createdAt)
       - Date.parse(right.startedAt ?? right.createdAt) || left.id.localeCompare(right.id))
     .at(-1);
+}
+
+/** Selects the provider-owned child lifecycle when no local execution is active. */
+export function getCanonicalLifecycleTurn(threadId: string, record: ThreadRecord): AgentTurn | undefined {
+  const latest = latestCanonicalTurn(threadId, record);
   if (latest?.trigger.kind !== "child") return undefined;
-  if (record.runtimePhase !== "idle" && (latest.status === "Pending" || latest.status === "Running")) return undefined;
+  // A runtime phase stamped by this same canonical turn must not gate it out;
+  // only a locally-owned phase suppresses the child lifecycle.
+  const locallyOwnedPhase = getCanonicalRuntimeTurn(threadId, record)?.id === latest.id
+    ? "idle"
+    : record.runtimePhase;
+  if (locallyOwnedPhase === "running" || locallyOwnedPhase === "finalizing") return undefined;
+  if (locallyOwnedPhase !== "idle" && (latest.status === "Pending" || latest.status === "Running")) return undefined;
   return latest;
 }
 
-/** Resolves the lifecycle shared by transcript, Composer, and running-thread indicators. */
-export function getThreadRuntimePhase(threadId: string, record: ThreadRecord): TurnRuntimePhase {
-  const turn = getCanonicalLifecycleTurn(threadId, record);
-  if (!turn) return record.runtimePhase;
-  switch (turn.status) {
+/**
+ * Canonical turn that owns this record's runtime phase, once correlation is
+ * proven. A tracked local execution only follows the canonical turn carrying
+ * the same execution identity, so an older persisted turn cannot clear or
+ * resurrect a live run while a matching terminal can clear a stale one.
+ */
+export function getCanonicalRuntimeTurn(threadId: string, record: ThreadRecord): AgentTurn | undefined {
+  if (record.runtimePhase === "finalizing") return undefined;
+  const latest = latestCanonicalTurn(threadId, record);
+  if (!latest) return undefined;
+  if (record.turnExecutionId !== null) {
+    return latest.executionId === record.turnExecutionId ? latest : undefined;
+  }
+  // Optimistic sends run without an identity until turnStarted lands; a prior
+  // terminal turn must not cancel that window.
+  if (record.runtimePhase === "running") return undefined;
+  // Without an identity, canonical Pending or Running claims only idle
+  // records; terminal truth may claim any non-busy record.
+  if ((latest.status === "Pending" || latest.status === "Running") && record.runtimePhase !== "idle") return undefined;
+  return latest;
+}
+
+/** Runtime phase equivalent of one canonical turn status. */
+export function phaseForTurnStatus(status: AgentTurnStatus): TurnRuntimePhase {
+  switch (status) {
     case "Pending":
     case "Running": return "running";
     case "Completed": return "completed";
@@ -28,6 +57,17 @@ export function getThreadRuntimePhase(threadId: string, record: ThreadRecord): T
   }
 }
 
+/** Canonical-owned runtime phase for this record, or null when legacy owns it. */
+export function getCanonicalRuntimePhase(threadId: string, record: ThreadRecord): TurnRuntimePhase | null {
+  const turn = getCanonicalRuntimeTurn(threadId, record);
+  return turn ? phaseForTurnStatus(turn.status) : null;
+}
+
+/** Resolves the lifecycle shared by transcript, Composer, and running-thread indicators. */
+export function getThreadRuntimePhase(threadId: string, record: ThreadRecord): TurnRuntimePhase {
+  return getCanonicalRuntimePhase(threadId, record) ?? record.runtimePhase;
+}
+
 /** Whether the current lifecycle still owns execution, including final persistence. */
 export function isThreadRuntimeActive(threadId: string, record: ThreadRecord): boolean {
   const phase = getThreadRuntimePhase(threadId, record);

apps/web/src/stores/threadStore.ts

@@ -105,7 +105,7 @@ import {
 } from "./thread-store/usage";
 export { mergeProviderUsageSnapshot } from "./thread-store/usage";
 
-import { isThreadRuntimeActive } from "./thread-lifecycle";
+import { getCanonicalRuntimeTurn, isThreadRuntimeActive, phaseForTurnStatus } from "./thread-lifecycle";
 
 function deriveRunningThreadIds(records: Map<string, ThreadRecord>): Set<string> {
   return new Set(
@@ -2715,6 +2715,40 @@ export const useThreadStore = create<ThreadState>((zustandSet, get) => {
     return false;
   };
 
+  // Canonical lifecycle owns runtime state once its turn correlates to the
+  // record, so a dropped legacy terminal event cannot strand the running set
+  // or the composer gate.
+  const reconcileCanonicalRuntime = (
+    records: Map<string, ThreadRecord>,
+    runningThreadIds: Set<string>,
+    threadId: string,
+  ): { records: Map<string, ThreadRecord>; runningThreadIds: Set<string> } => {
+    const record = getThreadRecord(records, threadId);
+    const turn = getCanonicalRuntimeTurn(threadId, record);
+    if (!turn) return { records, runningThreadIds };
+    const phase = phaseForTurnStatus(turn.status);
+    let nextRecords = records;
+    const patch: Partial<ThreadRecord> = {};
+    if (record.runtimePhase !== phase) patch.runtimePhase = phase;
+    // Stamp the execution identity on live claims so later reconciles and the
+    // child lifecycle gate can keep correlating this turn to the record.
+    if (record.turnExecutionId === null && turn.executionId
+      && (turn.status === "Pending" || turn.status === "Running")) {
+      patch.turnExecutionId = turn.executionId;
+    }
+    if (Object.keys(patch).length > 0) {
+      nextRecords = patchThreadRecord(records, threadId, patch);
+    }
+    const running = phase === "running";
+    if (runningThreadIds.has(threadId) === running) {
+      return { records: nextRecords, runningThreadIds };
+    }
+    const nextRunning = new Set(runningThreadIds);
+    if (running) nextRunning.add(threadId);
+    else nextRunning.delete(threadId);
+    return { records: nextRecords, runningThreadIds: nextRunning };
+  };
+
   return {
     records: new Map<string, ThreadRecord>(),
     currentThreadId: null,
@@ -2728,6 +2762,7 @@ export const useThreadStore = create<ThreadState>((zustandSet, get) => {
     flushPendingTextDeltas();
     set((state) => {
       let records = state.records;
+      let runningThreadIds = state.runningThreadIds;
       let changed = false;
       for (const recovery of recoveries) {
         const current = getThreadRecord(records, recovery.threadId);
@@ -2738,9 +2773,14 @@ export const useThreadStore = create<ThreadState>((zustandSet, get) => {
           canonicalAgent: update.replica,
           ...(update.installedSnapshot ? recoverParentNarrative(recovery.threadId, update.replica.state) : {}),
         });
+        const reconciled = reconcileCanonicalRuntime(records, runningThreadIds, recovery.threadId);
+        records = reconciled.records;
+        runningThreadIds = reconciled.runningThreadIds;
         changed = true;
       }
-      return changed ? { records } : {};
+      return changed
+        ? { records, ...(runningThreadIds === state.runningThreadIds ? {} : { runningThreadIds }) }
+        : {};
     });
   },
 
@@ -2757,8 +2797,11 @@ export const useThreadStore = create<ThreadState>((zustandSet, get) => {
       const update = applyCanonicalPushEvents(current.canonicalAgent, threadId, events);
       if (update.replica === current.canonicalAgent) return {};
       accepted = true;
+      const records = patchThreadRecord(state.records, threadId, { canonicalAgent: update.replica });
+      const reconciled = reconcileCanonicalRuntime(records, state.runningThreadIds, threadId);
       return {
-        records: patchThreadRecord(state.records, threadId, { canonicalAgent: update.replica }),
+        records: reconciled.records,
+        ...(reconciled.runningThreadIds === state.runningThreadIds ? {} : { runningThreadIds: reconciled.runningThreadIds }),
       };
     });
     if (

apps/web/src/transport/__tests__/rpc-timeout-recovery.test.ts

@@ -49,6 +49,11 @@ class TimeoutSocket {
     const parsed: unknown = JSON.parse(raw);
     if (!isRpcRequest(parsed)) throw new Error("Expected an RPC request");
     this.requests.push(parsed);
+    // A socket that ignores the liveness probe is closed by the watchdog;
+    // these tests exercise RPC timeouts on a live connection instead.
+    if (parsed.method === "app.version") {
+      queueMicrotask(() => this.respond(parsed, "0.0.1"));
+    }
   }
 
   open(): void {