chore(deps): bump @earendil-works/pi-ai from 1.1.0 to 1.4.0

package.json

@@ -88,7 +88,8 @@
       "minimatch@9.0.9": "patches/minimatch@9.0.9.patch",
       "minimatch@10.2.6": "patches/minimatch@10.2.6.patch",
       "glob@11.1.0": "patches/glob@11.1.0.patch",
-      "glob@13.0.6": "patches/glob@13.0.6.patch"
+      "glob@13.0.6": "patches/glob@13.0.6.patch",
+      "@earendil-works/pi-coding-agent@1.4.0": "patches/@earendil-works__pi-coding-agent@1.4.0.patch"
     }
   }
 }

packages/runner-pi/package.json

@@ -6,16 +6,16 @@
   "types": "./dist/index.d.ts",
   "scripts": {
     "build:ci": "tsc -p tsconfig.json",
-    "typecheck:ci": "tsc -p tsconfig.json --noEmit",
-    "test:ci": "node --test test/*.test.mjs",
+    "typecheck:ci": "tsc -p tsconfig.json --noEmit && tsc -p tsconfig.test.json",
+    "test:ci": "node --test test/*.test.mjs test/*.test.ts",
     "build": "corepack pnpm --filter @ling-stack/plugin-sdk build && tsc -p tsconfig.json",
-    "typecheck": "corepack pnpm --filter @ling-stack/plugin-sdk build && tsc -p tsconfig.json --noEmit",
+    "typecheck": "corepack pnpm --filter @ling-stack/plugin-sdk build && corepack pnpm typecheck:ci",
     "lint": "echo \"lint not configured yet\"",
-    "test": "corepack pnpm --filter @ling-stack/plugin-sdk build && tsc -p tsconfig.json && node --test test/*.test.mjs"
+    "test": "corepack pnpm --filter @ling-stack/plugin-sdk build && tsc -p tsconfig.json && corepack pnpm test:ci"
   },
   "dependencies": {
-    "@earendil-works/pi-ai": "1.1.0",
-    "@earendil-works/pi-coding-agent": "1.1.0",
+    "@earendil-works/pi-ai": "1.4.0",
+    "@earendil-works/pi-coding-agent": "1.4.0",
     "@ling-stack/plugin-sdk": "workspace:*",
     "undici": "7.30.1",
     "zod": "4.6.7"

packages/runner-pi/src/adapter.ts

@@ -145,6 +145,12 @@ export class PiRunnerAdapter {
       if (event.type === "agent_settled") sdkSettlement = { aborted: event.aborted };
       if ((event.type === "message_start" || event.type === "message_end" || event.type === "message_update")
         && isPiHostResumeControl(event.message)) return;
+      if (event.type === "message_end" && event.message.role === "system" && session && frozenAuthority) {
+        // Pi reduces the system patch into state before notifying subscribers.
+        // Omit SDK bookkeeping only after proving prompt and tools are unchanged.
+        frozenPiConversation(session.state.messages, frozenAuthority);
+        return;
+      }
       const translatedEvents = translatePiEvents(event);
       // Buffer only Pi's exact reflection of this boundary's private control
       // error and its identity-paired empty turn_end. Nothing is discarded until

packages/runner-pi/test/session-authority.test.ts

@@ -0,0 +1,109 @@
+import assert from "node:assert/strict";
+import { mkdtemp, rm } from "node:fs/promises";
+import { createServer } from "node:http";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import test, { type TestContext } from "node:test";
+import type { SystemMessage } from "@earendil-works/pi-ai";
+import { AgentSession } from "@earendil-works/pi-coding-agent";
+import { PiRunnerAdapter } from "../dist/adapter.js";
+import { frozenPiConversation } from "../dist/session-transcript.js";
+import type { PiRunnerEvent, PiRunnerRunOptions } from "../dist/types.js";
+
+async function fixture(t: TestContext) {
+  const cwd = await mkdtemp(join(tmpdir(), "ling-pi-authority-"));
+  t.after(() => rm(cwd, { recursive: true, force: true }));
+  const requests: unknown[] = [];
+  const events: PiRunnerEvent[] = [];
+  let effects = 0;
+  const server = createServer(async (request, response) => {
+    const chunks: Buffer[] = [];
+    for await (const chunk of request) chunks.push(Buffer.from(chunk));
+    requests.push(JSON.parse(Buffer.concat(chunks).toString("utf8")));
+    response.writeHead(200, { "content-type": "text/event-stream" });
+    response.end('data: {"choices":[{"index":0,"delta":{"content":"Recovered."},"finish_reason":null}]}\n\n'
+      + 'data: {"choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}\n\ndata: [DONE]\n\n');
+  });
+  t.after(() => new Promise<void>((resolve, reject) => {
+    server.close(error => error ? reject(error) : resolve());
+    server.closeAllConnections();
+  }));
+  await new Promise<void>((resolve, reject) => {
+    server.once("error", reject);
+    server.listen(0, "127.0.0.1", resolve);
+  });
+  const address = server.address();
+  assert.ok(address && typeof address !== "string");
+  const runner = new PiRunnerAdapter();
+  const options: PiRunnerRunOptions = {
+    cwd, agentDir: join(cwd, "agent"), runId: "saved-run", runnerAttemptId: "restored-attempt",
+    task: { id: "saved-task", title: "Recover", prompt: "Report the saved read without repeating it." },
+    provider: { provider: "authority-fixture", model: "fixture", api: "openai-completions",
+      baseUrl: `http://127.0.0.1:${address.port}/v1`, apiKey: "fixture-no-network", reasoning: false },
+    tools: ["read"], onEvent: event => { events.push(event); },
+    onToolBefore: () => { effects += 1; return { allowed: true }; },
+    onTaskContinuationBoundary: async () => {},
+  };
+  const preparedInput = await runner.prepareTaskInput(options);
+  const modelView = [
+    { role: "user", content: [{ type: "text", text: preparedInput.userPrompt }], timestamp: 0 },
+    { role: "assistant", content: [{ type: "toolCall", id: "saved-read", name: "read", arguments: { path: "saved.txt" } }],
+      api: "openai-completions", provider: options.provider.provider, model: options.provider.model, stopReason: "toolUse",
+      usage: { input: 10, output: 5, cacheRead: 0, cacheWrite: 0, totalTokens: 15,
+        cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 } }, timestamp: 1 },
+    { role: "toolResult", toolCallId: "saved-read", toolName: "read", isError: false,
+      content: [{ type: "text", text: "The original durable result." }], timestamp: 2 },
+  ];
+  return { runner, options: { ...options, preparedInput,
+    ordinaryProgressRecovery: { schemaVersion: 1 as const, modelView } }, modelView, events, requests, effects: () => effects };
+}
+
+test("a restored SDK system patch preserves frozen authority without adding Host recovery effects", { timeout: 15_000 }, async t => {
+  const f = await fixture(t);
+  let admissions = 0;
+  const result = await f.runner.runTask({ ...f.options, onTaskContinuationBoundary: async point => {
+    if (point !== "before_provider_request") return;
+    admissions += 1;
+    assert.deepEqual(f.events.map(event => event.type), ["run_start", "turn_start"],
+      "only canonical recovery bookkeeping may precede the first provider request");
+  } });
+  assert.equal(admissions, 1);
+  assert.equal(f.requests.length, 1);
+  assert.equal(f.effects(), 0);
+  assert.ok(result.messages.filter(message => (message as { role: string }).role === "system").length >= 2,
+    "the SDK transcript retains its system patch");
+  const conversation = frozenPiConversation(result.messages, f.options.preparedInput);
+  assert.deepEqual(conversation.slice(0, -1), f.modelView);
+  assert.equal((conversation.at(-1) as { role: string }).role, "assistant");
+  assert.equal(f.events.some(event => event.type === "message_end" && event.role === "system"), false);
+  assert.equal(f.events.some(event => event.type === "run_end"), true);
+});
+
+const drifts: { name: string; patch: Partial<SystemMessage> }[] = [
+  { name: "prompt", patch: { content: "Unsealed replacement instructions." } },
+  { name: "tool declaration", patch: { toolsAdded: [{ name: "read", description: "Changed read authority", parameters: { type: "object" } }] } },
+  { name: "tool removal", patch: { toolsRemoved: [{ name: "read" }] } },
+];
+
+for (const { name, patch } of drifts) {
+  test(`a restored ${name} change cannot be omitted as bookkeeping or reach a provider`, { timeout: 15_000 }, async t => {
+    const subscribe = AgentSession.prototype.subscribe;
+    let injected = false;
+    t.mock.method(AgentSession.prototype, "subscribe", function (this: AgentSession, listener: Parameters<AgentSession["subscribe"]>[0]) {
+      return subscribe.call(this, event => {
+        if (event.type === "message_end" && event.message.role === "system") {
+          const message = { ...event.message, ...patch };
+          this.agent.state.messages = this.agent.state.messages.map(value => value === event.message ? message : value);
+          injected = true;
+          listener({ ...event, message });
+        } else listener(event);
+      });
+    });
+    const f = await fixture(t);
+    await assert.rejects(f.runner.runTask(f.options));
+    assert.equal(injected, true);
+    assert.equal(f.requests.length, 0);
+    assert.equal(f.effects(), 0);
+    assert.equal(f.events.some(event => event.type === "run_end"), false);
+  });
+}

packages/runner-pi/test/summary-budget.test.ts

@@ -0,0 +1,134 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import { InMemoryCredentialStore, InMemoryModelsStore, type Model } from "@earendil-works/pi-ai";
+import { stream } from "@earendil-works/pi-ai/api/openai-completions";
+import { compact, generateBranchSummary, generateSummaryWithUsage, ModelRuntime, type SessionEntry } from "@earendil-works/pi-coding-agent";
+import { createPiTaskBoundary } from "../dist/task-boundary.js";
+import type { PiRunnerRunOptions } from "../dist/types.js";
+
+const messages = [{ role: "user" as const, content: "Preserve the completed work.", timestamp: 0 }];
+
+async function fixture({ outputLimit = 1_024, quota = 204, finishReason = "stop", finishAt = 1 } = {}) {
+  const model: Model<"openai-completions"> = {
+    provider: "summary-budget-fixture", id: "fixture", name: "fixture", api: "openai-completions",
+    baseUrl: "https://fixture.invalid/v1", reasoning: false, input: ["text"], contextWindow: 32_768,
+    maxTokens: outputLimit, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
+  };
+  const reservations: { maxOutputTokens: number; payload: unknown }[] = [];
+  const requests: { max_tokens?: number; max_completion_tokens?: number }[] = [];
+  let settlements = 0;
+  const options: PiRunnerRunOptions = {
+    cwd: process.cwd(), task: { id: "summary-budget", title: "Summary", prompt: "Preserve the completed work." },
+    runId: "summary-run", runnerAttemptId: "summary-attempt",
+    provider: { provider: model.provider, model: model.id, api: model.api, baseUrl: model.baseUrl,
+      contextWindow: model.contextWindow, maxTokens: model.maxTokens, apiKey: "fixture-no-network" },
+    providerBudget: {
+      requiresTokenUpperBound: true, requiresCostUpperBound: false,
+      reserve(request) {
+        reservations.push(request);
+        if (request.maxOutputTokens > quota) throw new Error("summary quota exhausted");
+        return { reservationId: `summary-${reservations.length}`, reservedTokens: 32_768 + quota };
+      },
+      settle() { settlements += 1; },
+      settleUnknown() { assert.fail("the fixture supplies measured usage"); },
+    },
+  };
+  const runtime = await ModelRuntime.create({ credentials: new InMemoryCredentialStore(), modelsStore: new InMemoryModelsStore(),
+    modelsPath: null, allowModelNetwork: false, refreshOnCreate: false });
+  runtime.registerProvider(model.provider, {
+    api: model.api, apiKey: options.provider.apiKey,
+    streamSimple: (requestModel, context, requestOptions) => stream(requestModel as Model<"openai-completions">, context, {
+      ...requestOptions,
+      fetch: async (_url, init) => {
+        requests.push(JSON.parse(String(init?.body)));
+        const records = [
+          { choices: [{ index: 0, delta: { content: "Completed work retained." }, finish_reason: null }] },
+          { choices: [{ index: 0, delta: {}, finish_reason: requests.length === finishAt ? finishReason : "stop" }] },
+          { choices: [], usage: { prompt_tokens: 20, completion_tokens: 16, total_tokens: 36 } },
+        ];
+        return new Response(records.map(record => `data: ${JSON.stringify(record)}\n\n`).join("") + "data: [DONE]\n\n", {
+          headers: { "content-type": "text/event-stream" },
+        });
+      },
+    }),
+  });
+  const boundary = createPiTaskBoundary(options);
+  boundary.bindModelRuntime(runtime);
+  const signal = boundary.options.signal;
+  assert.ok(signal);
+  async function admitted<T>(operation: Promise<T>): Promise<T> {
+    const result = await operation;
+    boundary.assertOpen();
+    return result;
+  }
+  return { model, runtime, boundary, reservations, requests, settlements: () => settlements,
+    summary: (reserveTokens = 256, previousSummary?: string) => admitted(generateSummaryWithUsage(messages, model, reserveTokens,
+      options.provider.apiKey, undefined, signal, undefined, previousSummary, "off", runtime.streamSimple)),
+    compact: (split = false) => admitted(compact({ firstKeptEntryId: "retained-entry", messagesToSummarize: messages,
+      turnPrefixMessages: split ? messages : [], isSplitTurn: split, tokensBefore: 500,
+      fileOps: { read: new Set(), written: new Set(), edited: new Set() },
+      settings: { enabled: true, reserveTokens: 256, keepRecentTokens: 64 } }, model, options.provider.apiKey,
+      undefined, undefined, signal, "off", runtime.streamSimple)),
+    branch: () => admitted(generateBranchSummary([{ type: "message", id: "abandoned-entry", parentId: null,
+      timestamp: new Date(0).toISOString(), message: messages[0] } satisfies SessionEntry], {
+      model, apiKey: options.provider.apiKey, reserveTokens: 256, signal, streamFn: runtime.streamSimple,
+    })),
+  };
+}
+
+for (const mode of ["summary", "update", "compact", "split", "branch"] as const) {
+  test(`${mode} summary reserves and sends the configured output cap`, { timeout: 15_000 }, async () => {
+    const f = await fixture();
+    const result = mode === "summary" ? await f.summary()
+      : mode === "update" ? await f.summary(256, "Earlier completed work.")
+      : mode === "branch" ? await f.branch() : await f.compact(mode === "split");
+    assert.match(("text" in result ? result.text : result.summary) ?? "", /Completed work retained\./);
+    assert.equal(f.requests.length, mode === "split" ? 2 : 1);
+    assert.equal(f.reservations.length, f.requests.length);
+    assert.equal(f.settlements(), f.requests.length);
+    for (const [index, request] of f.requests.entries()) {
+      assert.equal(request.max_tokens ?? request.max_completion_tokens, 204);
+      assert.equal(f.reservations[index].maxOutputTokens, 204);
+      assert.deepEqual(f.reservations[index].payload, request);
+    }
+    f.boundary.assertOpen();
+  });
+}
+
+test("the summary cap never exceeds the model output limit", async () => {
+  const f = await fixture({ outputLimit: 128, quota: 128 });
+  await f.summary();
+  assert.equal(f.reservations[0].maxOutputTokens, 128);
+  assert.equal(f.requests[0].max_tokens ?? f.requests[0].max_completion_tokens, 128);
+});
+
+test("an insufficient summary quota still rejects before transport", async () => {
+  const f = await fixture({ quota: 203 });
+  await assert.rejects(f.summary(), /summary quota exhausted/);
+  assert.equal(f.reservations.length, 1);
+  assert.equal(f.reservations[0].maxOutputTokens, 204);
+  assert.equal(f.requests.length, 0);
+  assert.throws(() => f.boundary.assertOpen(), /summary quota exhausted/);
+});
+
+for (const mode of ["summary", "update", "compact", "split", "branch"] as const) {
+  test(`${mode} output at the token cap cannot become a checkpoint`, async () => {
+    const requestCount = mode === "split" ? 2 : 1;
+    const f = await fixture({ finishReason: "length", finishAt: requestCount });
+    const operation = mode === "summary" ? f.summary()
+      : mode === "update" ? f.summary(256, "Earlier completed work.")
+      : mode === "branch" ? f.branch() : f.compact(mode === "split");
+    if (mode === "branch") {
+      assert.deepEqual(await operation, {
+        error: "Branch summarization failed: generation hit the token cap and the summary is incomplete",
+      });
+    } else {
+      await assert.rejects(operation, /summary is incomplete/);
+    }
+    assert.equal(f.requests.length, requestCount);
+    assert.equal(f.settlements(), requestCount);
+    for (const request of f.requests) {
+      assert.equal(request.max_tokens ?? request.max_completion_tokens, 204);
+    }
+  });
+}

packages/runner-pi/tsconfig.test.json

@@ -0,0 +1,12 @@
+{
+  "compilerOptions": {
+    "module": "NodeNext",
+    "moduleResolution": "NodeNext",
+    "noEmit": true,
+    "skipLibCheck": true,
+    "strict": true,
+    "target": "ES2022",
+    "types": ["node"]
+  },
+  "include": ["test/**/*.ts"]
+}