diff --git a/.github/workflows/check-all.yml b/.github/workflows/check-all.yml
index 422aa2e..6423825 100644
--- a/.github/workflows/check-all.yml
+++ b/.github/workflows/check-all.yml
@@ -7,7 +7,10 @@ jobs:
       - uses: actions/checkout@v4
       - uses: actions/setup-node@v4
         with:
-          node-version: "20"
+          # core/package.json requires >=22; the Pi SDK (0.80.10) needs
+          # >=22.19 and is ESM-only. Node 20 here meant CI never matched
+          # the engine the sidecar actually runs on.
+          node-version: "22"
       - name: Install project deps (CI parity with local node_modules)
         run: |
           for pj in $(find . -name node_modules -prune -o -name package.json -print); do
diff --git a/INVARIANTS.md b/INVARIANTS.md
index d6135ab..763e945 100644
--- a/INVARIANTS.md
+++ b/INVARIANTS.md
@@ -19,6 +19,7 @@ matters more than guard code — a wrong invariant that passes is worse than non
 | 11 | **extensions ≠ kernel** — bundled extensions use only the public extension API; no private kernel imports | arch-lint: extensions/ may import only core/api/; CI builds each bundled extension against the published API types | pending |
 | 12 | **extension ≠ trusted** — every extension registration is bound to declared grants; an extension's tools pass the same policy dispatcher as the model's | contract test: extension-registered tool with no grant is blocked | pending |
 | 13 | **capture ≠ egress** — no screen capture (or derived crop) reaches a cloud endpoint without passing the redaction stage; redaction emits an auditable event | contract test: cloud-bound ContextBundle without a matching redaction event fails; arch-lint: model-client module imports images only from the redaction module's output type | pending |
+| 14 | **turn ≠ ambient** — every Pi run is driven through `Session.runTurn`/`followUp` (budget gate in, `finishTurn` out); no loop-, queue-, or steer-originated turn exists. Past exhaustion, exactly one salvage turn is legal and it is audited (`salvage.issued` precedes it) | arch-lint: only `core/src/session/` may import the PiLoop seam, so no outside module can drive a turn around `Session`; seam contract test — `steer()` with no turn in flight throws in the double (adapter: drops, never calls `sendUserMessage` while idle); the existing abort-drains-queue test; salvage-bypass contract test lands with the adapter | pending |
 
 ## Naming triggers (add an invariant when one fires)
 
diff --git a/biome.json b/biome.json
index 03cb9ac..d0db5f6 100644
--- a/biome.json
+++ b/biome.json
@@ -6,7 +6,8 @@
     "useIgnoreFile": true
   },
   "files": {
-    "ignoreUnknown": false
+    "ignoreUnknown": false,
+    "includes": ["**"]
   },
   "formatter": {
     "enabled": true,
diff --git a/core/src/eventlog/types.ts b/core/src/eventlog/types.ts
index fab3d57..62dca85 100644
--- a/core/src/eventlog/types.ts
+++ b/core/src/eventlog/types.ts
@@ -191,6 +191,10 @@ export interface TaskWaitingData {
 }
 export interface TaskResumedData {
   readonly answerRef?: string;
+  // §3.1 amendment 2026-07-19 #2: absent ⇒ "answered". "cancelled" is the
+  // abort path — the wait ended without an answer, so a fold consumer must
+  // not read the resume as a user response.
+  readonly cause?: "answered" | "cancelled";
 }
 export type TaskReattachedData = Record<string, never>; // recovery.ts always appends {}
 export interface TaskCompletedData {
@@ -244,7 +248,7 @@ export interface ApprovalResolvedData {
   readonly toolCallId?: string;
   readonly argsHash?: string;
   readonly decision?: string; // "auto_denied" | "auto_approved" | "approved" | "denied" | "cancelled"
-  readonly decidedBy?: string; // "background_policy" | "ledger" | "user" | "recovery"
+  readonly decidedBy?: string; // "background_policy" | "ledger" | "user" | "recovery" | "abort" (§3.1 amendment 2026-07-19 #1)
   readonly scope?: "once" | "task" | "persistent";
 }
 export interface ApprovalStaleDecisionData {
diff --git a/core/src/session/abort.test.ts b/core/src/session/abort.test.ts
new file mode 100644
index 0000000..4cd6f1d
--- /dev/null
+++ b/core/src/session/abort.test.ts
@@ -0,0 +1,172 @@
+// Session-level abort oracle (M1-K5a) — the seam proven against its real
+// consumer, not just the double: Session.abort() delegates to the loop's
+// cooperative abort; an aborted turn resolves checkpointSafe: true /
+// exitReason "aborted", records turn.ended + cost.recorded (aborted turns burned
+// tokens — DESIGN §8), and DOES checkpoint — the run is durably finalized
+// (pi-internals: synthetic stopReason "aborted"), so the cursor is valid;
+// the poisoned-cursor guard (event-log §5.2) targets failed turns.
+import { afterAll, describe, expect, it } from "vitest";
+import { makeSessionFixture, waitForPending } from "./fixture.js";
+import type { Session } from "./index.js";
+import { createForegroundSession } from "./index.js";
+
+const fx = makeSessionFixture();
+const writer = fx.newWriter();
+afterAll(() => writer.close());
+
+fx.store.addGrant(fx.grant("files.read", [fx.proj]));
+fx.store.addGrant(fx.grant("files.write", [fx.proj]));
+
+// A read tool the test can hold open mid-execution, so the abort lands
+// while a dispatched call is genuinely in flight (registered before any
+// session is created).
+let release: (() => void) | null = null;
+let entered: (() => void) | null = null;
+fx.registry.register({
+  name: "slow_read",
+  origin: "kernel",
+  effect: "read",
+  requiredCapability: "files.read",
+  paramBinding: { kind: "paths", extract: (a) => [(a as { path: string }).path] },
+  execute: async () => {
+    entered?.();
+    await new Promise<void>((r) => {
+      release = r;
+    });
+    return "slow:ok";
+  },
+});
+
+function foreground(loopName: string): {
+  session: Session;
+  taskId: string;
+  loop: ReturnType<typeof fx.newLoop>;
+} {
+  const loop = fx.newLoop(loopName);
+  const taskId = fx.newTask(writer, loop.jsonlPath);
+  const session = createForegroundSession({
+    taskStreamId: taskId,
+    registry: fx.registry,
+    writer,
+    store: fx.store,
+    loop,
+    home: fx.home,
+    clock: fx.clock,
+  });
+  return { session, taskId, loop };
+}
+
+describe("Session.abort (grants §7.3 breach action / teardown / M2 barge-in)", () => {
+  it("abort mid-turn: in-flight tool completes gated, sibling never dispatches, aborted turn checkpoints", async () => {
+    const { session, taskId, loop } = foreground("abort-mid");
+    loop.queueTurn({
+      calls: [
+        { toolCallId: "a1", tool: "slow_read", args: { path: `${fx.proj}/src/a.ts` } },
+        { toolCallId: "a2", tool: "read_file", args: { path: `${fx.proj}/src/a.ts` } },
+      ],
+      stats: { usd: 0.2 },
+    });
+    const seen = new Promise<void>((r) => {
+      entered = r;
+    });
+    const turn = session.runTurn("long job");
+    await seen; // slow_read's executor is running
+    const abortP = session.abort();
+    release?.();
+    const result = await turn;
+    await abortP;
+
+    expect(result.checkpointSafe).toBe(true);
+    expect(result.exitReason).toBe("aborted");
+    const events = writer.envelopes(taskId);
+    const t = events.map((e) => e.type);
+    // The in-flight call ran to completion through the gate (cooperative
+    // abort awaits it); the queued sibling a2 never started.
+    expect(t.filter((x) => x === "tool.call_started")).toHaveLength(1);
+    expect(events.find((e) => e.type === "tool.call_ended")?.data.ok).toBe(true);
+    // Turn accounting rows exist — an aborted turn still counts (§5.1 rule 3).
+    expect(events.find((e) => e.type === "turn.ended")?.data.exitReason).toBe("aborted");
+    expect(events.filter((e) => e.type === "cost.recorded")).toHaveLength(1);
+    // The K5a ruling: an aborted turn's JSONL entries are durable — the
+    // cursor advances (contrast the failed-turn guard in runtime.test.ts).
+    expect(t.at(-1)).toBe("task.checkpointed");
+    expect(writer.resumeCursor(taskId)?.lastEntryId).toBe(result.lastEntryId);
+  });
+
+  it("§10.19 cancelled on abort: pending approval ⇒ cancelled row, promise errors, abort resolves, late decision mints nothing", async () => {
+    const { session, taskId, loop } = foreground("abort-parked");
+    loop.queueTurn({
+      calls: [{ toolCallId: "p1", tool: "write_file", args: { path: `${fx.proj}/src/p.ts` } }],
+    });
+    const turn = session.runTurn("write it");
+    await waitForPending(session);
+    const card = session.pendingApprovals()[0];
+    if (card === undefined) throw new Error("no pending card");
+    expect(writer.fold(taskId).state).toBe("waiting_for_user");
+
+    // The regression this pins hangs abort() forever (the parked dispatch
+    // promise never settles, so the loop never reaches idle) — race a
+    // timeout so the failure is loud, not a suite hang.
+    const settled = await Promise.race([
+      session.abort().then(() => "resolved"),
+      new Promise<string>((r) => setTimeout(() => r("timed out"), 2000)),
+    ]);
+    expect(settled).toBe("resolved");
+    const result = await turn;
+    expect(result.checkpointSafe).toBe(true);
+    expect(result.exitReason).toBe("aborted");
+
+    // §6.1 cancelled arrow: the audit pair closes, the parked promise
+    // settles as an error result to the model, the table empties, and the
+    // task leaves waiting_for_user (task.resumed) so the aborted turn's
+    // accounting rows fold from a coherent state.
+    expect(session.pendingApprovals()).toHaveLength(0);
+    const events = writer.envelopes(taskId);
+    const resolved = events.filter((e) => e.type === "approval.resolved");
+    expect(resolved).toHaveLength(1);
+    expect(resolved[0]?.data).toMatchObject({
+      toolCallId: "p1",
+      argsHash: card.argsHash,
+      decision: "cancelled",
+      decidedBy: "abort",
+    });
+    // event-log §3.1 amendment 2026-07-19 #2: the resume that closes an
+    // aborted wait carries cause "cancelled" — nobody answered, and a fold
+    // consumer must not read it as a user response.
+    const resumed = events.filter((e) => e.type === "task.resumed");
+    expect(resumed).toHaveLength(1);
+    expect(resumed[0]?.data).toMatchObject({ cause: "cancelled" });
+    expect(writer.fold(taskId).state).toBe("running");
+    expect(fx.executed.filter((e) => e.tool === "write_file")).toHaveLength(0);
+    expect(loop.outcomes.at(-1)).toMatchObject({ ok: false });
+    expect(writer.resumeCursor(taskId)?.lastEntryId).toBe(result.lastEntryId);
+
+    // §10.19 late-decision clause: the cancelled toolCallId is gone from
+    // the table — a decision for it is stale-logged, executes nothing, and
+    // mints nothing (no ledger extension, no store writes).
+    const grantsBefore = fx.store.grants().length;
+    await session.resolveApproval({
+      toolCallId: "p1",
+      argsHash: card.argsHash,
+      decision: "approved",
+      scope: "persistent",
+    });
+    expect(
+      writer.envelopes(taskId).filter((e) => e.type === "approval.stale_decision"),
+    ).toHaveLength(1);
+    expect(fx.executed.filter((e) => e.tool === "write_file")).toHaveLength(0);
+    expect(fx.store.standingApprovals()).toHaveLength(0);
+    expect(fx.store.grants()).toHaveLength(grantsBefore);
+  });
+
+  it("abort with no turn in flight is a no-op; the next turn runs and checkpoints normally", async () => {
+    const { session, taskId, loop } = foreground("abort-idle");
+    await session.abort();
+    loop.queueTurn({
+      calls: [{ toolCallId: "n1", tool: "read_file", args: { path: `${fx.proj}/src/a.ts` } }],
+    });
+    const r = await session.runTurn("normal");
+    expect(r).toMatchObject({ checkpointSafe: true, exitReason: "completed" });
+    expect(writer.resumeCursor(taskId)?.lastEntryId).toBe(r.lastEntryId);
+  });
+});
diff --git a/core/src/session/approval.ts b/core/src/session/approval.ts
index 66a2e86..974723b 100644
--- a/core/src/session/approval.ts
+++ b/core/src/session/approval.ts
@@ -176,6 +176,44 @@ export class ApprovalMachine {
     });
   }
 
+  // grants §6.1: `cancelled` fires on session abort — a parked call cannot
+  // outlive its session (§10.19). For every parked record: append the
+  // cancelled approval.resolved (same row shape recovery.ts appends for the
+  // restart producer), drop it from the table, and settle the held promise
+  // as the §6.1 "error result to model" arrow. One task.resumed closes the
+  // task.waiting the first park opened (event-log §4: waiting_for_user →
+  // running) so the aborted turn's accounting rows fold from a coherent
+  // state — no new event type needed. Deleting before settling is what
+  // makes §10.19's late-decision clause hold: a decision for a cancelled
+  // toolCallId hits resolve()'s unknown-id guard and mints nothing.
+  // grants §6.1: session abort is the escape hatch — "a parked call cannot
+  // outlive its session". Cancel before the loop's own abort, or the parked
+  // dispatch promise never settles and the loop never reaches idle
+  // (deadlock). `decidedBy: "abort"` and the `cause: "cancelled"` on
+  // task.resumed below are event-log §3.1 amendment 2026-07-19 #1/#2:
+  // "recovery" would mislabel a live abort as the boot pass, and a bare
+  // task.resumed reads as "the user answered" when nobody did.
+  cancelAll(append: Appender): void {
+    if (this.pending.size === 0) return;
+    const records = [...this.pending.values()];
+    this.pending.clear();
+    for (const p of records) {
+      append("approval.resolved", {
+        toolCallId: p.card.toolCallId,
+        argsHash: p.card.argsHash,
+        decision: "cancelled",
+        decidedBy: "abort",
+      });
+    }
+    append("task.resumed", {
+      answerRef: records[records.length - 1]?.card.toolCallId,
+      cause: "cancelled",
+    });
+    for (const p of records) {
+      p.settle({ ok: false, error: "approval cancelled: session aborted" });
+    }
+  }
+
   // The user decision resolves a parked call (grants §6.2). A decision whose
   // toolCallId is unknown or whose argsHash mismatches is ignored and logged
   // — a stale card can never approve anything and never mints a grant.
diff --git a/core/src/session/background.test.ts b/core/src/session/background.test.ts
index 9e2698e..781ec94 100644
--- a/core/src/session/background.test.ts
+++ b/core/src/session/background.test.ts
@@ -52,7 +52,7 @@ describe("background session (golden: auto-deny, never awaiting_approval)", () =
       ],
     });
     const result = await session.runTurn("nightly run");
-    expect(result.ok).toBe(true);
+    expect(result.checkpointSafe).toBe(true);
 
     // npm test executed via the schedule-time standing approval.
     expect(fx.executed.some((e) => e.tool === "run_command")).toBe(true);
@@ -155,7 +155,7 @@ describe("push seam robustness (runtime.ts review M1)", () => {
     });
 
     const result = await session.runTurn("push-throws run");
-    expect(result.ok).toBe(true); // the turn survived the throwing sink
+    expect(result.checkpointSafe).toBe(true); // the turn survived the throwing sink
 
     const events = writer.envelopes(taskId);
     const resolved = events.find(
diff --git a/core/src/session/budget.test.ts b/core/src/session/budget.test.ts
index 45487ac..84e7376 100644
--- a/core/src/session/budget.test.ts
+++ b/core/src/session/budget.test.ts
@@ -121,7 +121,7 @@ describe("§10.13: cost.recorded totals match getSessionStats()", () => {
     const { session, taskId, loop } = foreground("cost-failed");
     loop.queueTurn({ calls: [], fail: true, stats: { usd: 0.1 } });
     const result = await session.runTurn("failing");
-    expect(result.ok).toBe(false);
+    expect(result.checkpointSafe).toBe(false);
     const events = writer.envelopes(taskId);
     expect(events.find((e) => e.type === "turn.ended")?.data.exitReason).toBe("error");
     expect(costRows(events)).toHaveLength(1);
@@ -188,7 +188,7 @@ describe("§7.3 R2: refuse to START a turn once a ceiling is already exhausted",
     const { session, taskId, loop } = foreground("budget-start-fg", { maxToolCalls: 1 });
     loop.queueTurn({ calls: [read("s1")] }); // exhausts the ceiling on this turn
     const first = await session.runTurn("a");
-    expect(first.ok).toBe(true);
+    expect(first.checkpointSafe).toBe(true);
     expect(writer.envelopes(taskId).filter((e) => e.type === "budget.exhausted")).toHaveLength(1);
 
     // A queued turn that WOULD dispatch s2 if the loop ever ran it.
@@ -196,7 +196,7 @@ describe("§7.3 R2: refuse to START a turn once a ceiling is already exhausted",
     const executedBefore = fx.executed.length;
     const second = await session.followUp("b");
 
-    expect(second.ok).toBe(false);
+    expect(second.checkpointSafe).toBe(false);
     expect(second.exitReason).toBe("budget_exhausted");
     expect(second.lastEntryId).toBeNull();
     expect(fx.executed.length).toBe(executedBefore); // s2 never dispatched
@@ -211,7 +211,7 @@ describe("§7.3 R2: refuse to START a turn once a ceiling is already exhausted",
     expect(events.filter((e) => e.type === "cost.recorded")).toHaveLength(1);
     // Still exactly one budget.exhausted — the kind was already recorded.
     expect(events.filter((e) => e.type === "budget.exhausted")).toHaveLength(1);
-    // No checkpoint past the refusal (result.ok is false).
+    // No checkpoint past the refusal (result.checkpointSafe is false).
     expect(writer.resumeCursor(taskId)?.lastEntryId).toBe(first.lastEntryId);
   });
 
@@ -219,14 +219,14 @@ describe("§7.3 R2: refuse to START a turn once a ceiling is already exhausted",
     const { session, taskId, loop } = background("budget-start-bg", { maxToolCalls: 1 });
     loop.queueTurn({ calls: [read("bg1")] });
     const first = await session.runTurn("nightly");
-    expect(first.ok).toBe(true);
+    expect(first.checkpointSafe).toBe(true);
     expect(writer.envelopes(taskId).filter((e) => e.type === "budget.exhausted")).toHaveLength(1);
 
     loop.queueTurn({ calls: [read("bg2")] });
     const executedBefore = fx.executed.length;
     const second = await session.followUp("nightly continuation");
 
-    expect(second.ok).toBe(false);
+    expect(second.checkpointSafe).toBe(false);
     expect(second.exitReason).toBe("budget_exhausted");
     expect(fx.executed.length).toBe(executedBefore);
   });
diff --git a/core/src/session/budget.ts b/core/src/session/budget.ts
index 412d706..63b8323 100644
--- a/core/src/session/budget.ts
+++ b/core/src/session/budget.ts
@@ -3,16 +3,27 @@
 // runtime.ts (file-size gate): the dispatch/approval-machine core stays
 // there; this module owns "is this session out of budget" plus what it
 // costs to log a turn.
-// SPEC-NOTE (grants.md §7.3): the hard-layer backstop — subscribing to
-// agent_settled/tool_execution_start/end, calling session.abort() on
-// breach, and the wall-clock TIMER that reaches sessions parked in
-// awaiting_approval (round-1 A11b) — needs the real Pi adapter's
-// event/abort seam, which the PiLoop interface deliberately does not carry
-// yet. What is decidable at this layer — per-call/per-turn accounting, the
-// budget.* rows, AND refusing a new turn's ENTRY once a kind is already
-// exhausted (exhaustedKinds below, gated in runtime.ts before
-// loop.start/followUp runs) — is implemented here; the mid-turn abort
-// watcher lands with the Pi adapter.
+// SPEC-NOTE (grants.md §7.3): the PiLoop seam now carries what the
+// hard-layer backstop consumes — onEvent (tool_start/tool_end/settled)
+// and abort() (K5a) — so the watcher is UNBLOCKED but still deferred to
+// the Pi-adapter work item: the subscription wiring, the wall-clock TIMER
+// that reaches sessions parked in awaiting_approval (round-1 A11b), the
+// breach → Session.abort() call, and its own timeout around abort (a tool
+// ignoring its signal hangs, pi-internals) all land there. What is
+// decidable at this layer — per-call/per-turn accounting, the budget.*
+// rows, AND refusing a new turn's ENTRY once a kind is already exhausted
+// (exhaustedKinds below, gated in runtime.ts before loop.start/followUp
+// runs) — is implemented here.
+//
+// The one salvage call at exhaustion (DESIGN §4) does NOT just "land with
+// the adapter" alongside that watcher wiring: runtime.ts's budgetGate()
+// refuses ANY new turn once exhaustedKinds is non-empty, with no bypass —
+// so a salvage turn issued after exhaustion would be dropped by the very
+// gate this file implements, contradicting DESIGN.md:140 ("never a
+// dropped turn"). The salvage path is otherwise already specified —
+// event-log.md §3.1 defines a salvage.issued {} event type — it just has
+// no way through budgetGate yet. The Pi-adapter work item needs a
+// budgetGate bypass for the single salvage call; it is not built here.
 import type { Envelope } from "../eventlog/index.js";
 import type { PiLoop, SessionStats, TurnResult } from "./piloop.js";
 
@@ -117,6 +128,6 @@ export class BudgetTracker {
   // this call, so there is no cost delta to record — only the refusal.
   refuseTurn(append: Appender): TurnResult {
     append("turn.ended", { exitReason: "budget_exhausted" });
-    return { ok: false, lastEntryId: null, exitReason: "budget_exhausted" };
+    return { checkpointSafe: false, lastEntryId: null, exitReason: "budget_exhausted" };
   }
 }
diff --git a/core/src/session/index.ts b/core/src/session/index.ts
index 429d82a..7a7f403 100644
--- a/core/src/session/index.ts
+++ b/core/src/session/index.ts
@@ -11,12 +11,15 @@ export {
 } from "./jsonl.js";
 export {
   type Dispatch,
+  type ExitReason,
   type PiLoop,
+  type PiLoopEvent,
   type PiToolCall,
   readJsonlEntries,
   type SessionStats,
   type ToolOutcome,
   type TurnResult,
+  type Unsubscribe,
 } from "./piloop.js";
 export {
   type ApprovalDecision,
diff --git a/core/src/session/jsonl.test.ts b/core/src/session/jsonl.test.ts
index a2caa0c..7ccb808 100644
--- a/core/src/session/jsonl.test.ts
+++ b/core/src/session/jsonl.test.ts
@@ -19,20 +19,22 @@ afterAll(() => writer.close());
 
 const dirs = [fx.sessionsDir, fx.archiveDir];
 
-function completedTask(name: string): { taskId: string; jsonlPath: string } {
+async function completedTask(name: string): Promise<{ taskId: string; jsonlPath: string }> {
   const loop = fx.newLoop(name);
   loop.queueTurn({ calls: [] });
   const taskId = fx.newTask(writer, loop.jsonlPath);
-  // Materialize the JSONL file (a turn writes entries).
-  void loop.start("x", async () => ({ ok: true, result: null }));
+  // Materialize the JSONL file (a turn writes entries). Awaited: the loop
+  // contract makes no synchronous-write promise (K5a — the fake defers a
+  // microtask, like the real async prompt()).
+  await loop.start("x", async () => ({ ok: true, result: null }));
   writer.append(taskId, { type: "task.completed", actor: "user", data: { resultRef: null } });
   return { taskId, jsonlPath: loop.jsonlPath };
 }
 
 describe("JSONL lifecycle (invariant 8 cross-store guard)", () => {
-  it("compaction moves the file to the archive dir — content intact, count conserved, only purge deletes", () => {
-    const a = completedTask("jl-a");
-    const b = completedTask("jl-b");
+  it("compaction moves the file to the archive dir — content intact, count conserved, only purge deletes", async () => {
+    const a = await completedTask("jl-a");
+    const b = await completedTask("jl-b");
     const contentA = readFileSync(a.jsonlPath, "utf8");
     const before = countJsonlFiles(dirs);
     expect(before).toBe(2);
@@ -74,8 +76,8 @@ describe("JSONL lifecycle (invariant 8 cross-store guard)", () => {
     expect(res.archivedJsonlPath).toBeNull();
   });
 
-  it("purge deletes live AND archived JSONL — recorded first as system.purge", () => {
-    const c = completedTask("jl-c"); // one live file alongside the archived two
+  it("purge deletes live AND archived JSONL — recorded first as system.purge", async () => {
+    const c = await completedTask("jl-c"); // one live file alongside the archived two
     expect(countJsonlFiles(dirs)).toBe(3);
 
     const res = purgeAllWithJsonl(writer, dirs);
@@ -89,10 +91,10 @@ describe("JSONL lifecycle (invariant 8 cross-store guard)", () => {
     expect(row?.envelope.data.scope).toBe("all");
   });
 
-  it("guard: across compaction cycles, the cross-store file count never decreases absent a purge", () => {
+  it("guard: across compaction cycles, the cross-store file count never decreases absent a purge", async () => {
     const counts: number[] = [countJsonlFiles(dirs)];
     for (const name of ["jl-d", "jl-e", "jl-f"]) {
-      const t = completedTask(name);
+      const t = await completedTask(name);
       counts.push(countJsonlFiles(dirs));
       compactTaskWithJsonl(writer, t.taskId, fx.archiveDir);
       counts.push(countJsonlFiles(dirs));
diff --git a/core/src/session/piloop.test.ts b/core/src/session/piloop.test.ts
new file mode 100644
index 0000000..358f1b7
--- /dev/null
+++ b/core/src/session/piloop.test.ts
@@ -0,0 +1,370 @@
+// PiLoop seam contract oracle (M1-K5a) — the abort/onEvent surface the real
+// Pi adapter must satisfy, pinned against recon ground truth
+// (specs/recon/pi-types-0.80.10.md, pi-internals.md): cooperative abort
+// awaits in-flight tools and resolves at idle; events fan out synchronously
+// in subscription order; a throwing subscriber never tears the loop
+// (runtime.ts onActivity precedent, d35c9ad). Exercised on FakePiLoop — the
+// double is the executable statement of the contract the adapter inherits.
+import { describe, expect, it } from "vitest";
+import { makeSessionFixture } from "./fixture.js";
+import type { Dispatch, PiLoopEvent } from "./piloop.js";
+import { readJsonlEntries } from "./piloop.js";
+
+const fx = makeSessionFixture();
+
+const okDispatch: Dispatch = () => Promise.resolve({ ok: true, result: "r" });
+
+const call = (id: string) => ({ toolCallId: id, tool: "read_file", args: {} });
+
+describe("PiLoop.abort (cooperative, pi-types §Abort-is-async)", () => {
+  it("abort with no turn running resolves immediately and leaves the loop usable", async () => {
+    const loop = fx.newLoop("idle-abort");
+    await loop.abort(); // already idle — resolves, poisons nothing
+    loop.queueTurn({ calls: [call("a1")] });
+    const result = await loop.start("go", okDispatch);
+    expect(result).toMatchObject({ checkpointSafe: true, exitReason: "completed" });
+  });
+
+  it("abort mid-turn awaits the in-flight call, skips queued siblings, finalizes durably (checkpointSafe: true, exitReason aborted)", async () => {
+    const loop = fx.newLoop("coop-abort");
+    loop.queueTurn({ calls: [call("c1"), call("c2")] });
+    let release!: () => void;
+    const gate = new Promise<void>((r) => {
+      release = r;
+    });
+    const dispatched: string[] = [];
+    let abortDone = false;
+    let abortPromise: Promise<void> = Promise.resolve();
+    const dispatch: Dispatch = async (c) => {
+      dispatched.push(c.toolCallId);
+      if (c.toolCallId === "c1") {
+        abortPromise = loop.abort().then(() => {
+          abortDone = true;
+        });
+        await gate; // the in-flight tool keeps running after abort is requested
+      }
+      return { ok: true, result: "r" };
+    };
+    const turn = loop.start("go", dispatch);
+    await new Promise((r) => setImmediate(r));
+    // Cooperative: abort() has NOT resolved while a dispatched tool runs —
+    // in-flight results are awaited, never discarded (pi-internals). A tool
+    // ignoring its signal hangs; the budget layer needs its own timeout.
+    expect(abortDone).toBe(false);
+    release();
+    const result = await turn;
+    await abortPromise;
+    expect(result.checkpointSafe).toBe(true);
+    expect(result.exitReason).toBe("aborted");
+    expect(dispatched).toEqual(["c1"]); // c2 was never started
+    // The run is finalized on disk with a synthetic aborted entry
+    // (pi-internals: stopReason "aborted") — lastEntryId is a real,
+    // durable cursor.
+    const entries = readJsonlEntries(loop.jsonlPath);
+    expect(entries.at(-1)).toMatchObject({
+      id: result.lastEntryId,
+      kind: "turnEnd",
+      aborted: true,
+    });
+  });
+
+  it("repeated aborts are idempotent — every promise resolves at idle, one aborted finalization, next turn clean", async () => {
+    const loop = fx.newLoop("re-abort");
+    loop.queueTurn({ calls: [call("r1")] });
+    const aborts: Promise<void>[] = [];
+    const dispatch: Dispatch = (c) => {
+      aborts.push(loop.abort(), loop.abort());
+      return okDispatch(c);
+    };
+    const result = await loop.start("go", dispatch);
+    await Promise.all(aborts);
+    expect(result.exitReason).toBe("aborted");
+    expect(readJsonlEntries(loop.jsonlPath).filter((e) => e.aborted === true)).toHaveLength(1);
+    // The abort applied to the current operation only — a fresh turn runs.
+    loop.queueTurn({ calls: [call("r2")] });
+    expect((await loop.start("again", okDispatch)).exitReason).toBe("completed");
+  });
+});
+
+describe("PiLoop.onEvent (narrow typed stream, surface-protocol §7 + grants §7.3)", () => {
+  it("tool_start/tool_end wrap each dispatch, settled closes the turn; unsubscribe stops delivery", async () => {
+    const loop = fx.newLoop("events");
+    const seen: PiLoopEvent[] = [];
+    const unsub = loop.onEvent((e) => seen.push(e));
+    loop.queueTurn({ calls: [call("e1"), call("e2")] });
+    await loop.start("go", okDispatch);
+    expect(seen.map((e) => e.kind)).toEqual([
+      "tool_start",
+      "tool_end",
+      "tool_start",
+      "tool_end",
+      "settled",
+    ]);
+    expect(seen[0]).toEqual({ kind: "tool_start", toolCallId: "e1", tool: "read_file" });
+    expect(seen[1]).toEqual({ kind: "tool_end", toolCallId: "e1", tool: "read_file", ok: true });
+    unsub();
+    loop.queueTurn({ calls: [call("e3")] });
+    await loop.followUp("more", okDispatch);
+    expect(seen).toHaveLength(5);
+  });
+
+  it("settled fires for failed turns too (agent_settled = idle, not success); a crash never settles", async () => {
+    const loop = fx.newLoop("settled-fail");
+    const seen: string[] = [];
+    loop.onEvent((e) => seen.push(e.kind));
+    loop.queueTurn({ calls: [], fail: true });
+    const r = await loop.start("go", okDispatch);
+    expect(r.checkpointSafe).toBe(false);
+    expect(seen).toEqual(["settled"]);
+    loop.queueTurn({ calls: [], crashBeforeReturn: true });
+    await expect(loop.followUp("crash", okDispatch)).rejects.toThrow();
+    expect(seen).toEqual(["settled"]); // no settle for a dead process
+  });
+
+  it("a throwing subscriber is swallowed — later listeners still fire and the turn completes (runtime.ts onActivity precedent)", async () => {
+    const loop = fx.newLoop("throwing-sub");
+    const seen: string[] = [];
+    loop.onEvent(() => {
+      throw new Error("surface down");
+    });
+    loop.onEvent((e) => seen.push(e.kind));
+    loop.queueTurn({ calls: [call("t1")] });
+    const result = await loop.start("go", okDispatch);
+    expect(result.checkpointSafe).toBe(true);
+    expect(seen).toEqual(["tool_start", "tool_end", "settled"]);
+  });
+
+  it("scripted emission: the surface-protocol §7 message variants reach listeners with our shapes", () => {
+    const loop = fx.newLoop("scripted");
+    const seen: PiLoopEvent[] = [];
+    loop.onEvent((e) => seen.push(e));
+    loop.emit({ kind: "message_start", messageId: "m1" });
+    loop.emit({
+      kind: "message_delta",
+      messageId: "m1",
+      blockIndex: 0,
+      deltaKind: "text",
+      delta: "hel",
+    });
+    loop.emit({
+      kind: "message_delta",
+      messageId: "m1",
+      blockIndex: 1,
+      deltaKind: "thinking",
+      delta: "hmm",
+    });
+    loop.emit({ kind: "tool_update", toolCallId: "x1", update: { pct: 50 } });
+    loop.emit({ kind: "message_end", messageId: "m1", message: { role: "assistant" } });
+    expect(seen.map((e) => e.kind)).toEqual([
+      "message_start",
+      "message_delta",
+      "message_delta",
+      "tool_update",
+      "message_end",
+    ]);
+    expect(seen[1]).toEqual({
+      kind: "message_delta",
+      messageId: "m1",
+      blockIndex: 0,
+      deltaKind: "text",
+      delta: "hel",
+    });
+    expect(seen[4]).toEqual({
+      kind: "message_end",
+      messageId: "m1",
+      message: { role: "assistant" },
+    });
+  });
+
+  it("listeners fire in subscription order (docstring's fan-out guarantee — item 5: reversing broke no test before this)", async () => {
+    const loop = fx.newLoop("fanout-order");
+    const order: string[] = [];
+    loop.onEvent((e) => order.push(`A:${e.kind}`));
+    loop.onEvent((e) => order.push(`B:${e.kind}`));
+    loop.queueTurn({ calls: [call("f1")] });
+    await loop.start("go", okDispatch);
+    expect(order.length).toBeGreaterThan(0);
+    // Every event must reach listener A before listener B — a fan-out that
+    // reversed subscription order would flip every pair here.
+    for (let i = 0; i < order.length; i += 2) {
+      expect(order[i]?.startsWith("A:")).toBe(true);
+      expect(order[i + 1]?.startsWith("B:")).toBe(true);
+      expect(order[i]?.slice(2)).toBe(order[i + 1]?.slice(2));
+    }
+  });
+
+  it("token-keyed listeners: subscribing the same function twice gives two independent unsubscribes (item 4)", async () => {
+    const loop = fx.newLoop("dup-listener-fn");
+    const seen: string[] = [];
+    const fn = (e: PiLoopEvent) => seen.push(e.kind);
+    const unsubA = loop.onEvent(fn);
+    loop.onEvent(fn); // same function reference, second subscription
+    unsubA(); // must remove only ONE of the two subscriptions
+    loop.queueTurn({ calls: [call("d1")] });
+    await loop.start("go", okDispatch);
+    // If the two subscriptions had collapsed into one (a Set<listener>
+    // bug), unsubA() would have killed both and nothing would be seen.
+    expect(seen).toEqual(["tool_start", "tool_end", "settled"]);
+  });
+
+  it("two concurrent calls emit tool_start in source order but tool_end in COMPLETION order (item 1)", async () => {
+    const loop = fx.newLoop("completion-order");
+    const seen: PiLoopEvent[] = [];
+    loop.onEvent((e) => seen.push(e));
+    loop.queueTurn({ concurrent: true, calls: [call("c1"), call("c2")] });
+    let releaseC1!: () => void;
+    const c1Gate = new Promise<void>((r) => {
+      releaseC1 = r;
+    });
+    const dispatch: Dispatch = async (c) => {
+      if (c.toolCallId === "c1") {
+        await c1Gate; // c1 resolves LAST
+      }
+      return { ok: true, result: "r" };
+    };
+    const turn = loop.start("go", dispatch);
+    await new Promise((r) => setImmediate(r));
+    // c2 has no gate — it settles first.
+    releaseC1();
+    await turn;
+    const starts = seen
+      .filter((e) => e.kind === "tool_start")
+      .map((e) => (e as { toolCallId: string }).toolCallId);
+    const ends = seen
+      .filter((e) => e.kind === "tool_end")
+      .map((e) => (e as { toolCallId: string }).toolCallId);
+    expect(starts).toEqual(["c1", "c2"]);
+    expect(ends).toEqual(["c2", "c1"]);
+  });
+
+  it("a rejecting dispatch in a concurrent turn propagates through Promise.all and leaks no unhandled rejection (item 1)", async () => {
+    const loop = fx.newLoop("concurrent-reject");
+    loop.queueTurn({ concurrent: true, calls: [call("r1"), call("r2")] });
+    const unhandled: unknown[] = [];
+    const onUnhandledRejection = (reason: unknown) => unhandled.push(reason);
+    process.on("unhandledRejection", onUnhandledRejection);
+    try {
+      const dispatch: Dispatch = async (c) => {
+        if (c.toolCallId === "r1") {
+          // decideCall()/append() can both throw in production (runtime.ts
+          // dispatch) — this stands in for that rejection.
+          throw new Error("decideCall blew up");
+        }
+        return { ok: true, result: "r" };
+      };
+      // The rejection still propagates — settled.then()'s onRejected
+      // no-op (item 1's fix) only swallows the DERIVED promise, never
+      // `settled` itself, which Promise.all still awaits and rejects on.
+      await expect(loop.start("go", dispatch)).rejects.toThrow("decideCall blew up");
+      // Give a leaked unhandled rejection (pre-fix behavior) several ticks
+      // to surface before asserting none did.
+      await new Promise((r) => setImmediate(r));
+      await new Promise((r) => setImmediate(r));
+      expect(unhandled).toEqual([]);
+    } finally {
+      process.off("unhandledRejection", onUnhandledRejection);
+    }
+  });
+});
+
+describe("PiLoop re-entry guard (item 2)", () => {
+  it("a second turn started while one is in flight is refused, keeping inFlight from being clobbered", async () => {
+    const loop = fx.newLoop("no-reentry");
+    loop.queueTurn({ calls: [call("o1")] });
+    let release!: () => void;
+    const gate = new Promise<void>((r) => {
+      release = r;
+    });
+    const dispatch: Dispatch = async (c) => {
+      await gate;
+      return { ok: true, result: c.toolCallId };
+    };
+    const turn1 = loop.start("go", dispatch);
+    await new Promise((r) => setImmediate(r));
+    // turn1 is genuinely in flight — a second call must be refused, not
+    // silently overwrite `inFlight` (the bug: turn 1's finally would then
+    // clear turn 2's slot, so abort() resolves claiming idle while turn 2
+    // keeps dispatching).
+    expect(() => loop.followUp("again", okDispatch)).toThrow(/in flight/);
+    release();
+    const result1 = await turn1;
+    expect(result1.checkpointSafe).toBe(true);
+    expect(result1.exitReason).toBe("completed");
+    // The slot is clean afterward — a fresh turn runs normally.
+    loop.queueTurn({ calls: [call("o2")] });
+    expect((await loop.start("again", okDispatch)).exitReason).toBe("completed");
+  });
+});
+
+describe("PiLoop steer queue cleared on abort (item 3)", () => {
+  it("abort() clears queued steer text (clearQueue) so a stray steer can't fire a turn past the abort", async () => {
+    const loop = fx.newLoop("clear-queue");
+    loop.queueTurn({ calls: [call("q1")] });
+    let release!: () => void;
+    const gate = new Promise<void>((r) => {
+      release = r;
+    });
+    let abortPromise: Promise<void> = Promise.resolve();
+    const dispatch: Dispatch = async (c) => {
+      loop.steer("stop that");
+      abortPromise = loop.abort();
+      await gate; // the in-flight tool keeps running after abort is requested
+      return { ok: true, result: c.toolCallId };
+    };
+    const turn = loop.start("go", dispatch);
+    await new Promise((r) => setImmediate(r));
+    // abort() drains the queue synchronously, before the turn even settles.
+    expect(loop.queuedSteers).toEqual([]);
+    release();
+    const result = await turn;
+    await abortPromise;
+    expect(result.exitReason).toBe("aborted");
+    // The audit trail ("was it steered") survives even though the queue
+    // ("is it still pending") was cleared.
+    expect(loop.steered).toEqual(["stop that"]);
+    expect(loop.queuedSteers).toEqual([]);
+  });
+
+  it("steering an idle loop throws (item 3: real sendUserMessage always triggers a turn, an ungated one)", async () => {
+    const loop = fx.newLoop("steer-idle");
+    expect(() => loop.steer("hello?")).toThrow(/no turn in flight/);
+    expect(loop.steered).toEqual([]);
+    expect(loop.queuedSteers).toEqual([]);
+  });
+
+  it("steering mid-turn still queues as before", async () => {
+    const loop = fx.newLoop("steer-mid-turn");
+    loop.queueTurn({ calls: [call("m1")] });
+    let release!: () => void;
+    const gate = new Promise<void>((r) => {
+      release = r;
+    });
+    const dispatch: Dispatch = async (c) => {
+      loop.steer("still running");
+      await gate;
+      return { ok: true, result: c.toolCallId };
+    };
+    const turn = loop.start("go", dispatch);
+    await new Promise((r) => setImmediate(r));
+    expect(loop.queuedSteers).toEqual(["still running"]);
+    release();
+    await turn;
+    expect(loop.steered).toEqual(["still running"]);
+  });
+});
+
+describe("PiLoop.dispose (item 9 — not abort, teardown)", () => {
+  it("dispose() clears all listeners; abort() does not", async () => {
+    const loop = fx.newLoop("dispose");
+    const seen: string[] = [];
+    loop.onEvent((e) => seen.push(e.kind));
+    loop.queueTurn({ calls: [call("p1")] });
+    await loop.start("go", okDispatch);
+    expect(seen.length).toBeGreaterThan(0);
+    loop.dispose();
+    loop.queueTurn({ calls: [call("p2")] });
+    await loop.start("again", okDispatch);
+    // No new events after dispose — the listener is gone, not merely idle.
+    expect(seen).toEqual(["tool_start", "tool_end", "settled"]);
+  });
+});
diff --git a/core/src/session/piloop.ts b/core/src/session/piloop.ts
index 82d9110..9965dba 100644
--- a/core/src/session/piloop.ts
+++ b/core/src/session/piloop.ts
@@ -14,18 +14,103 @@ export type PiToolCall = {
 
 export type ToolOutcome = { ok: true; result: unknown } | { ok: false; error: string };
 
-// The session runtime's gated dispatcher, handed to the loop per turn. The
-// loop must await it for every tool call the model emits.
+// The session runtime's gated dispatcher. NOT "handed to the loop per
+// turn" in the way that reads on Pi: real Pi wires tool execution at
+// SESSION CONSTRUCTION (AgentSessionConfig.customTools), not per prompt()/
+// followUp() call — there is no per-turn hook. The adapter must register
+// each tool ONCE, around a MUTABLE current-dispatch reference that
+// start()/followUp() swap in before driving the turn; this is only sound
+// because turns never overlap (enforced below — FakePiLoop.runTurn
+// refuses re-entry; pi-types-0.80.10, agent-session.d.ts). The loop must
+// await this for every tool call the model emits.
 export type Dispatch = (call: PiToolCall) => Promise<ToolOutcome>;
 
+// The seam's closed vocabulary for how a turn ended — narrow so nothing
+// downstream can compare against a typo'd literal (event-log.md §3.1's
+// turn.ended {exitReason} shape).
+export type ExitReason = "completed" | "aborted" | "error" | "budget_exhausted";
+
 export type TurnResult = {
-  ok: boolean;
+  // checkpointSafe: true iff lastEntryId is a durable, checkpoint-safe
+  // cursor — NOT "the model finished what was asked". An ABORTED turn is
+  // checkpointSafe: true (its JSONL entries are durably on disk, pi-
+  // internals synthetic stopReason "aborted"); only a FAILED turn is
+  // false — the poisoned-cursor guard (event-log §5.2) targets failures,
+  // not aborts.
+  checkpointSafe: boolean;
   // Id of the last JSONL entry the loop reports committed (Pi's format,
   // opaque to us — event-log.md §5.2). null when nothing was written.
   lastEntryId: string | null;
-  exitReason: string;
+  exitReason: ExitReason;
 };
 
+// Loop event seam — OUR union, never Pi's AgentSessionEvent re-exported
+// (pi.md §⚠2: Pi minors are breaking; this wrapper IS the insulation).
+// Narrow by design: each admitted variant has a named consumer; everything
+// else in Pi's union is deliberately excluded —
+//   · turn_start/turn_end/agent_start/agent_end: TurnResult already reports
+//     turn completion, and agent_end is a trap (willRetry: true means the
+//     run is NOT over — only agent_settled is, pi-types-0.80.10).
+//   · compaction_*/auto_retry_*: excluded NOT because getSessionStats()
+//     exposes counters for them — it does not (specs/recon/
+//     pi-types-0.80.10.md "Stats — and the four fields Pi does NOT give
+//     us": real SessionStats carries no model/route/compactions/retries
+//     at all). The exclusion still stands: the adapter holds the Pi
+//     session directly and can subscribe to compaction_end/auto_retry_end
+//     internally to accumulate its OWN compactions/retries counters,
+//     supplying model/route from our routing decision — no PiLoopEvent
+//     consumer needs the mid-turn events themselves.
+//   · entry_appended: checkpointing consumes TurnResult.lastEntryId; one
+//     writer (invariant 7) means nobody tails entries live.
+//   · queue_update/session_info_changed/thinking_level_changed: no consumer
+//     yet; widen only when an M2 surface names one.
+export type PiLoopEvent =
+  // grants §7.3 hard layer: the backstop watcher counts tool executions
+  // and needs the settle signal to close the audit record.
+  | { kind: "tool_start"; toolCallId: string; tool: string }
+  | { kind: "tool_end"; toolCallId: string; tool: string; ok: boolean }
+  // agent_settled — fires once when the loop is fully idle; the ONLY
+  // end-of-run signal (see agent_end note above). The settlement anchor
+  // for the deferred budget watcher. FakePiLoop emits `settled` once per
+  // driveTurn, and under THIS seam's own rules that matches real Pi
+  // exactly one-for-one: the adapter must drive turns via prompt(), never
+  // Pi's own followUp() (start()/followUp() doc above), and abort() drains
+  // any queued steer/followUp text (item 3) — so the queue real Pi's
+  // agent_settled watches for is always empty at seam-turn boundaries.
+  // Hypothetical, not a live risk: IF a future adapter ever queued via
+  // Pi's own followUp() against these rules, real agent_settled would not
+  // fire between the queuing turn and the one that drains it, and a
+  // watcher counting one `settled` per seam-turn would undercount.
+  | { kind: "settled" }
+  // surface-protocol §7 stream bridge: message_start/delta/message_end/
+  // tool_update (renamed from tool_progress to match §7's wire name and
+  // this variant's own comment). Deltas carry type+index+delta ONLY —
+  // never Pi's full `partial` snapshot per event (pi-internals "Streaming
+  // granularity"); blockIndex = Pi's contentIndex. The one full message
+  // rides message_end. toolArgs deltas are model-authored JSON text —
+  // untrusted, display-only, never parse-and-act (pi-types-0.80.10).
+  // `message` and `update` are opaque payloads (like PiToolCall.args): the
+  // bridge serializes them and applies its own caps (50 KB on tool_update
+  // is surface policy, not seam mechanism). messageId (surface-protocol
+  // §7) is minted by the ADAPTER, not Pi: real AssistantMessage has no
+  // stable id, only an optional provider-specific responseId (pi-types-
+  // 0.80.10, pi-ai types.d.ts) — so the adapter must mint one id per
+  // message and hold it stable across that message's start/delta*/end, or
+  // two bridge consumers watching the same stream would mint different
+  // ids for the same message.
+  | { kind: "message_start"; messageId: string }
+  | {
+      kind: "message_delta";
+      messageId: string;
+      blockIndex: number;
+      deltaKind: "text" | "thinking" | "toolArgs";
+      delta: string;
+    }
+  | { kind: "message_end"; messageId: string; message: unknown }
+  | { kind: "tool_update"; toolCallId: string; update: unknown };
+
+export type Unsubscribe = () => void;
+
 // Cumulative session stats (pi-internals "Cost tracking": getSessionStats
 // sums per-assistant-turn usage; attribution is per turn, never per tool
 // call). The runtime diffs successive snapshots to produce the per-turn
@@ -46,9 +131,81 @@ export interface PiLoop {
   // in the resume cursor (event-log.md §5.2, §9).
   readonly jsonlPath: string;
   start(prompt: string, dispatch: Dispatch): Promise<TurnResult>;
+  // NAME COLLISION WARNING: real AgentSession.followUp() QUEUES a message
+  // for delivery after the CURRENT run finishes — it does not drive a
+  // turn (see clearQueue() below for how a queued one gets dropped on
+  // abort). OUR followUp() drives a whole new turn, same as start(), just
+  // against an existing session. An adapter author pattern-matching the
+  // name will reach for the wrong Pi method; the adapter's followUp()
+  // must call Pi's prompt(), never AgentSession.followUp().
   followUp(text: string, dispatch: Dispatch): Promise<TurnResult>;
+  // CONTRACT (item 3): legal only while a turn is IN FLIGHT. Steering an
+  // idle loop is a caller error, not a no-op — real AgentSession.
+  // sendUserMessage "Always triggers a turn" (agent-session.d.ts), so
+  // steering nothing would mint a turn that never passes budgetGate(),
+  // is never checkpointed, and emits no turn.ended/cost.recorded: a whole
+  // turn outside the audit trail and outside budget enforcement (the same
+  // hole the abort/clearQueue fix above closes for a QUEUED steer, here
+  // for the IDLE case). The correct idle path is followUp() — it drives
+  // through Session.followUp and therefore through budgetGate. Adapter
+  // obligation: never call Pi's sendUserMessage while the loop is idle;
+  // drop the text or reject instead. FakePiLoop.steer throws in this
+  // state so the double can't certify a world where the hole is harmless.
+  //
+  // Stays fire-and-forget `void` although the real sendUserMessage is
+  // async: its promise resolves when the text is QUEUED, not acted on
+  // (pi-types-0.80.10), no consumer branches on acceptance, and threading
+  // a promise up through Session.steer to the surface handler buys
+  // nothing. The adapter must attach its own rejection handler so a
+  // failed send becomes a log line, never an unhandled rejection.
   steer(text: string): void;
+  // Cooperative abort (pi-types-0.80.10 "Abort is async"): resolves when
+  // the loop is idle. In-flight tools are AWAITED, not discarded — a tool
+  // ignoring the abort hangs this promise, so the §7.3 watcher still
+  // needs its own timeout around it. A parked approval inside dispatch
+  // does NOT hang it: Session.abort() cancels the parked table BEFORE
+  // calling this (grants §6.1 cancelled-on-abort, §10.19 — a parked call
+  // cannot outlive its session), settling every dispatch promise the loop
+  // is awaiting. A bare loop.abort() with an uncancelled park would never
+  // settle — always abort through Session.
+  // Also clears any queued steer()/followUp() text — real Pi ships
+  // clearQueue() specifically for this path ("Useful for restoring to
+  // editor when user aborts", pi-types-0.80.10 agent-session.d.ts):
+  // sendUserMessage ALWAYS triggers a turn, so a steer left queued past
+  // an abort would fire a brand-new turn that never passes through
+  // budgetGate() — a whole turn missing from the audit trail past an
+  // exhausted budget. The adapter's abort() must call clearQueue() (or
+  // equivalent) before resolving.
+  // ADAPTER OBLIGATION, not a guarantee Pi documents: Pi states no
+  // idempotency or never-rejects contract on its abort(). OUR abort()
+  // must be idempotent (resolves immediately when no turn is running) and
+  // must never reject — the adapter is responsible for de-duplicating
+  // concurrent aborts and swallowing any rejection from the underlying
+  // Session.abort() call to satisfy this. The in-flight start()/
+  // followUp() resolves with exitReason "aborted" and checkpointSafe: true
+  // (see TurnResult — the aborted run is durably finalized and checkpoint-
+  // safe). Consumers: the grants §7.3 budget backstop on breach, M2 voice
+  // barge-in. NOT teardown — see dispose() below for releasing resources.
+  abort(): Promise<void>;
+  // Subscription over constructor-sink: the budget watcher and the surface
+  // stream bridge are independent consumers with independent lifetimes —
+  // each needs its own unsubscribe (mirrors Pi's subscribe(listener) =>
+  // unsubscribe). Delivery is synchronous, events in emission order to
+  // each listener, listeners in subscription order. Across PARALLEL tool
+  // executions tool_end arrives in completion order (pi core types) —
+  // consumers key on toolCallId, never position. A throwing listener is
+  // swallowed and later listeners still fire: same rule as the onActivity
+  // push seam (runtime.ts append(), d35c9ad) — a surface hiccup must not
+  // tear the loop or the audit trail; the adapter must wrap OUR listeners
+  // so a throw never propagates into Pi's emit loop.
+  onEvent(listener: (event: PiLoopEvent) => void): Unsubscribe;
   getSessionStats(): SessionStats;
+  // Release resources permanently (real Pi: "Remove all listeners and
+  // disconnect from agent"). NOT what abort() does — abort() stops the
+  // in-flight turn and leaves the loop usable for the next one; dispose()
+  // ends the loop's life. Conflating the two is a leak vector given
+  // pi-internals' process-global hazards.
+  dispose(): void;
 }
 
 // ---------------------------------------------------------------------------
@@ -61,8 +218,8 @@ export interface PiLoop {
 
 export type FakeTurn = {
   calls: readonly PiToolCall[];
-  // Turn errors after its calls ran ⇒ TurnResult.ok false (the caller must
-  // NOT checkpoint — poisoned-cursor guard).
+  // Turn errors after its calls ran ⇒ TurnResult.checkpointSafe false (the
+  // caller must NOT checkpoint — poisoned-cursor guard).
   fail?: boolean;
   crashBeforeReturn?: boolean;
   // Dispatch this turn's calls via Promise.all (pi-internals: prep
@@ -94,7 +251,22 @@ export class FakePiLoop implements PiLoop {
   private readonly turns: FakeTurn[] = [];
   private entrySeq: number;
   readonly outcomes: ToolOutcome[] = [];
+  // Append-only record of every steer() call, for "was this steered"
+  // assertions — never cleared, unlike the pending queue below.
   readonly steered: string[] = [];
+  // Models Pi's steer/followUp queue (agent-session.d.ts clearQueue()).
+  // abort() drains this (item 3) so a stray queued steer can't fire a
+  // turn past the abort; `steered` above stays intact as the audit trail.
+  private queued: string[] = [];
+  // Token-keyed, not a Set<listener>: a Set collapses two subscriptions
+  // that pass the SAME function reference into one, and unsubscribing
+  // either would kill both — contradicting onEvent's "each consumer needs
+  // its own unsubscribe" (PiLoop.onEvent doc). A Map preserves insertion
+  // (subscription) order for fan-out, same as the Set did.
+  private readonly listeners = new Map<number, (event: PiLoopEvent) => void>();
+  private nextListenerToken = 0;
+  private inFlight: Promise<TurnResult> | null = null;
+  private aborting = false;
 
   constructor(readonly jsonlPath: string) {
     // Continue entry numbering across reattach — a resumed loop appends
@@ -106,6 +278,58 @@ export class FakePiLoop implements PiLoop {
     this.turns.push(turn);
   }
 
+  onEvent(listener: (event: PiLoopEvent) => void): Unsubscribe {
+    const token = this.nextListenerToken++;
+    this.listeners.set(token, listener);
+    return () => {
+      this.listeners.delete(token);
+    };
+  }
+
+  // Public so tests can script the message-stream variants directly; the
+  // tool_start/tool_end/settled variants are also emitted by runTurn
+  // itself. The catch is the seam contract, not fake convenience: a
+  // throwing subscriber never tears the loop (PiLoop.onEvent). Map
+  // iteration order is insertion order, so this delivers in subscription
+  // order (PiLoop.onEvent doc; pinned by the fan-out-order test).
+  emit(event: PiLoopEvent): void {
+    for (const listener of [...this.listeners.values()]) {
+      try {
+        listener(event);
+      } catch {
+        /* throwing subscriber swallowed — see onEvent contract */
+      }
+    }
+  }
+
+  // Cooperative abort per the PiLoop contract: flags the running turn and
+  // resolves when it settles — the in-flight dispatch is awaited (a parked
+  // approval keeps this pending, faithfully). Idle ⇒ resolves immediately
+  // and poisons nothing. Also drains the queued-steer table (item 3;
+  // mirrors clearQueue()) unconditionally — idle or not, nothing queued
+  // should survive an abort to fire a turn that skipped budgetGate().
+  abort(): Promise<void> {
+    this.queued = [];
+    if (this.inFlight === null) return Promise.resolve();
+    this.aborting = true;
+    return this.inFlight.then(
+      () => undefined,
+      () => undefined,
+    );
+  }
+
+  // Pending (not yet cleared) steer/followUp queue — what abort() above
+  // drains. Distinct from `steered`, which never shrinks.
+  get queuedSteers(): readonly string[] {
+    return this.queued;
+  }
+
+  // Real Pi: "Remove all listeners and disconnect from agent." Interface +
+  // fake only for K5a (PiLoop.dispose doc) — no Session consumer wired.
+  dispose(): void {
+    this.listeners.clear();
+  }
+
   private write(entry: Record<string, unknown>): string {
     this.entrySeq += 1;
     const id = `e-${String(this.entrySeq).padStart(6, "0")}`;
@@ -113,7 +337,15 @@ export class FakePiLoop implements PiLoop {
     return id;
   }
 
-  private async runTurn(text: string, dispatch: Dispatch): Promise<TurnResult> {
+  // Item 4c: this used to be wrapped in its own re-entry guard mirroring
+  // runTurn's, but driveTurn has exactly one call site (runTurn, below) and
+  // runTurn already refuses re-entry SYNCHRONOUSLY, before `inFlight` is
+  // even touched — so a second guard here could never fire. Two guards for
+  // one invariant (item 8: the real loop can never be mid-turn twice —
+  // Dispatch's mutable current-dispatch binding depends on it) just risks
+  // confusing which one an adapter author should replicate; runTurn's is
+  // the one that matters.
+  private async driveTurn(text: string, dispatch: Dispatch): Promise<TurnResult> {
     const turn = this.turns.shift();
     if (turn === undefined) {
       throw new Error("FakePiLoop: no scripted turn queued");
@@ -121,31 +353,119 @@ export class FakePiLoop implements PiLoop {
     let lastEntryId: string | null = this.write({ kind: "user", text });
     if (turn.concurrent === true) {
       // Executors Promise.all: a parked call must not block its siblings.
-      const outcomes = await Promise.all(turn.calls.map((call) => dispatch(call)));
+      // Under abort everything here has already STARTED, so all are
+      // awaited (pi-internals: "stops queuing but awaits started tools").
+      // tool_execution_end fires in COMPLETION order, independent of
+      // source position (pi core types.d.ts ToolExecutionMode "parallel":
+      // "emitted in tool completion order after each tool is finalized,
+      // while tool-result message artifacts are emitted later in
+      // assistant source order") — only the message artifacts (our JSONL
+      // "toolCall" rows below) stay source-ordered. Emitting tool_end
+      // from outcomes.entries() index order was the double being SAFER
+      // than Pi — the dangerous direction: a watcher pairing the n-th
+      // start with the n-th end passes every test here and then inverts
+      // against real Pi. Attach a per-call .then() so each tool_end fires
+      // as ITS OWN dispatch settles, not when the batch does.
+      const dispatched = turn.calls.map((c) => {
+        this.emit({ kind: "tool_start", toolCallId: c.toolCallId, tool: c.tool });
+        const settled = dispatch(c);
+        // Item 1: `.then(onFulfilled)` with no onRejected returns a NEW
+        // (derived) promise that rejects right along with `settled` when
+        // dispatch rejects (decideCall()/append() can both throw) — and
+        // nothing else holds that derived promise, so its rejection is
+        // unhandled even though `settled` itself is awaited below via
+        // Promise.all. The no-op second arg keeps the derived promise from
+        // ever rejecting; the original rejection still propagates through
+        // `settled` → Promise.all → this function's caller unchanged.
+        settled.then(
+          (outcome) => {
+            this.emit({ kind: "tool_end", toolCallId: c.toolCallId, tool: c.tool, ok: outcome.ok });
+          },
+          () => {},
+        );
+        return settled;
+      });
+      const outcomes = await Promise.all(dispatched);
       for (const [i, outcome] of outcomes.entries()) {
         this.outcomes.push(outcome);
         const call = turn.calls[i];
+        // Message artifacts (unlike tool_end above) stay source-ordered.
         lastEntryId = this.write({ kind: "toolCall", tool: call?.tool, ok: outcome.ok });
       }
     } else {
       for (const call of turn.calls) {
+        // Cooperative abort: calls not yet started are skipped; the one
+        // in flight below is awaited, never discarded.
+        if (this.aborting) break;
+        this.emit({ kind: "tool_start", toolCallId: call.toolCallId, tool: call.tool });
         const outcome = await dispatch(call);
         this.outcomes.push(outcome);
+        this.emit({
+          kind: "tool_end",
+          toolCallId: call.toolCallId,
+          tool: call.tool,
+          ok: outcome.ok,
+        });
         lastEntryId = this.write({ kind: "toolCall", tool: call.tool, ok: outcome.ok });
       }
     }
     this.accumulate(turn.stats);
+    if (this.aborting) {
+      // pi-internals: the run is finalized with a synthetic assistant
+      // stopReason "aborted" — entries durably on disk, hence
+      // checkpointSafe: true (see TurnResult).
+      lastEntryId = this.write({ kind: "turnEnd", aborted: true });
+      this.emit({ kind: "settled" });
+      return { checkpointSafe: true, lastEntryId, exitReason: "aborted" };
+    }
     lastEntryId = this.write({ kind: "turnEnd" });
     if (turn.crashBeforeReturn === true) {
+      // A dead process never settles — no `settled` on this path.
       throw new Error("FakePiLoop: simulated crash between turn end and checkpoint");
     }
+    // agent_settled fires whenever the loop goes fully idle — failed runs
+    // included (settle ≠ success).
+    this.emit({ kind: "settled" });
     return {
-      ok: turn.fail !== true,
+      checkpointSafe: turn.fail !== true,
       lastEntryId,
       exitReason: turn.fail === true ? "error" : "completed",
     };
   }
 
+  private runTurn(text: string, dispatch: Dispatch): Promise<TurnResult> {
+    // The real loop can never run two turns at once (item 8: Dispatch's
+    // mutable current-dispatch binding is only sound because start()/
+    // followUp() never overlap). Refuse synchronously, before touching
+    // `inFlight`, so a re-entrant call can never clobber the identity the
+    // `.finally` below depends on — refusing beats silently modeling a
+    // state Pi cannot reach.
+    if (this.inFlight !== null) {
+      throw new Error("FakePiLoop: cannot start a turn while one is already in flight");
+    }
+    // Defer the drive one microtask so inFlight is set before the first
+    // dispatch runs — an abort() issued from inside a dispatched tool must
+    // see the running turn, not an idle loop (the real prompt() is async
+    // and never runs the loop synchronously either).
+    const run = Promise.resolve()
+      .then(() => this.driveTurn(text, dispatch))
+      .finally(() => {
+        // Guard by IDENTITY, not "clear inFlight unconditionally": the bug
+        // this pins — with two overlapping turns, turn 1's finally
+        // clearing turn 2's slot so abort() resolves claiming idle while
+        // turn 2 keeps dispatching — is refused above already, but this
+        // stays a defensive backstop against a stale finally clearing a
+        // LATER turn's slot/abort flag out from under it.
+        if (this.inFlight === run) {
+          this.inFlight = null;
+          // Abort applies to the current operation only — never the next turn.
+          this.aborting = false;
+        }
+      });
+    this.inFlight = run;
+    return run;
+  }
+
   start(prompt: string, dispatch: Dispatch): Promise<TurnResult> {
     return this.runTurn(prompt, dispatch);
   }
@@ -154,8 +474,20 @@ export class FakePiLoop implements PiLoop {
     return this.runTurn(text, dispatch);
   }
 
+  // Throws when idle (item 3, PiLoop.steer contract above): `inFlight` is
+  // non-null for exactly the lifetime of a running turn, so this is the
+  // same "is a turn running" test abort()/runTurn() use. Mid-turn steering
+  // still queues as before.
   steer(text: string): void {
+    if (this.inFlight === null) {
+      throw new Error(
+        "FakePiLoop: steer() called with no turn in flight — real " +
+          "sendUserMessage always triggers a turn; the caller must use " +
+          "followUp() instead so the turn passes through budgetGate()",
+      );
+    }
     this.steered.push(text);
+    this.queued.push(text);
   }
 
   private stats: SessionStats = {
diff --git a/core/src/session/recovery.test.ts b/core/src/session/recovery.test.ts
index 677e25e..677c9e9 100644
--- a/core/src/session/recovery.test.ts
+++ b/core/src/session/recovery.test.ts
@@ -45,7 +45,7 @@ describe("crash/resume golden", () => {
       calls: [{ toolCallId: "t1", tool: "read_file", args: { path: `${fx.proj}/src/a.ts` } }],
     });
     const turn1 = await s.runTurn("first");
-    expect(turn1.ok).toBe(true);
+    expect(turn1.checkpointSafe).toBe(true);
     const checkpointed = writer.resumeCursor(taskId);
     expect(checkpointed?.lastEntryId).toBe(turn1.lastEntryId);
 
@@ -81,7 +81,7 @@ describe("crash/resume golden", () => {
     const cont = await s2.followUp(
       "the previous turn was interrupted by a restart; continue or summarize",
     );
-    expect(cont.ok).toBe(true);
+    expect(cont.checkpointSafe).toBe(true);
     const cursor = w2.resumeCursor(taskId);
     expect(cursor?.lastEntryId).not.toBe(turn1.lastEntryId);
     expect(cursor && turn1.lastEntryId && cursor.lastEntryId > turn1.lastEntryId).toBe(true);
diff --git a/core/src/session/runtime.test.ts b/core/src/session/runtime.test.ts
index 6f943a2..30fcf68 100644
--- a/core/src/session/runtime.test.ts
+++ b/core/src/session/runtime.test.ts
@@ -75,7 +75,7 @@ describe("foreground session with approvals (golden)", () => {
       scope: "task",
     });
     const result = await turn;
-    expect(result.ok).toBe(true);
+    expect(result.checkpointSafe).toBe(true);
 
     const events = writer.envelopes(taskId);
     const t = types(events);
@@ -112,7 +112,7 @@ describe("foreground session with approvals (golden)", () => {
       calls: [{ toolCallId: "t3", tool: "write_file", args: { path: `${fx.proj}/docs/b.md` } }],
     });
     const second = await session.runTurn("again");
-    expect(second.ok).toBe(true);
+    expect(second.checkpointSafe).toBe(true);
     const after = writer.envelopes(taskId);
     const auto = after.filter(
       (e) => e.type === "approval.resolved" && e.data.decision === "auto_approved",
@@ -204,7 +204,7 @@ describe("foreground session with approvals (golden)", () => {
       calls: [{ toolCallId: "p2", tool: "write_file", args: { path: `${fx.proj}/src/q.ts` } }],
     });
     const r = await next.runTurn("write again");
-    expect(r.ok).toBe(true);
+    expect(r.checkpointSafe).toBe(true);
     const events = writer.envelopes(nextTask);
     expect(types(events)).not.toContain("approval.requested");
     expect(events.find((e) => e.type === "approval.resolved")?.data.decidedBy).toBe("ledger");
@@ -221,7 +221,7 @@ describe("foreground session with approvals (golden)", () => {
 
     loop.queueTurn({ calls: [], fail: true });
     const failed = await session.runTurn("failing turn");
-    expect(failed.ok).toBe(false);
+    expect(failed.checkpointSafe).toBe(false);
     // Cursor did NOT advance to the failed turn's entries.
     expect(writer.resumeCursor(taskId)?.lastEntryId).toBe(first.lastEntryId);
   });
diff --git a/core/src/session/runtime.ts b/core/src/session/runtime.ts
index 19471ff..be177d9 100644
--- a/core/src/session/runtime.ts
+++ b/core/src/session/runtime.ts
@@ -146,9 +146,10 @@ export class Session {
     return this.budgetTracker.refuseTurn((t, d) => this.append(t, d));
   }
 
-  // Turn drivers. Checkpoint after each SUCCESSFUL turn only — never a
-  // failed one (poisoned-cursor guard, event-log §5.2). A loop crash
-  // propagates without checkpointing.
+  // Turn drivers. Checkpoint whenever the result is checkpoint-safe
+  // (TurnResult.checkpointSafe) — an ABORTED turn included, a FAILED one
+  // never (poisoned-cursor guard, event-log §5.2). A loop crash propagates
+  // without checkpointing.
   async runTurn(prompt: string): Promise<TurnResult> {
     return this.budgetGate() ?? this.finishTurn(await this.loop.start(prompt, this.dispatch));
   }
@@ -157,16 +158,44 @@ export class Session {
     return this.budgetGate() ?? this.finishTurn(await this.loop.followUp(text, this.dispatch));
   }
 
+  // Forwards unconditionally — surface-facing handling of an idle-steer
+  // rejection is the M2 barge-in work item, not this seam. The seam itself
+  // rejects an idle steer (PiLoop.steer contract, item 3): FakePiLoop
+  // throws, and the real adapter must reject/drop before ever calling
+  // Pi's sendUserMessage while idle.
   steer(text: string): void {
     this.loop.steer(text);
   }
 
+  // Session abort — the grants §7.3 watcher's breach action and M2 voice
+  // barge-in land here. NOT teardown (piloop.ts PiLoop.dispose() is the
+  // resource-release path; conflating the two is a leak vector given
+  // pi-internals' process-global hazards). Parked approvals are
+  // cancelled FIRST (grants §6.1: `cancelled` fires on session abort — a
+  // parked call cannot outlive its session; §10.19): each held promise
+  // settles as an error result to the model, so the loop's dispatch
+  // promises can drain. Cancelling after the loop.abort() await would
+  // deadlock — the cooperative abort resolves only at idle, and the loop
+  // cannot reach idle while a dispatch promise is unsettled. Then the
+  // in-flight runTurn/followUp resolves through finishTurn with exitReason
+  // "aborted" and checkpointSafe: true, so the aborted turn records turn.ended +
+  // cost.recorded (it burned tokens — DESIGN §8) and DOES checkpoint: its
+  // JSONL entries are durably finalized (pi-internals: synthetic
+  // stopReason "aborted"), so the cursor is valid — the §5.2 poisoned-
+  // cursor guard is about FAILED turns. Cooperative: resolves at idle,
+  // after in-flight tools settle; the watcher wraps this in its own
+  // timeout (piloop.ts contract).
+  async abort(): Promise<void> {
+    this.approvals.cancelAll((t, d) => this.append(t, d));
+    return this.loop.abort();
+  }
+
   private finishTurn(result: TurnResult): TurnResult {
     // §10.13 / event-log §3.1: every completed turn — failed ones included
     // (a failed turn still counts against budget, §5.1 rule 3) — records
     // turn.ended and its cost delta. A crash mid-turn throws before this.
     this.budgetTracker.recordTurn((t, d) => this.append(t, d), this.loop, result);
-    if (result.ok && result.lastEntryId !== null) {
+    if (result.checkpointSafe && result.lastEntryId !== null) {
       this.writer.checkpoint(this.taskStreamId, this.actor, {
         piSessionPath: this.loop.jsonlPath,
         lastEntryId: result.lastEntryId,
diff --git a/core/src/surface/generated/protocol.ts b/core/src/surface/generated/protocol.ts
index cd7ce83..397c7aa 100644
--- a/core/src/surface/generated/protocol.ts
+++ b/core/src/surface/generated/protocol.ts
@@ -6,27 +6,93 @@
 
 export type AdminPurgeBody = { scope: "all" };
 export type AdminPurgeDoneBody = { systemPurgeRowid: number };
-export type ConsentApprovalCardBody = { consentId: string; shownHash: string; toolCallId: string; tool: string; capabilityFamily: string | null; argsPreview: string; pendingCommands: string[]; sessionId: string; taskId: string; label: string; promotable?: boolean; };
-export type ConsentCancelledBody = { consentId: string; reason: "resolved_elsewhere" | "session_ended" | "recovery" | "superseded"; };
-export type ConsentJobEnableBody = { consentId: string; shownHash: string; jobKey: string; defHash: string; jobDef: unknown; paramNeeds: unknown; };
-export type ConsentPreviewCardBody = { consentId: string; shownHash: string; taskId: string; snapshot: unknown };
-export type ConsentTrustPromptBody = { consentId: string; shownHash: string; extensionName: string; version: string; manifestSummary: string; manifestHash: string; contentHash: string; };
+export type ConsentApprovalCardBody = {
+  consentId: string;
+  shownHash: string;
+  toolCallId: string;
+  tool: string;
+  capabilityFamily: string | null;
+  argsPreview: string;
+  pendingCommands: string[];
+  sessionId: string;
+  taskId: string;
+  label: string;
+  promotable?: boolean;
+};
+export type ConsentCancelledBody = {
+  consentId: string;
+  reason: "resolved_elsewhere" | "session_ended" | "recovery" | "superseded";
+};
+export type ConsentJobEnableBody = {
+  consentId: string;
+  shownHash: string;
+  jobKey: string;
+  defHash: string;
+  jobDef: unknown;
+  paramNeeds: unknown;
+};
+export type ConsentPreviewCardBody = {
+  consentId: string;
+  shownHash: string;
+  taskId: string;
+  snapshot: unknown;
+};
+export type ConsentTrustPromptBody = {
+  consentId: string;
+  shownHash: string;
+  extensionName: string;
+  version: string;
+  manifestSummary: string;
+  manifestHash: string;
+  contentHash: string;
+};
 export type CoreStatusBody = { state: "recovering" | "ready" | "degraded"; detail?: string };
-export type DecisionApprovalBody = { consentId: string; shownHash: string; decision: "approved" | "denied"; scope: "once" | "task" | "persistent"; promote?: boolean; approvedCommands?: string[]; };
-export type DecisionJobEnableBody = { consentId: string; shownHash: string; approve: boolean; params?: unknown; };
-export type DecisionPreviewBody = { consentId: string; shownHash: string; accept: boolean; scope?: "run" | "persistent"; requestEdit?: unknown; };
+export type DecisionApprovalBody = {
+  consentId: string;
+  shownHash: string;
+  decision: "approved" | "denied";
+  scope: "once" | "task" | "persistent";
+  promote?: boolean;
+  approvedCommands?: string[];
+};
+export type DecisionJobEnableBody = {
+  consentId: string;
+  shownHash: string;
+  approve: boolean;
+  params?: unknown;
+};
+export type DecisionPreviewBody = {
+  consentId: string;
+  shownHash: string;
+  accept: boolean;
+  scope?: "run" | "persistent";
+  requestEdit?: unknown;
+};
 export type DecisionTrustBody = { consentId: string; shownHash: string; accept: boolean };
 export type Envelope = { id: string; type: string; replyTo?: string; body: unknown };
 export type FocusSetBody = { sessionId: string | null };
 export type HelloBody = { schemaVersion: number; appVersion: string; lastRowid?: number };
 export type LogEventBody = { rowid: number; envelope: unknown };
 export type LogReplayDoneBody = { switchoverRowid: number };
-export type PanelActionBody = { panelId: string; actionId: string; context?: { editedBody?: string; itemId?: string }; };
-export type PanelActionRejectedBody = { panelId: string; actionId: string; reason: "session_ended" | "extension_disabled" | "unknown_action" | "stale_edit"; };
+export type PanelActionBody = {
+  panelId: string;
+  actionId: string;
+  context?: { editedBody?: string; itemId?: string };
+};
+export type PanelActionRejectedBody = {
+  panelId: string;
+  actionId: string;
+  reason: "session_ended" | "extension_disabled" | "unknown_action" | "stale_edit";
+};
 export type PanelDismissedBody = { panelId: string };
 export type PixelPoint = { x: number; y: number };
 export type PixelRect = { x: number; y: number; w: number; h: number };
 export type ProtocolErrorBody = { message: string; missingRef?: string };
-export type SessionAnnounceBody = { sessionId: string; taskId: string; label: string; profile: string; };
+export type SessionAnnounceBody = {
+  sessionId: string;
+  taskId: string;
+  label: string;
+  profile: string;
+};
 export type SessionRenameBody = { sessionId: string; label: string };
 export type UpdateRequiredBody = { minAppVersion: string; minCoreVersion: string; message: string };
diff --git a/scripts/arch-checks.mjs b/scripts/arch-checks.mjs
index 97f7d92..818d855 100644
--- a/scripts/arch-checks.mjs
+++ b/scripts/arch-checks.mjs
@@ -92,6 +92,23 @@ const RULES = [
     reason: "only core/src/surface/wsserver.ts may import the ws library — one WS entry point",
     spec: "specs/surface-protocol.md §1",
   },
+  {
+    // INVARIANTS.md row 14 (turn ≠ ambient): a module holding the PiLoop can
+    // call start/followUp directly and mint a turn that never passes
+    // budgetGate() on the way in or finishTurn() on the way out — no
+    // turn.ended, no cost.recorded, no checkpoint, past an exhausted budget.
+    // Confining the seam's import to core/src/session/ means only the module
+    // that owns Session can name a loop at all; outside consumers (surface,
+    // scheduler, extensions) drive turns through Session. The composition
+    // root that constructs a loop is the expected future exception — add it
+    // here deliberately when it lands, don't widen the glob.
+    name: "turn-driving-stays-in-session",
+    forbidden: [{ from: "core/src/**", to: "**/session/piloop*" }],
+    allow: ["core/src/session/**"],
+    reason:
+      "only core/src/session/ may import the PiLoop seam — every Pi run is driven through Session.runTurn/followUp (budget gate in, finishTurn out)",
+    spec: "INVARIANTS.md row 14; specs/grants.md §7.3",
+  },
 ];
 
 // Textual (non-import) rules: {name, forbidden: [{from: path glob,
@@ -262,6 +279,24 @@ function run() {
 
 function selfTest() {
   const cases = [
+    {
+      name: "violation: a surface module imports the PiLoop seam (invariant 14)",
+      file: "core/src/surface/server.ts",
+      source: 'import type { PiLoop } from "../session/piloop.js";\n',
+      expectViolations: 1,
+    },
+    {
+      name: "ok: session/ owns the seam",
+      file: "core/src/session/runtime.ts",
+      source: 'import type { PiLoop } from "./piloop.js";\n',
+      expectViolations: 0,
+    },
+    {
+      name: "ok: outside modules drive turns through the session public surface",
+      file: "core/src/surface/server.ts",
+      source: 'import type { Session } from "../session/index.js";\n',
+      expectViolations: 0,
+    },
     {
       name: "violation: non-db module imports better-sqlite3",
       file: "core/src/eventlog/writer.ts",
diff --git a/scripts/gen-protocol.mjs b/scripts/gen-protocol.mjs
index 5f2cea2..a7563eb 100644
--- a/scripts/gen-protocol.mjs
+++ b/scripts/gen-protocol.mjs
@@ -20,12 +20,14 @@
 // Usage: node scripts/gen-protocol.mjs [--out <path>] [--ts <path>] [--self-test]
 // Default output: client/Packages/GillyProtocol/Sources/Generated/ProtocolTypes.swift
 
-import { readFileSync, writeFileSync, mkdirSync } from "node:fs";
+import { execFileSync } from "node:child_process";
+import { mkdirSync, readFileSync, writeFileSync } from "node:fs";
 import { dirname, join } from "node:path";
 import { fileURLToPath } from "node:url";
 
 const ROOT = join(dirname(fileURLToPath(import.meta.url)), "..");
 const SCHEMA = join(ROOT, "specs", "protocol", "schema.ts");
+const BIOME_BIN = join(ROOT, "core", "node_modules", ".bin", "biome");
 const DEFAULT_OUT = join(
   ROOT,
   "client",
@@ -47,13 +49,58 @@ function fail(msg) {
 // keyword slipping through produces Swift that does not compile, or
 // worse, compiles to something else.
 const SWIFT_KEYWORDS = new Set([
-  "default", "case", "in", "repeat", "class", "struct", "enum", "protocol",
-  "extension", "func", "var", "let", "if", "else", "for", "while", "return",
-  "import", "public", "private", "internal", "static", "init", "self", "Self",
-  "Type", "associatedtype", "operator", "where", "guard", "defer", "do",
-  "catch", "throw", "throws", "try", "as", "is", "nil", "true", "false",
-  "super", "switch", "break", "continue", "fallthrough", "subscript",
-  "typealias", "inout", "indirect", "any", "some",
+  "default",
+  "case",
+  "in",
+  "repeat",
+  "class",
+  "struct",
+  "enum",
+  "protocol",
+  "extension",
+  "func",
+  "var",
+  "let",
+  "if",
+  "else",
+  "for",
+  "while",
+  "return",
+  "import",
+  "public",
+  "private",
+  "internal",
+  "static",
+  "init",
+  "self",
+  "Self",
+  "Type",
+  "associatedtype",
+  "operator",
+  "where",
+  "guard",
+  "defer",
+  "do",
+  "catch",
+  "throw",
+  "throws",
+  "try",
+  "as",
+  "is",
+  "nil",
+  "true",
+  "false",
+  "super",
+  "switch",
+  "break",
+  "continue",
+  "fallthrough",
+  "subscript",
+  "typealias",
+  "inout",
+  "indirect",
+  "any",
+  "some",
 ]);
 
 // Backtick-escape a field name for Swift emission when it collides with
@@ -69,8 +116,7 @@ function parseSchema(source) {
 
   // Every remaining non-whitespace content must be consumed by the
   // restricted grammar: `export type Name = { field: number; ... };`
-  const aliasRe =
-    /export\s+type\s+([A-Za-z_$][A-Za-z0-9_$]*)\s*=\s*\{([^{}]*)\}\s*;/g;
+  const aliasRe = /export\s+type\s+([A-Za-z_$][A-Za-z0-9_$]*)\s*=\s*\{([^{}]*)\}\s*;/g;
 
   const types = [];
   let consumed = stripped;
@@ -78,7 +124,9 @@ function parseSchema(source) {
   while ((match = aliasRe.exec(stripped)) !== null) {
     const [full, name, body] = match;
     if (name.startsWith("$")) {
-      fail(`type ${name}: "$"-prefixed identifiers are FORBIDDEN — "$" has no clean Swift spelling (property-wrapper projections). Rename the type in schema.ts.`);
+      fail(
+        `type ${name}: "$"-prefixed identifiers are FORBIDDEN — "$" has no clean Swift spelling (property-wrapper projections). Rename the type in schema.ts.`,
+      );
     }
     const fields = [];
     const fieldSrc = body.trim();
@@ -88,11 +136,15 @@ function parseSchema(source) {
         if (piece === "") continue;
         const fieldMatch = /^([A-Za-z_$][A-Za-z0-9_$]*)\s*:\s*number$/.exec(piece);
         if (!fieldMatch) {
-          fail(`type ${name}: unsupported field declaration "${piece}" — generator only understands "name: number"`);
+          fail(
+            `type ${name}: unsupported field declaration "${piece}" — generator only understands "name: number"`,
+          );
         }
         const field = fieldMatch[1];
         if (field.startsWith("$")) {
-          fail(`type ${name}: field "${field}" — "$"-prefixed identifiers are FORBIDDEN ("$" has no clean Swift spelling; property-wrapper projections). Rename the field in schema.ts.`);
+          fail(
+            `type ${name}: field "${field}" — "$"-prefixed identifiers are FORBIDDEN ("$" has no clean Swift spelling; property-wrapper projections). Rename the field in schema.ts.`,
+          );
         }
         fields.push(field);
       }
@@ -120,9 +172,7 @@ function render(types) {
     .map(({ name, fields }) => {
       const props = fields.map((f) => `  public var ${swiftIdent(f)}: Int`).join("\n");
       const params = fields.map((f) => `${swiftIdent(f)}: Int`).join(", ");
-      const assigns = fields
-        .map((f) => `    self.${swiftIdent(f)} = ${swiftIdent(f)}`)
-        .join("\n");
+      const assigns = fields.map((f) => `    self.${swiftIdent(f)} = ${swiftIdent(f)}`).join("\n");
       return `public struct ${name}: Sendable, Hashable, Codable {
 ${props}
 
@@ -183,6 +233,26 @@ function swiftSubsetSource(source) {
     .join("\n");
 }
 
+// Run generated TS through biome's own formatter (item 14: the committed
+// generated file must stay under the SAME lint/assist gate as everything
+// else — biome.json no longer excludes src/surface/generated/**, so what
+// this writes must already match what `biome check` expects). The virtual
+// filename is stable and repo-root-relative regardless of the actual
+// --ts output path (which may be a throwaway tmp file, e.g. the parity
+// test) so config resolution (biome.json at ROOT) is unaffected by where
+// the caller happens to write the result.
+function formatTs(text) {
+  return execFileSync(
+    BIOME_BIN,
+    ["format", "--stdin-file-path=core/src/surface/generated/protocol.ts"],
+    {
+      cwd: ROOT,
+      input: text,
+      encoding: "utf8",
+    },
+  );
+}
+
 function renderTs(decls) {
   const body = decls.map((d) => `export type ${d.name} = ${d.rhs};`).join("\n");
   return `// DO NOT EDIT.
@@ -209,8 +279,7 @@ function selfTest() {
     {
       name: "every listed Swift keyword is escaped in props",
       schema: `export type Kw = { ${[...SWIFT_KEYWORDS].map((k) => `${k}: number`).join("; ")} };`,
-      check: (out) =>
-        [...SWIFT_KEYWORDS].every((k) => out.includes(`public var \`${k}\`: Int`)),
+      check: (out) => [...SWIFT_KEYWORDS].every((k) => out.includes(`public var \`${k}\`: Int`)),
     },
     {
       name: "$-prefixed field is rejected",
@@ -241,7 +310,7 @@ function selfTest() {
       // A rich (non-number-record) type coexists in schema.ts: Swift skips it.
       name: "swift subset keeps only number-records; rich types dropped",
       schema:
-        'export type PixelRect = { x: number; y: number };\nexport type FocusSetBody = { sessionId: string | null };',
+        "export type PixelRect = { x: number; y: number };\nexport type FocusSetBody = { sessionId: string | null };",
       check: (out) => out.includes("struct PixelRect") && !out.includes("FocusSetBody"),
       renderSwiftFromSubset: true,
     },
@@ -308,10 +377,12 @@ const tsOut = flagValue("--ts");
 
 try {
   if (tsOut !== null) {
-    // TS target: every alias, verbatim.
+    // TS target: every alias, verbatim, then through biome's formatter —
+    // item 14: the committed output must already satisfy the same
+    // formatter/linter the rest of the repo is gated on.
     const decls = splitTopLevelTypes(source);
     mkdirSync(dirname(tsOut), { recursive: true });
-    writeFileSync(tsOut, renderTs(decls));
+    writeFileSync(tsOut, formatTs(renderTs(decls)));
     console.error(`gen-protocol: wrote ${decls.length} TS type(s) to ${tsOut}`);
   } else {
     // Swift target: the pure-number-record subset only.
diff --git a/specs/event-log.md b/specs/event-log.md
index c01f96b..8b35167 100644
--- a/specs/event-log.md
+++ b/specs/event-log.md
@@ -125,7 +125,10 @@ their consent semantics)
 · `task.approved` {resolvedModel, toolNames, grants: [{family, params}],
 budget, estDuration} (the DESIGN §8 preview snapshot) · `task.started` ·
 `task.waiting` {cause: approval|ask_user, refId} · `task.resumed`
-{answerRef} · `task.reattached` (recovery only) · `task.completed`
+{answerRef, cause?: answered|cancelled} (amendment 2026-07-19 #2:
+`cancelled` is the abort path — the wait ended without an answer, so a fold
+consumer must not read a resume as a user response; absent ⇒ `answered`) ·
+`task.reattached` (recovery only) · `task.completed`
 {resultRef} · `task.failed` {reason, detail} · `task.cancelled` {reason:
 user_abort|stale_at_boot|timeout}.
 
@@ -152,7 +155,7 @@ unpaired `approval.resolved {auto_denied}` — see amendment #2 below) ·
 `approval.requested` {toolCallId, tool, capabilityFamily, argsHash,
 resolvedArgs, sessionId} · `approval.resolved` {toolCallId, argsHash,
 decision: approved|denied|cancelled|auto_approved|auto_denied, scope?,
-decidedBy: user|ledger|background_policy|recovery} — **every** §6-machine
+decidedBy: user|ledger|background_policy|recovery|abort} — **every** §6-machine
 outcome is logged, including automatic ones (round-1 A17); auto_* are
 appended as an unpaired resolved (no requested precedes them) ·
 `approval.stale_decision` {toolCallId, argsHash} (grants §6.2 "ignored
@@ -556,6 +559,21 @@ task.approved grants params, bounded write-args, task.compacted/renamed
 golden background sequence gains the synthetic approval + continuation
 turn; §10.11 gains task.* -excludes-step-types delivery case.
 
+Amendment batch (approved 2026-07-19, M1-K5a PiLoop-seam review): **#1**
+`approval.resolved.decidedBy` gains `abort` (§3.1). grants §6.1 mandates
+that session abort produce `cancelled` rows, but the vocabulary here was a
+closed list with no tag for that producer — the two locked specs were
+inconsistent. `recovery` would mislabel a live abort as the boot pass and
+`user` would be an audit lie when the §7.3 budget watcher is the aborter,
+so the enum gains one member rather than overloading an existing one.
+**#2** `task.resumed` gains `cause?: answered|cancelled` (§3.1).
+Cancelling the parked table on abort must un-stick the `waiting_for_user`
+fold state, and `task.resumed` is the only legal transition out of it
+(§4) — but bare, it reads as "the user answered" when nobody did. The
+discriminator keeps the audit trail honest; absent means `answered`, so
+every pre-amendment row stays valid. Both land with the K5a abort path
+(`ApprovalMachine.cancelAll`).
+
 Amendment batch (approved 2026-07-18 #2, M1-K3 review round 1): **#3a**
 `blockedAt` gains `registration` (step-1 unknown tool) and `liveness`
 (step-2 extension disabled or lost grant) — closed enum stays closed,
diff --git a/specs/recon/pi-types-0.80.10.md b/specs/recon/pi-types-0.80.10.md
new file mode 100644
index 0000000..1405d6e
--- /dev/null
+++ b/specs/recon/pi-types-0.80.10.md
@@ -0,0 +1,103 @@
+# Recon: Pi SDK type ground truth @ 0.80.10
+
+Transcribed July 19, 2026 from the published `.d.ts` of
+`@earendil-works/pi-coding-agent@0.80.10` (+ `pi-agent-core`, `pi-ai` at the
+same version). This is the exact surface the `PiLoop` seam wraps — pi.md /
+pi-internals.md are prose recon of the *source*; this file is the *typed API*
+we must not expose raw (minor = breaking, pi.md §⚠2 — pin `0.80.10` exactly).
+
+## Turn driving is not request/response
+
+`prompt(text, options?): Promise<void>` — resolves when the prompt is
+*accepted*, not when the turn is done; there is no TurnResult. Completion must
+be derived from events (`agent_settled`) plus `AgentState`
+(`pendingToolCalls`, `errorMessage`, `aborted`) / `waitForIdle()`. That is the
+structural reason the seam needs `onEvent`: without it an adapter cannot
+implement `start()/followUp(): Promise<TurnResult>` at all.
+
+`sendUserMessage(content, {deliverAs: "steer"|"followUp"}): Promise<void>` —
+steering is async too (our `steer()` is currently sync/void).
+
+## Abort is async
+
+`abort(): Promise<void>` — "Abort current operation and wait for agent to
+become idle." Cooperative (pi-internals): in-flight tools are awaited, not
+discarded; the run finalizes with a synthetic assistant message
+`stopReason: "aborted"`. A tool ignoring its signal hangs ⇒ the budget layer
+still needs its own timeout. Narrower aborts also exist and are NOT what we
+want for the seam: `abortCompaction()`, `abortRetry()`, `abortBranchSummary()`,
+`abortBash()`.
+
+## Subscription
+
+`subscribe(listener: (event: AgentSessionEvent) => void): () => void` —
+multiple listeners, returns an unsubscribe. `dispose()` tears the session down.
+
+## `AgentSessionEvent` = `AgentEvent` (with `agent_end` replaced) ∪ session events
+
+Core `AgentEvent` (pi-agent-core `types.d.ts`):
+
+| type | payload |
+|---|---|
+| `agent_start` | — |
+| `turn_start` | — |
+| `turn_end` | `message`, `toolResults` |
+| `message_start` / `message_end` | `message` |
+| `message_update` | `message`, `assistantMessageEvent` (the token-level delta) |
+| `tool_execution_start` | `toolCallId`, `toolName`, `args` |
+| `tool_execution_update` | + `partialResult` |
+| `tool_execution_end` | `toolCallId`, `toolName`, `result`, `isError` |
+| `agent_end` (session override) | `messages`, `willRetry` |
+
+Session-only additions: `agent_settled` (no payload — fires once when fully
+idle; the budget/audit close signal) · `queue_update {steering, followUp}` ·
+`compaction_start {reason}` / `compaction_end {reason, result, aborted,
+willRetry, errorMessage?}` · `auto_retry_start {attempt, maxAttempts, delayMs,
+errorMessage}` / `auto_retry_end {success, attempt, finalError?}` ·
+`entry_appended {entry}` · `session_info_changed` · `thinking_level_changed`.
+
+Note `agent_end` carries `willRetry` and `compaction_end`/`auto_retry_end`
+exist as pairs: an `agent_end` with `willRetry: true` is NOT the end of the
+run — only `agent_settled` is.
+
+## Token-level deltas: `AssistantMessageEvent` (pi-ai)
+
+`start` · `{text,thinking,toolcall}_start {contentIndex}` ·
+`{text,thinking,toolcall}_delta {contentIndex, delta}` ·
+`text_end/thinking_end {contentIndex, content}` ·
+`toolcall_end {contentIndex, toolCall}` · `done {reason: stop|length|toolUse,
+message}` · `error {reason: aborted|error, error}`.
+
+**Every event carries a full `partial` snapshot** — forward `type` +
+`contentIndex` + `delta` only (surface-protocol §7 `stream.delta` uses
+`blockIndex` = `contentIndex`); the one full message goes over
+`stream.message_end`. `toolcall_delta` is raw JSON-argument text, so it is
+model-authored data — treat as untrusted for display, never parse-and-act.
+
+## Stats — and the four fields Pi does NOT give us
+
+```ts
+interface SessionStats {           // agent-session.d.ts:150
+  sessionFile: string | undefined; sessionId: string;
+  userMessages: number; assistantMessages: number;
+  toolCalls: number; toolResults: number; totalMessages: number;
+  tokens: { input; output; cacheRead; cacheWrite; total: number };
+  cost: number; contextUsage?: ContextUsage;
+}
+```
+
+It aggregates over ALL session entries *including history compacted away*, so
+totals stay billing-accurate across compaction — consistent with the
+cumulative-snapshot diffing in `session/budget.ts`.
+
+**But our `SessionStats` (piloop.ts) is NOT this type.** Pi's has no `model`,
+no `route`, no `compactions`, no `retries`. The adapter must supply all four
+itself: `model`/`route` are OUR routing decision (Pi has no notion of a
+profile), and the `compactions`/`retries` counters must be accumulated by
+subscribing to `compaction_end` / `auto_retry_end` — they are NOT readable
+from a stats call. Anything that reasons "getSessionStats() already exposes
+the counters" is wrong (caught in K5a review B2). Note also `sessionFile` is
+`string | undefined`: `SessionManager.inMemory()` sessions have no JSONL path
+at all.
+
+`getContextUsage()` is live context pressure (separate concern).