From 217feed6d60c07791651b3749d48dcdbad6a83c2 Mon Sep 17 00:00:00 2001 From: Akash Aedavelli Date: Sun, 19 Jul 2026 22:00:50 -0500 Subject: [PATCH 1/9] recon: transcribe Pi SDK 0.80.10 typed surface for the PiLoop seam --- specs/recon/pi-types-0.80.10.md | 82 +++++++++++++++++++++++++++++++++ 1 file changed, 82 insertions(+) create mode 100644 specs/recon/pi-types-0.80.10.md 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..c5429e5 --- /dev/null +++ b/specs/recon/pi-types-0.80.10.md @@ -0,0 +1,82 @@ +# 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` — 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` at all. + +`sendUserMessage(content, {deliverAs: "steer"|"followUp"}): Promise` — +steering is async too (our `steer()` is currently sync/void). + +## Abort is async + +`abort(): Promise` — "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 + +`getSessionStats()` 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`. +`getContextUsage()` is live context pressure (separate concern). From e1ab11d2b269cbbd28a580911be0df865bb15d5a Mon Sep 17 00:00:00 2001 From: Akash Aedavelli Date: Sun, 19 Jul 2026 22:14:04 -0500 Subject: [PATCH 2/9] session: extend the PiLoop seam with abort() and a narrow typed onEvent stream (M1-K5a) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Unblocks the real Pi adapter and the deferred grants $7.3 hard budget layer without building either. abort() is cooperative Promise matching Pi 0.80.10 ground truth: in-flight tools are awaited, idle aborts resolve immediately, repeats are idempotent, and the in-flight turn resolves ok: true / exitReason "aborted" — an aborted run is durably finalized (synthetic stopReason), so the resume cursor advances; the event-log $5.2 poisoned-cursor guard stays scoped to failed turns. onEvent carries our own union (tool_start/tool_end/settled for the budget watcher; message_start/delta/end + tool_progress for the surface $7 stream bridge), never Pi's raw event types; each admitted variant names its consumer and each exclusion its reason. Sync fan-out in subscription order; a throwing subscriber is swallowed (same rule as the onActivity push seam). FakePiLoop implements both and auto-emits around dispatch; Session gains abort() and an oracle proving aborted-turn checkpoint semantics against the real consumer. jsonl.test's helper now awaits loop.start — the fake defers a microtask like the real async prompt(), so nothing may rely on synchronous JSONL writes. biome now skips src/surface/generated (the DO-NOT-EDIT emitted module tripped format checks; generated code is guarded by its generator and the round-trip oracle, not the linter). --- biome.json | 3 +- core/src/session/abort.test.ts | 105 ++++++++++++++++++++ core/src/session/budget.ts | 22 +++-- core/src/session/index.ts | 2 + core/src/session/jsonl.test.ts | 22 +++-- core/src/session/piloop.test.ts | 161 ++++++++++++++++++++++++++++++ core/src/session/piloop.ts | 170 +++++++++++++++++++++++++++++++- core/src/session/runtime.ts | 14 +++ 8 files changed, 477 insertions(+), 22 deletions(-) create mode 100644 core/src/session/abort.test.ts create mode 100644 core/src/session/piloop.test.ts diff --git a/biome.json b/biome.json index 03cb9ac..eef4c30 100644 --- a/biome.json +++ b/biome.json @@ -6,7 +6,8 @@ "useIgnoreFile": true }, "files": { - "ignoreUnknown": false + "ignoreUnknown": false, + "includes": ["**", "!**/src/surface/generated/**"] }, "formatter": { "enabled": true, diff --git a/core/src/session/abort.test.ts b/core/src/session/abort.test.ts new file mode 100644 index 0000000..1af018d --- /dev/null +++ b/core/src/session/abort.test.ts @@ -0,0 +1,105 @@ +// 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 ok: 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 } 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])); + +// 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((r) => { + release = r; + }); + return "slow:ok"; + }, +}); + +function foreground(loopName: string): { + session: Session; + taskId: string; + loop: ReturnType; +} { + 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((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.ok).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("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({ ok: true, exitReason: "completed" }); + expect(writer.resumeCursor(taskId)?.lastEntryId).toBe(r.lastEntryId); + }); +}); diff --git a/core/src/session/budget.ts b/core/src/session/budget.ts index 412d706..a5ed6be 100644 --- a/core/src/session/budget.ts +++ b/core/src/session/budget.ts @@ -3,16 +3,18 @@ // 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, its own timeout around abort (a tool +// ignoring its signal hangs, pi-internals), and the one salvage call at +// exhaustion (DESIGN §4) 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. import type { Envelope } from "../eventlog/index.js"; import type { PiLoop, SessionStats, TurnResult } from "./piloop.js"; diff --git a/core/src/session/index.ts b/core/src/session/index.ts index 429d82a..fd26e64 100644 --- a/core/src/session/index.ts +++ b/core/src/session/index.ts @@ -12,11 +12,13 @@ export { export { type Dispatch, 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..b74585c --- /dev/null +++ b/core/src/session/piloop.test.ts @@ -0,0 +1,161 @@ +// 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({ ok: true, exitReason: "completed" }); + }); + + it("abort mid-turn awaits the in-flight call, skips queued siblings, finalizes durably (ok: true, exitReason aborted)", async () => { + const loop = fx.newLoop("coop-abort"); + loop.queueTurn({ calls: [call("c1"), call("c2")] }); + let release!: () => void; + const gate = new Promise((r) => { + release = r; + }); + const dispatched: string[] = []; + let abortDone = false; + let abortPromise: Promise = 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.ok).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[] = []; + 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.ok).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.ok).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" }); + loop.emit({ kind: "message_delta", blockIndex: 0, deltaKind: "text", delta: "hel" }); + loop.emit({ kind: "message_delta", blockIndex: 1, deltaKind: "thinking", delta: "hmm" }); + loop.emit({ kind: "tool_progress", toolCallId: "x1", update: { pct: 50 } }); + loop.emit({ kind: "message_end", message: { role: "assistant" } }); + expect(seen.map((e) => e.kind)).toEqual([ + "message_start", + "message_delta", + "message_delta", + "tool_progress", + "message_end", + ]); + expect(seen[1]).toEqual({ + kind: "message_delta", + blockIndex: 0, + deltaKind: "text", + delta: "hel", + }); + expect(seen[4]).toEqual({ kind: "message_end", message: { role: "assistant" } }); + }); +}); diff --git a/core/src/session/piloop.ts b/core/src/session/piloop.ts index 82d9110..0750eab 100644 --- a/core/src/session/piloop.ts +++ b/core/src/session/piloop.ts @@ -19,6 +19,16 @@ export type ToolOutcome = { ok: true; result: unknown } | { ok: false; error: st export type Dispatch = (call: PiToolCall) => Promise; export type TurnResult = { + // ok = "the run finalized coherently and lastEntryId is a durable, + // checkpoint-safe cursor" — NOT "the model finished what was asked". + // An ABORTED turn is ok: true with exitReason "aborted": pi-internals, + // the run is finalized with a synthetic assistant stopReason "aborted" + // and its JSONL entries are durably on disk, so the resume cursor may + // advance. The poisoned-cursor guard (event-log §5.2 "never advanced on + // a failed turn") targets FAILED turns, whose trailing JSONL state is + // not trustworthy; refusing to checkpoint an aborted turn would instead + // leave the §5.1 truncation check blind to entries that are durably + // present — losing a valid cursor, the other half of the same bug. ok: 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. @@ -26,6 +36,50 @@ export type TurnResult = { exitReason: string; }; +// 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_*: getSessionStats() exposes the counters and +// budget.ts diffs snapshots per turn for includesCompaction/ +// includesRetry — no consumer needs the mid-turn events. +// · 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. + | { kind: "settled" } + // surface-protocol §7 stream bridge: message_start/delta/message_end/ + // tool_update. 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). + | { kind: "message_start" } + | { + kind: "message_delta"; + blockIndex: number; + deltaKind: "text" | "thinking" | "toolArgs"; + delta: string; + } + | { kind: "message_end"; message: unknown } + | { kind: "tool_progress"; 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 @@ -47,7 +101,35 @@ export interface PiLoop { readonly jsonlPath: string; start(prompt: string, dispatch: Dispatch): Promise; followUp(text: string, dispatch: Dispatch): Promise; + // 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 + // (or a parked approval inside dispatch) ignoring the abort hangs this + // promise, so the §7.3 watcher still needs its own timeout around it. + // Idempotent; resolves immediately when no turn is running; never + // rejects. The in-flight start()/followUp() resolves with exitReason + // "aborted" and ok: true (see TurnResult — the aborted run is durably + // finalized and checkpoint-safe). Consumers: the grants §7.3 budget + // backstop on breach, uninstall/teardown, M2 voice barge-in. + abort(): Promise; + // 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; } @@ -95,6 +177,9 @@ export class FakePiLoop implements PiLoop { private entrySeq: number; readonly outcomes: ToolOutcome[] = []; readonly steered: string[] = []; + private readonly listeners = new Set<(event: PiLoopEvent) => void>(); + private inFlight: Promise | null = null; + private aborting = false; constructor(readonly jsonlPath: string) { // Continue entry numbering across reattach — a resumed loop appends @@ -106,6 +191,38 @@ export class FakePiLoop implements PiLoop { this.turns.push(turn); } + onEvent(listener: (event: PiLoopEvent) => void): Unsubscribe { + this.listeners.add(listener); + return () => this.listeners.delete(listener); + } + + // 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). + emit(event: PiLoopEvent): void { + for (const listener of [...this.listeners]) { + 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. + abort(): Promise { + if (this.inFlight === null) return Promise.resolve(); + this.aborting = true; + return this.inFlight.then( + () => undefined, + () => undefined, + ); + } + private write(entry: Record): string { this.entrySeq += 1; const id = `e-${String(this.entrySeq).padStart(6, "0")}`; @@ -113,7 +230,7 @@ export class FakePiLoop implements PiLoop { return id; } - private async runTurn(text: string, dispatch: Dispatch): Promise { + private async driveTurn(text: string, dispatch: Dispatch): Promise { const turn = this.turns.shift(); if (turn === undefined) { throw new Error("FakePiLoop: no scripted turn queued"); @@ -121,24 +238,59 @@ 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. + // Under abort everything here has already STARTED, so all are + // awaited (pi-internals: "stops queuing but awaits started tools"). + for (const c of turn.calls) { + this.emit({ kind: "tool_start", toolCallId: c.toolCallId, tool: c.tool }); + } const outcomes = await Promise.all(turn.calls.map((call) => dispatch(call))); for (const [i, outcome] of outcomes.entries()) { this.outcomes.push(outcome); const call = turn.calls[i]; lastEntryId = this.write({ kind: "toolCall", tool: call?.tool, ok: outcome.ok }); + if (call !== undefined) { + this.emit({ + kind: "tool_end", + toolCallId: call.toolCallId, + 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 ok: true + // and a checkpoint-safe lastEntryId (see TurnResult). + lastEntryId = this.write({ kind: "turnEnd", aborted: true }); + this.emit({ kind: "settled" }); + return { ok: 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, lastEntryId, @@ -146,6 +298,22 @@ export class FakePiLoop implements PiLoop { }; } + private runTurn(text: string, dispatch: Dispatch): Promise { + // 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(() => { + 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 { return this.runTurn(prompt, dispatch); } diff --git a/core/src/session/runtime.ts b/core/src/session/runtime.ts index 19471ff..fc2b7da 100644 --- a/core/src/session/runtime.ts +++ b/core/src/session/runtime.ts @@ -161,6 +161,20 @@ export class Session { this.loop.steer(text); } + // Session abort — the grants §7.3 watcher's breach action, uninstall/ + // teardown, and M2 voice barge-in all land here. Pure delegation: the + // in-flight runTurn/followUp resolves through finishTurn with exitReason + // "aborted" and ok: 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 (including a parked approval) settle; the + // watcher wraps this in its own timeout (piloop.ts contract). + async abort(): Promise { + 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 From 6ede6e952e0917a93d3c8165c9300ca410d299ab Mon Sep 17 00:00:00 2001 From: Akash Aedavelli Date: Sun, 19 Jul 2026 22:21:24 -0500 Subject: [PATCH 3/9] ci: run check-all on Node 22 to match the engine core requires --- .github/workflows/check-all.yml | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) 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 From 3676f7258ae3da4a8c28c093c5924da30a95419c Mon Sep 17 00:00:00 2001 From: Akash Aedavelli Date: Sun, 19 Jul 2026 22:29:11 -0500 Subject: [PATCH 4/9] =?UTF-8?q?recon:=20transcribe=20the=20real=20SessionS?= =?UTF-8?q?tats=20shape=20=E2=80=94=20Pi=20supplies=20none=20of=20model/ro?= =?UTF-8?q?ute/compactions/retries?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- specs/recon/pi-types-0.80.10.md | 29 +++++++++++++++++++++++++---- 1 file changed, 25 insertions(+), 4 deletions(-) diff --git a/specs/recon/pi-types-0.80.10.md b/specs/recon/pi-types-0.80.10.md index c5429e5..1405d6e 100644 --- a/specs/recon/pi-types-0.80.10.md +++ b/specs/recon/pi-types-0.80.10.md @@ -74,9 +74,30 @@ message}` · `error {reason: aborted|error, error}`. `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 +## 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. -`getSessionStats()` 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`. `getContextUsage()` is live context pressure (separate concern). From f209213a5bbfc03c261318ac9dfab28b984579f1 Mon Sep 17 00:00:00 2001 From: Akash Aedavelli Date: Sun, 19 Jul 2026 22:34:55 -0500 Subject: [PATCH 5/9] =?UTF-8?q?session:=20cancel=20parked=20approvals=20on?= =?UTF-8?q?=20abort=20(grants=20=C2=A76.1/=C2=A710.19)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Session.abort() was pure delegation to loop.abort(), so a foreground session parked in awaiting_approval hung forever on abort: the dispatch promise never settled, the loop never reached idle, and the task stayed waiting_for_user with an unpaired approval.requested — a silent divergence from grants §6.1 ('cancelled fires on: session abort … a parked call cannot outlive its session') and the §10.19 oracle. ApprovalMachine.cancelAll now, per parked record: appends approval.resolved {decision: cancelled, decidedBy: abort} (the row shape recovery.ts already emits for the restart producer), drops the record, and settles the held promise as the §6.1 error-result-to-model arrow. One task.resumed closes the task.waiting the first park opened so the aborted turn's accounting rows fold from a coherent state. Session.abort cancels FIRST, then awaits loop.abort() — the reverse order would still deadlock, since the cooperative abort resolves only at idle. 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, is stale-logged, executes nothing, and mints nothing. The piloop.ts/runtime.ts contract comments had codified the hang as intended behavior; rewritten so the real Pi adapter isn't built against a spec-wrong contract. SPEC-NOTE in cancelAll flags the one amendment question: event-log §3.1's decidedBy vocabulary lacks a tag for the abort producer grants §6.1 requires — 'abort' extends it rather than mislabeling a live abort as the boot recovery pass. Oracle: abort.test.ts pins §10.19 end to end — cancelled row appended, promise settles as error, pendingApprovals empties, abort() resolves (raced against a timeout so a regression fails loudly instead of hanging the suite), late decision ignored with no ledger or store writes. --- core/src/eventlog/types.ts | 2 +- core/src/session/abort.test.ts | 64 +++++++++++++++++++++++++++++++++- core/src/session/approval.ts | 33 ++++++++++++++++++ core/src/session/piloop.ts | 9 +++-- core/src/session/runtime.ts | 13 +++++-- 5 files changed, 114 insertions(+), 7 deletions(-) diff --git a/core/src/eventlog/types.ts b/core/src/eventlog/types.ts index fab3d57..4b34c2c 100644 --- a/core/src/eventlog/types.ts +++ b/core/src/eventlog/types.ts @@ -244,7 +244,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" (see approval.ts cancelAll SPEC-NOTE) readonly scope?: "once" | "task" | "persistent"; } export interface ApprovalStaleDecisionData { diff --git a/core/src/session/abort.test.ts b/core/src/session/abort.test.ts index 1af018d..004e0d8 100644 --- a/core/src/session/abort.test.ts +++ b/core/src/session/abort.test.ts @@ -6,7 +6,7 @@ // (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 } from "./fixture.js"; +import { makeSessionFixture, waitForPending } from "./fixture.js"; import type { Session } from "./index.js"; import { createForegroundSession } from "./index.js"; @@ -15,6 +15,7 @@ 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 @@ -92,6 +93,67 @@ describe("Session.abort (grants §7.3 breach action / teardown / M2 barge-in)", 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((r) => setTimeout(() => r("timed out"), 2000)), + ]); + expect(settled).toBe("resolved"); + const result = await turn; + expect(result.ok).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", + }); + expect(events.filter((e) => e.type === "task.resumed")).toHaveLength(1); + 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(); diff --git a/core/src/session/approval.ts b/core/src/session/approval.ts index 66a2e86..5724084 100644 --- a/core/src/session/approval.ts +++ b/core/src/session/approval.ts @@ -176,6 +176,39 @@ 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. + // SPEC-NOTE (event-log §3.1): the decidedBy vocabulary there enumerates + // user|ledger|background_policy|recovery, but grants §6.1 names session + // abort as a `cancelled` producer without a tag for it — "abort" extends + // the vocabulary rather than mislabeling a live abort as the boot + // recovery pass. Flagged for spec amendment. + 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 }); + 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/piloop.ts b/core/src/session/piloop.ts index 0750eab..800205f 100644 --- a/core/src/session/piloop.ts +++ b/core/src/session/piloop.ts @@ -110,8 +110,13 @@ export interface PiLoop { 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 - // (or a parked approval inside dispatch) ignoring the abort hangs this - // promise, so the §7.3 watcher still needs its own timeout around it. + // 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. // Idempotent; resolves immediately when no turn is running; never // rejects. The in-flight start()/followUp() resolves with exitReason // "aborted" and ok: true (see TurnResult — the aborted run is durably diff --git a/core/src/session/runtime.ts b/core/src/session/runtime.ts index fc2b7da..7ee138a 100644 --- a/core/src/session/runtime.ts +++ b/core/src/session/runtime.ts @@ -162,16 +162,23 @@ export class Session { } // Session abort — the grants §7.3 watcher's breach action, uninstall/ - // teardown, and M2 voice barge-in all land here. Pure delegation: the + // teardown, and M2 voice barge-in all land here. 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 ok: 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 (including a parked approval) settle; the - // watcher wraps this in its own timeout (piloop.ts contract). + // after in-flight tools settle; the watcher wraps this in its own + // timeout (piloop.ts contract). async abort(): Promise { + this.approvals.cancelAll((t, d) => this.append(t, d)); return this.loop.abort(); } From 7aab4fc2a4ceb262b2dac3c935ff4dcb292259d7 Mon Sep 17 00:00:00 2001 From: Akash Aedavelli Date: Sun, 19 Jul 2026 22:55:29 -0500 Subject: [PATCH 6/9] PiLoop seam: fix tool_end ordering, turn re-entry, abort queue drain (K5a fixer 2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Correctness fixes from the two K5a adversarial reviews: - FakePiLoop's concurrent branch now emits tool_end in completion order (per-call .then()), not source order — the double was silently safer than real Pi, which inverts against a watcher that pairs start/end by index. - runTurn refuses re-entry synchronously, before touching `inFlight`, so a second overlapping turn can never clobber the identity the .finally guard depends on; driveTurn also refuses re-entry as a backstop. - abort() now drains the queued-steer table (mirrors Pi's clearQueue()), so a steer queued mid-turn can't fire a budget-gate-skipping turn after the abort. `steered` stays as the append-only audit record. - onEvent listeners are now token-keyed (Map) instead of a Set, so two subscriptions of the same function get independent unsubscribes. - Added a fan-out subscription-order oracle (previously unpinned). - Renamed tool_progress -> tool_update and added messageId to the message_start/delta/end variants (surface-protocol §7), documenting that the adapter mints and holds the id stable since Pi's AssistantMessage has none. Comment corrections (rationale was misleading the future adapter author, not the behavior): - The compaction/auto_retry exclusion cited a stats field (model/route/compactions/retries) that real SessionStats does not have — corrected against specs/recon/pi-types-0.80.10.md. - Documented that Pi binds tool execution at session construction, not per-turn, and that the seam's Dispatch is a mutable current-dispatch swap that only works because turns never overlap. - Added PiLoop.dispose() (interface + fake only) and stopped calling abort() "teardown" in both piloop.ts and runtime.ts. - Flagged the followUp() name collision with Pi's queueing method. - Reworded abort()'s idempotency/never-rejects clause as an adapter obligation, not a fact Pi documents. - Corrected budget.ts's SPEC-NOTE: budgetGate() has no bypass today, so the DESIGN §4 salvage call would be dropped, not "land with the adapter" for free. - Noted that FakePiLoop's settled fires once per turn while real agent_settled does not fire between a turn and a queued follow-up. Gate: dropped the biome.json generated-code exclusion (it silenced lint and assist, not just formatting) and instead run gen-protocol.mjs's TS output through biome's own formatter before writing it, so the committed file stays under the same gate as hand-written code. Regenerated core/src/surface/generated/protocol.ts through the new formatting step; parity test and self-test both still pass. --- biome.json | 2 +- core/src/session/budget.ts | 23 ++- core/src/session/piloop.test.ts | 167 ++++++++++++++++++- core/src/session/piloop.ts | 216 ++++++++++++++++++++----- core/src/session/runtime.ts | 6 +- core/src/surface/generated/protocol.ts | 88 ++++++++-- scripts/gen-protocol.mjs | 113 ++++++++++--- 7 files changed, 524 insertions(+), 91 deletions(-) diff --git a/biome.json b/biome.json index eef4c30..d0db5f6 100644 --- a/biome.json +++ b/biome.json @@ -7,7 +7,7 @@ }, "files": { "ignoreUnknown": false, - "includes": ["**", "!**/src/surface/generated/**"] + "includes": ["**"] }, "formatter": { "enabled": true, diff --git a/core/src/session/budget.ts b/core/src/session/budget.ts index a5ed6be..27e2352 100644 --- a/core/src/session/budget.ts +++ b/core/src/session/budget.ts @@ -8,13 +8,22 @@ // 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, its own timeout around abort (a tool -// ignoring its signal hangs, pi-internals), and the one salvage call at -// exhaustion (DESIGN §4) 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. +// 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"; diff --git a/core/src/session/piloop.test.ts b/core/src/session/piloop.test.ts index b74585c..c6059b8 100644 --- a/core/src/session/piloop.test.ts +++ b/core/src/session/piloop.test.ts @@ -138,24 +138,177 @@ describe("PiLoop.onEvent (narrow typed stream, surface-protocol §7 + grants §7 const loop = fx.newLoop("scripted"); const seen: PiLoopEvent[] = []; loop.onEvent((e) => seen.push(e)); - loop.emit({ kind: "message_start" }); - loop.emit({ kind: "message_delta", blockIndex: 0, deltaKind: "text", delta: "hel" }); - loop.emit({ kind: "message_delta", blockIndex: 1, deltaKind: "thinking", delta: "hmm" }); - loop.emit({ kind: "tool_progress", toolCallId: "x1", update: { pct: 50 } }); - loop.emit({ kind: "message_end", message: { role: "assistant" } }); + 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_progress", + "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", message: { role: "assistant" } }); + 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 + // 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((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"]); + }); +}); + +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((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.ok).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((r) => { + release = r; + }); + let abortPromise: Promise = 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([]); + }); +}); + +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 800205f..edd4d68 100644 --- a/core/src/session/piloop.ts +++ b/core/src/session/piloop.ts @@ -14,8 +14,15 @@ 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; export type TurnResult = { @@ -43,9 +50,15 @@ export type TurnResult = { // · 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_*: getSessionStats() exposes the counters and -// budget.ts diffs snapshots per turn for includesCompaction/ -// includesRetry — no consumer needs the mid-turn events. +// · 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 @@ -57,26 +70,39 @@ export type PiLoopEvent = | { 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. + // for the deferred budget watcher. Semantics differ from the double: + // FakePiLoop emits `settled` once per driveTurn (once per turn); real + // agent_settled fires only when Pi is FULLY idle — with a queued + // followUp pending, Pi does NOT settle between the turn that queued it + // and the one that drains it. A watcher counting one `settled` per turn + // against the real adapter will undercount turns. | { kind: "settled" } // surface-protocol §7 stream bridge: message_start/delta/message_end/ - // tool_update. 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). - | { kind: "message_start" } + // 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"; message: unknown } - | { kind: "tool_progress"; toolCallId: string; update: unknown }; + | { kind: "message_end"; messageId: string; message: unknown } + | { kind: "tool_update"; toolCallId: string; update: unknown }; export type Unsubscribe = () => void; @@ -100,6 +126,13 @@ export interface PiLoop { // in the resume cursor (event-log.md §5.2, §9). readonly jsonlPath: string; start(prompt: string, dispatch: Dispatch): Promise; + // 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; // Stays fire-and-forget `void` although the real sendUserMessage is // async: its promise resolves when the text is QUEUED, not acted on @@ -117,11 +150,24 @@ export interface PiLoop { // 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. - // Idempotent; resolves immediately when no turn is running; never - // rejects. The in-flight start()/followUp() resolves with exitReason - // "aborted" and ok: true (see TurnResult — the aborted run is durably - // finalized and checkpoint-safe). Consumers: the grants §7.3 budget - // backstop on breach, uninstall/teardown, M2 voice barge-in. + // 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 ok: 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; // Subscription over constructor-sink: the budget watcher and the surface // stream bridge are independent consumers with independent lifetimes — @@ -136,6 +182,12 @@ export interface PiLoop { // 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; } // --------------------------------------------------------------------------- @@ -181,10 +233,26 @@ 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[] = []; - private readonly listeners = new Set<(event: PiLoopEvent) => void>(); + // 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: 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 void>(); + private nextListenerToken = 0; private inFlight: Promise | null = null; private aborting = false; + // True for the duration of driveTurn's body — guards re-entry (item 2): + // the real loop can never run two turns at once (Dispatch's mutable + // current-dispatch binding is only sound because turns never overlap). + private driving = false; constructor(readonly jsonlPath: string) { // Continue entry numbering across reattach — a resumed loop appends @@ -197,16 +265,21 @@ export class FakePiLoop implements PiLoop { } onEvent(listener: (event: PiLoopEvent) => void): Unsubscribe { - this.listeners.add(listener); - return () => this.listeners.delete(listener); + 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). + // 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]) { + for (const listener of [...this.listeners.values()]) { try { listener(event); } catch { @@ -218,8 +291,11 @@ export class FakePiLoop implements PiLoop { // 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. + // 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 { + this.queued = []; if (this.inFlight === null) return Promise.resolve(); this.aborting = true; return this.inFlight.then( @@ -228,6 +304,18 @@ export class FakePiLoop implements PiLoop { ); } + // 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 { this.entrySeq += 1; const id = `e-${String(this.entrySeq).padStart(6, "0")}`; @@ -236,6 +324,23 @@ export class FakePiLoop implements PiLoop { } private async driveTurn(text: string, dispatch: Dispatch): Promise { + // Re-entry is already refused in runTurn (before inFlight is even + // touched — item 2), but guard the drive itself too: the real loop can + // never be mid-turn twice (item 8's mutable current-dispatch binding + // depends on it), and the double should refuse that state rather than + // silently model it wrongly. + if (this.driving) { + throw new Error("FakePiLoop: driveTurn re-entered while a turn is already in flight"); + } + this.driving = true; + try { + return await this.driveTurnBody(text, dispatch); + } finally { + this.driving = false; + } + } + + private async driveTurnBody(text: string, dispatch: Dispatch): Promise { const turn = this.turns.shift(); if (turn === undefined) { throw new Error("FakePiLoop: no scripted turn queued"); @@ -245,22 +350,31 @@ export class FakePiLoop implements PiLoop { // Executors Promise.all: a parked call must not block its siblings. // Under abort everything here has already STARTED, so all are // awaited (pi-internals: "stops queuing but awaits started tools"). - for (const c of turn.calls) { + // 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 outcomes = await Promise.all(turn.calls.map((call) => dispatch(call))); + const settled = dispatch(c); + 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 }); - if (call !== undefined) { - this.emit({ - kind: "tool_end", - toolCallId: call.toolCallId, - tool: call.tool, - ok: outcome.ok, - }); - } } } else { for (const call of turn.calls) { @@ -304,6 +418,15 @@ export class FakePiLoop implements PiLoop { } private runTurn(text: string, dispatch: Dispatch): Promise { + // 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 @@ -311,9 +434,17 @@ export class FakePiLoop implements PiLoop { const run = Promise.resolve() .then(() => this.driveTurn(text, dispatch)) .finally(() => { - this.inFlight = null; - // Abort applies to the current operation only — never the next turn. - this.aborting = false; + // 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; @@ -329,6 +460,7 @@ export class FakePiLoop implements PiLoop { steer(text: string): void { this.steered.push(text); + this.queued.push(text); } private stats: SessionStats = { diff --git a/core/src/session/runtime.ts b/core/src/session/runtime.ts index 7ee138a..52eb185 100644 --- a/core/src/session/runtime.ts +++ b/core/src/session/runtime.ts @@ -161,8 +161,10 @@ export class Session { this.loop.steer(text); } - // Session abort — the grants §7.3 watcher's breach action, uninstall/ - // teardown, and M2 voice barge-in all land here. Parked approvals are + // 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 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/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 ] [--ts ] [--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. From 91d736684cf6cc093264c489edb2009cef0af7e1 Mon Sep 17 00:00:00 2001 From: Akash Aedavelli Date: Sun, 19 Jul 2026 23:17:13 -0500 Subject: [PATCH 7/9] PiLoop seam: close K5a's last three findings (unhandled rejection, TurnResult naming, idle-steer hole) - Fix an unhandled rejection in the concurrent dispatch branch of FakePiLoop.driveTurn: the per-call .then() had no onRejected, so a rejecting dispatch (decideCall()/append() can both throw) leaked an unhandled rejection on the derived promise even though Promise.all still awaited and propagated it correctly. Regression test asserts the turn still rejects and zero unhandledRejection events fire. - Rename TurnResult.ok to checkpointSafe across every producer/consumer (piloop.ts, runtime.ts, budget.ts, FakePiLoop, all session tests) so the field name carries what it means (a durable, checkpoint-safe cursor - not "the model succeeded"); shrink the justifying comment now that the name does the work. Narrow exitReason to a closed ExitReason union and export it. - Close the steer-while-idle hole: document on PiLoop.steer that steering is only legal while a turn is in flight (an idle steer would mint a turn through real sendUserMessage that skips budgetGate() and the audit trail entirely - the same hole the abort/clearQueue fix closed for queued steers, left open for idle). FakePiLoop.steer now throws when idle instead of shelving the steer inertly. Tests pin both the idle-throw and that mid-turn steering still queues as before. - Comment coherence: reword runtime.ts's checkpoint comment to the checkpoint-safe framing (aborted included, failed excluded); rescope the settled-undercount warning to the hypothetical it actually is under this seam's own rules; delete the unreachable driving guard in driveTurn (runTurn already refuses re-entry synchronously before it would run) rather than keep two guards for one invariant. - Add invariant 14 (turn != ambient) to INVARIANTS.md covering the steer-idle hole, queue drain, and the future salvage bypass, guarded by the new steer-while-idle contract test plus the existing abort-drains-queue test. --- INVARIANTS.md | 1 + core/src/session/abort.test.ts | 10 +-- core/src/session/background.test.ts | 4 +- core/src/session/budget.test.ts | 12 +-- core/src/session/budget.ts | 2 +- core/src/session/index.ts | 1 + core/src/session/piloop.test.ts | 68 +++++++++++++-- core/src/session/piloop.ts | 127 +++++++++++++++++----------- core/src/session/recovery.test.ts | 4 +- core/src/session/runtime.test.ts | 8 +- core/src/session/runtime.ts | 16 ++-- 11 files changed, 172 insertions(+), 81 deletions(-) diff --git a/INVARIANTS.md b/INVARIANTS.md index d6135ab..b63b285 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) | 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/core/src/session/abort.test.ts b/core/src/session/abort.test.ts index 004e0d8..51d3ed4 100644 --- a/core/src/session/abort.test.ts +++ b/core/src/session/abort.test.ts @@ -1,7 +1,7 @@ // 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 ok: true / exitReason -// "aborted", records turn.ended + cost.recorded (aborted turns burned +// 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. @@ -76,7 +76,7 @@ describe("Session.abort (grants §7.3 breach action / teardown / M2 barge-in)", const result = await turn; await abortP; - expect(result.ok).toBe(true); + expect(result.checkpointSafe).toBe(true); expect(result.exitReason).toBe("aborted"); const events = writer.envelopes(taskId); const t = events.map((e) => e.type); @@ -113,7 +113,7 @@ describe("Session.abort (grants §7.3 breach action / teardown / M2 barge-in)", ]); expect(settled).toBe("resolved"); const result = await turn; - expect(result.ok).toBe(true); + expect(result.checkpointSafe).toBe(true); expect(result.exitReason).toBe("aborted"); // §6.1 cancelled arrow: the audit pair closes, the parked promise @@ -161,7 +161,7 @@ describe("Session.abort (grants §7.3 breach action / teardown / M2 barge-in)", calls: [{ toolCallId: "n1", tool: "read_file", args: { path: `${fx.proj}/src/a.ts` } }], }); const r = await session.runTurn("normal"); - expect(r).toMatchObject({ ok: true, exitReason: "completed" }); + expect(r).toMatchObject({ checkpointSafe: true, exitReason: "completed" }); expect(writer.resumeCursor(taskId)?.lastEntryId).toBe(r.lastEntryId); }); }); 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 27e2352..63b8323 100644 --- a/core/src/session/budget.ts +++ b/core/src/session/budget.ts @@ -128,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 fd26e64..7a7f403 100644 --- a/core/src/session/index.ts +++ b/core/src/session/index.ts @@ -11,6 +11,7 @@ export { } from "./jsonl.js"; export { type Dispatch, + type ExitReason, type PiLoop, type PiLoopEvent, type PiToolCall, diff --git a/core/src/session/piloop.test.ts b/core/src/session/piloop.test.ts index c6059b8..358f1b7 100644 --- a/core/src/session/piloop.test.ts +++ b/core/src/session/piloop.test.ts @@ -22,10 +22,10 @@ describe("PiLoop.abort (cooperative, pi-types §Abort-is-async)", () => { await loop.abort(); // already idle — resolves, poisons nothing loop.queueTurn({ calls: [call("a1")] }); const result = await loop.start("go", okDispatch); - expect(result).toMatchObject({ ok: true, exitReason: "completed" }); + expect(result).toMatchObject({ checkpointSafe: true, exitReason: "completed" }); }); - it("abort mid-turn awaits the in-flight call, skips queued siblings, finalizes durably (ok: true, exitReason aborted)", async () => { + 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; @@ -54,7 +54,7 @@ describe("PiLoop.abort (cooperative, pi-types §Abort-is-async)", () => { release(); const result = await turn; await abortPromise; - expect(result.ok).toBe(true); + 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 @@ -114,7 +114,7 @@ describe("PiLoop.onEvent (narrow typed stream, surface-protocol §7 + grants §7 loop.onEvent((e) => seen.push(e.kind)); loop.queueTurn({ calls: [], fail: true }); const r = await loop.start("go", okDispatch); - expect(r.ok).toBe(false); + expect(r.checkpointSafe).toBe(false); expect(seen).toEqual(["settled"]); loop.queueTurn({ calls: [], crashBeforeReturn: true }); await expect(loop.followUp("crash", okDispatch)).rejects.toThrow(); @@ -130,7 +130,7 @@ describe("PiLoop.onEvent (narrow typed stream, surface-protocol §7 + grants §7 loop.onEvent((e) => seen.push(e.kind)); loop.queueTurn({ calls: [call("t1")] }); const result = await loop.start("go", okDispatch); - expect(result.ok).toBe(true); + expect(result.checkpointSafe).toBe(true); expect(seen).toEqual(["tool_start", "tool_end", "settled"]); }); @@ -236,6 +236,35 @@ describe("PiLoop.onEvent (narrow typed stream, surface-protocol §7 + grants §7 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)", () => { @@ -259,7 +288,7 @@ describe("PiLoop re-entry guard (item 2)", () => { expect(() => loop.followUp("again", okDispatch)).toThrow(/in flight/); release(); const result1 = await turn1; - expect(result1.ok).toBe(true); + 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")] }); @@ -295,6 +324,33 @@ describe("PiLoop steer queue cleared on abort (item 3)", () => { 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((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)", () => { diff --git a/core/src/session/piloop.ts b/core/src/session/piloop.ts index edd4d68..9965dba 100644 --- a/core/src/session/piloop.ts +++ b/core/src/session/piloop.ts @@ -25,22 +25,23 @@ export type ToolOutcome = { ok: true; result: unknown } | { ok: false; error: st // await this for every tool call the model emits. export type Dispatch = (call: PiToolCall) => Promise; +// 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 = "the run finalized coherently and lastEntryId is a durable, - // checkpoint-safe cursor" — NOT "the model finished what was asked". - // An ABORTED turn is ok: true with exitReason "aborted": pi-internals, - // the run is finalized with a synthetic assistant stopReason "aborted" - // and its JSONL entries are durably on disk, so the resume cursor may - // advance. The poisoned-cursor guard (event-log §5.2 "never advanced on - // a failed turn") targets FAILED turns, whose trailing JSONL state is - // not trustworthy; refusing to checkpoint an aborted turn would instead - // leave the §5.1 truncation check blind to entries that are durably - // present — losing a valid cursor, the other half of the same bug. - 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 @@ -70,12 +71,16 @@ export type PiLoopEvent = | { 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. Semantics differ from the double: - // FakePiLoop emits `settled` once per driveTurn (once per turn); real - // agent_settled fires only when Pi is FULLY idle — with a queued - // followUp pending, Pi does NOT settle between the turn that queued it - // and the one that drains it. A watcher counting one `settled` per turn - // against the real adapter will undercount turns. + // 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 @@ -134,6 +139,19 @@ export interface PiLoop { // 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; + // 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 @@ -164,8 +182,8 @@ export interface PiLoop { // 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 ok: true (see - // TurnResult — the aborted run is durably finalized and checkpoint- + // 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; @@ -200,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 @@ -249,10 +267,6 @@ export class FakePiLoop implements PiLoop { private nextListenerToken = 0; private inFlight: Promise | null = null; private aborting = false; - // True for the duration of driveTurn's body — guards re-entry (item 2): - // the real loop can never run two turns at once (Dispatch's mutable - // current-dispatch binding is only sound because turns never overlap). - private driving = false; constructor(readonly jsonlPath: string) { // Continue entry numbering across reattach — a resumed loop appends @@ -323,24 +337,15 @@ export class FakePiLoop implements PiLoop { return id; } + // 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 { - // Re-entry is already refused in runTurn (before inFlight is even - // touched — item 2), but guard the drive itself too: the real loop can - // never be mid-turn twice (item 8's mutable current-dispatch binding - // depends on it), and the double should refuse that state rather than - // silently model it wrongly. - if (this.driving) { - throw new Error("FakePiLoop: driveTurn re-entered while a turn is already in flight"); - } - this.driving = true; - try { - return await this.driveTurnBody(text, dispatch); - } finally { - this.driving = false; - } - } - - private async driveTurnBody(text: string, dispatch: Dispatch): Promise { const turn = this.turns.shift(); if (turn === undefined) { throw new Error("FakePiLoop: no scripted turn queued"); @@ -364,9 +369,20 @@ export class FakePiLoop implements PiLoop { const dispatched = turn.calls.map((c) => { this.emit({ kind: "tool_start", toolCallId: c.toolCallId, tool: c.tool }); const settled = dispatch(c); - settled.then((outcome) => { - this.emit({ kind: "tool_end", toolCallId: c.toolCallId, tool: c.tool, ok: outcome.ok }); - }); + // 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); @@ -396,11 +412,11 @@ export class FakePiLoop implements PiLoop { this.accumulate(turn.stats); if (this.aborting) { // pi-internals: the run is finalized with a synthetic assistant - // stopReason "aborted" — entries durably on disk, hence ok: true - // and a checkpoint-safe lastEntryId (see TurnResult). + // stopReason "aborted" — entries durably on disk, hence + // checkpointSafe: true (see TurnResult). lastEntryId = this.write({ kind: "turnEnd", aborted: true }); this.emit({ kind: "settled" }); - return { ok: true, lastEntryId, exitReason: "aborted" }; + return { checkpointSafe: true, lastEntryId, exitReason: "aborted" }; } lastEntryId = this.write({ kind: "turnEnd" }); if (turn.crashBeforeReturn === true) { @@ -411,7 +427,7 @@ export class FakePiLoop implements PiLoop { // included (settle ≠ success). this.emit({ kind: "settled" }); return { - ok: turn.fail !== true, + checkpointSafe: turn.fail !== true, lastEntryId, exitReason: turn.fail === true ? "error" : "completed", }; @@ -458,7 +474,18 @@ 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); } 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 52eb185..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 { return this.budgetGate() ?? this.finishTurn(await this.loop.start(prompt, this.dispatch)); } @@ -157,6 +158,11 @@ 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); } @@ -172,7 +178,7 @@ export class Session { // 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 ok: true, so the aborted turn records turn.ended + + // "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- @@ -189,7 +195,7 @@ export class Session { // (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, From 3a2548e46b0b02413b8fb2f8195c542fcd963d93 Mon Sep 17 00:00:00 2001 From: Akash Aedavelli Date: Sun, 19 Jul 2026 23:22:01 -0500 Subject: [PATCH 8/9] =?UTF-8?q?spec:=20amend=20event-log=20=C2=A73.1=20for?= =?UTF-8?q?=20the=20abort=20path=20(decidedBy=20+=3D=20abort,=20task.resum?= =?UTF-8?q?ed=20cause)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit grants §6.1 mandates that session abort produce cancelled approval rows, but §3.1's decidedBy vocabulary was a closed list with no tag for that producer, and task.resumed — the only legal transition out of waiting_for_user — reads as "the user answered" when an abort ends the wait. Both amendments approved 2026-07-19 after the K5a seam review; cause is optional so every existing row stays valid. Code and the abort oracle now cite the amendment instead of flagging the gap. --- core/src/eventlog/types.ts | 6 +++++- core/src/session/abort.test.ts | 7 ++++++- core/src/session/approval.ts | 17 +++++++++++------ specs/event-log.md | 22 ++++++++++++++++++++-- 4 files changed, 42 insertions(+), 10 deletions(-) diff --git a/core/src/eventlog/types.ts b/core/src/eventlog/types.ts index 4b34c2c..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; // 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" | "abort" (see approval.ts cancelAll SPEC-NOTE) + 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 index 51d3ed4..4cd6f1d 100644 --- a/core/src/session/abort.test.ts +++ b/core/src/session/abort.test.ts @@ -130,7 +130,12 @@ describe("Session.abort (grants §7.3 breach action / teardown / M2 barge-in)", decision: "cancelled", decidedBy: "abort", }); - expect(events.filter((e) => e.type === "task.resumed")).toHaveLength(1); + // 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 }); diff --git a/core/src/session/approval.ts b/core/src/session/approval.ts index 5724084..974723b 100644 --- a/core/src/session/approval.ts +++ b/core/src/session/approval.ts @@ -186,11 +186,13 @@ export class ApprovalMachine { // 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. - // SPEC-NOTE (event-log §3.1): the decidedBy vocabulary there enumerates - // user|ledger|background_policy|recovery, but grants §6.1 names session - // abort as a `cancelled` producer without a tag for it — "abort" extends - // the vocabulary rather than mislabeling a live abort as the boot - // recovery pass. Flagged for spec amendment. + // 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()]; @@ -203,7 +205,10 @@ export class ApprovalMachine { decidedBy: "abort", }); } - append("task.resumed", { answerRef: records[records.length - 1]?.card.toolCallId }); + 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" }); } 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, From 0f3c616a6bb3348289d2a367d60d4a6a59cf3981 Mon Sep 17 00:00:00 2001 From: Akash Aedavelli Date: Sun, 19 Jul 2026 23:25:58 -0500 Subject: [PATCH 9/9] =?UTF-8?q?arch:=20guard=20invariant=2014's=20headline?= =?UTF-8?q?=20clause=20=E2=80=94=20PiLoop=20seam=20imports=20stay=20in=20s?= =?UTF-8?q?ession/?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Row 14 named three guards, none of which covered its main sentence: nothing stopped a future module from holding the loop and calling start/followUp directly, minting a turn that skips budgetGate on the way in and finishTurn on the way out. Confining the seam's import to core/src/session/ makes that drift fail the gate instead of passing it. Three self-test cases; 10 rules now. --- INVARIANTS.md | 2 +- scripts/arch-checks.mjs | 35 +++++++++++++++++++++++++++++++++++ 2 files changed, 36 insertions(+), 1 deletion(-) diff --git a/INVARIANTS.md b/INVARIANTS.md index b63b285..763e945 100644 --- a/INVARIANTS.md +++ b/INVARIANTS.md @@ -19,7 +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) | 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 | +| 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/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",