PiLoop seam: abort() + typed event stream (M1-K5a)

.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

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)
 

biome.json

@@ -6,7 +6,8 @@
     "useIgnoreFile": true
   },
   "files": {
-    "ignoreUnknown": false
+    "ignoreUnknown": false,
+    "includes": ["**"]
   },
   "formatter": {
     "enabled": true,

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 {

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);
+  });
+});

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.

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(

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,15 +188,15 @@ 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.
     loop.queueTurn({ calls: [read("s2")] });
     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,22 +211,22 @@ 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);
   });
 
   it("background: an autonomous session at its maxToolCalls ceiling refuses a further turn too", async () => {
     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);
   });

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" };
   }
 }