diff --git a/src/node/services/attemptLedger.test.ts b/src/node/services/attemptLedger.test.ts
index 583524aca48..a707d48ce73 100644
--- a/src/node/services/attemptLedger.test.ts
+++ b/src/node/services/attemptLedger.test.ts
@@ -4,8 +4,10 @@ import * as os from "node:os";
 import * as path from "node:path";
 import { afterEach, beforeEach, describe, expect, mock, test } from "bun:test";
 
+import { TASK_REPORTED_QUEUED_SEND_UNSENT_MESSAGE } from "@/constants/agentMessaging";
+import { Ok } from "@/common/types/result";
 import type { TaskAttemptOutcome } from "@/common/types/tasks";
-import type { ProjectsConfig } from "@/node/config";
+import type { ProjectsConfig, Workspace as WorkspaceConfigEntry } from "@/node/config";
 import type { AgentTaskIndex } from "@/node/services/agentTaskIndex";
 import { AttemptLedger, type AttemptLedgerDeps } from "@/node/services/attemptLedger";
 import { readSubagentAttemptSettlementReceiptStrict } from "@/node/services/subagentAttemptSettlements";
@@ -15,8 +17,11 @@ import type {
   OwnedTaskAttempt,
   SettlementReceiptDecision,
   StreamEndDecision,
+  WorkspaceStopRecord,
 } from "@/node/services/taskService";
+import { findWorkspaceEntry } from "@/node/services/taskUtils";
 import { MutexMap } from "@/node/utils/concurrency/mutexMap";
+import { isTaskAttemptId } from "@/node/utils/taskAttemptId";
 
 const TASK = "task-a";
 const PARENT = "parent-a";
@@ -30,7 +35,10 @@ function send(attemptId: string): AdmittedSend {
 
 /** A config whose only workspace is TASK, with this row. */
 function configWithRow(row: Record<string, unknown> | undefined): ProjectsConfig {
-  const workspaces = row == null ? [] : [{ id: TASK, ...row }];
+  return configWithRows(row == null ? [] : [{ id: TASK, ...row }]);
+}
+
+function configWithRows(workspaces: Array<Record<string, unknown>>): ProjectsConfig {
   return { projects: new Map([["/project", { workspaces }]]) } as unknown as ProjectsConfig;
 }
 
@@ -39,13 +47,20 @@ interface HarnessState {
   sessionsDir: string;
   stopRecordAttemptId: string | undefined;
   hasStopRecord: boolean;
-  sends: Set<AdmittedSend>;
   activeTurn: symbol | undefined;
   liveRegistration: unknown;
   streaming: boolean;
   streamEnd: StreamEndDecision | undefined;
   owners: string[];
   outcomes: Array<TaskAttemptOutcome<AgentTaskReport>>;
+  stopInProgress: boolean;
+  currentStopRecord: WorkspaceStopRecord | undefined;
+  /** Config reads (lenient and strict) per call; the next one wins when set. */
+  nextConfigs: Array<ProjectsConfig | Error>;
+  /** Config edits throw this. */
+  editError: Error | undefined;
+  /** Config edits run on a copy that is never saved (a write that failed silently). */
+  dropEdits: boolean;
 }
 
 /**
@@ -59,13 +74,25 @@ function createHarness() {
     sessionsDir: "/unused",
     stopRecordAttemptId: undefined,
     hasStopRecord: false,
-    sends: new Set(),
     activeTurn: undefined,
     liveRegistration: undefined,
     streaming: false,
     streamEnd: undefined,
     owners: [PARENT, ROOT],
     outcomes: [],
+    stopInProgress: false,
+    currentStopRecord: undefined,
+    nextConfigs: [],
+    editError: undefined,
+    dropEdits: false,
+  };
+  /** Like Config.editConfig: the edit applies to the stored config, unless it fails. */
+  const editConfig = (fn: (config: ProjectsConfig) => ProjectsConfig): Promise<void> => {
+    if (state.editError != null) return Promise.reject(state.editError);
+    if (state.config instanceof Error) return Promise.reject(state.config);
+    const edited = fn(state.dropEdits ? structuredClone(state.config) : state.config);
+    if (!state.dropEdits) state.config = edited;
+    return Promise.resolve();
   };
   const eventLocks = new MutexMap<string>();
   const inspect = mock(
@@ -79,13 +106,22 @@ function createHarness() {
     config: () => ({
       // Like the real Config: an unreadable file throws only for a strict read.
       loadConfigOrDefault: (options?: { throwOnError?: boolean }) => {
+        const next = state.nextConfigs.shift();
+        if (next != null) state.config = next;
         if (!(state.config instanceof Error)) return state.config;
         if (options?.throwOnError === true) throw state.config;
         return configWithRow(undefined);
       },
       sessionsDir: state.sessionsDir,
+      editConfig,
+    }),
+    aiService: () => ({
+      isStreaming: () => state.streaming,
+      stopStream: (taskId, options) => {
+        events.push(`stopStream:${taskId}:abandonPartial=${String(options?.abandonPartial)}`);
+        return Promise.resolve(Ok(undefined));
+      },
     }),
-    aiService: () => ({ isStreaming: () => state.streaming }),
     workspaceService: () => ({ getActiveTurnGeneration: () => state.activeTurn }),
     turnManager: () =>
       ({
@@ -96,8 +132,35 @@ function createHarness() {
     listAncestorWorkspaceIdsUsingParentById: () => state.owners,
     stopRecordAttemptId: () => state.stopRecordAttemptId,
     hasStopRecord: () => state.hasStopRecord,
-    admittedSends: () => state.sends,
+    isWorkspaceStopInProgress: () => state.stopInProgress,
+    isCurrentStopRecord: (_workspaceId, record) => state.currentStopRecord === record,
+    recheckWorkspaceStopRelease: (workspaceId) => events.push(`recheckStop:${workspaceId}`),
+    // Like TaskService.currentTaskAttemptId: the entry, else a lenient read of the row.
+    currentTaskAttemptId: (taskId, entry) => {
+      const id =
+        entry?.taskAttemptId ??
+        findWorkspaceEntry(deps.config().loadConfigOrDefault(), taskId)?.workspace.taskAttemptId;
+      return isTaskAttemptId(id) ? id : undefined;
+    },
+    // Like TaskService.editWorkspaceEntry: the updater runs on the row inside one config edit.
+    editWorkspaceEntry: async (workspaceId, updater) => {
+      let found = false;
+      await editConfig((config) => {
+        const row = findWorkspaceEntry(config, workspaceId)?.workspace;
+        if (row != null) {
+          updater(row, config);
+          found = true;
+        }
+        return config;
+      });
+      return found;
+    },
+    settleChildGoalPause: (workspaceId) => {
+      events.push(`settleGoalPause:${workspaceId}`);
+      return Promise.resolve();
+    },
     findStreamEndDecision: () => state.streamEnd,
+    pruneStreamEndDecisions: (taskId) => events.push(`prune:${taskId}`),
     discardStreamEndDecisions: (taskId) => events.push(`discardStreamEnds:${taskId}`),
     inspectAttemptOutcome: (taskId, options) => {
       events.push("inspect");
@@ -303,10 +366,11 @@ describe("AttemptLedger", () => {
     });
 
     test("an obligation of this attempt blocks; a successor's does not", () => {
-      const { ledger, state, owned } = closedOwner();
-      state.sends.add(send(OTHER));
+      const { ledger, owned } = closedOwner();
+      const sends = new Set([send(OTHER)]);
+      ledger.admittedSendsByTaskId.set(TASK, sends);
       expect(ledger.attemptCannotStillReport(TASK, owned)).toBe(true);
-      state.sends.add(send(ATTEMPT));
+      sends.add(send(ATTEMPT));
       expect(ledger.attemptCannotStillReport(TASK, owned)).toBe(false);
     });
 
@@ -523,12 +587,12 @@ describe("AttemptLedger", () => {
       state.hasStopRecord = true;
       ledger.recheckResumeFailureSettlement(TASK);
       state.hasStopRecord = false;
-      state.sends.add(send(ATTEMPT));
+      ledger.admittedSendsByTaskId.set(TASK, new Set([send(ATTEMPT)]));
       ledger.recheckResumeFailureSettlement(TASK);
       expect(ledger.resumeFailureSettlementByTaskId.has(TASK)).toBe(true);
       expect(events).toEqual([]);
 
-      state.sends.clear();
+      ledger.admittedSendsByTaskId.delete(TASK);
       ledger.recheckResumeFailureSettlement(TASK);
       expect(ledger.resumeFailureSettlementByTaskId.has(TASK)).toBe(false);
       await Promise.resolve();
@@ -651,4 +715,608 @@ describe("AttemptLedger", () => {
       expect(inspect).toHaveBeenCalledTimes(1);
     });
   });
+
+  describe("send admission", () => {
+    function admit(ledger: AttemptLedger, expectedAttemptId?: string) {
+      const admission = ledger.admitTaskWorkspaceTurn(TASK, {
+        acceptanceOrigin: "automatic",
+        ...(expectedAttemptId != null ? { expectedAttemptId } : {}),
+      });
+      if (admission.kind !== "admitted") throw new Error(`not admitted: ${admission.kind}`);
+      return admission.token;
+    }
+
+    function onlySend(ledger: AttemptLedger): AdmittedSend {
+      const sends = [...ledger.sendsOf(TASK)];
+      expect(sends).toHaveLength(1);
+      return sends[0];
+    }
+
+    /** A stop record that captured `captured` while it was pending (by reference). */
+    function capture(captured: AdmittedSend): WorkspaceStopRecord {
+      const record = {
+        pendingAdmissions: new Set([captured]),
+        capturedTurns: new Set<symbol>(),
+        cleanupInFlight: 0,
+      } as unknown as WorkspaceStopRecord;
+      captured.capturedBy.add(record);
+      return record;
+    }
+
+    test("an admitted send is the object the task's set holds; it is owned only under the owner's id", () => {
+      const { ledger } = createHarness();
+      begin(ledger, OTHER);
+      const unownedToken = admit(ledger);
+      const unowned = onlySend(ledger);
+      expect(unowned.token).toBe(unownedToken);
+      expect(unowned.attemptId).toBe(ATTEMPT);
+      expect(unowned.attempt).toBeUndefined();
+      expect(ledger.sendsOf(TASK)).toBe(ledger.admittedSendsByTaskId.get(TASK)!);
+
+      const owner = begin(ledger, ATTEMPT);
+      admit(ledger, ATTEMPT);
+      const [, owned] = [...ledger.sendsOf(TASK)];
+      expect(owned.attempt).toBe(owner);
+    });
+
+    test("an empty workspace id is refused; any other id is classified", () => {
+      const { ledger } = createHarness();
+      expect(() => ledger.admitTaskWorkspaceTurn("", { acceptanceOrigin: "manual" })).toThrow();
+      expect(ledger.admitTaskWorkspaceTurn("t", { acceptanceOrigin: "manual" })).toEqual({
+        kind: "not-a-task",
+      });
+    });
+
+    test("refusals and non-task sends register no obligation", () => {
+      const cases: Array<{
+        name: string;
+        arrange: (h: ReturnType<typeof createHarness>) => void;
+        expectedAttemptId?: string;
+        kind: "refused" | "not-a-task";
+      }> = [
+        {
+          name: "unreadable",
+          arrange: (h) => (h.state.config = new Error("EIO")),
+          kind: "refused",
+        },
+        {
+          name: "row gone for a decided send",
+          arrange: (h) => (h.state.config = configWithRow(undefined)),
+          expectedAttemptId: ATTEMPT,
+          kind: "refused",
+        },
+        {
+          name: "no row for an ordinary send",
+          arrange: (h) => (h.state.config = configWithRow(undefined)),
+          kind: "not-a-task",
+        },
+        {
+          name: "no parent",
+          arrange: (h) => (h.state.config = configWithRow({ taskAttemptId: ATTEMPT })),
+          kind: "not-a-task",
+        },
+        {
+          name: "retired",
+          arrange: (h) =>
+            (h.state.config = configWithRow({
+              parentWorkspaceId: PARENT,
+              taskAttemptId: ATTEMPT,
+              taskAttemptRetiredBy: { runId: "run", stepId: "step" },
+            })),
+          kind: "refused",
+        },
+        {
+          name: "pre-identity",
+          arrange: (h) => (h.state.config = configWithRow({ parentWorkspaceId: PARENT })),
+          kind: "not-a-task",
+        },
+        { name: "superseded", arrange: () => undefined, expectedAttemptId: OTHER, kind: "refused" },
+        { name: "stopping", arrange: (h) => (h.state.stopInProgress = true), kind: "refused" },
+        {
+          name: "closed",
+          arrange: (h) => h.ledger.closeAttemptAdmission(TASK, ATTEMPT, undefined, "closed"),
+          kind: "refused",
+        },
+      ];
+      for (const c of cases) {
+        const harness = createHarness();
+        c.arrange(harness);
+        const admission = harness.ledger.admitTaskWorkspaceTurn(TASK, {
+          acceptanceOrigin: "manual",
+          ...(c.expectedAttemptId != null ? { expectedAttemptId: c.expectedAttemptId } : {}),
+        });
+        expect({ name: c.name, kind: admission.kind }).toEqual({ name: c.name, kind: c.kind });
+        expect(harness.ledger.sendsOf(TASK).size).toBe(0);
+      }
+    });
+
+    test("disposal before admission discharges the send, then prunes, then rechecks only the current stop", () => {
+      const { ledger, state, events } = createHarness();
+      const token = admit(ledger);
+      const pending = onlySend(ledger);
+      const current = capture(pending);
+      const replaced = capture(pending);
+      state.currentStopRecord = current;
+
+      token.onDisposed("refused");
+      expect(pending.state).toBe("discharged");
+      expect(ledger.admittedSendsByTaskId.has(TASK)).toBe(false);
+      expect(current.pendingAdmissions.size).toBe(0);
+      expect(replaced.pendingAdmissions.size).toBe(0);
+      expect(pending.capturedBy.size).toBe(0);
+      expect(events).toEqual([`prune:${TASK}`, `recheckStop:${TASK}`]);
+
+      token.onDisposed("refused");
+      ledger.dischargeAdmittedSend(pending);
+      expect(events).toEqual([`prune:${TASK}`, `recheckStop:${TASK}`]);
+    });
+
+    test("the first admission binds the turn and owes each capturing stop one more stopStream", async () => {
+      const { ledger, events } = createHarness();
+      const token = admit(ledger);
+      const pending = onlySend(ledger);
+      const record = capture(pending);
+      const turn = Symbol("turn");
+
+      token.onAdmitted(turn);
+      expect(pending.state).toBe("admitted");
+      expect(pending.turnId).toBe(turn);
+      expect(record.pendingAdmissions.has(pending)).toBe(false);
+      expect(record.capturedTurns.has(turn)).toBe(true);
+      expect(record.cleanupInFlight).toBe(1);
+      expect(pending.capturedBy.size).toBe(0);
+      await new Promise((resolve) => setTimeout(resolve, 0));
+      expect(record.cleanupInFlight).toBe(0);
+      expect(events).toEqual([`stopStream:${TASK}:abandonPartial=false`, `recheckStop:${TASK}`]);
+
+      // First admission wins; an admitted send belongs to its turn, not to a disposal.
+      token.onAdmitted(Symbol("other"));
+      token.onDisposed("no-work");
+      expect(pending.turnId).toBe(turn);
+      expect(pending.state).toBe("admitted");
+      expect(token.admissionStale()).toBe(false);
+    });
+
+    test("a discharged send is never re-admitted", () => {
+      const { ledger } = createHarness();
+      const token = admit(ledger);
+      const pending = onlySend(ledger);
+      token.onDisposed("canceled-before-admission");
+      token.onAdmitted(Symbol("late"));
+      expect(pending.state).toBe("discharged");
+      expect(pending.turnId).toBeUndefined();
+      expect(token.admissionStale()).toBe(true);
+    });
+
+    test("staleness and the dequeue gate follow the attempt, its owner and its stream end", () => {
+      const { ledger, state } = createHarness();
+      begin(ledger, ATTEMPT);
+      const token = admit(ledger);
+      const decision = (outcome: StreamEndDecision["outcome"]): StreamEndDecision => ({
+        attempt: undefined,
+        attemptId: ATTEMPT,
+        messageId: "msg-1",
+        outcome,
+        turn: undefined,
+      });
+      expect(token.admissionStale()).toBe(false);
+      expect(token.resolveDispatch?.()).toBe("proceed");
+
+      state.streamEnd = decision("pending");
+      expect(token.admissionStale()).toBe(true);
+      token.onEnqueued();
+      // Enqueued: the queue's gate holds or refuses instead of the staleness probe.
+      expect(token.admissionStale()).toBe(false);
+      expect(token.resolveDispatch?.()).toBe("hold");
+      state.streamEnd = decision("published");
+      expect(token.resolveDispatch?.()).toEqual({
+        refuse: TASK_REPORTED_QUEUED_SEND_UNSENT_MESSAGE,
+      });
+      state.streamEnd = decision("nonreport");
+      expect(token.resolveDispatch?.()).toBe("proceed");
+
+      begin(ledger, ATTEMPT);
+      expect(token.admissionStale()).toBe(true);
+    });
+
+    test("a pending send reads stale once its row rotates, a closure, a claim or a removal lands", () => {
+      const row = { parentWorkspaceId: PARENT, taskAttemptId: ATTEMPT };
+      const cases: Array<{ name: string; arrange: (h: ReturnType<typeof createHarness>) => void }> =
+        [
+          {
+            name: "rotated",
+            arrange: (h) => (h.state.config = configWithRow({ ...row, taskAttemptId: OTHER })),
+          },
+          {
+            name: "closed",
+            arrange: (h) => h.ledger.closeAttemptAdmission(TASK, ATTEMPT, undefined, "closed"),
+          },
+          {
+            name: "retired",
+            arrange: (h) =>
+              (h.state.config = configWithRow({ ...row, taskAttemptRetiredBy: { runId: "run" } })),
+          },
+          {
+            name: "removing",
+            arrange: (h) =>
+              (h.state.config = configWithRow({ ...row, pendingRemoval: { pid: 1 } })),
+          },
+        ];
+      for (const c of cases) {
+        const harness = createHarness();
+        const token = admit(harness.ledger);
+        expect({ name: c.name, stale: token.admissionStale() }).toEqual({
+          name: c.name,
+          stale: false,
+        });
+        c.arrange(harness);
+        expect({ name: c.name, stale: token.admissionStale() }).toEqual({
+          name: c.name,
+          stale: true,
+        });
+      }
+    });
+
+    test("an admitted send is neither a pending admission nor a live reawakening once its turn ends", () => {
+      const { ledger } = createHarness();
+      admit(ledger).onAdmitted(Symbol("turn"));
+      expect(ledger.hasPendingAdmissions(TASK)).toBe(false);
+      expect(ledger.hasLiveReawakening(TASK)).toBe(false);
+    });
+
+    test("pending admissions and live reawakenings", () => {
+      const { ledger, state } = createHarness();
+      expect(ledger.hasLiveReawakening(TASK)).toBe(false);
+      ledger.beginReawakening(TASK, ATTEMPT);
+      expect(ledger.hasLiveReawakening(TASK)).toBe(true);
+      ledger.endReawakening(TASK, OTHER);
+      expect(ledger.hasLiveReawakening(TASK)).toBe(true);
+      ledger.endReawakening(TASK, ATTEMPT);
+      expect(ledger.hasLiveReawakening(TASK)).toBe(false);
+
+      admit(ledger);
+      expect(ledger.hasPendingAdmissions(TASK)).toBe(true);
+      expect(ledger.hasLiveReawakening(TASK)).toBe(true);
+      // A send that reads stale is no continuation to protect.
+      state.stopInProgress = true;
+      expect(ledger.hasLiveReawakening(TASK)).toBe(false);
+      state.activeTurn = Symbol("turn");
+      expect(ledger.hasLiveReawakening(TASK)).toBe(true);
+      state.config = configWithRow({ parentWorkspaceId: PARENT, taskAttemptId: OTHER });
+      expect(ledger.hasPendingAdmissions(TASK)).toBe(false);
+    });
+
+    test("a stream event belongs to the turn's direct admission before an inherited or owned one", () => {
+      const { ledger, state } = createHarness();
+      const owner = begin(ledger, OTHER);
+      expect(ledger.resolveStreamAttemptAtEvent(TASK)).toEqual({
+        ownedAttempt: owner,
+        unownedAttemptId: undefined,
+      });
+      const turn = Symbol("turn");
+      state.activeTurn = turn;
+      const token = admit(ledger);
+      token.onAdmitted(turn);
+      expect(ledger.resolveStreamAttemptAtEvent(TASK)).toEqual({
+        ownedAttempt: undefined,
+        unownedAttemptId: ATTEMPT,
+      });
+    });
+
+    test("an inherited obligation counts only when the turn has no direct admission", () => {
+      const { ledger, state } = createHarness();
+      const turn = Symbol("turn");
+      state.activeTurn = turn;
+      // Rebound from a superseded predecessor (recordWorkspaceTurnSuperseded) under OTHER.
+      admit(ledger).onAdmitted(turn);
+      const inherited = onlySend(ledger) as AdmittedSend & { attemptId: string };
+      Object.assign(inherited, { attemptId: OTHER, inherited: true });
+      expect(ledger.resolveStreamAttemptAtEvent(TASK).unownedAttemptId).toBe(OTHER);
+
+      admit(ledger).onAdmitted(turn);
+      expect(ledger.resolveStreamAttemptAtEvent(TASK).unownedAttemptId).toBe(ATTEMPT);
+      // Another turn's admission says nothing about this one.
+      state.activeTurn = Symbol("next");
+      state.config = configWithRow({ parentWorkspaceId: PARENT, taskAttemptId: OTHER });
+      expect(ledger.resolveStreamAttemptAtEvent(TASK).unownedAttemptId).toBe(OTHER);
+    });
+  });
+
+  describe("lineage and unowned settlement proofs", () => {
+    let sessionsDir: string;
+    beforeEach(async () => {
+      sessionsDir = await fs.mkdtemp(path.join(os.tmpdir(), "attempt-ledger-proof-"));
+    });
+    afterEach(async () => {
+      await fs.rm(sessionsDir, { recursive: true, force: true });
+    });
+
+    const interruptedRow = (attemptId: string) =>
+      configWithRow({
+        parentWorkspaceId: PARENT,
+        taskAttemptId: attemptId,
+        taskStatus: "interrupted",
+      });
+
+    test("the owner's closing settlement is re-read after the closure wait", async () => {
+      const { ledger } = createHarness();
+      const owner = begin(ledger, ATTEMPT);
+      ledger.closeAttemptAdmission(TASK, ATTEMPT, owner, "stopping");
+      const entry: WorkspaceConfigEntry = {
+        path: "/project/task",
+        parentWorkspaceId: PARENT,
+        taskAttemptId: ATTEMPT,
+      };
+      const lineage = ledger.evaluateAttemptLineage(TASK, entry);
+      ledger.settleOwnedTaskAttempt(TASK, owner, "stopped");
+      expect(await lineage).toEqual({
+        proven: true,
+        reason: "predecessor settled by this process",
+      });
+    });
+
+    test("a receipt in the parent's dir proves the predecessor; one naming another parent does not", async () => {
+      const { ledger, state } = createHarness();
+      state.sessionsDir = sessionsDir;
+      const entry: WorkspaceConfigEntry = {
+        path: "/project/task",
+        parentWorkspaceId: PARENT,
+        taskAttemptId: ATTEMPT,
+      };
+      expect((await ledger.evaluateAttemptLineage(TASK, entry)).proven).toBe(false);
+      const target = { attemptId: ATTEMPT, parentWorkspaceId: PARENT, owners: [PARENT] };
+      expect(await ledger.writeSettlementReceipt(TASK, target, "idle-settled")).toBe(true);
+      expect((await ledger.evaluateAttemptLineage(TASK, entry)).proven).toBe(true);
+      expect(
+        (await ledger.evaluateAttemptLineage(TASK, { ...entry, parentWorkspaceId: ROOT })).proven
+      ).toBe(false);
+    });
+
+    test("a receipt copy that names another parent proves nothing", async () => {
+      const { ledger, state } = createHarness();
+      state.sessionsDir = sessionsDir;
+      // The parent's receipt fans out to ROOT's dir too, naming PARENT.
+      const target = { attemptId: ATTEMPT, parentWorkspaceId: PARENT, owners: [PARENT, ROOT] };
+      expect(await ledger.writeSettlementReceipt(TASK, target, "idle-settled")).toBe(true);
+      const entry: WorkspaceConfigEntry = {
+        path: "/project/task",
+        parentWorkspaceId: ROOT,
+        taskAttemptId: ATTEMPT,
+      };
+      expect(await ledger.evaluateAttemptLineage(TASK, entry)).toEqual({
+        proven: false,
+        reason: `settlement receipt names parent ${PARENT}, expected ${ROOT}`,
+      });
+      state.config = configWithRow({
+        parentWorkspaceId: ROOT,
+        taskAttemptId: ATTEMPT,
+        taskStatus: "interrupted",
+      });
+      expect(await ledger.readUnownedSettlementProof(TASK)).toEqual({
+        kind: "unproven",
+        reason: `settlement receipt names parent ${PARENT}`,
+      });
+    });
+
+    test("no id, marked lineage, an ineligible owner or no parent is unproven even with a receipt", async () => {
+      const { ledger, state } = createHarness();
+      state.sessionsDir = sessionsDir;
+      const target = { attemptId: ATTEMPT, parentWorkspaceId: PARENT, owners: [PARENT] };
+      expect(await ledger.writeSettlementReceipt(TASK, target, "idle-settled")).toBe(true);
+      const entry: WorkspaceConfigEntry = {
+        path: "/project/task",
+        parentWorkspaceId: PARENT,
+        taskAttemptId: ATTEMPT,
+      };
+      const unprovenEntries: WorkspaceConfigEntry[] = [
+        { ...entry, taskAttemptId: undefined },
+        { ...entry, taskAttemptUnproven: true },
+        { ...entry, parentWorkspaceId: undefined },
+      ];
+      for (const unproven of unprovenEntries) {
+        expect(await ledger.evaluateAttemptLineage(TASK, unproven)).toMatchObject({
+          proven: false,
+        });
+      }
+      begin(ledger, ATTEMPT, false);
+      expect(await ledger.evaluateAttemptLineage(TASK, entry)).toEqual({
+        proven: false,
+        reason: "predecessor owned but not receipt-eligible",
+      });
+    });
+
+    test("an unowned proof needs the receipt and the same row before and after reading it", async () => {
+      const { ledger, state } = createHarness();
+      state.sessionsDir = sessionsDir;
+      state.config = interruptedRow(ATTEMPT);
+      expect(await ledger.readUnownedSettlementProof(TASK)).toMatchObject({ kind: "unproven" });
+      const target = { attemptId: ATTEMPT, parentWorkspaceId: PARENT, owners: [PARENT] };
+      expect(await ledger.writeSettlementReceipt(TASK, target, "idle-settled")).toBe(true);
+      expect(await ledger.readUnownedSettlementProof(TASK)).toEqual({
+        kind: "proven",
+        attemptId: ATTEMPT,
+      });
+
+      // Another writer admitted, re-ran or marked the attempt while the receipt was read.
+      const changed = { parentWorkspaceId: PARENT, taskAttemptId: ATTEMPT };
+      for (const latest of [
+        interruptedRow(OTHER),
+        configWithRow({ ...changed, taskStatus: "running" }),
+        configWithRow({ ...changed, taskStatus: "interrupted", taskAttemptUnproven: true }),
+      ]) {
+        state.nextConfigs = [interruptedRow(ATTEMPT), latest];
+        expect(await ledger.readUnownedSettlementProof(TASK)).toEqual({
+          kind: "unproven",
+          reason: "the attempt changed while its receipt was read",
+        });
+      }
+      // A strict re-read that fails is no evidence (never the lenient empty view).
+      state.nextConfigs = [interruptedRow(ATTEMPT), new Error("EIO")];
+      expect(await ledger.readUnownedSettlementProof(TASK)).toEqual({
+        kind: "unproven",
+        reason: "config unreadable: EIO",
+      });
+    });
+  });
+
+  describe("workflow claims", () => {
+    const CLAIMANT = { runId: "run", stepId: "step", inputHash: "hash" };
+    const claimableRow = () => ({
+      parentWorkspaceId: PARENT,
+      taskAttemptId: ATTEMPT,
+      taskStatus: "interrupted",
+      workflowTask: { runId: CLAIMANT.runId, stepId: CLAIMANT.stepId },
+    });
+    const rowOf = (state: HarnessState, id = TASK) =>
+      findWorkspaceEntry(state.config as ProjectsConfig, id)?.workspace;
+
+    test("a granted claim retires the attempt, is confirmed, and then settles the goal pause", async () => {
+      const { ledger, state, events } = createHarness();
+      state.config = configWithRow(claimableRow());
+      const claim = await ledger.claimRetiredAttempt(TASK, ATTEMPT, CLAIMANT);
+      if (!claim.success) throw new Error(claim.error);
+      expect(rowOf(state)?.taskAttemptRetiredBy).toMatchObject({
+        runId: CLAIMANT.runId,
+        attemptId: ATTEMPT,
+        nonce: claim.data.nonce,
+      });
+      expect(events).toEqual([`settleGoalPause:${TASK}`]);
+
+      // The same step re-stamps with a fresh nonce.
+      const again = await ledger.claimRetiredAttempt(TASK, ATTEMPT, CLAIMANT);
+      if (!again.success) throw new Error(again.error);
+      expect(again.data.nonce).not.toBe(claim.data.nonce);
+    });
+
+    test("a live attempt, a refused row, a failed or lost write and an unconfirmed claim refuse", async () => {
+      const cases: Array<{ name: string; arrange: (h: ReturnType<typeof createHarness>) => void }> =
+        [
+          { name: "attempt is live", arrange: (h) => (h.state.activeTurn = Symbol("turn")) },
+          { name: "attempt is live", arrange: (h) => (h.state.streaming = true) },
+          { name: "attempt is live", arrange: (h) => (h.state.liveRegistration = {}) },
+          { name: "attempt is live", arrange: (h) => (h.state.stopInProgress = true) },
+          {
+            name: "attempt is running",
+            arrange: (h) =>
+              (h.state.config = configWithRow({ ...claimableRow(), taskStatus: "running" })),
+          },
+          {
+            name: "the task does not belong to this workflow step",
+            arrange: (h) =>
+              (h.state.config = configWithRow({
+                ...claimableRow(),
+                workflowTask: { runId: CLAIMANT.runId, stepId: "other-step" },
+              })),
+          },
+          {
+            name: "the retired attempt already has replacement",
+            arrange: (h) =>
+              (h.state.config = configWithRow({
+                ...claimableRow(),
+                taskAttemptRetiredBy: { ...CLAIMANT, attemptId: ATTEMPT, replacementTaskId: "r" },
+              })),
+          },
+          {
+            name: "the attempt was retired by another workflow step",
+            arrange: (h) =>
+              (h.state.config = configWithRow({
+                ...claimableRow(),
+                taskAttemptRetiredBy: { ...CLAIMANT, attemptId: ATTEMPT, stepId: "other-step" },
+              })),
+          },
+          {
+            name: "claim lost: the task now names attempt",
+            arrange: (h) =>
+              (h.state.config = configWithRow({ ...claimableRow(), taskAttemptId: OTHER })),
+          },
+          {
+            name: "task record not found",
+            arrange: (h) => (h.state.config = configWithRow(undefined)),
+          },
+          { name: "claim write failed", arrange: (h) => (h.state.editError = new Error("EIO")) },
+          { name: "claim lost", arrange: (h) => (h.state.dropEdits = true) },
+          {
+            // The edit lands, then the strict re-read fails.
+            name: "claim unconfirmed: config unreadable",
+            arrange: (h) => h.state.nextConfigs.push(new Error("EIO")),
+          },
+        ];
+      for (const c of cases) {
+        const harness = createHarness();
+        harness.state.config = configWithRow(claimableRow());
+        c.arrange(harness);
+        const claim = await harness.ledger.claimRetiredAttempt(TASK, ATTEMPT, CLAIMANT);
+        expect({ name: c.name, success: claim.success }).toEqual({ name: c.name, success: false });
+        if (!claim.success) expect(claim.error.startsWith(c.name)).toBe(true);
+        expect(harness.events).not.toContain(`settleGoalPause:${TASK}`);
+      }
+    });
+
+    test("empty ids are refused; one-character ids are not", async () => {
+      const { ledger } = createHarness();
+      const rejected = (promise: Promise<unknown>) =>
+        promise.then(
+          () => false,
+          () => true
+        );
+      const claimant = { runId: "r", stepId: "s", inputHash: "" };
+      expect(await rejected(ledger.claimRetiredAttempt("", ATTEMPT, claimant))).toBe(true);
+      expect(
+        await rejected(ledger.claimRetiredAttempt("t", ATTEMPT, { ...claimant, runId: "" }))
+      ).toBe(true);
+      expect(
+        await rejected(ledger.claimRetiredAttempt("t", ATTEMPT, { ...claimant, stepId: "" }))
+      ).toBe(true);
+      expect(await ledger.claimRetiredAttempt("t", ATTEMPT, claimant)).toEqual({
+        success: false,
+        error: "task record not found",
+      });
+      expect(await rejected(ledger.tombstoneUnpublishedReservation("", "t"))).toBe(true);
+      expect(await rejected(ledger.tombstoneUnpublishedReservation("p", ""))).toBe(true);
+      expect(await ledger.tombstoneUnpublishedReservation("p", "t")).toEqual({
+        success: false,
+        error: "parent record not found",
+      });
+    });
+
+    test("a tombstone is written on the parent once and confirmed by a strict re-read", async () => {
+      const { ledger, state } = createHarness();
+      state.config = configWithRows([{ id: PARENT }]);
+      expect(await ledger.tombstoneUnpublishedReservation(PARENT, TASK)).toEqual(Ok(undefined));
+      expect(await ledger.tombstoneUnpublishedReservation(PARENT, TASK)).toEqual(Ok(undefined));
+      expect(rowOf(state, PARENT)?.taskReservationTombstones).toEqual([TASK]);
+    });
+
+    test("an owned attempt, a published task, a missing parent and a failed or lost write refuse", async () => {
+      const cases: Array<{ name: string; arrange: (h: ReturnType<typeof createHarness>) => void }> =
+        [
+          { name: "this process owns an attempt for the task", arrange: (h) => begin(h.ledger) },
+          {
+            name: "the task was published",
+            arrange: (h) => (h.state.config = configWithRows([{ id: PARENT }, { id: TASK }])),
+          },
+          {
+            name: "parent record not found",
+            arrange: (h) => (h.state.config = configWithRows([])),
+          },
+          {
+            name: "tombstone write failed",
+            arrange: (h) => (h.state.editError = new Error("EIO")),
+          },
+          { name: "tombstone lost", arrange: (h) => (h.state.dropEdits = true) },
+          {
+            name: "tombstone unconfirmed: config unreadable",
+            arrange: (h) => h.state.nextConfigs.push(new Error("EIO")),
+          },
+        ];
+      for (const c of cases) {
+        const harness = createHarness();
+        harness.state.config = configWithRows([{ id: PARENT }]);
+        c.arrange(harness);
+        const result = await harness.ledger.tombstoneUnpublishedReservation(PARENT, TASK);
+        expect({ name: c.name, success: result.success }).toEqual({ name: c.name, success: false });
+        if (!result.success) expect(result.error.startsWith(c.name)).toBe(true);
+      }
+    });
+  });
 });
diff --git a/src/node/services/attemptLedger.ts b/src/node/services/attemptLedger.ts
index 53f0ef98f6c..8b0e54418ab 100644
--- a/src/node/services/attemptLedger.ts
+++ b/src/node/services/attemptLedger.ts
@@ -1,13 +1,26 @@
 import assert from "node:assert/strict";
+import { randomUUID } from "node:crypto";
 import * as path from "node:path";
+import {
+  SEND_ADMISSION_STALE_MESSAGE,
+  TASK_ATTEMPT_SETTLED_SEND_BLOCKED_MESSAGE,
+  TASK_REPORT_OUTCOME_INDETERMINATE_UNSENT_MESSAGE,
+  TASK_REPORTED_QUEUED_SEND_UNSENT_MESSAGE,
+  WORKSPACE_STOP_IN_PROGRESS_SEND_BLOCKED_MESSAGE,
+  pendingRemovalAdmissionMessage,
+  retiredAttemptMessage,
+} from "@/constants/agentMessaging";
+import { Err, Ok, type Result } from "@/common/types/result";
 import { getErrorMessage } from "@/common/utils/errors";
 import type { TaskAttemptOutcome, TaskAttemptSettlement } from "@/common/types/tasks";
-import type { Config } from "@/node/config";
+import type { Config, ProjectsConfig, Workspace as WorkspaceConfigEntry } from "@/node/config";
 import type { AgentTaskIndex } from "@/node/services/agentTaskIndex";
 import type { AIService } from "@/node/services/aiService";
 import { log } from "@/node/services/log";
 import {
+  readSubagentAttemptSettlementReceiptStrict,
   writeSubagentAttemptSettlementReceipt,
+  type SubagentAttemptSettlementReceiptReadResult,
   type SubagentAttemptSettlementSource,
 } from "@/node/services/subagentAttemptSettlements";
 import type {
@@ -16,26 +29,144 @@ import type {
   AttemptSettlementEntry,
   OwnedTaskAttempt,
   SettlementReceiptDecision,
+  StreamAttemptOrigin,
   StreamEndDecision,
+  WorkspaceStopRecord,
 } from "@/node/services/taskService";
-import { coerceNonEmptyString, findWorkspaceEntry, rowSupersedes } from "@/node/services/taskUtils";
-import type { WorkspaceHost } from "@/node/services/taskWorkspaceSeam";
+import {
+  coerceNonEmptyString,
+  findWorkspaceEntry,
+  markChildGoalPauseOwed,
+  rowSupersedes,
+} from "@/node/services/taskUtils";
+import type {
+  QueuedDispatchDecision,
+  TaskTurnAdmission,
+  TurnAcceptanceOrigin,
+  TurnAdmissionToken,
+  WorkspaceHost,
+} from "@/node/services/taskWorkspaceSeam";
 import type { WorkspaceTurnManager } from "@/node/services/workspaceTurnManager";
 import type { MutexMap } from "@/node/utils/concurrency/mutexMap";
 import { raceWithAbortAndTimeout } from "@/node/utils/concurrency/withTimeout";
-import { assertTaskAttemptId } from "@/node/utils/taskAttemptId";
+import { assertTaskAttemptId, isTaskAttemptId } from "@/node/utils/taskAttemptId";
+
+/**
+ * Bound on waiting for an owned predecessor's in-flight settlement write before deciding a
+ * successor's lineage (evaluateAttemptLineage), and on a failed direct launch's wait for the
+ * owners its stop cascade captured before its rollback (TaskService.create). Elapsing the bound
+ * is a decision — unproven lineage, deferred rollback — not an error; the write or the cascade
+ * keeps its own ownership and completes on its own.
+ */
+export const ATTEMPT_CLOSURE_SETTLE_WAIT_MS = 5_000;
+
+/** Lineage proof (evaluateAttemptLineage) from the predecessor's receipt in the parent's dir. */
+function lineageProofFromReceipt(
+  receipt: SubagentAttemptSettlementReceiptReadResult,
+  parentWorkspaceId: string
+): { proven: boolean; reason: string } {
+  if (receipt.kind === "found") {
+    // The strict reader binds task and attempt id to the path; the parent is bound here, to
+    // the entry: a receipt naming another parent at this path was copied or misplaced, never
+    // written by this task's owner (the same fail-closed outcome as an unreadable receipt).
+    if (receipt.receipt.parentWorkspaceId !== parentWorkspaceId) {
+      return {
+        proven: false,
+        reason: `settlement receipt names parent ${receipt.receipt.parentWorkspaceId}, expected ${parentWorkspaceId}`,
+      };
+    }
+    return { proven: true, reason: "settlement receipt found" };
+  }
+  return {
+    proven: false,
+    reason:
+      receipt.kind === "unreadable"
+        ? `settlement receipt unreadable: ${receipt.error}`
+        : "no settlement receipt for the predecessor attempt",
+  };
+}
+
+/**
+ * Row step of readUnownedSettlementProof: the row names an interrupted, proven attempt with a
+ * parent, so the parent's receipt copy can prove it; else why it cannot.
+ */
+function unownedSettlementCandidate(
+  row: WorkspaceConfigEntry | undefined
+):
+  | { kind: "candidate"; attemptId: string; parentWorkspaceId: string }
+  | { kind: "unproven"; reason: string } {
+  const unproven = (reason: string) => ({ kind: "unproven" as const, reason });
+  if (row == null) return unproven("no task record");
+  const attemptId = row.taskAttemptId;
+  if (!isTaskAttemptId(attemptId)) return unproven("no attempt id (pre-identity entry)");
+  if (row.taskStatus !== "interrupted") return unproven(`status ${row.taskStatus ?? "missing"}`);
+  if (row.taskAttemptUnproven === true) return unproven("lineage marked unproven");
+  const parentWorkspaceId = coerceNonEmptyString(row.parentWorkspaceId);
+  if (parentWorkspaceId == null) return unproven("no parent");
+  return { kind: "candidate", attemptId, parentWorkspaceId };
+}
+
+/** Receipt step of readUnownedSettlementProof: why the read receipt proves nothing, if it does not. */
+function unownedSettlementReceiptGap(
+  receipt: SubagentAttemptSettlementReceiptReadResult,
+  attemptId: string,
+  parentWorkspaceId: string
+): string | undefined {
+  if (receipt.kind === "not_found") return `no settlement receipt for ${attemptId}`;
+  if (receipt.kind === "unreadable") {
+    return `settlement receipt unreadable: ${receipt.error}`;
+  }
+  if (receipt.receipt.parentWorkspaceId !== parentWorkspaceId) {
+    return `settlement receipt names parent ${receipt.receipt.parentWorkspaceId}`;
+  }
+  return undefined;
+}
+
+/**
+ * Why a workflow claim on `attemptId` must be refused (see TaskService.claimRetiredAttempt), or
+ * undefined when it may be granted (or re-stamped, for the same run/step before any replacement).
+ */
+function retiredAttemptClaimRefusal(
+  row: WorkspaceConfigEntry,
+  attemptId: string,
+  claimant: { runId: string; stepId: string }
+): string | undefined {
+  if (row.taskAttemptId !== attemptId) {
+    return `claim lost: the task now names attempt ${row.taskAttemptId ?? "none"}`;
+  }
+  if (row.taskStatus !== "interrupted") return `attempt is ${row.taskStatus ?? "running"}`;
+  const task = row.workflowTask;
+  if (task?.runId !== claimant.runId || task.stepId !== claimant.stepId) {
+    return "the task does not belong to this workflow step";
+  }
+  const existing = row.taskAttemptRetiredBy;
+  if (existing == null) return undefined;
+  if (existing.replacementTaskId != null) {
+    return `the retired attempt already has replacement ${existing.replacementTaskId}`;
+  }
+  if (
+    existing.runId !== claimant.runId ||
+    existing.stepId !== claimant.stepId ||
+    existing.attemptId !== attemptId
+  ) {
+    return "the attempt was retired by another workflow step";
+  }
+  return undefined;
+}
+
+const NO_SENDS: ReadonlySet<AdmittedSend> = new Set();
 
 /**
  * Everything the attempt ledger reads from TaskService. Each entry is a call-time closure over
  * TaskService, never a captured value or bound method: tests swap TaskService fields and patch
  * its methods after construction. Methods tests spy on (closeAttemptAdmission,
  * decideSettlementReceipt, writeSettlementReceipt, inspectAttemptOutcome) and the other
- * TaskService facades are called back through TaskService so every call reaches the spy. Stop,
- * send and stream-end state stays with its owner and is reached only through these closures.
+ * TaskService facades are called back through TaskService so every call reaches the spy. Stop
+ * and stream-end state stays with its owner and is reached only through these closures.
  */
 export interface AttemptLedgerDeps {
-  config: () => Pick<Config, "loadConfigOrDefault" | "sessionsDir">;
-  aiService: () => Pick<AIService, "isStreaming">;
+  config: () => Pick<Config, "loadConfigOrDefault" | "sessionsDir" | "editConfig">;
+  aiService: () => Pick<AIService, "isStreaming" | "stopStream">;
   workspaceService: () => Pick<WorkspaceHost, "getActiveTurnGeneration">;
   turnManager: () => Pick<WorkspaceTurnManager, "getLiveWorkspaceTurnRegistration">;
   workspaceEventLocks: () => Pick<MutexMap<string>, "withLock">;
@@ -48,9 +179,23 @@ export interface AttemptLedgerDeps {
   stopRecordAttemptId: (taskId: string) => string | undefined;
   /** TaskStopLatch.hasStopRecord. */
   hasStopRecord: (taskId: string) => boolean;
-  /** The task's live set of send obligations (TaskService.admittedSendsByTaskId). */
-  admittedSends: (taskId: string) => ReadonlySet<AdmittedSend> | undefined;
+  /** TaskStopLatch.isWorkspaceStopInProgress (through the TaskService facade). */
+  isWorkspaceStopInProgress: (workspaceId: string) => boolean;
+  /** TaskStopLatch.isCurrentStopRecord. */
+  isCurrentStopRecord: (workspaceId: string, record: WorkspaceStopRecord) => boolean;
+  /** TaskStopLatch.recheckWorkspaceStopRelease. */
+  recheckWorkspaceStopRelease: (workspaceId: string) => void;
+  /** TaskService.currentTaskAttemptId: the persisted attempt id of the task's row. */
+  currentTaskAttemptId: (taskId: string, entry?: WorkspaceConfigEntry) => string | undefined;
+  editWorkspaceEntry: (
+    workspaceId: string,
+    updater: (workspace: WorkspaceConfigEntry, config: ProjectsConfig) => void,
+    options?: { allowMissing?: boolean }
+  ) => Promise<boolean>;
+  settleChildGoalPause: (workspaceId: string) => Promise<void>;
   findStreamEndDecision: (taskId: string, attemptId: string) => StreamEndDecision | undefined;
+  /** Drop the stream-end decisions no send obligation can still read (TaskService). */
+  pruneStreamEndDecisions: (taskId: string) => void;
   /** Drop every stream-end decision of the task (TaskService.streamEndDecisionsByTaskId). */
   discardStreamEndDecisions: (taskId: string) => void;
   inspectAttemptOutcome: (
@@ -85,8 +230,10 @@ export interface AttemptLedgerDeps {
  * Attempt identity and settlement state: the attempt this process owns per task, the
  * closure/settlement of exactly the attempt it names, the listeners waiting on either, the
  * failed reawakens still owed a settlement, and the settlement receipts written for owned
- * attempts. Stop records, send obligations and stream-end decisions stay with their owners and
- * are reached through the deps.
+ * attempts. Send admission: the send obligations bound to an attempt (AdmittedSend), the
+ * reawakenings in flight, lineage and unowned-settlement proofs, and the workflow claims that
+ * retire an attempt or tombstone an unpublished reservation. Stop records and stream-end
+ * decisions stay with their owners and are reached through the deps.
  */
 export class AttemptLedger {
   /**
@@ -119,9 +266,38 @@ export class AttemptLedger {
    * proof that a workspace is not a task.
    */
   readonly currentAttemptIdByTaskId = new Map<string, string>();
+  /** Outstanding send obligations per task (see AdmittedSend); discharged entries are removed. */
+  readonly admittedSendsByTaskId = new Map<string, Set<AdmittedSend>>();
+  /**
+   * Reawakenings (parent reactivation or manual resume) that published a fresh attempt and have
+   * not yet handed it a send obligation (reactivation: until createWorkspaceTurn returns; manual
+   * resume: until reawakenInterruptedTask returns to the send that admits it; a reactivation in
+   * the remaining hand-off tick only makes that send, which names its attempt, read stale). With
+   * pending admissions and the live turn it is the "another reawakening is live" evidence each
+   * path rechecks under the mutex (see hasLiveReawakening): neither path holds the other's locks,
+   * and rotating an attempt a live continuation runs under drops that continuation's report.
+   */
+  private readonly reawakeningsInFlight = new Map<string, string>();
 
   constructor(private readonly deps: AttemptLedgerDeps) {}
 
+  /** The task's live set of send obligations (never a copy); empty when it has none. */
+  sendsOf(taskId: string): ReadonlySet<AdmittedSend> {
+    return this.admittedSendsByTaskId.get(taskId) ?? NO_SENDS;
+  }
+
+  /** A reawakening published `attemptId` and has not yet handed it a send (see the field). */
+  beginReawakening(taskId: string, attemptId: string): void {
+    this.reawakeningsInFlight.set(taskId, attemptId);
+  }
+
+  /** The reawakening of `attemptId` ended; a later reawakening's entry stays. */
+  endReawakening(taskId: string, attemptId: string): void {
+    if (this.reawakeningsInFlight.get(taskId) === attemptId) {
+      this.reawakeningsInFlight.delete(taskId);
+    }
+  }
+
   /** This process's owned attempt for the task, if any. */
   ownedAttempt(taskId: string): OwnedTaskAttempt | undefined {
     return this.ownedAttemptByTaskId.get(taskId);
@@ -345,7 +521,7 @@ export class AttemptLedger {
       this.isAttemptClosed(taskId, attemptId) ||
       this.deps.stopRecordAttemptId(taskId) === attemptId;
     if (!closed) return false;
-    for (const send of this.deps.admittedSends(taskId) ?? []) {
+    for (const send of this.admittedSendsByTaskId.get(taskId) ?? []) {
       if (send.attemptId === attemptId) return false;
     }
     if (
@@ -692,4 +868,589 @@ export class AttemptLedger {
       }
     );
   }
+
+  /**
+   * Dequeue-gate consultation for an enqueued obligation (see TurnAdmissionToken.resolveDispatch):
+   * held while its attempt's stream-end is undecided, refused once that stream turned out to be
+   * the terminal report (or its outcome is unknowable), otherwise on to the ordinary gates.
+   */
+  private resolveEnqueuedSendDispatch(send: AdmittedSend): QueuedDispatchDecision {
+    if (send.state !== "enqueued") return "proceed";
+    const decision = this.deps.findStreamEndDecision(send.taskId, send.attemptId);
+    switch (decision?.outcome) {
+      case "pending":
+        return "hold";
+      case "published":
+        return { refuse: TASK_REPORTED_QUEUED_SEND_UNSENT_MESSAGE };
+      case "indeterminate":
+        return { refuse: TASK_REPORT_OUTCOME_INDETERMINATE_UNSENT_MESSAGE };
+      case "nonreport":
+      case undefined:
+        return "proceed";
+    }
+  }
+
+  /**
+   * Common send-authorization predicate (no ownership): the id is still the task's current
+   * attempt, nothing closed it, no stop cascade holds the workspace and no workflow claim retired
+   * it. Ownership is a separate question, checked only where it applies.
+   */
+  attemptAdmissionOpen(taskId: string, attemptId: string): boolean {
+    const entry = findWorkspaceEntry(this.deps.config().loadConfigOrDefault(), taskId)?.workspace;
+    if (this.deps.currentTaskAttemptId(taskId, entry) !== attemptId) return false;
+    if (this.isAttemptClosed(taskId, attemptId)) return false;
+    if (this.deps.isWorkspaceStopInProgress(taskId)) return false;
+    return entry?.taskAttemptRetiredBy == null && entry?.pendingRemoval == null;
+  }
+
+  /**
+   * Who a stream event belongs to, read in the event's own tick. The live turn's ADMITTED attempt
+   * decides first: local ownership counts only when it names that same attempt. A process can
+   * still hold a stale owned attempt A after another backend rotated the row to B and a local send
+   * was admitted for B (unowned) — preferring the owner would treat B's stream as A's and discard
+   * its report/recovery as superseded. Without an admitted obligation (legacy turns) the owner,
+   * else the persisted id, applies as before.
+   */
+  resolveStreamAttemptAtEvent(taskId: string): StreamAttemptOrigin {
+    const owned = this.ownedAttempt(taskId);
+    const turn = this.deps.workspaceService().getActiveTurnGeneration(taskId);
+    // The turn's DIRECT admission decides; an obligation rebound from a superseded predecessor
+    // (inherited) shares the turn id but not its attempt, and counts only when the turn has no
+    // direct admission (a successor started without its own obligation).
+    let admitted: string | undefined;
+    let inherited: string | undefined;
+    if (turn != null) {
+      for (const send of this.admittedSendsByTaskId.get(taskId) ?? []) {
+        if (send.state !== "admitted" || send.turnId !== turn) continue;
+        if (send.inherited !== true) {
+          admitted = send.attemptId;
+          break;
+        }
+        inherited ??= send.attemptId;
+      }
+    }
+    admitted ??= inherited;
+    if (owned != null && (admitted == null || owned.attemptId === admitted)) {
+      return { ownedAttempt: owned, unownedAttemptId: undefined };
+    }
+    return {
+      ownedAttempt: undefined,
+      unownedAttemptId: admitted ?? this.deps.currentTaskAttemptId(taskId),
+    };
+  }
+  /** Sends admitted against the task's current attempt that have not claimed a turn yet. */
+  hasPendingAdmissions(taskId: string): boolean {
+    const current = this.deps.currentTaskAttemptId(taskId);
+    for (const send of this.admittedSendsByTaskId.get(taskId) ?? []) {
+      if (send.attemptId !== current) continue;
+      if (send.state === "pending" || send.state === "enqueued") return true;
+    }
+    return false;
+  }
+
+  /**
+   * Another reawakening's attempt is live in this process: published and still launching, or
+   * handed to work that has not ended (a pending/enqueued admission, the live turn, a stream).
+   * Synchronous; both reawakening paths evaluate it under the mutex BEFORE their own publish, so
+   * their own work is never counted. Evidence, not ownership: an owned attempt that nothing runs
+   * (a reactivation whose launch was refused, a follow-up turn that ended) stays reawakenable.
+   */
+  hasLiveReawakening(taskId: string): boolean {
+    if (
+      this.reawakeningsInFlight.has(taskId) ||
+      this.deps.workspaceService().getActiveTurnGeneration(taskId) != null ||
+      this.deps.aiService().isStreaming(taskId)
+    ) {
+      return true;
+    }
+    // Only sends that can still start a turn: one bound to a closed, superseded or stopped
+    // attempt reads stale at every later gate and is no continuation to protect.
+    for (const send of this.admittedSendsByTaskId.get(taskId) ?? []) {
+      if ((send.state === "pending" || send.state === "enqueued") && !send.token.admissionStale()) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+  dischargeAdmittedSend(send: AdmittedSend): void {
+    if (send.state === "discharged") return;
+    send.state = "discharged";
+    const sends = this.admittedSendsByTaskId.get(send.taskId);
+    sends?.delete(send);
+    if (sends?.size === 0) this.admittedSendsByTaskId.delete(send.taskId);
+    for (const record of send.capturedBy) {
+      record.pendingAdmissions.delete(send);
+    }
+    send.capturedBy.clear();
+    this.deps.pruneStreamEndDecisions(send.taskId);
+  }
+
+  /**
+   * Register one send obligation against `attemptId` and return its token. Every lifecycle event
+   * updates the obligation object and the records that captured it by reference — never
+   * whichever record currently occupies the task's slot.
+   */
+  private createAdmittedSend(
+    taskId: string,
+    attemptId: string,
+    attempt: OwnedTaskAttempt | undefined
+  ): AdmittedSend {
+    assertTaskAttemptId(attemptId, "createAdmittedSend");
+    const send: AdmittedSend = {
+      taskId,
+      attemptId,
+      ...(attempt != null ? { attempt } : {}),
+      state: "pending",
+      capturedBy: new Set(),
+      token: {
+        resolveDispatch: () => this.resolveEnqueuedSendDispatch(send),
+        admissionStale: () => this.isAdmittedSendStale(send),
+        onEnqueued: () => {
+          if (send.state === "pending") send.state = "enqueued";
+        },
+        onAdmitted: (turnId) => this.recordAdmittedSendTurn(send, turnId),
+        onDisposed: (kind) => this.disposeSendBeforeAdmission(send, kind),
+      },
+    };
+    const sends = this.admittedSendsByTaskId.get(taskId) ?? new Set<AdmittedSend>();
+    sends.add(send);
+    this.admittedSendsByTaskId.set(taskId, sends);
+    return send;
+  }
+
+  /** TurnAdmissionToken.admissionStale for `send` (see createAdmittedSend). */
+  private isAdmittedSendStale(send: AdmittedSend): boolean {
+    const { taskId, attemptId } = send;
+    // Admitted obligations belong to their turn: rotation, closure and stops no longer
+    // concern the send itself (the turn is captured by the stop record instead).
+    if (send.state === "admitted") return false;
+    if (send.state === "discharged") return true;
+    if (!this.attemptAdmissionOpen(taskId, attemptId)) return true;
+    if (send.attempt != null && this.ownedAttempt(taskId) !== send.attempt) {
+      return true;
+    }
+    // A send still in its preflight while the attempt's last stream is undecided, or was
+    // decided a report (or unknowable): it must not start a turn under that attempt. The
+    // queue's own gate handles enqueued entries (hold/refuse) before reaching here.
+    if (send.state === "pending") {
+      return this.streamEndBlocksNewTurn(taskId, attemptId);
+    }
+    return false;
+  }
+
+  /** The attempt's last stream is undecided, or was decided a report (or unknowable). */
+  private streamEndBlocksNewTurn(taskId: string, attemptId: string): boolean {
+    const decision = this.deps.findStreamEndDecision(taskId, attemptId);
+    return decision != null && decision.outcome !== "nonreport";
+  }
+
+  /** TurnAdmissionToken.onAdmitted for `send`: bind it to `turnId` (see createAdmittedSend). */
+  private recordAdmittedSendTurn(send: AdmittedSend, turnId: symbol): void {
+    const { taskId, attemptId } = send;
+    if (send.state === "admitted") {
+      // The dequeued entry is adopted by a second sendMessage naming the same turn.
+      if (send.turnId !== turnId) {
+        log.warn("[task-attempt] admitted send re-admitted under another turn; keeping first", {
+          taskId,
+          attemptId,
+        });
+      }
+      return;
+    }
+    if (send.state === "discharged") {
+      log.warn("[task-attempt] disposed send reported admitted; ignoring", {
+        taskId,
+        attemptId,
+      });
+      return;
+    }
+    send.state = "admitted";
+    send.turnId = turnId;
+    // A record that captured this obligation while pending now waits on the turn instead,
+    // and owes one more stopStream so the late turn is actually stopped: Phase B's single
+    // stopStream targeted the execution captured at Phase A, not this one.
+    for (const record of send.capturedBy) {
+      record.pendingAdmissions.delete(send);
+      record.capturedTurns.add(turnId);
+      record.cleanupInFlight += 1;
+      void this.deps
+        .aiService()
+        .stopStream(taskId, { abandonPartial: false })
+        .catch((error: unknown) => {
+          log.debug("[task-attempt] late-turn stopStream threw", { taskId, error });
+        })
+        .finally(() => {
+          record.cleanupInFlight -= 1;
+          this.deps.recheckWorkspaceStopRelease(taskId);
+        });
+    }
+    send.capturedBy.clear();
+  }
+
+  /** TurnAdmissionToken.onDisposed for `send`: no turn will exist (see createAdmittedSend). */
+  private disposeSendBeforeAdmission(
+    send: AdmittedSend,
+    kind: Parameters<TurnAdmissionToken["onDisposed"]>[0]
+  ): void {
+    const { taskId, attemptId } = send;
+    if (send.state === "admitted" || send.state === "discharged") return;
+    log.debug("[task-attempt] send obligation disposed before admission", {
+      taskId,
+      attemptId,
+      kind,
+    });
+    const records = [...send.capturedBy];
+    this.dischargeAdmittedSend(send);
+    for (const record of records) {
+      if (this.deps.isCurrentStopRecord(taskId, record)) {
+        this.deps.recheckWorkspaceStopRelease(taskId);
+      }
+    }
+    this.recheckResumeFailureSettlement(taskId);
+  }
+
+  /**
+   * Admission fence for a send into a workspace (AgentTaskIntegration.admitTaskWorkspaceTurn),
+   * evaluated synchronously at the session handoff. Not a task, or a pre-identity task entry →
+   * no obligation (the next admission stamps an id). A task whose current attempt is retired,
+   * stopping, or closed refuses; otherwise the send is bound to the current attempt — owned when
+   * this process owns exactly that id, unowned otherwise (startup re-drives, prior-process
+   * attempts). Ownership is never acquired here: a send authorizes work, not receipt authority.
+   */
+  admitTaskWorkspaceTurn(
+    workspaceId: string,
+    options: { acceptanceOrigin: TurnAcceptanceOrigin; expectedAttemptId?: string }
+  ): TaskTurnAdmission {
+    assert(workspaceId.length > 0, "admitTaskWorkspaceTurn: workspaceId must be non-empty");
+    let entry: WorkspaceConfigEntry | undefined;
+    try {
+      entry = findWorkspaceEntry(
+        this.deps.config().loadConfigOrDefault({ throwOnError: true }),
+        workspaceId
+      )?.workspace;
+    } catch (error: unknown) {
+      // Unreadable registry: the classification itself is indeterminate. Absence from this
+      // process's identity map (currentAttemptIdByTaskId) is no proof of "not a task" — a task
+      // loaded from disk and never rotated here has no entry — so EVERY send refuses here (fail
+      // closed) rather than bypassing the attempt fence. That deliberately includes root/non-task
+      // workspaces: without the registry nothing can tell them apart, and such a send could not
+      // run anyway (lenient readers then see no workspaces, so its metadata cannot be resolved);
+      // refusing here just fails it before anything is persisted. Sends resume once the registry
+      // reads again.
+      return {
+        kind: "refused",
+        message: `Workspace registry unreadable; send refused: ${getErrorMessage(error)}`,
+      };
+    }
+    // A send decided for a specific attempt whose row is gone (unpublished, removed) is stale,
+    // never an ordinary send into a workspace that is not a task.
+    if (entry == null && options.expectedAttemptId != null) {
+      return { kind: "refused", message: SEND_ADMISSION_STALE_MESSAGE };
+    }
+    // Checked before the task classification: another backend's removal of a root workspace
+    // deletes its checkout too (#4478).
+    if (entry?.pendingRemoval != null) {
+      return { kind: "refused", message: pendingRemovalAdmissionMessage(entry.pendingRemoval) };
+    }
+    if (!entry?.parentWorkspaceId) return { kind: "not-a-task" };
+    // The claim is monotonic and stronger than the id: a retired task refuses even when its id
+    // is missing or malformed (fail closed on partial state), so check it before the id.
+    if (entry.taskAttemptRetiredBy != null) {
+      return { kind: "refused", message: retiredAttemptMessage(entry.taskAttemptRetiredBy) };
+    }
+    return this.admitSendToCurrentAttempt(workspaceId, entry, options);
+  }
+  /**
+   * Attempt step of admitTaskWorkspaceTurn for a task entry that is not retired: refuse a send
+   * the current attempt cannot take, else bind it to that attempt.
+   */
+  private admitSendToCurrentAttempt(
+    workspaceId: string,
+    entry: WorkspaceConfigEntry,
+    options: { acceptanceOrigin: TurnAcceptanceOrigin; expectedAttemptId?: string }
+  ): TaskTurnAdmission {
+    const attemptId = this.deps.currentTaskAttemptId(workspaceId, entry);
+    if (attemptId == null) {
+      log.debug("[task-attempt] send into a pre-identity task entry carries no obligation", {
+        workspaceId,
+      });
+      return { kind: "not-a-task" };
+    }
+    // A send decided for one attempt never binds to a successor another writer admitted since:
+    // the decision is stale, not the row (see TaskWorkspaceSeam.admitTaskWorkspaceTurn).
+    if (options.expectedAttemptId != null && attemptId !== options.expectedAttemptId) {
+      log.info("[task-attempt] send refused: decided for a superseded attempt", {
+        workspaceId,
+        expectedAttemptId: options.expectedAttemptId,
+        attemptId,
+        acceptanceOrigin: options.acceptanceOrigin,
+      });
+      return { kind: "refused", message: SEND_ADMISSION_STALE_MESSAGE };
+    }
+    if (this.deps.isWorkspaceStopInProgress(workspaceId)) {
+      return { kind: "refused", message: WORKSPACE_STOP_IN_PROGRESS_SEND_BLOCKED_MESSAGE };
+    }
+    if (this.isAttemptClosed(workspaceId, attemptId)) {
+      log.info("[task-attempt] send refused: attempt settled", {
+        workspaceId,
+        attemptId,
+        acceptanceOrigin: options.acceptanceOrigin,
+      });
+      return { kind: "refused", message: TASK_ATTEMPT_SETTLED_SEND_BLOCKED_MESSAGE };
+    }
+    const owned = this.ownedAttempt(workspaceId);
+    const send = this.createAdmittedSend(
+      workspaceId,
+      attemptId,
+      owned?.attemptId === attemptId ? owned : undefined
+    );
+    return { kind: "admitted", token: send.token };
+  }
+
+  /**
+   * Lineage proof for a reawaken/reactivation of `entry`'s current attempt (P11): proven iff the
+   * predecessor is settled by this process's owner (recorded `settled`) or by a durable receipt.
+   * Anything else — no id, marked lineage, unsettled/closing owner, missing or unreadable receipt —
+   * is unproven and makes the successor ineligible for receipts. Never throws.
+   */
+  async evaluateAttemptLineage(
+    taskId: string,
+    entry: WorkspaceConfigEntry
+  ): Promise<{ proven: boolean; reason: string }> {
+    const attemptId = entry.taskAttemptId;
+    if (!isTaskAttemptId(attemptId)) {
+      return { proven: false, reason: "predecessor has no attempt id (pre-upgrade entry)" };
+    }
+    if (entry.taskAttemptUnproven === true) {
+      return { proven: false, reason: "lineage marked unproven" };
+    }
+    const owned = this.ownedAttempt(taskId);
+    if (owned?.attemptId === attemptId) {
+      if (!owned.receiptEligible) {
+        return { proven: false, reason: "predecessor owned but not receipt-eligible" };
+      }
+      if (this.ownedAttemptSettlementPhase(taskId, owned) === "closing") {
+        // The producer's awaited write is in flight; wait for its outcome (bounded, no lock).
+        await this.waitForAttemptClosureToSettle(taskId, ATTEMPT_CLOSURE_SETTLE_WAIT_MS);
+      }
+      if (this.ownedAttemptSettlementPhase(taskId, owned) === "settled") {
+        return { proven: true, reason: "predecessor settled by this process" };
+      }
+    }
+    const parentWorkspaceId = coerceNonEmptyString(entry.parentWorkspaceId);
+    if (parentWorkspaceId == null) return { proven: false, reason: "entry has no parent" };
+    const receipt = await readSubagentAttemptSettlementReceiptStrict(
+      path.join(this.deps.config().sessionsDir, parentWorkspaceId),
+      taskId,
+      attemptId
+    );
+    return lineageProofFromReceipt(receipt, parentWorkspaceId);
+  }
+  /** Phase of this process's settlement entry for `owned`; undefined when it records another. */
+  private ownedAttemptSettlementPhase(
+    taskId: string,
+    owned: OwnedTaskAttempt
+  ): AttemptSettlementEntry["phase"] | undefined {
+    const settlement = this.attemptSettlement(taskId);
+    return settlement?.attempt === owned ? settlement.phase : undefined;
+  }
+
+  /**
+   * Cross-process evidence (G2) that the task's CURRENT attempt ended without a report, for an
+   * attempt this process does not own (another backend, or this one before a restart). Called
+   * only after the report artifact read positively found none: a receipt never outranks a report.
+   *
+   * Attempt identity: the receipt must name the row's current taskAttemptId. The workflow journal
+   * names only the child task, and every task admission (launch, reawaken, re-drive) rotates the
+   * id, so the current id is the only attempt that can still run; a receipt for an older attempt
+   * says nothing about it. A plain send binds to the current id instead: another backend's turn
+   * can therefore run under an attempt this receipt settled (#4545) — its late report still wins
+   * unless a claim retired the attempt first (publishAgentTaskReport). Only the PARENT's receipt copy counts (the copy lineage reads; ancestor copies are fan-out
+   * and never evidence on their own), and it must name that parent. Every read is strict, and
+   * the row is re-read after the receipt so an admission landing in between is never missed.
+   */
+  async readUnownedSettlementProof(
+    taskId: string
+  ): Promise<{ kind: "proven"; attemptId: string } | { kind: "unproven"; reason: string }> {
+    const unproven = (reason: string) => ({ kind: "unproven" as const, reason });
+    const readRow = () =>
+      findWorkspaceEntry(this.deps.config().loadConfigOrDefault({ throwOnError: true }), taskId)
+        ?.workspace;
+    let row: WorkspaceConfigEntry | undefined;
+    try {
+      row = readRow();
+    } catch (error: unknown) {
+      return unproven(`config unreadable: ${getErrorMessage(error)}`);
+    }
+    const candidate = unownedSettlementCandidate(row);
+    if (candidate.kind === "unproven") return candidate;
+    const { attemptId, parentWorkspaceId } = candidate;
+    const receipt = await readSubagentAttemptSettlementReceiptStrict(
+      path.join(this.deps.config().sessionsDir, parentWorkspaceId),
+      taskId,
+      attemptId
+    );
+    const receiptGap = unownedSettlementReceiptGap(receipt, attemptId, parentWorkspaceId);
+    if (receiptGap != null) return unproven(receiptGap);
+    let latest: WorkspaceConfigEntry | undefined;
+    try {
+      latest = readRow();
+    } catch (error: unknown) {
+      return unproven(`config unreadable: ${getErrorMessage(error)}`);
+    }
+    if (
+      latest?.taskAttemptId !== attemptId ||
+      latest.taskStatus !== "interrupted" ||
+      latest.taskAttemptUnproven === true
+    ) {
+      return unproven("the attempt changed while its receipt was read");
+    }
+    return { kind: "proven", attemptId };
+  }
+
+  /**
+   * Workflow claim that retires `attemptId` of `taskId` so exactly one replacement can be
+   * published for it (G2). A pure compare-and-swap fence: the classifier (readAttemptOutcome)
+   * decided the attempt ended without a report; the claim only makes that decision exclusive.
+   * Granted while the row still names the attempt, is interrupted, belongs to this run/step and
+   * has no replacement yet; an earlier claim of the same run/step is re-stamped with a fresh
+   * nonce (a stale runner holding the old nonce can then no longer publish). Every admission
+   * already refuses a retired attempt. Confirmed by a strict re-read, since a config edit that
+   * fails to write is not always surfaced to its caller.
+   *
+   * A spent claim is never re-granted, even when its replacementTaskId names a row that is gone
+   * (a replacement whose launch failed sanitization is unpublished). That is safe because the
+   * workflow checkpoints the replacement (createMany's onTaskReserved) BEFORE the publishing
+   * commit that sets replacementTaskId: once it is set, the journal already names the
+   * replacement, so no runner classifies this attempt again. Recovery is retry-from-checkpoint's
+   * fresh reservation (WorkflowRunner.replacementRace.test.ts).
+   */
+  async claimRetiredAttempt(
+    taskId: string,
+    attemptId: string,
+    claimant: { runId: string; stepId: string; inputHash: string }
+  ): Promise<Result<{ nonce: string }, string>> {
+    assert(taskId.length > 0, "claimRetiredAttempt: taskId must be non-empty");
+    assertTaskAttemptId(attemptId, "claimRetiredAttempt");
+    assert(
+      claimant.runId.length > 0 && claimant.stepId.length > 0,
+      "claimRetiredAttempt: claimant"
+    );
+    return await this.deps.workspaceEventLocks().withLock(taskId, async () => {
+      if (
+        this.deps.isWorkspaceStopInProgress(taskId) ||
+        this.deps.turnManager().getLiveWorkspaceTurnRegistration(taskId) != null ||
+        this.deps.aiService().isStreaming(taskId) ||
+        this.deps.workspaceService().getActiveTurnGeneration(taskId) != null
+      ) {
+        return Err("attempt is live");
+      }
+      const nonce = randomUUID();
+      let refusal: string | undefined = "task record not found";
+      try {
+        await this.deps.editWorkspaceEntry(
+          taskId,
+          (ws) => {
+            refusal = retiredAttemptClaimRefusal(ws, attemptId, claimant);
+            if (refusal != null) return;
+            markChildGoalPauseOwed(ws);
+            ws.taskAttemptRetiredBy = {
+              runId: claimant.runId,
+              stepId: claimant.stepId,
+              inputHash: claimant.inputHash,
+              childTaskId: taskId,
+              attemptId,
+              mode: "no-report",
+              at: new Date().toISOString(),
+              nonce,
+            };
+          },
+          { allowMissing: true }
+        );
+      } catch (error: unknown) {
+        return Err(`claim write failed: ${getErrorMessage(error)}`);
+      }
+      if (refusal != null) return Err(refusal);
+      let row: WorkspaceConfigEntry | undefined;
+      try {
+        row = findWorkspaceEntry(
+          this.deps.config().loadConfigOrDefault({ throwOnError: true }),
+          taskId
+        )?.workspace;
+      } catch (error: unknown) {
+        return Err(`claim unconfirmed: config unreadable: ${getErrorMessage(error)}`);
+      }
+      if (row?.taskAttemptId !== attemptId || row.taskAttemptRetiredBy?.nonce !== nonce) {
+        return Err("claim lost");
+      }
+      await this.deps.settleChildGoalPause(taskId);
+      return Ok({ nonce });
+    });
+  }
+  /**
+   * A workflow step's started checkpoint is written in createMany's onTaskReserved, BEFORE
+   * commitReservations publishes the child's row. A crash in between leaves a checkpoint naming a
+   * task that never existed. Replacing it outright is unsafe: the reserving backend may only be
+   * stalled (lease expired, process alive), and its late commit would then publish a second child
+   * for the step, or re-run a step the replacement already recorded. So the replacing runner first
+   * tombstones the ID on the parent's row; commitReservations checks the tombstone inside its own
+   * config write and refuses, which makes "replace" and "late commit" mutually exclusive.
+   *
+   * Granted only while a strict read shows no row for the task (checked inside the same write) and
+   * this process owns no attempt for it (an own reservation may still be before its commit).
+   * Idempotent. Confirmed by a strict re-read, since a config edit that fails to write is not
+   * always surfaced to its caller.
+   */
+  async tombstoneUnpublishedReservation(
+    parentWorkspaceId: string,
+    taskId: string
+  ): Promise<Result<void, string>> {
+    assert(parentWorkspaceId.length > 0, "tombstoneUnpublishedReservation: parentWorkspaceId");
+    assert(taskId.length > 0, "tombstoneUnpublishedReservation: taskId");
+    return await this.deps.workspaceEventLocks().withLock(taskId, async () => {
+      if (this.hasOwnedAttempt(taskId)) {
+        return Err("this process owns an attempt for the task");
+      }
+      let refusal: string | undefined;
+      try {
+        await this.deps.config().editConfig((config) => {
+          refusal = undefined;
+          if (findWorkspaceEntry(config, taskId) != null) {
+            refusal = "the task was published";
+            return config;
+          }
+          const parent = findWorkspaceEntry(config, parentWorkspaceId)?.workspace;
+          if (parent == null) {
+            refusal = "parent record not found";
+            return config;
+          }
+          const tombstones = parent.taskReservationTombstones ?? [];
+          if (!tombstones.includes(taskId)) {
+            parent.taskReservationTombstones = [...tombstones, taskId];
+          }
+          return config;
+        });
+      } catch (error: unknown) {
+        return Err(`tombstone write failed: ${getErrorMessage(error)}`);
+      }
+      if (refusal != null) return Err(refusal);
+      let config: ProjectsConfig;
+      try {
+        config = this.deps.config().loadConfigOrDefault({ throwOnError: true });
+      } catch (error: unknown) {
+        return Err(`tombstone unconfirmed: config unreadable: ${getErrorMessage(error)}`);
+      }
+      if (
+        findWorkspaceEntry(config, taskId) != null ||
+        findWorkspaceEntry(
+          config,
+          parentWorkspaceId
+        )?.workspace.taskReservationTombstones?.includes(taskId) !== true
+      ) {
+        return Err("tombstone lost");
+      }
+      return Ok(undefined);
+    });
+  }
 }
diff --git a/src/node/services/taskService.attemptAdmission.test.ts b/src/node/services/taskService.attemptAdmission.test.ts
index 919ad531335..088a068b77a 100644
--- a/src/node/services/taskService.attemptAdmission.test.ts
+++ b/src/node/services/taskService.attemptAdmission.test.ts
@@ -89,11 +89,11 @@ interface Internals {
     >;
     currentAttemptIdByTaskId: Map<string, string>;
     resumeFailureSettlementByTaskId: Map<string, unknown>;
+    admittedSendsByTaskId: Map<
+      string,
+      Set<{ state: string; turnId?: symbol; attemptId: string; attempt?: unknown }>
+    >;
   };
-  admittedSendsByTaskId: Map<
-    string,
-    Set<{ state: string; turnId?: symbol; attemptId: string; attempt?: unknown }>
-  >;
   taskStopLatch: {
     workspaceStopRecords: Map<
       string,
@@ -408,7 +408,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
       expect(entryOf(config, spawnedId)?.taskAttemptId).toBe(reserved!.taskAttemptId);
       expect(svc.attemptLedger.ownedAttemptByTaskId.get(spawnedId)).toBe(owned);
       // The fake host disposed the token (no session → no turn): the obligation is discharged.
-      expect(svc.admittedSendsByTaskId.get(spawnedId)).toBeUndefined();
+      expect(svc.attemptLedger.admittedSendsByTaskId.get(spawnedId)).toBeUndefined();
       expect(await taskService.readAttemptOutcome(spawnedId, requesting)).toEqual({
         kind: "live",
         executionId: spawnedId,
@@ -750,8 +750,8 @@ describe("TaskService attempt identity and send admission (G1)", () => {
       const sendToken = admitted(
         taskService.admitTaskWorkspaceTurn(taskId, { acceptanceOrigin: "manual" })
       );
-      expect(svc.admittedSendsByTaskId.get(taskId)?.size).toBe(1);
-      for (const send of svc.admittedSendsByTaskId.get(taskId) ?? []) {
+      expect(svc.attemptLedger.admittedSendsByTaskId.get(taskId)?.size).toBe(1);
+      for (const send of svc.attemptLedger.admittedSendsByTaskId.get(taskId) ?? []) {
         expect(send.attemptId).toBe(continuedAttemptId);
       }
       // The continuation runs and settles without a second report (a reported child never turns
@@ -759,7 +759,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
       const turn = Symbol("manual-continuation");
       sendToken.onAdmitted(turn);
       host.settleTurn(taskId, turn);
-      expect(svc.admittedSendsByTaskId.get(taskId)).toBeUndefined();
+      expect(svc.attemptLedger.admittedSendsByTaskId.get(taskId)).toBeUndefined();
       // ...so a resume (same rescue, same fence) continues THAT attempt rather than rotating again.
       expect(await taskService.markInterruptedTaskRunning(taskId)).toBe(false);
       expect(entryOf(config, taskId)?.taskAttemptId).toBe(continuedAttemptId);
@@ -809,8 +809,8 @@ describe("TaskService attempt identity and send admission (G1)", () => {
       const sendToken = admitted(
         taskService.admitTaskWorkspaceTurn(taskId, { acceptanceOrigin: "manual" })
       );
-      expect(svc.admittedSendsByTaskId.get(taskId)?.size).toBe(1);
-      for (const send of svc.admittedSendsByTaskId.get(taskId) ?? []) {
+      expect(svc.attemptLedger.admittedSendsByTaskId.get(taskId)?.size).toBe(1);
+      for (const send of svc.attemptLedger.admittedSendsByTaskId.get(taskId) ?? []) {
         expect(send.attemptId).toBe(firstAttemptId);
       }
       sendToken.onDisposed("no-work");
@@ -1358,7 +1358,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
           attempt: owned,
           phase: "settled",
         });
-        expect(svc.admittedSendsByTaskId.get(taskId)).toBeUndefined();
+        expect(svc.attemptLedger.admittedSendsByTaskId.get(taskId)).toBeUndefined();
         expect(await taskService.readAttemptOutcome(taskId, requesting)).toMatchObject({
           kind: "terminal-no-report",
         });
@@ -1440,7 +1440,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
           expect(dispatchCalls[0].turnAdmission).toBeDefined();
           expect(dispatchCalls[0].stale).toBe(false);
           // The follow-up's turn is bound to the attempt rotated here, so a Stop captures it.
-          const sends = [...svc.admittedSendsByTaskId.get(taskId)!];
+          const sends = [...svc.attemptLedger.admittedSendsByTaskId.get(taskId)!];
           expect(sends.map((s) => s.attemptId)).toEqual([row.taskAttemptId!]);
           expect(sends[0].state).toBe("admitted");
           expect(sendMessage.mock.calls.filter((call) => call[0] === taskId)).toHaveLength(0);
@@ -1454,7 +1454,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
           expect(dispatchCalls[0].stale).toBe(true);
         }
         expect(sendMessage.mock.calls.filter((call) => call[0] === taskId)).toHaveLength(0);
-        expect(svc.admittedSendsByTaskId.get(taskId)).toBeUndefined();
+        expect(svc.attemptLedger.admittedSendsByTaskId.get(taskId)).toBeUndefined();
         expect(svc.attemptLedger.ownedAttemptByTaskId.has(taskId)).toBe(false);
         if (variant.startsWith("rotated")) expect(row.taskAttemptId).toBe(foreign);
         else expect(row.taskAttemptRetiredBy).toEqual(claim);
@@ -1487,7 +1487,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
       await taskService.recoverInterruptedTasks();
       expect(tokens).toHaveLength(1);
       // No obligation is left pending for a turn that will never exist...
-      expect(svc.admittedSendsByTaskId.get(taskId)).toBeUndefined();
+      expect(svc.attemptLedger.admittedSendsByTaskId.get(taskId)).toBeUndefined();
       // ...so a Stop has nothing to wait for and releases its latch.
       await raceWithTimeout(taskService.terminateAllDescendantAgentTasks(rootId), 2_000);
       expect(taskService.isWorkspaceStopInProgress(taskId)).toBe(false);
@@ -1818,9 +1818,9 @@ describe("TaskService attempt identity and send admission (G1)", () => {
       expect(winner).toMatch(ATTEMPT_ID);
       expect(winner).not.toBe(previous);
       expect(racing).toHaveLength(1);
-      expect([...svc.admittedSendsByTaskId.get(taskId)!].map((s) => s.attemptId)).toEqual([
-        previous,
-      ]);
+      expect(
+        [...svc.attemptLedger.admittedSendsByTaskId.get(taskId)!].map((s) => s.attemptId)
+      ).toEqual([previous]);
       expect(racing[0].admissionStale()).toBe(true);
       racing[0].onDisposed("refused");
 
@@ -1836,7 +1836,9 @@ describe("TaskService attempt identity and send admission (G1)", () => {
       const next = admitted(
         taskService.admitTaskWorkspaceTurn(taskId, { acceptanceOrigin: "manual" })
       );
-      expect([...svc.admittedSendsByTaskId.get(taskId)!].map((s) => s.attemptId)).toEqual([winner]);
+      expect(
+        [...svc.attemptLedger.admittedSendsByTaskId.get(taskId)!].map((s) => s.attemptId)
+      ).toEqual([winner]);
       expect(next.admissionStale()).toBe(false);
       next.onDisposed("no-work");
     });
@@ -1974,7 +1976,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
       const gapSend = admitted(
         taskService.admitTaskWorkspaceTurn(taskId, { acceptanceOrigin: "automatic" })
       );
-      const sends = [...svc.admittedSendsByTaskId.get(taskId)!];
+      const sends = [...svc.attemptLedger.admittedSendsByTaskId.get(taskId)!];
       expect(sends.map((s) => s.attemptId)).toEqual([fresh]);
       expect(sends[0].attempt).toBeUndefined();
       gate.resolve();
@@ -2070,9 +2072,9 @@ describe("TaskService attempt identity and send admission (G1)", () => {
       const token = admitted(
         taskService.admitTaskWorkspaceTurn(taskId, { acceptanceOrigin: "manual" })
       );
-      expect([...svc.admittedSendsByTaskId.get(taskId)!].map((s) => s.attemptId)).toEqual([
-        successor,
-      ]);
+      expect(
+        [...svc.attemptLedger.admittedSendsByTaskId.get(taskId)!].map((s) => s.attemptId)
+      ).toEqual([successor]);
       token.onDisposed("no-work");
     });
 
@@ -2177,7 +2179,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
       const afterRotation = admitted(
         taskService.admitTaskWorkspaceTurn(taskId, { acceptanceOrigin: "manual" })
       );
-      const sends = [...svc.admittedSendsByTaskId.get(taskId)!];
+      const sends = [...svc.attemptLedger.admittedSendsByTaskId.get(taskId)!];
       expect(sends).toHaveLength(1);
       expect(sends[0].attemptId).toBe(foreign);
       expect(sends[0].attempt).toBeUndefined();
@@ -2332,7 +2334,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
             // compressed settle bound can outrun; the record still releases on its own (nothing
             // below disposes a token or settles a turn), so wait for it instead of racing it.
             await waitForStopRelease(taskService, spawnedId);
-            expect(svc.admittedSendsByTaskId.has(spawnedId)).toBe(false);
+            expect(svc.attemptLedger.admittedSendsByTaskId.has(spawnedId)).toBe(false);
             expect(svc.attemptLedger.attemptSettlementByTaskId.get(spawnedId)).toMatchObject({
               phase: "settled",
             });
@@ -2366,7 +2368,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
             phase: "settled",
             source: "stop-settled",
           });
-          expect(svc.admittedSendsByTaskId.has(spawnedId)).toBe(false);
+          expect(svc.attemptLedger.admittedSendsByTaskId.has(spawnedId)).toBe(false);
           expect(await taskService.readAttemptOutcome(spawnedId, requesting)).toMatchObject({
             kind: "terminal-no-report",
           });
@@ -2572,7 +2574,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
             phase: "closing",
             source: "launch-failed",
           });
-          const obligations = [...(svc.admittedSendsByTaskId.get(spawnedId) ?? [])];
+          const obligations = [...(svc.attemptLedger.admittedSendsByTaskId.get(spawnedId) ?? [])];
           expect(obligations).toHaveLength(1);
           expect(obligations[0]).toMatchObject({
             attemptId: mintedAttemptId,
@@ -2588,7 +2590,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
             liveTurn = undefined;
             host.settleTurn(spawnedId, racingTurn);
           }
-          expect(svc.admittedSendsByTaskId.has(spawnedId)).toBe(false);
+          expect(svc.attemptLedger.admittedSendsByTaskId.has(spawnedId)).toBe(false);
           expect(svc.attemptLedger.attemptSettlementByTaskId.get(spawnedId)).toMatchObject({
             attemptId: mintedAttemptId,
             phase: "closing",
@@ -2807,7 +2809,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
         taskService.admitTaskWorkspaceTurn(owned, { acceptanceOrigin: "automatic" })
       );
       expect(token.admissionStale()).toBe(false);
-      const [send] = [...svc.admittedSendsByTaskId.get(owned)!];
+      const [send] = [...svc.attemptLedger.admittedSendsByTaskId.get(owned)!];
       expect(send).toMatchObject({ state: "pending", attemptId });
       // A closure recorded for the current id refuses further sends and marks the token stale.
       svc.closeAttemptAdmission(owned, attemptId, undefined, "test");
@@ -2818,7 +2820,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
       });
       token.onDisposed("refused");
       expect(send.state).toBe("discharged");
-      expect(svc.admittedSendsByTaskId.get(owned)).toBeUndefined();
+      expect(svc.attemptLedger.admittedSendsByTaskId.get(owned)).toBeUndefined();
       // A disposed token is terminal even though nothing else changed.
       expect(token.admissionStale()).toBe(true);
     });
@@ -2844,10 +2846,9 @@ describe("TaskService attempt identity and send admission (G1)", () => {
       const turn = Symbol("turn");
       adopted.onAdmitted(turn);
       adopted.onAdmitted(turn); // idempotent per turn
-      expect([...svc.admittedSendsByTaskId.get(taskId)!].map((s) => s.state).sort()).toEqual([
-        "admitted",
-        "pending",
-      ]);
+      expect(
+        [...svc.attemptLedger.admittedSendsByTaskId.get(taskId)!].map((s) => s.state).sort()
+      ).toEqual(["admitted", "pending"]);
       const { stopped } = await startStopAwaitingRelease(taskService, rootId, taskId);
       // The admitted turn AND the pending obligation keep the record: the Stop cannot release
       // until the turn settles and the pending send is dispositioned.
@@ -2858,7 +2859,9 @@ describe("TaskService attempt identity and send admission (G1)", () => {
       host.settleTurn(taskId, turn);
       await settle();
       expect(taskService.isWorkspaceStopInProgress(taskId)).toBe(true);
-      expect([...svc.admittedSendsByTaskId.get(taskId)!].map((s) => s.state)).toEqual(["pending"]);
+      expect([...svc.attemptLedger.admittedSendsByTaskId.get(taskId)!].map((s) => s.state)).toEqual(
+        ["pending"]
+      );
       // The pending send is stale (stop in progress) and gets refused at its gate.
       expect(pending.admissionStale()).toBe(true);
       pending.onDisposed("refused");
@@ -2866,7 +2869,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
       expect(taskService.isWorkspaceStopInProgress(taskId)).toBe(false);
       // The Stop resolves only now, once its latch dropped (#6727).
       expect((await stopped).success).toBe(true);
-      expect(svc.admittedSendsByTaskId.get(taskId)).toBeUndefined();
+      expect(svc.attemptLedger.admittedSendsByTaskId.get(taskId)).toBeUndefined();
       expect(await taskService.readAttemptOutcome(taskId, requesting)).toMatchObject({
         kind: "terminal-no-report",
       });
@@ -2927,7 +2930,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
       host.settleTurn(taskId, late);
       await settle();
       expect(taskService.isWorkspaceStopInProgress(taskId)).toBe(false);
-      expect(svc.admittedSendsByTaskId.get(taskId)).toBeUndefined();
+      expect(svc.attemptLedger.admittedSendsByTaskId.get(taskId)).toBeUndefined();
       expect(svc.attemptLedger.attemptSettlementByTaskId.get(taskId)?.phase).toBe("settled");
       expect(await taskService.readAttemptOutcome(taskId, requesting)).toMatchObject({
         kind: "terminal-no-report",
@@ -2971,7 +2974,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
       host.supersedeTurn(taskId, first, second);
       expect(record.capturedTurns.has(first)).toBe(false);
       expect(record.capturedTurns.has(second)).toBe(true);
-      const [send] = [...svc.admittedSendsByTaskId.get(taskId)!];
+      const [send] = [...svc.attemptLedger.admittedSendsByTaskId.get(taskId)!];
       expect(send.turnId).toBe(second);
       // A settlement for the superseded generation is not evidence about the live successor.
       host.settleTurn(taskId, first);
@@ -3013,7 +3016,9 @@ describe("TaskService attempt identity and send admission (G1)", () => {
           message: WORKSPACE_STOP_IN_PROGRESS_SEND_BLOCKED_MESSAGE,
         }
       );
-      expect(internals(taskService).admittedSendsByTaskId.get(taskId)).toBeUndefined();
+      expect(
+        internals(taskService).attemptLedger.admittedSendsByTaskId.get(taskId)
+      ).toBeUndefined();
     });
 
     test("markTaskLaunchFailed closes the attempt for its id; a later send under that id is refused", async () => {
@@ -3264,7 +3269,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
       host.settleTurn(taskId, turn);
       await settle();
       expect(taskService.isWorkspaceStopInProgress(taskId)).toBe(false);
-      expect(svc.admittedSendsByTaskId.get(taskId)).toBeUndefined();
+      expect(svc.attemptLedger.admittedSendsByTaskId.get(taskId)).toBeUndefined();
       expect(await taskService.readAttemptOutcome(taskId, requesting)).toMatchObject({
         kind: "terminal-no-report",
       });
@@ -3421,7 +3426,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
           kind: "refused",
           message: retiredAttemptMessage(claim),
         });
-        expect(internals(taskService).admittedSendsByTaskId.has(taskId)).toBe(false);
+        expect(internals(taskService).attemptLedger.admittedSendsByTaskId.has(taskId)).toBe(false);
       }
     );
 
@@ -3475,7 +3480,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
           });
           expect(admission.kind).toBe("refused");
           if (admission.kind === "refused") expect(admission.message).toContain("unreadable");
-          expect(svc.admittedSendsByTaskId.has(workspaceId)).toBe(false);
+          expect(svc.attemptLedger.admittedSendsByTaskId.has(workspaceId)).toBe(false);
         } finally {
           load.mockRestore();
         }
@@ -3540,7 +3545,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
         expect(sendMessage).not.toHaveBeenCalled();
         // The fenced token was minted for R by the re-drive; a throw before the handoff must not
         // leave it pending — a pending obligation nobody holds keeps R's stop cascade waiting.
-        expect(svc.admittedSendsByTaskId.has(taskId)).toBe(false);
+        expect(svc.attemptLedger.admittedSendsByTaskId.has(taskId)).toBe(false);
         // R's row is untouched apart from the rotation itself (no budget charged).
         expect(entryOf(config, taskId)).toMatchObject({
           taskStatus: "awaiting_report",
@@ -3606,7 +3611,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
         expect(rotatedAttemptId).toMatch(ATTEMPT_ID);
         expect(sendMessage).not.toHaveBeenCalled();
         // R's fenced obligation is disposed either way (no pending obligation outlives the helper).
-        expect(svc.admittedSendsByTaskId.has(taskId)).toBe(false);
+        expect(svc.attemptLedger.admittedSendsByTaskId.has(taskId)).toBe(false);
         const failure = await readSubagentFailureArtifact(
           path.join(config.sessionsDir, rootId),
           taskId
@@ -3686,13 +3691,15 @@ describe("TaskService attempt identity and send admission (G1)", () => {
               const messageId = `assistant-${completions.length}`;
               // Which obligations (and whose) the stream starts under.
               streamStarts.push(
-                [...(ledger.svc?.admittedSendsByTaskId.get(taskId) ?? [])].map((send) => ({
-                  attemptId: send.attemptId,
-                  state: send.state,
-                  owned:
-                    ledger.svc?.attemptLedger.ownedAttemptByTaskId.get(taskId)?.attemptId ===
-                    send.attemptId,
-                }))
+                [...(ledger.svc?.attemptLedger.admittedSendsByTaskId.get(taskId) ?? [])].map(
+                  (send) => ({
+                    attemptId: send.attemptId,
+                    state: send.state,
+                    owned:
+                      ledger.svc?.attemptLedger.ownedAttemptByTaskId.get(taskId)?.attemptId ===
+                      send.attemptId,
+                  })
+                )
               );
               aiEmitter.emit("stream-start", {
                 type: "stream-start",
@@ -3759,7 +3766,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
             // No prompt reached the stream, nothing is bound to B, R's obligation was disposed
             // at the refusal, and B's row (budget included) is exactly as its writer left it.
             expect(streamStarts).toHaveLength(0);
-            expect(svc.admittedSendsByTaskId.has(taskId)).toBe(false);
+            expect(svc.attemptLedger.admittedSendsByTaskId.has(taskId)).toBe(false);
             expect(entryOf(config, taskId)).toMatchObject({
               taskStatus: "awaiting_report",
               taskAttemptId: foreignAttemptId,
@@ -3982,7 +3989,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
       turn2
     );
     session.activeTurn = turn2;
-    const sends = [...(svc.admittedSendsByTaskId.get(taskId) ?? [])];
+    const sends = [...(svc.attemptLedger.admittedSendsByTaskId.get(taskId) ?? [])];
     expect(sends.map((send) => send.attemptId)).toEqual([attemptA, foreign]);
     // Turn 2's stream belongs to B (its direct admission), not to A's inherited obligation.
     expect(
@@ -4022,7 +4029,7 @@ describe("TaskService attempt identity and send admission (G1)", () => {
       // A manual send gets a turn-admission token bound to the healed attempt (never not-a-task).
       const admission = taskService.admitTaskWorkspaceTurn(taskId, { acceptanceOrigin: "manual" });
       expect(admission.kind).toBe("admitted");
-      const pending = [...(svc.admittedSendsByTaskId.get(taskId) ?? [])];
+      const pending = [...(svc.attemptLedger.admittedSendsByTaskId.get(taskId) ?? [])];
       expect(pending.map((send) => send.attemptId)).toEqual([healed!.taskAttemptId!]);
       if (admission.kind === "admitted") admission.token.onDisposed("no-work");
       // The Stop/settlement fence applies to it: a closed attempt refuses further sends.
diff --git a/src/node/services/taskService.reportDecisionHold.test.ts b/src/node/services/taskService.reportDecisionHold.test.ts
index 6735f89adf4..1a9d617d2ad 100644
--- a/src/node/services/taskService.reportDecisionHold.test.ts
+++ b/src/node/services/taskService.reportDecisionHold.test.ts
@@ -52,8 +52,8 @@ interface Internals {
   attemptLedger: {
     ownedAttemptByTaskId: Map<string, { attemptId?: string }>;
     attemptSettlementByTaskId: Map<string, { attemptId?: string; phase: string }>;
+    admittedSendsByTaskId: Map<string, Set<{ state: string; attemptId: string }>>;
   };
-  admittedSendsByTaskId: Map<string, Set<{ state: string; attemptId: string }>>;
   streamEndDecisionsByTaskId: Map<string, Array<{ attemptId: string; outcome: string }>>;
   taskStopLatch: { workspaceStopRecords: Map<string, unknown> };
   emitWorkspaceMetadata: (workspaceId: string) => Promise<void>;
@@ -128,7 +128,9 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
   const entryOf = (config: Config, id: string) => findWorkspaceInConfig(config, id);
   /** Obligations that could still start a turn (an admitted one belongs to its turn until it settles). */
   const outstanding = (svc: Internals, id: string) =>
-    [...(svc.admittedSendsByTaskId.get(id) ?? [])].filter((send) => send.state !== "admitted");
+    [...(svc.attemptLedger.admittedSendsByTaskId.get(id) ?? [])].filter(
+      (send) => send.state !== "admitted"
+    );
   async function until(condition: () => boolean, label: string): Promise<void> {
     const deadline = Date.now() + 5_000;
     while (!condition() && Date.now() < deadline) {
@@ -847,7 +849,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
       });
       // The follow-up went out (turn admitted); any recovery nudge queues behind it; the decision
       // has no reader left.
-      const sends = [...(svc.admittedSendsByTaskId.get(childId) ?? [])];
+      const sends = [...(svc.attemptLedger.admittedSendsByTaskId.get(childId) ?? [])];
       expect(sends.some((send) => send.state === "admitted" && send.attemptId === attemptA)).toBe(
         true
       );
@@ -1132,7 +1134,7 @@ describe("report-decision hold for queued follow-ups (real host)", () => {
       expect(completions).toHaveLength(1);
       expect(entryOf(config, childId)?.taskAttemptId).toBe("att_00000000000000b4");
       expect(stack.heldTexts()).toEqual(["follow-up text"]);
-      const sends = [...(svc.admittedSendsByTaskId.get(childId) ?? [])];
+      const sends = [...(svc.attemptLedger.admittedSendsByTaskId.get(childId) ?? [])];
       expect(sends.some((send) => send.attemptId === "att_00000000000000b4")).toBe(false);
       expect(outstanding(svc, childId).filter((send) => send.attemptId === attemptA)).toHaveLength(
         0
diff --git a/src/node/services/taskService.staleAttemptEffects.test.ts b/src/node/services/taskService.staleAttemptEffects.test.ts
index 163247e8978..5ddcfe94eb0 100644
--- a/src/node/services/taskService.staleAttemptEffects.test.ts
+++ b/src/node/services/taskService.staleAttemptEffects.test.ts
@@ -41,7 +41,7 @@ const A = "att_00000000000000a1";
 const B = "att_00000000000000b1";
 
 interface Internals {
-  admittedSendsByTaskId: Map<string, Set<{ attemptId: string; state: string }>>;
+  attemptLedger: { admittedSendsByTaskId: Map<string, Set<{ attemptId: string; state: string }>> };
   workspaceEventLocks: MutexMap<string>;
 }
 const internals = (service: TaskService) => service as unknown as Internals;
@@ -117,7 +117,7 @@ describe("stale attempt effects after another backend admits a successor (#4414)
       ): Promise<Result<void>> => {
         await beforeSend?.();
         const taskService = ledger.taskService!;
-        const sends = internals(taskService).admittedSendsByTaskId;
+        const sends = internals(taskService).attemptLedger.admittedSendsByTaskId;
         const callerToken = internal?.turnAdmission;
         if (callerToken != null) {
           // A caller-minted token: the real host refuses it at its admission gates once stale.
diff --git a/src/node/services/taskService.ts b/src/node/services/taskService.ts
index 382842abde3..7815f493c96 100644
--- a/src/node/services/taskService.ts
+++ b/src/node/services/taskService.ts
@@ -42,7 +42,6 @@ import {
   type AgentTaskStatus,
   type BackgroundableForegroundWaiter,
   type QueueCutAttributionSnapshot,
-  type QueuedDispatchDecision,
   type ResolvedWorkspaceAiSettings,
   type TaskCreateArgs,
   type TaskKind,
@@ -68,12 +67,8 @@ import {
   type AgentTaskIndex,
   type AgentTaskWorkspaceEntry,
 } from "@/node/services/agentTaskIndex";
-import {
-  readSubagentAttemptSettlementReceiptStrict,
-  type SubagentAttemptSettlementReceiptReadResult,
-  type SubagentAttemptSettlementSource,
-} from "@/node/services/subagentAttemptSettlements";
-import { assertTaskAttemptId, isTaskAttemptId, newTaskAttemptId } from "@/node/utils/taskAttemptId";
+import type { SubagentAttemptSettlementSource } from "@/node/services/subagentAttemptSettlements";
+import { isTaskAttemptId, newTaskAttemptId } from "@/node/utils/taskAttemptId";
 import type { HistoryService } from "@/node/services/historyService";
 import type { InitStateManager } from "@/node/services/initStateManager";
 import { STRUCTURED_WORKFLOW_REPORT_PLACEHOLDER_MARKDOWN } from "@/common/constants/workflowReports";
@@ -96,10 +91,7 @@ import {
   taskRecoveryPromptDedupePrefix,
   TASK_RECOVERY_DIAGNOSTIC_MAX_CHARS,
   SEND_ADMISSION_STALE_MESSAGE,
-  TASK_ATTEMPT_SETTLED_SEND_BLOCKED_MESSAGE,
-  TASK_REPORT_OUTCOME_INDETERMINATE_UNSENT_MESSAGE,
   TASK_REAWAKEN_LOST_SEND_BLOCKED_MESSAGE,
-  TASK_REPORTED_QUEUED_SEND_UNSENT_MESSAGE,
   WORKSPACE_STOP_IN_PROGRESS_SEND_BLOCKED_MESSAGE,
   pendingRemovalAdmissionMessage,
   retiredAttemptMessage,
@@ -110,7 +102,8 @@ import {
   TASK_FAMILY_MESSAGE_MAX_CHARS,
 } from "@/constants/taskMessages";
 import { log } from "@/node/services/log";
-import { AttemptLedger } from "@/node/services/attemptLedger";
+import { ATTEMPT_CLOSURE_SETTLE_WAIT_MS, AttemptLedger } from "@/node/services/attemptLedger";
+export { ATTEMPT_CLOSURE_SETTLE_WAIT_MS } from "@/node/services/attemptLedger";
 import { TaskStopLatch } from "@/node/services/taskStopLatch";
 import { PeerSendParking } from "@/node/services/peerSendParking";
 import {
@@ -1385,15 +1378,6 @@ interface StreamEndTaskOrigin {
 /** Ended streams per task the goal arbitration remembers, so a duplicate delivery is a no-op. */
 const CHILD_GOAL_ARBITRATED_STREAMS_MAX = 32;
 
-/**
- * Bound on waiting for an owned predecessor's in-flight settlement write before deciding a
- * successor's lineage (evaluateAttemptLineage), and on a failed direct launch's wait for the
- * owners its stop cascade captured before its rollback (TaskService.create). Elapsing the bound
- * is a decision — unproven lineage, deferred rollback — not an error; the write or the cascade
- * keeps its own ownership and completes on its own.
- */
-export const ATTEMPT_CLOSURE_SETTLE_WAIT_MS = 5_000;
-
 /** See TaskStopLatch.workspaceStopRecords. */
 export interface WorkspaceStopRecord {
   /** Latch releases (one per overlapping cascade) run together once release conditions hold. */
@@ -1448,68 +1432,6 @@ export type SettlementReceiptDecision =
   | { kind: "unreadable" }
   | { kind: "write"; attemptId: string; parentWorkspaceId: string; owners: string[] };
 
-/** Lineage proof (evaluateAttemptLineage) from the predecessor's receipt in the parent's dir. */
-function lineageProofFromReceipt(
-  receipt: SubagentAttemptSettlementReceiptReadResult,
-  parentWorkspaceId: string
-): { proven: boolean; reason: string } {
-  if (receipt.kind === "found") {
-    // The strict reader binds task and attempt id to the path; the parent is bound here, to
-    // the entry: a receipt naming another parent at this path was copied or misplaced, never
-    // written by this task's owner (the same fail-closed outcome as an unreadable receipt).
-    if (receipt.receipt.parentWorkspaceId !== parentWorkspaceId) {
-      return {
-        proven: false,
-        reason: `settlement receipt names parent ${receipt.receipt.parentWorkspaceId}, expected ${parentWorkspaceId}`,
-      };
-    }
-    return { proven: true, reason: "settlement receipt found" };
-  }
-  return {
-    proven: false,
-    reason:
-      receipt.kind === "unreadable"
-        ? `settlement receipt unreadable: ${receipt.error}`
-        : "no settlement receipt for the predecessor attempt",
-  };
-}
-
-/**
- * Row step of readUnownedSettlementProof: the row names an interrupted, proven attempt with a
- * parent, so the parent's receipt copy can prove it; else why it cannot.
- */
-function unownedSettlementCandidate(
-  row: WorkspaceConfigEntry | undefined
-):
-  | { kind: "candidate"; attemptId: string; parentWorkspaceId: string }
-  | { kind: "unproven"; reason: string } {
-  const unproven = (reason: string) => ({ kind: "unproven" as const, reason });
-  if (row == null) return unproven("no task record");
-  const attemptId = row.taskAttemptId;
-  if (!isTaskAttemptId(attemptId)) return unproven("no attempt id (pre-identity entry)");
-  if (row.taskStatus !== "interrupted") return unproven(`status ${row.taskStatus ?? "missing"}`);
-  if (row.taskAttemptUnproven === true) return unproven("lineage marked unproven");
-  const parentWorkspaceId = coerceNonEmptyString(row.parentWorkspaceId);
-  if (parentWorkspaceId == null) return unproven("no parent");
-  return { kind: "candidate", attemptId, parentWorkspaceId };
-}
-
-/** Receipt step of readUnownedSettlementProof: why the read receipt proves nothing, if it does not. */
-function unownedSettlementReceiptGap(
-  receipt: SubagentAttemptSettlementReceiptReadResult,
-  attemptId: string,
-  parentWorkspaceId: string
-): string | undefined {
-  if (receipt.kind === "not_found") return `no settlement receipt for ${attemptId}`;
-  if (receipt.kind === "unreadable") {
-    return `settlement receipt unreadable: ${receipt.error}`;
-  }
-  if (receipt.receipt.parentWorkspaceId !== parentWorkspaceId) {
-    return `settlement receipt names parent ${receipt.receipt.parentWorkspaceId}`;
-  }
-  return undefined;
-}
-
 /**
  * Task-pinned send options for an automatic TaskService turn (required-report prompt today;
  * child goal turns later). A whitelist on purpose: per-send payload (attachments, muxMetadata,
@@ -1550,38 +1472,6 @@ function markedTerminalFailure(
   };
 }
 
-/**
- * Why a workflow claim on `attemptId` must be refused (see TaskService.claimRetiredAttempt), or
- * undefined when it may be granted (or re-stamped, for the same run/step before any replacement).
- */
-function retiredAttemptClaimRefusal(
-  row: WorkspaceConfigEntry,
-  attemptId: string,
-  claimant: { runId: string; stepId: string }
-): string | undefined {
-  if (row.taskAttemptId !== attemptId) {
-    return `claim lost: the task now names attempt ${row.taskAttemptId ?? "none"}`;
-  }
-  if (row.taskStatus !== "interrupted") return `attempt is ${row.taskStatus ?? "running"}`;
-  const task = row.workflowTask;
-  if (task?.runId !== claimant.runId || task.stepId !== claimant.stepId) {
-    return "the task does not belong to this workflow step";
-  }
-  const existing = row.taskAttemptRetiredBy;
-  if (existing == null) return undefined;
-  if (existing.replacementTaskId != null) {
-    return `the retired attempt already has replacement ${existing.replacementTaskId}`;
-  }
-  if (
-    existing.runId !== claimant.runId ||
-    existing.stepId !== claimant.stepId ||
-    existing.attemptId !== attemptId
-  ) {
-    return "the attempt was retired by another workflow step";
-  }
-  return undefined;
-}
-
 /**
  * The attempt a settlement/finalization write acts for (see rowSupersedes). A REQUIRED parameter
  * on every helper reachable from stream end, stream error, report publication and interrupted
@@ -1596,7 +1486,7 @@ type AttemptFence = string | null;
  * this process's owned attempt, else the attempt the stream's turn was admitted under (or the
  * persisted id). Both undefined = captured without an identity (a pre-identity row).
  */
-interface StreamAttemptOrigin {
+export interface StreamAttemptOrigin {
   ownedAttempt: OwnedTaskAttempt | undefined;
   unownedAttemptId: string | undefined;
 }
@@ -2031,18 +1921,6 @@ export class TaskService implements AgentTaskIntegration {
       attemptId: AttemptFence;
     }
   >();
-  /** Outstanding send obligations per task (see AdmittedSend); discharged entries are removed. */
-  private readonly admittedSendsByTaskId = new Map<string, Set<AdmittedSend>>();
-  /**
-   * Reawakenings (parent reactivation or manual resume) that published a fresh attempt and have
-   * not yet handed it a send obligation (reactivation: until createWorkspaceTurn returns; manual
-   * resume: until reawakenInterruptedTask returns to the send that admits it; a reactivation in
-   * the remaining hand-off tick only makes that send, which names its attempt, read stale). With
-   * pending admissions and the live turn it is the "another reawakening is live" evidence each
-   * path rechecks under the mutex (see hasLiveReawakening): neither path holds the other's locks,
-   * and rotating an attempt a live continuation runs under drops that continuation's report.
-   */
-  private readonly reawakeningsInFlight = new Map<string, string>();
   /** Per task, the stream-end decisions of this process's owned attempts (see StreamEndDecision). */
   private readonly streamEndDecisionsByTaskId = new Map<string, StreamEndDecision[]>();
   /** Tracks consecutive auto-resumes per workspace. Reset when a user message is sent. */
@@ -2403,9 +2281,9 @@ export class TaskService implements AgentTaskIntegration {
    * fail-closed shape (cleanup-pending), never a silent discharge.
    */
   private recordWorkspaceTurnSettled(workspaceId: string, turnGeneration: symbol): void {
-    for (const send of this.admittedSendsByTaskId.get(workspaceId) ?? []) {
+    for (const send of this.attemptLedger.sendsOf(workspaceId)) {
       if (send.state === "admitted" && send.turnId === turnGeneration) {
-        this.dischargeAdmittedSend(send);
+        this.attemptLedger.dischargeAdmittedSend(send);
       }
     }
     // A terminal decision held for this turn's late readers: only actual readers keep it now.
@@ -2428,7 +2306,7 @@ export class TaskService implements AgentTaskIntegration {
    */
   private recordWorkspaceTurnSuperseded(workspaceId: string, previous: symbol, next: symbol): void {
     if (previous === next) return;
-    for (const send of this.admittedSendsByTaskId.get(workspaceId) ?? []) {
+    for (const send of this.attemptLedger.sendsOf(workspaceId)) {
       if (send.state === "admitted" && send.turnId === previous) {
         send.turnId = next;
         send.inherited = true;
@@ -2546,7 +2424,7 @@ export class TaskService implements AgentTaskIntegration {
     const listener = () => {
       if (
         attemptId == null ||
-        !this.attemptAdmissionOpen(taskId, attemptId) ||
+        !this.attemptLedger.attemptAdmissionOpen(taskId, attemptId) ||
         taskOrigin.stopEpoch !== this.getWorkspaceStopEpoch(taskId) ||
         taskOrigin.executionId !== this.getAgentTaskExecutionId(taskId) ||
         // arbitrateChildGoalAtStreamEnd's status gate (its owed-pause gate needs no mirror: the
@@ -2677,7 +2555,7 @@ export class TaskService implements AgentTaskIntegration {
     const decisions = this.streamEndDecisionsByTaskId.get(taskId);
     if (decisions == null) return;
     const readers = new Set<string>();
-    for (const send of this.admittedSendsByTaskId.get(taskId) ?? []) {
+    for (const send of this.attemptLedger.sendsOf(taskId)) {
       if (send.state === "pending" || send.state === "enqueued") readers.add(send.attemptId);
     }
     const kept = decisions.filter(
@@ -2690,27 +2568,6 @@ export class TaskService implements AgentTaskIntegration {
     else if (kept.length !== decisions.length) this.streamEndDecisionsByTaskId.set(taskId, kept);
   }
 
-  /**
-   * Dequeue-gate consultation for an enqueued obligation (see TurnAdmissionToken.resolveDispatch):
-   * held while its attempt's stream-end is undecided, refused once that stream turned out to be
-   * the terminal report (or its outcome is unknowable), otherwise on to the ordinary gates.
-   */
-  private resolveEnqueuedSendDispatch(send: AdmittedSend): QueuedDispatchDecision {
-    if (send.state !== "enqueued") return "proceed";
-    const decision = this.findStreamEndDecision(send.taskId, send.attemptId);
-    switch (decision?.outcome) {
-      case "pending":
-        return "hold";
-      case "published":
-        return { refuse: TASK_REPORTED_QUEUED_SEND_UNSENT_MESSAGE };
-      case "indeterminate":
-        return { refuse: TASK_REPORT_OUTCOME_INDETERMINATE_UNSENT_MESSAGE };
-      case "nonreport":
-      case undefined:
-        return "proceed";
-    }
-  }
-
   /** Tests call this; AttemptLedger owns the listeners. */
   private notifyAttemptSettlementListeners(taskId: string): void {
     this.attemptLedger.notifyAttemptSettlementListeners(taskId);
@@ -2736,19 +2593,6 @@ export class TaskService implements AgentTaskIntegration {
     return isTaskAttemptId(persisted) ? persisted : undefined;
   }
 
-  /**
-   * Common send-authorization predicate (no ownership): the id is still the task's current
-   * attempt, nothing closed it, no stop cascade holds the workspace and no workflow claim retired
-   * it. Ownership is a separate question, checked only where it applies.
-   */
-  private attemptAdmissionOpen(taskId: string, attemptId: string): boolean {
-    const entry = findWorkspaceEntry(this.config.loadConfigOrDefault(), taskId)?.workspace;
-    if (this.currentTaskAttemptId(taskId, entry) !== attemptId) return false;
-    if (this.attemptLedger.isAttemptClosed(taskId, attemptId)) return false;
-    if (this.isWorkspaceStopInProgress(taskId)) return false;
-    return entry?.taskAttemptRetiredBy == null && entry?.pendingRemoval == null;
-  }
-
   /**
    * The attempt id a stream event belongs to, read in the event's own tick (see
    * resolveStreamAttemptAtEvent).
@@ -2758,75 +2602,9 @@ export class TaskService implements AgentTaskIntegration {
     return origin.ownedAttempt != null ? origin.ownedAttempt.attemptId : origin.unownedAttemptId;
   }
 
-  /**
-   * Who a stream event belongs to, read in the event's own tick. The live turn's ADMITTED attempt
-   * decides first: local ownership counts only when it names that same attempt. A process can
-   * still hold a stale owned attempt A after another backend rotated the row to B and a local send
-   * was admitted for B (unowned) — preferring the owner would treat B's stream as A's and discard
-   * its report/recovery as superseded. Without an admitted obligation (legacy turns) the owner,
-   * else the persisted id, applies as before.
-   */
+  /** Tests and the stream listeners call this (see AttemptLedger.resolveStreamAttemptAtEvent). */
   private resolveStreamAttemptAtEvent(taskId: string): StreamAttemptOrigin {
-    const owned = this.attemptLedger.ownedAttempt(taskId);
-    const turn = this.workspaceService.getActiveTurnGeneration(taskId);
-    // The turn's DIRECT admission decides; an obligation rebound from a superseded predecessor
-    // (inherited) shares the turn id but not its attempt, and counts only when the turn has no
-    // direct admission (a successor started without its own obligation).
-    let admitted: string | undefined;
-    let inherited: string | undefined;
-    if (turn != null) {
-      for (const send of this.admittedSendsByTaskId.get(taskId) ?? []) {
-        if (send.state !== "admitted" || send.turnId !== turn) continue;
-        if (send.inherited !== true) {
-          admitted = send.attemptId;
-          break;
-        }
-        inherited ??= send.attemptId;
-      }
-    }
-    admitted ??= inherited;
-    if (owned != null && (admitted == null || owned.attemptId === admitted)) {
-      return { ownedAttempt: owned, unownedAttemptId: undefined };
-    }
-    return {
-      ownedAttempt: undefined,
-      unownedAttemptId: admitted ?? this.currentTaskAttemptId(taskId),
-    };
-  }
-
-  /** Sends admitted against the task's current attempt that have not claimed a turn yet. */
-  private hasPendingAdmissions(taskId: string): boolean {
-    const current = this.currentTaskAttemptId(taskId);
-    for (const send of this.admittedSendsByTaskId.get(taskId) ?? []) {
-      if (send.attemptId !== current) continue;
-      if (send.state === "pending" || send.state === "enqueued") return true;
-    }
-    return false;
-  }
-
-  /**
-   * Another reawakening's attempt is live in this process: published and still launching, or
-   * handed to work that has not ended (a pending/enqueued admission, the live turn, a stream).
-   * Synchronous; both reawakening paths evaluate it under the mutex BEFORE their own publish, so
-   * their own work is never counted. Evidence, not ownership: an owned attempt that nothing runs
-   * (a reactivation whose launch was refused, a follow-up turn that ended) stays reawakenable.
-   */
-  private hasLiveReawakening(taskId: string): boolean {
-    if (
-      this.reawakeningsInFlight.has(taskId) ||
-      this.workspaceService.getActiveTurnGeneration(taskId) != null ||
-      this.aiService.isStreaming(taskId)
-    ) {
-      return true;
-    }
-    // Only sends that can still start a turn: one bound to a closed, superseded or stopped
-    // attempt reads stale at every later gate and is no continuation to protect.
-    for (const send of this.admittedSendsByTaskId.get(taskId) ?? []) {
-      if ((send.state === "pending" || send.state === "enqueued") && !send.token.admissionStale()) {
-        return true;
-      }
-    }
-    return false;
+    return this.attemptLedger.resolveStreamAttemptAtEvent(taskId);
   }
 
   /**
@@ -2858,291 +2636,20 @@ export class TaskService implements AgentTaskIntegration {
     return false;
   }
 
-  private dischargeAdmittedSend(send: AdmittedSend): void {
-    if (send.state === "discharged") return;
-    send.state = "discharged";
-    const sends = this.admittedSendsByTaskId.get(send.taskId);
-    sends?.delete(send);
-    if (sends?.size === 0) this.admittedSendsByTaskId.delete(send.taskId);
-    for (const record of send.capturedBy) {
-      record.pendingAdmissions.delete(send);
-    }
-    send.capturedBy.clear();
-    this.pruneStreamEndDecisions(send.taskId);
-  }
-
-  /**
-   * Register one send obligation against `attemptId` and return its token. Every lifecycle event
-   * updates the obligation object and the records that captured it by reference — never
-   * whichever record currently occupies the task's slot.
-   */
-  private createAdmittedSend(
-    taskId: string,
-    attemptId: string,
-    attempt: OwnedTaskAttempt | undefined
-  ): AdmittedSend {
-    assertTaskAttemptId(attemptId, "createAdmittedSend");
-    const send: AdmittedSend = {
-      taskId,
-      attemptId,
-      ...(attempt != null ? { attempt } : {}),
-      state: "pending",
-      capturedBy: new Set(),
-      token: {
-        resolveDispatch: () => this.resolveEnqueuedSendDispatch(send),
-        admissionStale: () => this.isAdmittedSendStale(send),
-        onEnqueued: () => {
-          if (send.state === "pending") send.state = "enqueued";
-        },
-        onAdmitted: (turnId) => this.recordAdmittedSendTurn(send, turnId),
-        onDisposed: (kind) => this.disposeSendBeforeAdmission(send, kind),
-      },
-    };
-    const sends = this.admittedSendsByTaskId.get(taskId) ?? new Set<AdmittedSend>();
-    sends.add(send);
-    this.admittedSendsByTaskId.set(taskId, sends);
-    return send;
-  }
-
-  /** TurnAdmissionToken.admissionStale for `send` (see createAdmittedSend). */
-  private isAdmittedSendStale(send: AdmittedSend): boolean {
-    const { taskId, attemptId } = send;
-    // Admitted obligations belong to their turn: rotation, closure and stops no longer
-    // concern the send itself (the turn is captured by the stop record instead).
-    if (send.state === "admitted") return false;
-    if (send.state === "discharged") return true;
-    if (!this.attemptAdmissionOpen(taskId, attemptId)) return true;
-    if (send.attempt != null && this.attemptLedger.ownedAttempt(taskId) !== send.attempt) {
-      return true;
-    }
-    // A send still in its preflight while the attempt's last stream is undecided, or was
-    // decided a report (or unknowable): it must not start a turn under that attempt. The
-    // queue's own gate handles enqueued entries (hold/refuse) before reaching here.
-    if (send.state === "pending") {
-      return this.streamEndBlocksNewTurn(taskId, attemptId);
-    }
-    return false;
-  }
-
-  /** The attempt's last stream is undecided, or was decided a report (or unknowable). */
-  private streamEndBlocksNewTurn(taskId: string, attemptId: string): boolean {
-    const decision = this.findStreamEndDecision(taskId, attemptId);
-    return decision != null && decision.outcome !== "nonreport";
-  }
-
-  /** TurnAdmissionToken.onAdmitted for `send`: bind it to `turnId` (see createAdmittedSend). */
-  private recordAdmittedSendTurn(send: AdmittedSend, turnId: symbol): void {
-    const { taskId, attemptId } = send;
-    if (send.state === "admitted") {
-      // The dequeued entry is adopted by a second sendMessage naming the same turn.
-      if (send.turnId !== turnId) {
-        log.warn("[task-attempt] admitted send re-admitted under another turn; keeping first", {
-          taskId,
-          attemptId,
-        });
-      }
-      return;
-    }
-    if (send.state === "discharged") {
-      log.warn("[task-attempt] disposed send reported admitted; ignoring", {
-        taskId,
-        attemptId,
-      });
-      return;
-    }
-    send.state = "admitted";
-    send.turnId = turnId;
-    // A record that captured this obligation while pending now waits on the turn instead,
-    // and owes one more stopStream so the late turn is actually stopped: Phase B's single
-    // stopStream targeted the execution captured at Phase A, not this one.
-    for (const record of send.capturedBy) {
-      record.pendingAdmissions.delete(send);
-      record.capturedTurns.add(turnId);
-      record.cleanupInFlight += 1;
-      void this.aiService
-        .stopStream(taskId, { abandonPartial: false })
-        .catch((error: unknown) => {
-          log.debug("[task-attempt] late-turn stopStream threw", { taskId, error });
-        })
-        .finally(() => {
-          record.cleanupInFlight -= 1;
-          this.taskStopLatch.recheckWorkspaceStopRelease(taskId);
-        });
-    }
-    send.capturedBy.clear();
-  }
-
-  /** TurnAdmissionToken.onDisposed for `send`: no turn will exist (see createAdmittedSend). */
-  private disposeSendBeforeAdmission(
-    send: AdmittedSend,
-    kind: Parameters<TurnAdmissionToken["onDisposed"]>[0]
-  ): void {
-    const { taskId, attemptId } = send;
-    if (send.state === "admitted" || send.state === "discharged") return;
-    log.debug("[task-attempt] send obligation disposed before admission", {
-      taskId,
-      attemptId,
-      kind,
-    });
-    const records = [...send.capturedBy];
-    this.dischargeAdmittedSend(send);
-    for (const record of records) {
-      if (this.taskStopLatch.isCurrentStopRecord(taskId, record)) {
-        this.taskStopLatch.recheckWorkspaceStopRelease(taskId);
-      }
-    }
-    this.attemptLedger.recheckResumeFailureSettlement(taskId);
-  }
-
-  /**
-   * Admission fence for a send into a workspace (AgentTaskIntegration.admitTaskWorkspaceTurn),
-   * evaluated synchronously at the session handoff. Not a task, or a pre-identity task entry →
-   * no obligation (the next admission stamps an id). A task whose current attempt is retired,
-   * stopping, or closed refuses; otherwise the send is bound to the current attempt — owned when
-   * this process owns exactly that id, unowned otherwise (startup re-drives, prior-process
-   * attempts). Ownership is never acquired here: a send authorizes work, not receipt authority.
-   */
+  /** AgentTaskIntegration (see AttemptLedger.admitTaskWorkspaceTurn). */
   admitTaskWorkspaceTurn(
     workspaceId: string,
     options: { acceptanceOrigin: TurnAcceptanceOrigin; expectedAttemptId?: string }
   ): TaskTurnAdmission {
-    assert(workspaceId.length > 0, "admitTaskWorkspaceTurn: workspaceId must be non-empty");
-    let entry: WorkspaceConfigEntry | undefined;
-    try {
-      entry = findWorkspaceEntry(
-        this.config.loadConfigOrDefault({ throwOnError: true }),
-        workspaceId
-      )?.workspace;
-    } catch (error: unknown) {
-      // Unreadable registry: the classification itself is indeterminate. Absence from this
-      // process's identity map (currentAttemptIdByTaskId) is no proof of "not a task" — a task
-      // loaded from disk and never rotated here has no entry — so EVERY send refuses here (fail
-      // closed) rather than bypassing the attempt fence. That deliberately includes root/non-task
-      // workspaces: without the registry nothing can tell them apart, and such a send could not
-      // run anyway (lenient readers then see no workspaces, so its metadata cannot be resolved);
-      // refusing here just fails it before anything is persisted. Sends resume once the registry
-      // reads again.
-      return {
-        kind: "refused",
-        message: `Workspace registry unreadable; send refused: ${getErrorMessage(error)}`,
-      };
-    }
-    // A send decided for a specific attempt whose row is gone (unpublished, removed) is stale,
-    // never an ordinary send into a workspace that is not a task.
-    if (entry == null && options.expectedAttemptId != null) {
-      return { kind: "refused", message: SEND_ADMISSION_STALE_MESSAGE };
-    }
-    // Checked before the task classification: another backend's removal of a root workspace
-    // deletes its checkout too (#4478).
-    if (entry?.pendingRemoval != null) {
-      return { kind: "refused", message: pendingRemovalAdmissionMessage(entry.pendingRemoval) };
-    }
-    if (!entry?.parentWorkspaceId) return { kind: "not-a-task" };
-    // The claim is monotonic and stronger than the id: a retired task refuses even when its id
-    // is missing or malformed (fail closed on partial state), so check it before the id.
-    if (entry.taskAttemptRetiredBy != null) {
-      return { kind: "refused", message: retiredAttemptMessage(entry.taskAttemptRetiredBy) };
-    }
-    return this.admitSendToCurrentAttempt(workspaceId, entry, options);
-  }
-
-  /**
-   * Attempt step of admitTaskWorkspaceTurn for a task entry that is not retired: refuse a send
-   * the current attempt cannot take, else bind it to that attempt.
-   */
-  private admitSendToCurrentAttempt(
-    workspaceId: string,
-    entry: WorkspaceConfigEntry,
-    options: { acceptanceOrigin: TurnAcceptanceOrigin; expectedAttemptId?: string }
-  ): TaskTurnAdmission {
-    const attemptId = this.currentTaskAttemptId(workspaceId, entry);
-    if (attemptId == null) {
-      log.debug("[task-attempt] send into a pre-identity task entry carries no obligation", {
-        workspaceId,
-      });
-      return { kind: "not-a-task" };
-    }
-    // A send decided for one attempt never binds to a successor another writer admitted since:
-    // the decision is stale, not the row (see TaskWorkspaceSeam.admitTaskWorkspaceTurn).
-    if (options.expectedAttemptId != null && attemptId !== options.expectedAttemptId) {
-      log.info("[task-attempt] send refused: decided for a superseded attempt", {
-        workspaceId,
-        expectedAttemptId: options.expectedAttemptId,
-        attemptId,
-        acceptanceOrigin: options.acceptanceOrigin,
-      });
-      return { kind: "refused", message: SEND_ADMISSION_STALE_MESSAGE };
-    }
-    if (this.isWorkspaceStopInProgress(workspaceId)) {
-      return { kind: "refused", message: WORKSPACE_STOP_IN_PROGRESS_SEND_BLOCKED_MESSAGE };
-    }
-    if (this.attemptLedger.isAttemptClosed(workspaceId, attemptId)) {
-      log.info("[task-attempt] send refused: attempt settled", {
-        workspaceId,
-        attemptId,
-        acceptanceOrigin: options.acceptanceOrigin,
-      });
-      return { kind: "refused", message: TASK_ATTEMPT_SETTLED_SEND_BLOCKED_MESSAGE };
-    }
-    const owned = this.attemptLedger.ownedAttempt(workspaceId);
-    const send = this.createAdmittedSend(
-      workspaceId,
-      attemptId,
-      owned?.attemptId === attemptId ? owned : undefined
-    );
-    return { kind: "admitted", token: send.token };
+    return this.attemptLedger.admitTaskWorkspaceTurn(workspaceId, options);
   }
 
-  /**
-   * Lineage proof for a reawaken/reactivation of `entry`'s current attempt (P11): proven iff the
-   * predecessor is settled by this process's owner (recorded `settled`) or by a durable receipt.
-   * Anything else — no id, marked lineage, unsettled/closing owner, missing or unreadable receipt —
-   * is unproven and makes the successor ineligible for receipts. Never throws.
-   */
-  private async evaluateAttemptLineage(
+  /** Tests spy on this (see AttemptLedger.evaluateAttemptLineage). */
+  private evaluateAttemptLineage(
     taskId: string,
     entry: WorkspaceConfigEntry
   ): Promise<{ proven: boolean; reason: string }> {
-    const attemptId = entry.taskAttemptId;
-    if (!isTaskAttemptId(attemptId)) {
-      return { proven: false, reason: "predecessor has no attempt id (pre-upgrade entry)" };
-    }
-    if (entry.taskAttemptUnproven === true) {
-      return { proven: false, reason: "lineage marked unproven" };
-    }
-    const owned = this.attemptLedger.ownedAttempt(taskId);
-    if (owned?.attemptId === attemptId) {
-      if (!owned.receiptEligible) {
-        return { proven: false, reason: "predecessor owned but not receipt-eligible" };
-      }
-      if (this.ownedAttemptSettlementPhase(taskId, owned) === "closing") {
-        // The producer's awaited write is in flight; wait for its outcome (bounded, no lock).
-        await this.attemptLedger.waitForAttemptClosureToSettle(
-          taskId,
-          ATTEMPT_CLOSURE_SETTLE_WAIT_MS
-        );
-      }
-      if (this.ownedAttemptSettlementPhase(taskId, owned) === "settled") {
-        return { proven: true, reason: "predecessor settled by this process" };
-      }
-    }
-    const parentWorkspaceId = coerceNonEmptyString(entry.parentWorkspaceId);
-    if (parentWorkspaceId == null) return { proven: false, reason: "entry has no parent" };
-    const receipt = await readSubagentAttemptSettlementReceiptStrict(
-      path.join(this.config.sessionsDir, parentWorkspaceId),
-      taskId,
-      attemptId
-    );
-    return lineageProofFromReceipt(receipt, parentWorkspaceId);
-  }
-
-  /** Phase of this process's settlement entry for `owned`; undefined when it records another. */
-  private ownedAttemptSettlementPhase(
-    taskId: string,
-    owned: OwnedTaskAttempt
-  ): AttemptSettlementEntry["phase"] | undefined {
-    const settlement = this.attemptLedger.attemptSettlement(taskId);
-    return settlement?.attempt === owned ? settlement.phase : undefined;
+    return this.attemptLedger.evaluateAttemptLineage(taskId, entry);
   }
 
   /** Tests spy on this; TaskStopLatch owns the wait. */
@@ -3272,203 +2779,21 @@ export class TaskService implements AgentTaskIntegration {
     );
   }
 
-  /**
-   * Cross-process evidence (G2) that the task's CURRENT attempt ended without a report, for an
-   * attempt this process does not own (another backend, or this one before a restart). Called
-   * only after the report artifact read positively found none: a receipt never outranks a report.
-   *
-   * Attempt identity: the receipt must name the row's current taskAttemptId. The workflow journal
-   * names only the child task, and every task admission (launch, reawaken, re-drive) rotates the
-   * id, so the current id is the only attempt that can still run; a receipt for an older attempt
-   * says nothing about it. A plain send binds to the current id instead: another backend's turn
-   * can therefore run under an attempt this receipt settled (#4545) — its late report still wins
-   * unless a claim retired the attempt first (publishAgentTaskReport). Only the PARENT's receipt copy counts (the copy lineage reads; ancestor copies are fan-out
-   * and never evidence on their own), and it must name that parent. Every read is strict, and
-   * the row is re-read after the receipt so an admission landing in between is never missed.
-   */
-  private async readUnownedSettlementProof(
-    taskId: string
-  ): Promise<{ kind: "proven"; attemptId: string } | { kind: "unproven"; reason: string }> {
-    const unproven = (reason: string) => ({ kind: "unproven" as const, reason });
-    const readRow = () =>
-      findWorkspaceEntry(this.config.loadConfigOrDefault({ throwOnError: true }), taskId)
-        ?.workspace;
-    let row: WorkspaceConfigEntry | undefined;
-    try {
-      row = readRow();
-    } catch (error: unknown) {
-      return unproven(`config unreadable: ${getErrorMessage(error)}`);
-    }
-    const candidate = unownedSettlementCandidate(row);
-    if (candidate.kind === "unproven") return candidate;
-    const { attemptId, parentWorkspaceId } = candidate;
-    const receipt = await readSubagentAttemptSettlementReceiptStrict(
-      path.join(this.config.sessionsDir, parentWorkspaceId),
-      taskId,
-      attemptId
-    );
-    const receiptGap = unownedSettlementReceiptGap(receipt, attemptId, parentWorkspaceId);
-    if (receiptGap != null) return unproven(receiptGap);
-    let latest: WorkspaceConfigEntry | undefined;
-    try {
-      latest = readRow();
-    } catch (error: unknown) {
-      return unproven(`config unreadable: ${getErrorMessage(error)}`);
-    }
-    if (
-      latest?.taskAttemptId !== attemptId ||
-      latest.taskStatus !== "interrupted" ||
-      latest.taskAttemptUnproven === true
-    ) {
-      return unproven("the attempt changed while its receipt was read");
-    }
-    return { kind: "proven", attemptId };
-  }
-
-  /**
-   * Workflow claim that retires `attemptId` of `taskId` so exactly one replacement can be
-   * published for it (G2). A pure compare-and-swap fence: the classifier (readAttemptOutcome)
-   * decided the attempt ended without a report; the claim only makes that decision exclusive.
-   * Granted while the row still names the attempt, is interrupted, belongs to this run/step and
-   * has no replacement yet; an earlier claim of the same run/step is re-stamped with a fresh
-   * nonce (a stale runner holding the old nonce can then no longer publish). Every admission
-   * already refuses a retired attempt. Confirmed by a strict re-read, since a config edit that
-   * fails to write is not always surfaced to its caller.
-   *
-   * A spent claim is never re-granted, even when its replacementTaskId names a row that is gone
-   * (a replacement whose launch failed sanitization is unpublished). That is safe because the
-   * workflow checkpoints the replacement (createMany's onTaskReserved) BEFORE the publishing
-   * commit that sets replacementTaskId: once it is set, the journal already names the
-   * replacement, so no runner classifies this attempt again. Recovery is retry-from-checkpoint's
-   * fresh reservation (WorkflowRunner.replacementRace.test.ts).
-   */
-  async claimRetiredAttempt(
+  /** Workflow adapter (see AttemptLedger.claimRetiredAttempt). */
+  claimRetiredAttempt(
     taskId: string,
     attemptId: string,
     claimant: { runId: string; stepId: string; inputHash: string }
   ): Promise<Result<{ nonce: string }, string>> {
-    assert(taskId.length > 0, "claimRetiredAttempt: taskId must be non-empty");
-    assertTaskAttemptId(attemptId, "claimRetiredAttempt");
-    assert(
-      claimant.runId.length > 0 && claimant.stepId.length > 0,
-      "claimRetiredAttempt: claimant"
-    );
-    return await this.workspaceEventLocks.withLock(taskId, async () => {
-      if (
-        this.isWorkspaceStopInProgress(taskId) ||
-        this.getWorkspaceTurnManager().getLiveWorkspaceTurnRegistration(taskId) != null ||
-        this.aiService.isStreaming(taskId) ||
-        this.workspaceService.getActiveTurnGeneration(taskId) != null
-      ) {
-        return Err("attempt is live");
-      }
-      const nonce = randomUUID();
-      let refusal: string | undefined = "task record not found";
-      try {
-        await this.editWorkspaceEntry(
-          taskId,
-          (ws) => {
-            refusal = retiredAttemptClaimRefusal(ws, attemptId, claimant);
-            if (refusal != null) return;
-            markChildGoalPauseOwed(ws);
-            ws.taskAttemptRetiredBy = {
-              runId: claimant.runId,
-              stepId: claimant.stepId,
-              inputHash: claimant.inputHash,
-              childTaskId: taskId,
-              attemptId,
-              mode: "no-report",
-              at: new Date().toISOString(),
-              nonce,
-            };
-          },
-          { allowMissing: true }
-        );
-      } catch (error: unknown) {
-        return Err(`claim write failed: ${getErrorMessage(error)}`);
-      }
-      if (refusal != null) return Err(refusal);
-      let row: WorkspaceConfigEntry | undefined;
-      try {
-        row = findWorkspaceEntry(
-          this.config.loadConfigOrDefault({ throwOnError: true }),
-          taskId
-        )?.workspace;
-      } catch (error: unknown) {
-        return Err(`claim unconfirmed: config unreadable: ${getErrorMessage(error)}`);
-      }
-      if (row?.taskAttemptId !== attemptId || row.taskAttemptRetiredBy?.nonce !== nonce) {
-        return Err("claim lost");
-      }
-      await this.settleChildGoalPause(taskId);
-      return Ok({ nonce });
-    });
+    return this.attemptLedger.claimRetiredAttempt(taskId, attemptId, claimant);
   }
 
-  /**
-   * A workflow step's started checkpoint is written in createMany's onTaskReserved, BEFORE
-   * commitReservations publishes the child's row. A crash in between leaves a checkpoint naming a
-   * task that never existed. Replacing it outright is unsafe: the reserving backend may only be
-   * stalled (lease expired, process alive), and its late commit would then publish a second child
-   * for the step, or re-run a step the replacement already recorded. So the replacing runner first
-   * tombstones the ID on the parent's row; commitReservations checks the tombstone inside its own
-   * config write and refuses, which makes "replace" and "late commit" mutually exclusive.
-   *
-   * Granted only while a strict read shows no row for the task (checked inside the same write) and
-   * this process owns no attempt for it (an own reservation may still be before its commit).
-   * Idempotent. Confirmed by a strict re-read, since a config edit that fails to write is not
-   * always surfaced to its caller.
-   */
-  async tombstoneUnpublishedReservation(
+  /** Workflow adapter (see AttemptLedger.tombstoneUnpublishedReservation). */
+  tombstoneUnpublishedReservation(
     parentWorkspaceId: string,
     taskId: string
   ): Promise<Result<void, string>> {
-    assert(parentWorkspaceId.length > 0, "tombstoneUnpublishedReservation: parentWorkspaceId");
-    assert(taskId.length > 0, "tombstoneUnpublishedReservation: taskId");
-    return await this.workspaceEventLocks.withLock(taskId, async () => {
-      if (this.attemptLedger.hasOwnedAttempt(taskId)) {
-        return Err("this process owns an attempt for the task");
-      }
-      let refusal: string | undefined;
-      try {
-        await this.config.editConfig((config) => {
-          refusal = undefined;
-          if (findWorkspaceEntry(config, taskId) != null) {
-            refusal = "the task was published";
-            return config;
-          }
-          const parent = findWorkspaceEntry(config, parentWorkspaceId)?.workspace;
-          if (parent == null) {
-            refusal = "parent record not found";
-            return config;
-          }
-          const tombstones = parent.taskReservationTombstones ?? [];
-          if (!tombstones.includes(taskId)) {
-            parent.taskReservationTombstones = [...tombstones, taskId];
-          }
-          return config;
-        });
-      } catch (error: unknown) {
-        return Err(`tombstone write failed: ${getErrorMessage(error)}`);
-      }
-      if (refusal != null) return Err(refusal);
-      let config: ProjectsConfig;
-      try {
-        config = this.config.loadConfigOrDefault({ throwOnError: true });
-      } catch (error: unknown) {
-        return Err(`tombstone unconfirmed: config unreadable: ${getErrorMessage(error)}`);
-      }
-      if (
-        findWorkspaceEntry(config, taskId) != null ||
-        findWorkspaceEntry(
-          config,
-          parentWorkspaceId
-        )?.workspace.taskReservationTombstones?.includes(taskId) !== true
-      ) {
-        return Err("tombstone lost");
-      }
-      return Ok(undefined);
-    });
+    return this.attemptLedger.tombstoneUnpublishedReservation(parentWorkspaceId, taskId);
   }
 
   /**
@@ -3604,7 +2929,7 @@ export class TaskService implements AgentTaskIntegration {
       }
       // Report first: this branch runs only once the report artifact read positively found none.
       if (reportPositivelyAbsent) {
-        const proof = await this.readUnownedSettlementProof(taskId);
+        const proof = await this.attemptLedger.readUnownedSettlementProof(taskId);
         if (proof.kind === "proven") {
           return {
             kind: "terminal-no-report",
@@ -3844,8 +3169,17 @@ export class TaskService implements AgentTaskIntegration {
         this.listAncestorWorkspaceIdsUsingParentById(parentById, taskId),
       stopRecordAttemptId: (taskId) => this.taskStopLatch.stopRecordAttemptId(taskId),
       hasStopRecord: (taskId) => this.taskStopLatch.hasStopRecord(taskId),
-      admittedSends: (taskId) => this.admittedSendsByTaskId.get(taskId),
+      isWorkspaceStopInProgress: (workspaceId) => this.isWorkspaceStopInProgress(workspaceId),
+      isCurrentStopRecord: (workspaceId, record) =>
+        this.taskStopLatch.isCurrentStopRecord(workspaceId, record),
+      recheckWorkspaceStopRelease: (workspaceId) =>
+        this.taskStopLatch.recheckWorkspaceStopRelease(workspaceId),
+      currentTaskAttemptId: (taskId, entry) => this.currentTaskAttemptId(taskId, entry),
+      editWorkspaceEntry: (workspaceId, updater, options) =>
+        this.editWorkspaceEntry(workspaceId, updater, options),
+      settleChildGoalPause: (workspaceId) => this.settleChildGoalPause(workspaceId),
       findStreamEndDecision: (taskId, attemptId) => this.findStreamEndDecision(taskId, attemptId),
+      pruneStreamEndDecisions: (taskId) => this.pruneStreamEndDecisions(taskId),
       discardStreamEndDecisions: (taskId) => {
         this.streamEndDecisionsByTaskId.delete(taskId);
       },
@@ -3877,7 +3211,7 @@ export class TaskService implements AgentTaskIntegration {
         this.attemptLedger.removeAttemptSettlementListener(taskId, listener),
       ownedAttempt: (taskId) => this.attemptLedger.ownedAttempt(taskId),
       currentTaskAttemptId: (taskId) => this.currentTaskAttemptId(taskId),
-      admittedSends: (taskId) => this.admittedSendsByTaskId.get(taskId),
+      admittedSends: (taskId) => this.attemptLedger.sendsOf(taskId),
       ownsReceiptEligibleAttempt: (taskId, attempt) =>
         this.attemptLedger.ownsReceiptEligibleAttempt(taskId, attempt),
       findStreamEndDecision: (taskId, attemptId) => this.findStreamEndDecision(taskId, attemptId),
@@ -7828,7 +7162,7 @@ export class TaskService implements AgentTaskIntegration {
       // work is that writer's and must not be sampled as ours.
       const liveUnderLaunchAttempt = (): boolean => {
         const activeTurn = this.workspaceService.getActiveTurnGeneration(taskId);
-        for (const send of this.admittedSendsByTaskId.get(taskId) ?? []) {
+        for (const send of this.attemptLedger.sendsOf(taskId)) {
           if (send.attemptId !== attemptId) continue;
           if (send.state === "pending" || send.state === "enqueued") return true;
           if (send.state === "admitted" && send.turnId != null && send.turnId === activeTurn) {
@@ -7894,7 +7228,7 @@ export class TaskService implements AgentTaskIntegration {
         ? this.workspaceService.getActiveTurnGeneration(taskId) != null ||
           this.getWorkspaceTurnManager().getLiveWorkspaceTurnRegistration(taskId) != null ||
           this.aiService.isStreaming(taskId) ||
-          this.hasPendingAdmissions(taskId)
+          this.attemptLedger.hasPendingAdmissions(taskId)
         : liveUnderLaunchAttempt();
       if (rowStillOurs && liveExecution) {
         this.taskStopLatch.beginWorkspaceStop(taskId);
@@ -8476,7 +7810,7 @@ export class TaskService implements AgentTaskIntegration {
       // A manual resume (reawakenInterruptedTask, which holds neither the event nor the tree
       // lock) may have published and started its own attempt since the caller found the child
       // inactive: rotating it would supersede that live turn and drop its report.
-      if (this.hasLiveReawakening(taskId)) {
+      if (this.attemptLedger.hasLiveReawakening(taskId)) {
         return Err({
           code: "send_failed" as const,
           message: TASK_REAWAKEN_LOST_SEND_BLOCKED_MESSAGE,
@@ -8547,7 +7881,7 @@ export class TaskService implements AgentTaskIntegration {
         receiptEligible: committedProven,
       });
       // Live until createWorkspaceTurn returns; its send's admission is the evidence after that.
-      this.reawakeningsInFlight.set(taskId, reactivationAttemptId);
+      this.attemptLedger.beginReawakening(taskId, reactivationAttemptId);
     }
     let execution: Awaited<ReturnType<WorkspaceTurnManager["createWorkspaceTurn"]>>;
     try {
@@ -8576,9 +7910,7 @@ export class TaskService implements AgentTaskIntegration {
           : {}),
       });
     } finally {
-      if (this.reawakeningsInFlight.get(taskId) === reactivationAttemptId) {
-        this.reawakeningsInFlight.delete(taskId);
-      }
+      this.attemptLedger.endReawakening(taskId, reactivationAttemptId);
     }
     if (!execution.success) {
       // The fresh attempt is already published (config and memory) and stays: it reads as an
@@ -16112,7 +15444,7 @@ export class TaskService implements AgentTaskIntegration {
       // and runs, so the row alone reads resumable: rotating that live attempt would supersede the
       // continuation and drop its report. The user's send loses and
       // can retry; once the continuation streams, sends queue behind it instead of reaching here.
-      if (this.hasLiveReawakening(workspaceId)) {
+      if (this.attemptLedger.hasLiveReawakening(workspaceId)) {
         log.debug("markInterruptedTaskRunning refused: another reawakening is live", {
           workspaceId,
         });
@@ -16190,15 +15522,13 @@ export class TaskService implements AgentTaskIntegration {
         receiptEligible: committedProven,
       });
       // Live until the caller's send binds to it (its admission is the evidence after that).
-      this.reawakeningsInFlight.set(workspaceId, attemptId);
+      this.attemptLedger.beginReawakening(workspaceId, attemptId);
     }
 
     try {
       await this.emitWorkspaceMetadata(workspaceId);
     } finally {
-      if (this.reawakeningsInFlight.get(workspaceId) === attemptId) {
-        this.reawakeningsInFlight.delete(workspaceId);
-      }
+      this.attemptLedger.endReawakening(workspaceId, attemptId);
     }
     return { kind: "reawakened", attemptId, statusChanged: !resumeSettledReportedTask };
   }
@@ -17266,7 +16596,8 @@ export class TaskService implements AgentTaskIntegration {
     if (
       status !== "running" ||
       this.isChildGoalPauseOwed(workspaceId) ||
-      (streamAttemptId != null && !this.attemptAdmissionOpen(workspaceId, streamAttemptId))
+      (streamAttemptId != null &&
+        !this.attemptLedger.attemptAdmissionOpen(workspaceId, streamAttemptId))
     ) {
       return "none";
     }
@@ -17546,7 +16877,11 @@ export class TaskService implements AgentTaskIntegration {
       .withLock(workspaceId, async () => {
         const status = findWorkspaceEntry(this.config.loadConfigOrDefault(), workspaceId)?.workspace
           .taskStatus;
-        if (status !== "running" || !this.attemptAdmissionOpen(workspaceId, attemptId)) return;
+        if (
+          status !== "running" ||
+          !this.attemptLedger.attemptAdmissionOpen(workspaceId, attemptId)
+        )
+          return;
         if (
           this.aiService.isStreaming(workspaceId) ||
           this.workspaceService.hasPendingQueuedOrPreparingTurn(workspaceId)
@@ -17676,7 +17011,7 @@ export class TaskService implements AgentTaskIntegration {
     return (
       workspace?.taskStatus === "running" &&
       workspace.taskGoalPauseOwed == null &&
-      this.attemptAdmissionOpen(workspaceId, attemptId)
+      this.attemptLedger.attemptAdmissionOpen(workspaceId, attemptId)
     );
   }
 
@@ -17921,7 +17256,7 @@ export class TaskService implements AgentTaskIntegration {
         if (
           this.aiService.isStreaming(workspaceId) ||
           this.workspaceService.hasPendingQueuedOrPreparingTurn(workspaceId) ||
-          this.hasPendingAdmissions(workspaceId) ||
+          this.attemptLedger.hasPendingAdmissions(workspaceId) ||
           isActiveWorkspaceTurnTaskStatus(ws.taskExecutionStatus)
         ) {
           return;
@@ -18177,7 +17512,7 @@ export class TaskService implements AgentTaskIntegration {
       this.workspaceService.getActiveTurnGeneration(workspaceId) != null ||
       this.getWorkspaceTurnManager().getLiveWorkspaceTurnRegistration(workspaceId) != null ||
       this.aiService.isStreaming(workspaceId) ||
-      this.hasPendingAdmissions(workspaceId);
+      this.attemptLedger.hasPendingAdmissions(workspaceId);
     let stopRecord: WorkspaceStopRecord | undefined;
     if (expectedAttemptId == null && sampleLiveExecution()) {
       await using _lock = await this.mutex.acquire();