chore: release pi-devin 0.2.1

Ports #6 with optional-field validation, conversion before cache writes, an upstream-specific cache directory, and startup regression tests.

Co-authored-by: dualface <dualface@gmail.com>

.github/workflows/test.yml

@@ -0,0 +1,22 @@
+name: Tests
+on:
+  push:
+  pull_request:
+permissions:
+  contents: read
+jobs:
+  test:
+    strategy:
+      fail-fast: false
+      matrix:
+        os: [ubuntu-latest, macos-latest, windows-latest]
+        node: [22, 24]
+    runs-on: ${{ matrix.os }}
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-node@v4
+        with:
+          node-version: ${{ matrix.node }}
+      - run: npm install --ignore-scripts --no-audit --no-fund
+      - run: npm run typecheck
+      - run: npm test

.gitignore

@@ -4,3 +4,4 @@ package-lock.json
 .DS_Store
 *.log
 .env
+.reference/

README.md

@@ -16,9 +16,9 @@ Those models are available through the local Devin CLI. This package uses that C
 
 ## Requirements
 
-- Pi Coding Agent 0.80+
+- Pi Coding Agent 0.86+
 - A signed-in [Devin CLI](https://docs.devin.ai/cli) (`devin auth status`)
-- Node 18+
+- Node 22.19+ (required by Pi 0.86)
 
 The CLI binary is resolved in this order:
 
@@ -27,6 +27,8 @@ The CLI binary is resolved in this order:
 3. Devin.app's bundled `devin` binary
 4. `which devin`
 
+On Windows, discovery checks the CLI installer and Devin Desktop locations, then uses `where.exe` to search PATH. Native executables and `.cmd`/`.bat` wrappers are supported, including paths with spaces. `DEVIN_CLI` takes precedence on every platform.
+
 ## Install
 
 From git:
@@ -35,12 +37,16 @@ From git:
 pi install git:github.com/kashyab12/pi-devin
 ```
 
-After npm publish:
+From npm:
 
 ```bash
 pi install npm:pi-devin
 ```
 
+Version 0.2.0 requires Pi 0.86+ and Node 22.19+. It restores system instructions and tools on the current Pi transcript format and sends system instructions through Devin's dedicated prompt field. Version 0.1.2 predates Pi 0.86 support.
+
+`pi-devin-local` is a separate npm package maintained in the [mizorewww/pi-devin fork](https://github.com/mizorewww/pi-devin). Changes merged here do not update that package. Install only one: both packages register the `devin` provider, so their registrations can overwrite each other.
+
 Local checkout:
 
 ```bash
@@ -65,6 +71,8 @@ Commands:
 - `/devin-status` — CLI path, version, auth
 - `/devin-refresh` — reload `devin models list --format json`
 
+The model catalog is cached for six hours in `$XDG_CACHE_HOME/pi-devin/models.json` (or `~/.cache/pi-devin/models.json`). A fresh cache avoids the CLI call at startup; an older cache remains available while it refreshes in the background. Set `PI_OFFLINE=1` to skip automatic catalog refreshes. `/devin-refresh` still requests a refresh explicitly.
+
 ## What this is / is not
 
 | This package | Not this package |

extensions/index.ts

@@ -2,6 +2,13 @@ import type { ExtensionAPI, ProviderModelConfig } from "@earendil-works/pi-codin
 import type { Api, Model, OAuthCredentials, OAuthLoginCallbacks } from "@earendil-works/pi-ai";
 import { authStatus, loginWithCli, readCredentials } from "../src/credentials.js";
 import { whichDevin, devinVersion } from "../src/cli.js";
+import {
+  type CachedDevinCatalog,
+  isUsableCatalog,
+  isCatalogCacheFresh,
+  readCatalogCache,
+  writeCatalogCache,
+} from "../src/catalog-cache.js";
 import { FALLBACK_MODELS, loadCliCatalog, modelsFromCatalog } from "../src/models.js";
 import { CLIENT_IDE, CLIENT_VERSION } from "../src/metadata.js";
 import { streamDevin } from "../src/stream.js";
@@ -10,6 +17,37 @@ const PROVIDER_ID = "devin";
 const PLACEHOLDER_BASE_URL = "https://server.codeium.com";
 
 let _pi: ExtensionAPI | null = null;
+let catalogRequest: Promise<ProviderModelConfig[]> | null = null;
+
+function isOffline(): boolean {
+  const value = process.env.PI_OFFLINE?.toLowerCase();
+  return value === "1" || value === "true" || value === "yes";
+}
+
+function refreshCatalog(pi: ExtensionAPI): Promise<ProviderModelConfig[]> {
+  if (!catalogRequest) {
+    const pending = loadCliCatalog().then((catalog) => {
+      if (!isUsableCatalog(catalog)) {
+        throw new Error("Devin CLI returned no usable model families");
+      }
+      const models = modelsFromCatalog(catalog);
+      try {
+        writeCatalogCache(catalog);
+      } catch (error) {
+        console.warn(`Devin: failed to cache model catalog: ${error instanceof Error ? error.message : String(error)}`);
+      }
+      return models;
+    });
+    catalogRequest = pending;
+    void pending.finally(() => {
+      if (catalogRequest === pending) catalogRequest = null;
+    }).catch(() => {});
+  }
+  return catalogRequest.then((models) => {
+    if (_pi === pi) registerDevinProvider(pi, models);
+    return models;
+  });
+}
 
 function registerDevinProvider(pi: ExtensionAPI, models: ProviderModelConfig[]): void {
   pi.registerProvider(PROVIDER_ID, {
@@ -23,8 +61,7 @@ function registerDevinProvider(pi: ExtensionAPI, models: ProviderModelConfig[]):
         const creds = await loginWithCli();
         if (_pi) {
           try {
-            const catalog = await loadCliCatalog();
-            registerDevinProvider(_pi, modelsFromCatalog(catalog));
+            await refreshCatalog(_pi);
           } catch {
             // keep current models
           }
@@ -57,26 +94,32 @@ function registerDevinProvider(pi: ExtensionAPI, models: ProviderModelConfig[]):
 
 export default async function (pi: ExtensionAPI): Promise<void> {
   _pi = pi;
-  registerDevinProvider(pi, FALLBACK_MODELS);
-
+  let cached: CachedDevinCatalog | null | undefined;
+  let initialModels = FALLBACK_MODELS;
   try {
-    if (readCredentials()) {
-      const catalog = await loadCliCatalog();
-      registerDevinProvider(pi, modelsFromCatalog(catalog));
-    }
-  } catch {
-    // fallback models already registered
+    cached = readCatalogCache();
+    if (cached) initialModels = modelsFromCatalog(cached.catalog);
+  } catch (error) {
+    cached = null;
+    console.warn(`Devin: failed to read model catalog cache: ${error instanceof Error ? error.message : String(error)}`);
   }
+  registerDevinProvider(pi, initialModels);
 
-  pi.on("session_start", async () => {
+  if (!isOffline()) {
     try {
-      if (!_pi || !readCredentials()) return;
-      const catalog = await loadCliCatalog();
-      registerDevinProvider(_pi, modelsFromCatalog(catalog));
+      if (readCredentials()) {
+        if (!cached) {
+          await refreshCatalog(pi);
+        } else if (!isCatalogCacheFresh(cached)) {
+          void refreshCatalog(pi).catch((error) => {
+            console.warn(`Devin: background model catalog refresh failed: ${error instanceof Error ? error.message : String(error)}`);
+          });
+        }
+      }
     } catch {
-      // keep current models
+      // Cached or fallback models stay registered when credential discovery or a cold refresh fails.
     }
-  });
+  }
 
   pi.registerCommand("devin-status", {
     description: "Show Devin CLI auth + binary status",
@@ -100,9 +143,7 @@ export default async function (pi: ExtensionAPI): Promise<void> {
     description: "Refresh Devin Local model catalog from `devin models list`",
     handler: async (_args, ctx) => {
       try {
-        const catalog = await loadCliCatalog();
-        const models = modelsFromCatalog(catalog);
-        registerDevinProvider(pi, models);
+        const models = await refreshCatalog(pi);
         ctx.ui.notify(`Devin: loaded ${models.length} families from the local CLI.`, "info");
       } catch (error) {
         ctx.ui.notify(
@@ -114,6 +155,6 @@ export default async function (pi: ExtensionAPI): Promise<void> {
   });
 
   pi.on("session_shutdown", async () => {
-    _pi = null;
+    if (_pi === pi) _pi = null;
   });
 }

package.json

@@ -1,6 +1,6 @@
 {
   "name": "pi-devin",
-  "version": "0.1.2",
+  "version": "0.2.1",
   "description": "Pi provider for Devin Local models. Auth and catalog come from the local Devin CLI; Pi stays the harness.",
   "type": "module",
   "keywords": [
@@ -29,17 +29,19 @@
     "LICENSE"
   ],
   "scripts": {
+    "test": "node test-runner.mjs",
     "typecheck": "tsc --noEmit"
   },
   "peerDependencies": {
-    "@earendil-works/pi-ai": "*",
-    "@earendil-works/pi-coding-agent": "*"
+    "@earendil-works/pi-ai": ">=0.86.0",
+    "@earendil-works/pi-coding-agent": ">=0.86.0"
   },
   "devDependencies": {
     "@types/node": "^24.13.3",
+    "tsx": "^4.23.15",
     "typescript": "^5.9.2"
   },
   "engines": {
-    "node": ">=18.0.0"
+    "node": ">=22.19.0"
   }
 }

src/catalog-cache.ts

@@ -0,0 +1,104 @@
+import { existsSync, mkdirSync, readFileSync, renameSync, unlinkSync, writeFileSync } from "node:fs";
+import { randomUUID } from "node:crypto";
+import { homedir } from "node:os";
+import { dirname, join } from "node:path";
+import type { DevinCatalog } from "./models.js";
+
+const CACHE_VERSION = 1;
+
+export const CATALOG_CACHE_MAX_AGE_MS = 6 * 60 * 60 * 1000;
+
+export interface CachedDevinCatalog {
+  catalog: DevinCatalog;
+  fetchedAt: number;
+}
+
+interface CatalogCacheFile extends CachedDevinCatalog {
+  version: typeof CACHE_VERSION;
+}
+
+function isRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === "object" && value !== null && !Array.isArray(value);
+}
+
+function optionalString(value: unknown): boolean {
+  return value === undefined || typeof value === "string";
+}
+
+function optionalTokenLimit(value: unknown): boolean {
+  return value === undefined || (typeof value === "number" && Number.isSafeInteger(value) && value > 0);
+}
+
+export function isUsableCatalog(value: unknown): value is DevinCatalog {
+  if (!isRecord(value) || !Array.isArray(value.families) || value.families.length === 0) return false;
+  return value.families.some((family) => isRecord(family) && Array.isArray(family.variants) && family.variants.length > 0)
+    && value.families.every((family) =>
+    isRecord(family)
+    && typeof family.family_label === "string"
+    && typeof family.family_uid === "string"
+    && typeof family.slug === "string"
+    && (family.aliases === undefined || (Array.isArray(family.aliases) && family.aliases.every((alias) => typeof alias === "string")))
+    && Array.isArray(family.variants)
+    && family.variants.every((variant) =>
+      isRecord(variant)
+      && typeof variant.model_uid === "string"
+      && variant.model_uid.trim().length > 0
+      && typeof variant.label === "string"
+      && optionalString(variant.cost_summary)
+      && optionalString(variant.cost_tier)
+      && optionalTokenLimit(variant.max_context_tokens)
+      && optionalTokenLimit(variant.max_output_tokens)
+      && (variant.is_new === undefined || typeof variant.is_new === "boolean")
+      && (variant.is_beta === undefined || typeof variant.is_beta === "boolean")
+    )
+  );
+}
+
+export function catalogCachePath(
+  env: NodeJS.ProcessEnv = process.env,
+  home: string = homedir(),
+): string {
+  const cacheRoot = env.XDG_CACHE_HOME || join(home, ".cache");
+  return join(cacheRoot, "pi-devin", "models.json");
+}
+
+export function readCatalogCache(path: string = catalogCachePath()): CachedDevinCatalog | null {
+  if (!existsSync(path)) return null;
+  const parsed: unknown = JSON.parse(readFileSync(path, "utf8"));
+  if (
+    !isRecord(parsed)
+    || parsed.version !== CACHE_VERSION
+    || typeof parsed.fetchedAt !== "number"
+    || !Number.isFinite(parsed.fetchedAt)
+    || !isUsableCatalog(parsed.catalog)
+  ) {
+    throw new Error(`Invalid Devin model catalog cache: ${path}`);
+  }
+  return { catalog: parsed.catalog, fetchedAt: parsed.fetchedAt };
+}
+
+export function writeCatalogCache(
+  catalog: DevinCatalog,
+  path: string = catalogCachePath(),
+  fetchedAt: number = Date.now(),
+): void {
+  if (!isUsableCatalog(catalog)) throw new Error("Refusing to cache an invalid Devin model catalog");
+  mkdirSync(dirname(path), { recursive: true, mode: 0o700 });
+  const temporaryPath = `${path}.${randomUUID()}.tmp`;
+  const body: CatalogCacheFile = { version: CACHE_VERSION, fetchedAt, catalog };
+  try {
+    writeFileSync(temporaryPath, `${JSON.stringify(body)}\n`, { mode: 0o600 });
+    renameSync(temporaryPath, path);
+  } finally {
+    if (existsSync(temporaryPath)) unlinkSync(temporaryPath);
+  }
+}
+
+export function isCatalogCacheFresh(
+  cached: CachedDevinCatalog,
+  now: number = Date.now(),
+  maxAgeMs: number = CATALOG_CACHE_MAX_AGE_MS,
+): boolean {
+  const age = now - cached.fetchedAt;
+  return age >= 0 && age <= maxAgeMs;
+}

src/cli.ts

@@ -1,4 +1,4 @@
-import { existsSync } from "node:fs";
+import { statSync } from "node:fs";
 import { homedir } from "node:os";
 import { join } from "node:path";
 import { execFile, spawn } from "node:child_process";
@@ -6,21 +6,50 @@ import { promisify } from "node:util";
 
 const execFileAsync = promisify(execFile);
 
-const KNOWN_BINS = [
-  process.env.DEVIN_CLI,
-  join(homedir(), ".local/bin/devin"),
-  join(homedir(), ".devin/bin/devin"),
-  "/opt/homebrew/bin/devin",
-  "/usr/local/bin/devin",
-  "/Applications/Devin.app/Contents/Resources/app/extensions/windsurf/devin/bin/devin",
-].filter((p): p is string => Boolean(p));
+function candidateBins(): string[] {
+  const localAppData = process.env.LOCALAPPDATA ?? join(homedir(), "AppData/Local");
+  return [
+    // Explicit override always wins. Read lazily so a value set after module
+    // load (e.g. by the host process) is still honoured.
+    process.env.DEVIN_CLI,
+    ...(process.platform === "win32" ? [
+    // Official installer location (%LOCALAPPDATA%\devin\cli\bin\devin.exe)
+    join(localAppData, "devin/cli/bin/devin.exe"),
+    // Devin Desktop (Windsurf-based) bundled CLI, per-user install
+    join(localAppData, "Programs/Devin/resources/app/extensions/windsurf/devin/bin/devin.exe"),
+    // Devin Desktop bundled CLI, machine-wide install
+    process.env.ProgramFiles &&
+      join(process.env.ProgramFiles, "Devin/resources/app/extensions/windsurf/devin/bin/devin.exe"),
+    // Set by the official Windows installer when it completes normally.
+    process.env.LOCALAPPDATA && join(process.env.LOCALAPPDATA, "devin/bin/devin.exe"),
+    // Manual fallback for installs that could not finish while Devin was running.
+    join(homedir(), ".devin-cli/devin.exe"),
+    ] : [
+    join(homedir(), ".local/bin/devin"),
+    join(homedir(), ".devin/bin/devin"),
+    "/opt/homebrew/bin/devin",
+    "/usr/local/bin/devin",
+    "/Applications/Devin.app/Contents/Resources/app/extensions/windsurf/devin/bin/devin",
+    ]),
+  ].filter((p): p is string => Boolean(p));
+}
 
 let cachedBin: string | null | undefined;
 
+function isFile(path: string): boolean {
+  try {
+    return statSync(path).isFile();
+  } catch {
+    return false;
+  }
+}
+
 export function findDevinBin(): string | null {
+  // An explicit override must take effect even after a successful discovery.
+  if (process.env.DEVIN_CLI && isFile(process.env.DEVIN_CLI)) return process.env.DEVIN_CLI;
   if (cachedBin !== undefined) return cachedBin;
-  for (const bin of KNOWN_BINS) {
-    if (existsSync(bin)) {
+  for (const bin of candidateBins()) {
+    if (isFile(bin)) {
       cachedBin = bin;
       return bin;
     }
@@ -36,15 +65,24 @@ export function clearDevinBinCache(): void {
 export async function whichDevin(): Promise<string | null> {
   const known = findDevinBin();
   if (known) return known;
-  try {
-    const { stdout } = await execFileAsync("/usr/bin/which", ["devin"], { timeout: 5_000 });
-    const path = stdout.trim();
-    if (path && existsSync(path)) {
-      cachedBin = path;
-      return path;
+  // Windows has no `which`; use `where` and probe the returned paths directly.
+  const isWindows = process.platform === "win32";
+  const locator = isWindows ? "where.exe" : "/usr/bin/which";
+  const candidates = isWindows ? ["devin.exe", "devin.cmd", "devin"] : ["devin"];
+  for (const candidate of candidates) {
+    try {
+      const { stdout } = await execFileAsync(locator, [candidate], { timeout: 5_000 });
+      // `where` may return several lines; take the first existing one.
+      for (const line of stdout.split(/\r?\n/)) {
+        const path = line.trim();
+        if (path && isFile(path)) {
+          cachedBin = path;
+          return path;
+        }
+      }
+    } catch {
+      // not on PATH
     }
-  } catch {
-    // not on PATH
   }
   return null;
 }
@@ -60,9 +98,18 @@ export async function runDevin(
     );
   }
 
+  // Batch wrappers need cmd.exe. Quote paths with spaces, and reject shell
+  // expansion characters rather than interpreting them as part of a command.
+  const wrapper = process.platform === "win32" && /\.(cmd|bat)$/i.test(bin);
+  if (wrapper && [bin, ...args].some((part) => /["%!\r\n]/.test(part))) {
+    throw new Error("Unsupported shell characters in Devin wrapper path or arguments. Set DEVIN_CLI to devin.exe.");
+  }
+  const executable = wrapper ? process.env.ComSpec || "cmd.exe" : bin;
+  const commandArgs = wrapper ? ["/d", "/s", "/c", `"${[bin, ...args].map((part) => `"${part}"`).join(" ")}"`] : args;
+
   if (opts.inheritStdio) {
     return await new Promise((resolve, reject) => {
-      const child = spawn(bin, args, { stdio: "inherit" });
+      const child = spawn(executable, commandArgs, { stdio: "inherit", windowsVerbatimArguments: wrapper });
       child.on("error", reject);
       child.on("close", (code) => {
         resolve({ stdout: "", stderr: "", code: code ?? 1 });
@@ -71,9 +118,10 @@ export async function runDevin(
   }
 
   try {
-    const { stdout, stderr } = await execFileAsync(bin, args, {
+    const { stdout, stderr } = await execFileAsync(executable, commandArgs, {
       timeout: opts.timeoutMs ?? 30_000,
       maxBuffer: 8 * 1024 * 1024,
+      windowsVerbatimArguments: wrapper,
     });
     return { stdout, stderr, code: 0 };
   } catch (error) {

src/context-map.ts

@@ -1,4 +1,12 @@
-import type { Context, Message, Tool } from "@earendil-works/pi-ai";
+import {
+  collapseSystemMessages,
+  getCurrentSystemPrompt,
+  getCurrentTools,
+  withoutInitialSystemMessage,
+  type Message,
+  type Tool,
+  type TranscriptContext,
+} from "@earendil-works/pi-ai";
 
 export interface ContentPart {
   type: "text" | "image";
@@ -8,7 +16,7 @@ export interface ContentPart {
 }
 
 export interface ChatHistoryItem {
-  role: "user" | "assistant" | "system" | "tool";
+  role: "user" | "assistant" | "tool";
   content: string | ContentPart[];
   tool_call_id?: string;
   tool_calls?: Array<{ id: string; name: string; arguments: string }>;
@@ -21,6 +29,8 @@ export interface ToolDef {
 }
 
 export interface MappedChat {
+  /** Sent in GetChatMessageRequest.prompt, separately from conversation history. */
+  systemPrompt?: string;
   messages: ChatHistoryItem[];
   tools: ToolDef[];
 }
@@ -39,13 +49,12 @@ function userContent(content: Message["content"]): string | ContentPart[] {
   return parts;
 }
 
-export function mapContextToChat(context: Context): MappedChat {
+export function mapContextToChat(context: TranscriptContext): MappedChat {
+  // Fold later system updates into the single prompt sent separately to Devin.
+  const transcript = collapseSystemMessages(context);
+  const systemPrompt = getCurrentSystemPrompt(transcript.messages);
   const messages: ChatHistoryItem[] = [];
-  if (context.systemPrompt) {
-    messages.push({ role: "system", content: context.systemPrompt });
-  }
-
-  for (const message of context.messages) {
+  for (const message of withoutInitialSystemMessage(transcript.messages)) {
     if (message.role === "user") {
       messages.push({ role: "user", content: userContent(message.content) });
       continue;
@@ -86,11 +95,11 @@ export function mapContextToChat(context: Context): MappedChat {
     }
   }
 
-  const tools: ToolDef[] = (context.tools ?? []).map((tool: Tool) => ({
+  const tools: ToolDef[] = getCurrentTools(transcript.messages).map((tool: Tool) => ({
     name: tool.name,
     description: tool.description,
     parameters: tool.parameters,
   }));
 
-  return { messages, tools };
+  return { systemPrompt: systemPrompt || undefined, messages, tools };
 }

src/metadata.ts

@@ -7,10 +7,11 @@ import {
 } from "./wire.js";
 
 /**
- * Cognition version-gates GetChatMessage against a known Windsurf/Devin Desktop
- * release. The Devin CLI build id (`3000.4.25`) is not that string and the
- * server answers: "Your Windsurf version is out of date."
- * Prefer the installed Devin.app product.json, then a current desktop release.
+ * Cognition gates Devin Local-only models (every GPT-5.6 variant: Sol, Terra,
+ * Luna) by client ide name. With ide="windsurf" GetChatMessage rejects them
+ * with "This model is only in Devin Local."; with ide="devin-desktop" the
+ * server serves them and the response header echoes the exact model
+ * (verified: "GPT-5.6 Sol High Thinking" for gpt-5-6-sol-high, 2026-08-29).
  */
 const FALLBACK_WINDSURF_VERSION = "3.6.27";
 const PRODUCT_JSON =
@@ -29,7 +30,7 @@ function desktopWindsurfVersion(): string {
 }
 
 export const CLIENT_VERSION = desktopWindsurfVersion();
-export const CLIENT_IDE = "windsurf";
+export const CLIENT_IDE = "devin-desktop";
 
 export interface MetadataInput {
   apiKey: string;

src/stream.ts

@@ -4,11 +4,12 @@ import {
   type Api,
   type AssistantMessage,
   type AssistantMessageEventStream,
-  type Context,
   type Model,
   type SimpleStreamOptions,
+  type TranscriptContext,
   calculateCost,
   createAssistantMessageEventStream,
+  parseStreamingJson,
 } from "@earendil-works/pi-ai";
 import { mapContextToChat, type ChatHistoryItem, type ContentPart, type ToolDef } from "./context-map.js";
 import { getCachedUserJwt } from "./jwt.js";
@@ -23,10 +24,9 @@ import {
   iterFields,
 } from "./wire.js";
 
-const SOURCE_BY_ROLE: Record<string, number> = {
+const SOURCE_BY_ROLE: Record<ChatHistoryItem["role"], number> = {
   user: 1,
   assistant: 2,
-  system: 1,
   tool: 4,
 };
 
@@ -50,40 +50,6 @@ function normalizeContent(content: string | ContentPart[]): ContentPart[] {
   return content;
 }
 
-function collapseSystemIntoUser(messages: ChatHistoryItem[]): ChatHistoryItem[] {
-  const out: ChatHistoryItem[] = [];
-  let pending: string[] = [];
-  const textOf = (content: string | ContentPart[]) =>
-    normalizeContent(content)
-      .filter((part) => part.type === "text" && part.text)
-      .map((part) => part.text as string)
-      .join("\n");
-
-  for (const message of messages) {
-    if (message.role === "system") {
-      const text = textOf(message.content);
-      if (text) pending.push(text);
-      continue;
-    }
-    if (message.role === "user" && pending.length > 0) {
-      const userParts = normalizeContent(message.content);
-      const userText = textOf(userParts);
-      const images = userParts.filter((part) => part.type === "image");
-      out.push({
-        role: "user",
-        content: [{ type: "text", text: `<system>\n${pending.join("\n\n")}\n</system>\n${userText}` }, ...images],
-      });
-      pending = [];
-      continue;
-    }
-    out.push(message);
-  }
-  if (pending.length > 0) {
-    out.push({ role: "user", content: `<system>\n${pending.join("\n\n")}\n</system>` });
-  }
-  return out;
-}
-
 function encodeImageData(img: { mimeType?: string; base64Data?: string }): Buffer {
   return Buffer.concat([
     encodeString(1, img.base64Data ?? ""),
@@ -144,6 +110,7 @@ function buildGetChatMessageRequest(args: {
   apiKey: string;
   userJwt: string;
   modelUid: string;
+  systemPrompt?: string;
   messages: ChatHistoryItem[];
   tools?: ToolDef[];
   cascadeId: string;
@@ -160,10 +127,10 @@ function buildGetChatMessageRequest(args: {
     requestId: args.requestId,
     triggerId: args.triggerId,
   });
-  const prompts = collapseSystemIntoUser(args.messages).map((message) =>
+  const prompts = args.messages.map((message) =>
     encodeMessage(
       3,
-      encodeChatMessagePrompt(normalizeContent(message.content), SOURCE_BY_ROLE[message.role] ?? 1, {
+      encodeChatMessagePrompt(normalizeContent(message.content), SOURCE_BY_ROLE[message.role], {
         toolCallId: message.role === "tool" ? message.tool_call_id : undefined,
         toolCalls: message.role === "assistant" ? message.tool_calls : undefined,
       }),
@@ -171,6 +138,7 @@ function buildGetChatMessageRequest(args: {
   );
   return Buffer.concat([
     encodeMessage(1, metadata),
+    ...(args.systemPrompt ? [encodeString(2, args.systemPrompt)] : []),
     ...prompts,
     encodeVarintField(7, 5),
     encodeMessage(8, encodeCompletionConfiguration(args.maxOutputTokens)),
@@ -244,12 +212,12 @@ function decodeUsage(buf: Buffer): CloudChatEvent | null {
     else if (metric.includes("cached") || metric.includes("cache_read")) cachedInputTokens = n;
     else if (metric.includes("cache_creation")) cacheCreationInputTokens = n;
   }
-  if (promptTokens === undefined && completionTokens === undefined) return null;
+  if ([promptTokens, completionTokens, cachedInputTokens, cacheCreationInputTokens].every((n) => n === undefined)) return null;
   return {
     kind: "usage",
     promptTokens,
     completionTokens,
-    totalTokens: (promptTokens ?? 0) + (completionTokens ?? 0),
+    totalTokens: (promptTokens ?? 0) + (completionTokens ?? 0) + (cachedInputTokens ?? 0) + (cacheCreationInputTokens ?? 0),
     cachedInputTokens,
     cacheCreationInputTokens,
   };
@@ -271,6 +239,7 @@ async function* streamChatEvents(args: {
   apiKey: string;
   host: string;
   modelUid: string;
+  systemPrompt?: string;
   messages: ChatHistoryItem[];
   tools?: ToolDef[];
   maxOutputTokens?: number;
@@ -283,6 +252,7 @@ async function* streamChatEvents(args: {
     apiKey: args.apiKey,
     userJwt,
     modelUid: args.modelUid,
+    systemPrompt: args.systemPrompt,
     messages: args.messages,
     tools: args.tools,
     cascadeId: ids.cascadeId,
@@ -390,7 +360,7 @@ async function* streamChatEvents(args: {
       // ignore
     }
     try {
-      void resp.body?.cancel();
+      await resp.body?.cancel();
     } catch {
       // ignore
     }
@@ -402,7 +372,7 @@ async function* streamChatEvents(args: {
 
 export function streamDevin(
   model: Model<Api>,
-  context: Context,
+  context: TranscriptContext,
   options?: SimpleStreamOptions,
 ): AssistantMessageEventStream {
   const stream = createAssistantMessageEventStream();
@@ -455,11 +425,7 @@ export function streamDevin(
       if (toolIndex < 0) return;
       const block = output.content[toolIndex];
       if (block.type === "toolCall") {
-        try {
-          block.arguments = JSON.parse(partialJson);
-        } catch {
-          // keep last parsed object
-        }
+        block.arguments = parseStreamingJson(partialJson);
         stream.push({
           type: "toolcall_end",
           contentIndex: toolIndex,
@@ -482,6 +448,7 @@ export function streamDevin(
         apiKey,
         host,
         modelUid,
+        systemPrompt: mapped.systemPrompt,
         messages: mapped.messages,
         tools: mapped.tools.length > 0 ? mapped.tools : undefined,
         maxOutputTokens: options?.maxTokens,
@@ -528,11 +495,7 @@ export function streamDevin(
           partialJson += event.argsDelta;
           const block = output.content[toolIndex];
           if (block.type === "toolCall") {
-            try {
-              block.arguments = JSON.parse(partialJson);
-            } catch {
-              // incomplete json
-            }
+            block.arguments = parseStreamingJson(partialJson);
           }
           stream.push({ type: "toolcall_delta", contentIndex: toolIndex, delta: event.argsDelta, partial: output });
         } else if (event.kind === "finish") {
@@ -542,11 +505,14 @@ export function streamDevin(
           output.stopReason =
             event.reason === "tool_calls" ? "toolUse" : event.reason === "length" ? "length" : "stop";
         } else if (event.kind === "usage") {
-          output.usage.input = event.promptTokens ?? 0;
-          output.usage.output = event.completionTokens ?? 0;
-          output.usage.cacheRead = event.cachedInputTokens ?? 0;
-          output.usage.cacheWrite = event.cacheCreationInputTokens ?? 0;
-          output.usage.totalTokens = event.totalTokens ?? output.usage.input + output.usage.output;
+          // Empty trailers must not erase usage already received. Metrics are
+          // snapshots, and a frame may contain only some of the components.
+          if (!event.totalTokens) continue;
+          output.usage.input = event.promptTokens ?? output.usage.input;
+          output.usage.output = event.completionTokens ?? output.usage.output;
+          output.usage.cacheRead = event.cachedInputTokens ?? output.usage.cacheRead;
+          output.usage.cacheWrite = event.cacheCreationInputTokens ?? output.usage.cacheWrite;
+          output.usage.totalTokens = output.usage.input + output.usage.output + output.usage.cacheRead + output.usage.cacheWrite;
           calculateCost(model, output.usage);
         }
       }

test-runner.mjs

@@ -0,0 +1,8 @@
+import { readdirSync } from "node:fs";
+import { spawnSync } from "node:child_process";
+import { join } from "node:path";
+
+const files = readdirSync("test").filter((file) => file.endsWith(".test.mjs")).sort().map((file) => join("test", file));
+const result = spawnSync(process.execPath, ["--import", "tsx", "--test", ...files], { stdio: "inherit" });
+if (result.error) throw result.error;
+process.exit(result.status ?? 1);

test/catalog-cache.test.mjs

@@ -0,0 +1,65 @@
+import assert from "node:assert/strict";
+import { mkdtempSync, rmSync, readFileSync, writeFileSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import test from "node:test";
+import {
+  catalogCachePath,
+  isUsableCatalog,
+  isCatalogCacheFresh,
+  readCatalogCache,
+  writeCatalogCache,
+} from "../src/catalog-cache.ts";
+
+const catalog = {
+  families: [{
+    family_label: "SWE-2",
+    family_uid: "swe-2",
+    slug: "swe-2",
+    variants: [{ model_uid: "swe-2-high", label: "SWE-2 High" }],
+  }],
+};
+
+test("writes and reads a validated catalog atomically", (t) => {
+  const directory = mkdtempSync(join(tmpdir(), "pi-devin-cache-"));
+  t.after(() => rmSync(directory, { recursive: true, force: true }));
+  const path = join(directory, "nested", "models.json");
+  writeCatalogCache(catalog, path, 1_000);
+  assert.deepEqual(readCatalogCache(path), { catalog, fetchedAt: 1_000 });
+  assert.equal(JSON.parse(readFileSync(path, "utf8")).version, 1);
+});
+
+test("rejects malformed cache content", (t) => {
+  const directory = mkdtempSync(join(tmpdir(), "pi-devin-cache-"));
+  t.after(() => rmSync(directory, { recursive: true, force: true }));
+  const path = join(directory, "models.json");
+  writeFileSync(path, JSON.stringify({ version: 1, fetchedAt: 1_000, catalog: { families: [{}] } }));
+  assert.throws(() => readCatalogCache(path), /Invalid Devin model catalog cache/);
+});
+
+test("rejects catalogs without usable model families", () => {
+  assert.equal(isUsableCatalog({ families: [] }), false);
+  assert.equal(isUsableCatalog({
+    families: [{ family_label: "Empty", family_uid: "empty", slug: "empty", variants: [] }],
+  }), false);
+});
+
+test("checks cache age and future timestamps", () => {
+  const cached = { catalog, fetchedAt: 1_000 };
+  assert.equal(isCatalogCacheFresh(cached, 1_500, 500), true);
+  assert.equal(isCatalogCacheFresh(cached, 1_501, 500), false);
+  assert.equal(isCatalogCacheFresh(cached, 999, 500), false);
+});
+
+test("uses XDG cache directory when configured", () => {
+  assert.equal(catalogCachePath({ XDG_CACHE_HOME: "/cache" }, "/home/test"), join("/cache", "pi-devin", "models.json"));
+  assert.equal(catalogCachePath({}, "/home/test"), join("/home/test", ".cache", "pi-devin", "models.json"));
+});
+
+for (const [field, value] of Object.entries({ cost_summary: {}, max_context_tokens: "1000", max_output_tokens: -1, model_uid: "", is_beta: "yes" })) {
+  test(`rejects invalid ${field} before model conversion`, () => {
+    const malformed = structuredClone(catalog);
+    malformed.families[0].variants[0][field] = value;
+    assert.equal(isUsableCatalog(malformed), false);
+  });
+}

test/catalog-startup.test.mjs

@@ -0,0 +1,101 @@
+import assert from "node:assert/strict";
+import test, { after } from "node:test";
+import fs from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { syncBuiltinESMExports } from "node:module";
+import { catalogCachePath, writeCatalogCache } from "../src/catalog-cache.ts";
+
+const directory = fs.mkdtempSync(join(tmpdir(), "pi-devin-catalog-test-"));
+after(() => fs.rmSync(directory, { recursive: true, force: true }));
+const cli = join(directory, process.platform === "win32" ? "devin.cmd" : "devin");
+const counter = join(directory, "calls");
+const catalog = { families: [{ family_label: "Test", family_uid: "test", slug: "test", variants: [{ model_uid: "test-high", label: "Test high", cost_summary: "$1/MTok In, $2/MTok Out" }] }] };
+const script = join(directory, "catalog.cjs");
+fs.writeFileSync(script, `const fs = require('node:fs');\nfs.appendFileSync(${JSON.stringify(counter)}, 'call\\n');\nsetTimeout(() => { console.log(process.env.TEST_CATALOG); process.exit(Number(process.env.TEST_CLI_EXIT || 0)); }, 30);\n`);
+fs.writeFileSync(cli, process.platform === "win32" ? `@echo off\r\n"${process.execPath}" "${script}"\r\n` : `#!/bin/sh\nexec "${process.execPath}" "${script}"\n`, { mode: 0o700 });
+const originalEnv = { ...process.env };
+process.env.DEVIN_CLI = cli;
+const { default: extension } = await import("../extensions/index.ts");
+const { clearDevinBinCache } = await import("../src/cli.ts");
+after(() => { if (originalEnv.DEVIN_CLI === undefined) delete process.env.DEVIN_CLI; else process.env.DEVIN_CLI = originalEnv.DEVIN_CLI; });
+
+function setup(t) {
+  clearDevinBinCache();
+  const root = fs.mkdtempSync(join(directory, "cache-"));
+  const previous = { ...process.env };
+  process.env.XDG_CACHE_HOME = root;
+  process.env.TEST_CATALOG = JSON.stringify(catalog);
+  delete process.env.PI_OFFLINE;
+  delete process.env.TEST_CLI_EXIT;
+  fs.writeFileSync(counter, "");
+  const read = fs.readFileSync, exists = fs.existsSync;
+  const isCredentials = (path) => String(path).replaceAll("\\", "/").endsWith("devin/credentials.toml");
+  t.mock.method(fs, "readFileSync", (path, ...args) => isCredentials(path) ? 'api_key = "synthetic-test-key"\n' : read(path, ...args));
+  t.mock.method(fs, "existsSync", (path) => isCredentials(path) || exists(path));
+  t.mock.method(console, "warn", () => {});
+  syncBuiltinESMExports();
+  const registrations = [], hooks = {}, commands = {};
+  const pi = { registerProvider: (_id, config) => registrations.push(config), on: (name, fn) => hooks[name] = fn, registerCommand: (name, command) => commands[name] = command };
+  t.after(async () => {
+    await hooks.session_shutdown?.();
+    t.mock.restoreAll(); syncBuiltinESMExports();
+    for (const key of ["XDG_CACHE_HOME", "PI_OFFLINE", "TEST_CATALOG", "TEST_CLI_EXIT"]) {
+      if (previous[key] === undefined) delete process.env[key]; else process.env[key] = previous[key];
+    }
+  });
+  return { registrations, hooks, commands, pi, path: catalogCachePath(), calls: () => fs.readFileSync(counter, "utf8").split("\n").filter(Boolean).length };
+}
+
+test("fresh cache registers once without invoking the CLI", async (t) => {
+  const x = setup(t); writeCatalogCache(catalog);
+  await extension(x.pi);
+  assert.equal(x.calls(), 0); assert.equal(x.registrations.length, 1);
+  assert.equal(x.registrations[0].models[0].id, "test-high");
+});
+
+test("malformed cached cost data cannot prevent provider and command registration", async (t) => {
+  const x = setup(t); process.env.PI_OFFLINE = "1";
+  const malformed = structuredClone(catalog); malformed.families[0].variants[0].cost_summary = {};
+  fs.mkdirSync(join(process.env.XDG_CACHE_HOME, "pi-devin"));
+  fs.writeFileSync(x.path, JSON.stringify({ version: 1, fetchedAt: Date.now(), catalog: malformed }));
+  await extension(x.pi);
+  assert.equal(x.registrations.length, 1); assert.ok(x.registrations[0].models.length);
+  assert.ok(x.commands["devin-refresh"]); assert.equal(x.calls(), 0);
+});
+
+test("cold startup loads and caches the CLI catalog", async (t) => {
+  const x = setup(t); await extension(x.pi);
+  assert.equal(x.calls(), 1); assert.equal(x.registrations.at(-1).models[0].id, "test-high");
+  assert.deepEqual(JSON.parse(fs.readFileSync(x.path, "utf8")).catalog, catalog);
+});
+
+test("stale cache shares its background request with manual refresh", async (t) => {
+  const x = setup(t); writeCatalogCache(catalog, x.path, 1);
+  await extension(x.pi);
+  const notifications = [];
+  await x.commands["devin-refresh"].handler("", { ui: { notify: (...args) => notifications.push(args) } });
+  assert.equal(x.calls(), 1); assert.equal(notifications.at(-1)[1], "info");
+  assert.equal(x.registrations.at(-1).models[0].id, "test-high");
+});
+
+test("failed refresh retains cached models and does not overwrite the cache", async (t) => {
+  const x = setup(t); writeCatalogCache(catalog, x.path, 1); process.env.TEST_CATALOG = '{"families":[{}]}';
+  await extension(x.pi);
+  const notifications = [];
+  await x.commands["devin-refresh"].handler("", { ui: { notify: (...args) => notifications.push(args) } });
+  assert.equal(notifications.at(-1)[1], "error"); assert.equal(x.registrations.length, 1);
+  assert.deepEqual(JSON.parse(fs.readFileSync(x.path, "utf8")).catalog, catalog);
+});
+
+test("offline cold startup uses fallback without running the CLI", async (t) => {
+  const x = setup(t); process.env.PI_OFFLINE = "true";
+  await extension(x.pi); assert.equal(x.calls(), 0); assert.ok(x.registrations[0].models.length);
+});
+
+test("shutdown prevents a late background result from registering a provider", async (t) => {
+  const x = setup(t); writeCatalogCache(catalog, x.path, 1);
+  await extension(x.pi); await x.hooks.session_shutdown();
+  await x.commands["devin-refresh"].handler("", { ui: { notify() {} } });
+  assert.equal(x.registrations.length, 1);
+});

test/cli.test.mjs

@@ -0,0 +1,73 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import { mkdtempSync, mkdirSync, writeFileSync, rmSync, realpathSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { dirname, join } from "node:path";
+import { spawnSync } from "node:child_process";
+import { clearDevinBinCache, findDevinBin, whichDevin, runDevin } from "../src/cli.ts";
+
+function fixture(t) {
+  const directory = mkdtempSync(join(tmpdir(), "pi devin cli "));
+  // Windows environment lookups ignore case; a spread object does not.
+  const keys = ["DEVIN_CLI", "LOCALAPPDATA", "ProgramFiles", "PATH"];
+  const original = Object.fromEntries(keys.map((key) => [key, process.env[key]]));
+  clearDevinBinCache();
+  t.after(() => {
+    for (const key of keys) {
+      if (original[key] === undefined) delete process.env[key]; else process.env[key] = original[key];
+    }
+    clearDevinBinCache(); rmSync(directory, { recursive: true, force: true });
+  });
+  return directory;
+}
+
+test("honors an override set after the discovery cache has already been populated", async (t) => {
+  const directory = fixture(t);
+  const first = join(directory, "first"), second = join(directory, "second");
+  writeFileSync(first, ""); writeFileSync(second, "");
+  process.env.DEVIN_CLI = first; assert.equal(findDevinBin(), first);
+  process.env.DEVIN_CLI = second; assert.equal(await whichDevin(), second);
+});
+
+test("runs a native executable directly with argument boundaries intact", async (t) => {
+  fixture(t); process.env.DEVIN_CLI = process.execPath;
+  const value = 'spaces & punctuation "stay literal"';
+  const result = await runDevin(["-e", "process.stdout.write(process.argv[1])", value]);
+  assert.equal(result.code, 0); assert.equal(result.stdout, value);
+});
+
+test("finds the Windows installer binary and ignores a directory override", { skip: process.platform !== "win32" }, (t) => {
+  const directory = fixture(t); process.env.LOCALAPPDATA = directory;
+  process.env.DEVIN_CLI = directory;
+  const binary = join(directory, "devin", "cli", "bin", "devin.exe");
+  mkdirSync(dirname(binary), { recursive: true }); writeFileSync(binary, "");
+  assert.equal(findDevinBin(), binary);
+});
+
+test("Windows PATH lookup uses where.exe and reads multiple results", { skip: process.platform !== "win32" }, async (t) => {
+  const directory = fixture(t); delete process.env.DEVIN_CLI;
+  process.env.LOCALAPPDATA = directory; process.env.ProgramFiles = directory;
+  const bin = join(directory, "path bin"); mkdirSync(bin);
+  const binary = join(bin, "devin.exe"); writeFileSync(binary, "");
+  const otherBin = join(directory, "other bin"); mkdirSync(otherBin);
+  writeFileSync(join(otherBin, "devin.exe"), "");
+  process.env.PATH = `${bin};${otherBin};${process.env.PATH}`;
+  const located = await whichDevin();
+  const probe = spawnSync("where.exe", ["devin.exe"], { encoding: "utf8" });
+  assert.ok(located, JSON.stringify({ status: probe.status, stdout: probe.stdout, stderr: probe.stderr, error: probe.error?.message }));
+  // where.exe expands Windows short paths such as RUNNER~1.
+  assert.equal(realpathSync.native(located), realpathSync.native(binary));
+  assert.ok(probe.stdout.trim().split(/\r?\n/).length >= 2);
+});
+
+test("Windows batch wrappers work in paths with spaces, including inherited stdio", { skip: process.platform !== "win32" }, async (t) => {
+  const directory = fixture(t), script = join(directory, "echo.cjs"), wrapper = join(directory, "devin.cmd");
+  writeFileSync(script, 'process.stdout.write(JSON.stringify(process.argv.slice(2)));');
+  writeFileSync(wrapper, `@echo off\r\n"${process.execPath}" "${script}" %*\r\n`);
+  process.env.DEVIN_CLI = wrapper;
+  const result = await runDevin(["models", "list", "--format", "json", "a b"]);
+  assert.equal(result.code, 0, result.stderr);
+  assert.deepEqual(JSON.parse(result.stdout), ["models", "list", "--format", "json", "a b"]);
+  assert.equal((await runDevin(["version"], { inheritStdio: true })).code, 0);
+  await assert.rejects(runDevin(["%PATH%"]), /Unsupported shell characters/);
+});

test/context-map.test.mjs

@@ -0,0 +1,53 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import { normalizeContext } from "@earendil-works/pi-ai";
+import { mapContextToChat } from "../src/context-map.ts";
+
+const tool = (name) => ({
+  name,
+  description: name,
+  parameters: { type: "object", properties: {} },
+});
+
+// Regression for #7: Pi 0.86 moves prompt and tools into system messages.
+test("preserves the prompt and tools from a normalized transcript", () => {
+  const read = tool("read");
+  const context = normalizeContext({
+    systemPrompt: "Inspect the repository.",
+    tools: [read],
+    messages: [{ role: "user", content: "Read README", timestamp: 1 }],
+  });
+  assert.deepEqual(mapContextToChat(context), {
+    systemPrompt: "Inspect the repository.",
+    messages: [
+      { role: "user", content: "Read README" },
+    ],
+    tools: [read],
+  });
+});
+
+test("applies later instruction, section, and tool changes", () => {
+  const bash = tool("bash");
+  const context = {
+    messages: [
+      { role: "system", content: "Inspect the repository.", sections: { policy: "Old policy", obsolete: "Remove me" }, toolsAdded: [tool("read")], timestamp: 0 },
+      { role: "user", content: "Continue", timestamp: 1 },
+      { role: "system", content: "Use bash.", sections: { policy: "Do not edit.", obsolete: null }, toolsRemoved: [{ name: "read" }], toolsAdded: [bash], timestamp: 2 },
+    ],
+  };
+  assert.deepEqual(mapContextToChat(context), {
+    systemPrompt: "Inspect the repository.\n\nUse bash.\n\nDo not edit.",
+    messages: [
+      { role: "user", content: "Continue" },
+    ],
+    tools: [bash],
+  });
+});
+
+test("accepts an empty transcript", () => {
+  assert.deepEqual(mapContextToChat(normalizeContext({ messages: [] })), {
+    systemPrompt: undefined,
+    messages: [],
+    tools: [],
+  });
+});

test/stream-abort.test.mjs

@@ -0,0 +1,104 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import { normalizeContext } from "@earendil-works/pi-ai";
+import { clearCachedUserJwt } from "../src/jwt.ts";
+import { streamDevin } from "../src/stream.ts";
+import { encodeMessage } from "../src/wire.ts";
+
+const model = {
+  id: "test-model",
+  name: "Test model",
+  api: "devin-local",
+  provider: "devin",
+  baseUrl: "https://devin.invalid",
+  reasoning: false,
+  input: ["text", "image"],
+  cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
+  contextWindow: 262_000,
+  maxTokens: 1_000,
+};
+const user = (content) => ({ role: "user", content, timestamp: 1 });
+
+test("aborting an in-flight stream reports aborted without leaking a rejection", async (t) => {
+  clearCachedUserJwt();
+  t.after(clearCachedUserJwt);
+
+  const rejections = [];
+  const onRejection = (reason) => rejections.push(reason);
+  process.on("unhandledRejection", onRejection);
+  t.after(() => process.off("unhandledRejection", onRejection));
+
+  const ac = new AbortController();
+  t.mock.method(globalThis, "fetch", async (url, options) => {
+    if (url === "https://devin.invalid/exa.auth_pb.AuthService/GetUserJwt") {
+      return new Response(encodeMessage(1, Buffer.from("eyJtest.jwt")));
+    }
+    assert.equal(url, "https://devin.invalid/exa.api_server_pb.ApiServerService/GetChatMessage");
+    // Mirror undici: an aborted fetch errors the response body with signal.reason.
+    const body = new ReadableStream({
+      start(controller) {
+        controller.enqueue(new TextEncoder().encode("partial"));
+        const onAbort = () => controller.error(options.signal.reason);
+        if (options.signal?.aborted) onAbort();
+        else options.signal?.addEventListener("abort", onAbort, { once: true });
+      },
+    });
+    return new Response(body);
+  });
+
+  const stream = streamDevin(model, normalizeContext({ messages: [user("hello")] }), {
+    apiKey: "synthetic-test-key",
+    env: { DEVIN_API_SERVER_URL: "https://devin.invalid" },
+    signal: ac.signal,
+  });
+
+  const events = [];
+  const consume = (async () => {
+    for await (const event of stream) events.push(event);
+  })();
+  ac.abort();
+  await consume;
+  // Give any floating rejection a chance to surface.
+  await new Promise((resolve) => setTimeout(resolve, 50));
+
+  const last = events.at(-1);
+  assert.equal(last?.type, "error");
+  assert.equal(last?.reason, "aborted");
+  assert.equal((await stream.result()).stopReason, "aborted");
+  assert.deepEqual(rejections, []);
+});
+
+test("real fetch cancellation after a response chunk leaves the process usable", { timeout: 10_000 }, async (t) => {
+  const { createServer } = await import("node:http");
+  const { once } = await import("node:events");
+  const { encodeString, encodeVarintField, frameConnectStream } = await import("../src/wire.ts");
+  clearCachedUserJwt(); t.after(clearCachedUserJwt);
+  let chats = 0;
+  const server = createServer((req, res) => {
+    if (req.url.endsWith("/GetUserJwt")) {
+      res.end(encodeString(1, "eyJtest.jwt")); return;
+    }
+    chats++;
+    res.writeHead(200, { "Content-Type": "application/connect+proto" });
+    res.write(frameConnectStream(encodeString(3, "started"), false));
+    if (chats > 1) {
+      res.write(frameConnectStream(encodeVarintField(5, 0), false));
+      res.end(Buffer.from([2, 0, 0, 0, 2, 123, 125]));
+    }
+  });
+  t.after(() => { server.closeAllConnections(); server.close(); });
+  server.listen(0, "127.0.0.1"); await once(server, "listening");
+  const ac = new AbortController();
+  const options = { apiKey: "synthetic-test-key", env: { DEVIN_API_SERVER_URL: `http://127.0.0.1:${server.address().port}` } };
+  const stream = streamDevin(model, normalizeContext({ messages: [user("hello")] }), { ...options, signal: ac.signal });
+  let sawText = false;
+  for await (const event of stream) {
+    if (event.type === "text_delta") { sawText = true; ac.abort(); }
+  }
+  assert.equal(sawText, true);
+  assert.equal((await stream.result()).stopReason, "aborted");
+  const next = streamDevin(model, normalizeContext({ messages: [user("again")] }), options);
+  for await (const event of next) assert.notEqual(event.type, "error");
+  assert.equal((await next.result()).stopReason, "stop");
+  assert.equal(chats, 2);
+});

test/stream-deltas.test.mjs

@@ -0,0 +1,86 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import { setImmediate } from "node:timers/promises";
+import { streamDevin } from "../src/stream.ts";
+import { clearCachedUserJwt } from "../src/jwt.ts";
+import { encodeMessage, encodeString, encodeTag, encodeVarintField, frameConnectStream } from "../src/wire.ts";
+
+const model = {
+  id: "test-model", name: "Test", api: "devin-local", provider: "devin",
+  baseUrl: "https://devin.invalid", reasoning: false, input: ["text"],
+  cost: { input: 1, output: 2, cacheRead: 0.1, cacheWrite: 1.25 },
+  contextWindow: 1_000_000, maxTokens: 1000,
+};
+
+async function run(t, frames) {
+  clearCachedUserJwt();
+  t.after(clearCachedUserJwt);
+  t.mock.method(globalThis, "fetch", async (url) => {
+    if (url === "https://devin.invalid/exa.auth_pb.AuthService/GetUserJwt") {
+      return new Response(encodeString(1, "eyJtest.jwt"));
+    }
+    assert.equal(url, "https://devin.invalid/exa.api_server_pb.ApiServerService/GetChatMessage");
+    const chunks = [...frames.map((frame) => frameConnectStream(frame, false)), Buffer.from([2, 0, 0, 0, 2, 123, 125])];
+    return new Response(new ReadableStream({
+      async pull(controller) {
+        await setImmediate();
+        if (chunks.length) controller.enqueue(chunks.shift());
+        else controller.close();
+      },
+    }));
+  });
+  const stream = streamDevin(model, { messages: [] }, {
+    apiKey: "synthetic-test-key", env: { DEVIN_API_SERVER_URL: "https://devin.invalid" },
+  });
+  const events = [];
+  for await (const event of stream) events.push(structuredClone(event));
+  const result = await stream.result();
+  assert.notEqual(result.stopReason, "error", result.errorMessage);
+  return { events, result };
+}
+
+function usage(metrics) {
+  return encodeMessage(28, Buffer.concat(Object.entries(metrics).map(([name, value]) => {
+    const float = Buffer.alloc(4);
+    float.writeFloatLE(value);
+    return encodeMessage(2, Buffer.concat([
+      encodeString(5, name), encodeMessage(4, Buffer.concat([encodeTag(2, 5), float])),
+    ]));
+  })));
+}
+
+const metrics = { input_tokens: 461, output_tokens: 262, cached_input_tokens: 228_162, cache_creation_input_tokens: 17 };
+
+test("counts all four usage components and preserves them after an empty frame (#3, #4)", async (t) => {
+  const { result } = await run(t, [usage(metrics), usage(Object.fromEntries(Object.keys(metrics).map((key) => [key, 0])))]);
+  assert.deepEqual({ ...result.usage, cost: undefined }, {
+    input: 461, output: 262, cacheRead: 228_162, cacheWrite: 17, totalTokens: 228_902, cost: undefined,
+  });
+  assert.equal(result.usage.cost.input, 461 / 1_000_000);
+  assert.equal(result.usage.cost.cacheRead, 228_162 * 0.1 / 1_000_000);
+});
+
+test("accepts cache-only usage and combines separate component snapshots without double counting", async (t) => {
+  const { result } = await run(t, [
+    usage({ cached_input_tokens: 58_905 }), usage({ input_tokens: 280, output_tokens: 118 }),
+    usage({ input_tokens: 280, output_tokens: 118 }), usage({ cache_creation_input_tokens: 17 }),
+  ]);
+  assert.equal(result.usage.totalTokens, 59_320);
+  assert.equal(result.usage.cacheRead, 58_905);
+  assert.equal(result.usage.cacheWrite, 17);
+});
+
+test("keeps growing tool arguments across split keys, strings, and Unicode escapes (#2)", async (t) => {
+  const fragments = ['{"path":"file.txt","cont', 'ent":"hello ', '\\u26', '3a"}'];
+  const frames = fragments.map((part, index) => encodeMessage(6, Buffer.concat([
+    ...(index === 0 ? [encodeString(1, "call-1"), encodeString(2, "write")] : []), encodeString(3, part),
+  ])));
+  frames.push(encodeVarintField(5, 10));
+  const { events, result } = await run(t, frames);
+  const deltas = events.filter((event) => event.type === "toolcall_delta");
+  assert.equal(deltas.length, 4);
+  for (const event of deltas) assert.equal(event.partial.content[0].arguments.path, "file.txt");
+  assert.equal(deltas[1].partial.content[0].arguments.content, "hello");
+  assert.deepEqual(result.content[0].arguments, { path: "file.txt", content: "hello ☺" });
+  assert.equal(result.stopReason, "toolUse");
+});

test/stream-request.test.mjs

@@ -0,0 +1,164 @@
+import assert from "node:assert/strict";
+import test from "node:test";
+import { gunzipSync } from "node:zlib";
+import { normalizeContext } from "@earendil-works/pi-ai";
+import { clearCachedUserJwt } from "../src/jwt.ts";
+import { streamDevin } from "../src/stream.ts";
+import { encodeMessage, encodeString, encodeVarintField, frameConnectStream, iterFields } from "../src/wire.ts";
+
+const model = {
+  id: "test-model",
+  name: "Test model",
+  api: "devin-local",
+  provider: "devin",
+  baseUrl: "https://devin.invalid",
+  reasoning: false,
+  input: ["text", "image"],
+  cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 },
+  contextWindow: 262_000,
+  maxTokens: 1_000,
+};
+const readTool = {
+  name: "read",
+  description: "Read a file",
+  parameters: { type: "object", properties: { path: { type: "string" } }, required: ["path"] },
+};
+const user = (content) => ({ role: "user", content, timestamp: 1 });
+const fields = (buffer) => [...iterFields(buffer)];
+const messages = (request) => request.filter((field) => field.num === 3).map((field) => fields(field.value));
+const stringField = (fields, number) => fields.find((field) => field.num === number)?.value.toString("utf8");
+
+function response(body) {
+  // Connect's end-of-stream frame contains the JSON trailer {}.
+  return new Response(Buffer.concat([frameConnectStream(body, false), Buffer.from([2, 0, 0, 0, 2, 123, 125])]));
+}
+
+function mockDevin(t, replies = [Buffer.concat([encodeString(3, "OK"), encodeVarintField(5, 0)])]) {
+  const requests = [];
+  clearCachedUserJwt();
+  t.after(clearCachedUserJwt);
+  t.mock.method(globalThis, "fetch", async (url, options) => {
+    if (url === "https://devin.invalid/exa.auth_pb.AuthService/GetUserJwt") {
+      return new Response(encodeMessage(1, Buffer.from("eyJtest.jwt")));
+    }
+    assert.equal(url, "https://devin.invalid/exa.api_server_pb.ApiServerService/GetChatMessage");
+    const frame = Buffer.from(options.body);
+    assert.equal(frame[0], 1, "request uses Connect gzip compression");
+    assert.equal(frame.readUInt32BE(1), frame.length - 5);
+    requests.push(fields(gunzipSync(frame.subarray(5))));
+    const reply = replies[requests.length - 1];
+    assert.ok(reply, "unexpected extra chat request");
+    return response(reply);
+  });
+  return requests;
+}
+
+async function complete(context) {
+  const stream = streamDevin(model, context, {
+    apiKey: "synthetic-test-key",
+    env: { DEVIN_API_SERVER_URL: "https://devin.invalid" },
+  });
+  for await (const event of stream) {
+    assert.notEqual(event.type, "error", event.error?.errorMessage);
+  }
+  return stream.result();
+}
+
+// #7: verify the normalized transcript survives the complete request-encoding path.
+test("encodes the current system prompt once in field 2, with unchanged user text and images", async (t) => {
+  const requests = mockDevin(t);
+  const text = "  Inspect this: <system>literal user text</system>\nこんにちは  ";
+  const image = { type: "image", mimeType: "image/png", data: "aW1hZ2UtZml4dHVyZQ==" };
+  const context = normalizeContext({
+    systemPrompt: "Initial instructions.",
+    tools: [readTool],
+    messages: [user([{ type: "text", text }, image])],
+  });
+  const tools = Array.from({ length: 90 }, (_, i) => ({ ...readTool, name: `read_${i}` }));
+  context.messages.push({
+    role: "system",
+    content: "Updated instructions.",
+    sections: { policy: "Do not edit." },
+    toolsRemoved: [{ name: "read" }],
+    toolsAdded: tools,
+    timestamp: 2,
+  });
+  await complete(context);
+  assert.equal(requests.length, 1);
+  const request = requests[0];
+  const promptFields = request.filter((field) => field.num === 2);
+  assert.equal(promptFields.length, 1);
+  assert.equal(promptFields[0].wire, 2);
+  assert.equal(promptFields[0].value.toString(), "Initial instructions.\n\nUpdated instructions.\n\nDo not edit.");
+  const history = messages(request);
+  assert.equal(history.length, 1);
+  assert.equal(history[0].find((field) => field.num === 2).value, 1n);
+  assert.equal(stringField(history[0], 3), text);
+  const images = history[0].filter((field) => field.num === 10);
+  assert.equal(images.length, 1);
+  assert.equal(stringField(fields(images[0].value), 1), image.data);
+  assert.equal(stringField(fields(images[0].value), 2), image.mimeType);
+  assert.deepEqual(request.filter((field) => field.num === 10).map((field) => {
+    const definition = fields(field.value);
+    return { name: stringField(definition, 1), description: stringField(definition, 2), parameters: JSON.parse(stringField(definition, 3)) };
+  }), tools);
+});
+
+test("omits an empty system prompt while preserving tools and user content", async (t) => {
+  const requests = mockDevin(t);
+  await complete(normalizeContext({ systemPrompt: "", tools: [readTool], messages: [user("Read probe.txt")] }));
+  assert.equal(requests[0].filter((field) => field.num === 2).length, 0);
+  assert.equal(requests[0].filter((field) => field.num === 10).length, 1);
+  assert.deepEqual(messages(requests[0]).map((message) => stringField(message, 3)), ["Read probe.txt"]);
+});
+
+test("sends a system-only transcript without inventing a user message", async (t) => {
+  const requests = mockDevin(t);
+  await complete(normalizeContext({ systemPrompt: "System-only instructions.", messages: [] }));
+  assert.equal(stringField(requests[0], 2), "System-only instructions.");
+  assert.deepEqual(messages(requests[0]), []);
+});
+
+test("encodes an empty transcript without a prompt or history", async (t) => {
+  const requests = mockDevin(t);
+  await complete(normalizeContext({ messages: [] }));
+  assert.equal(stringField(requests[0], 2), undefined);
+  assert.deepEqual(messages(requests[0]), []);
+});
+
+test("retains the system prompt and replays assistant tool calls and results on the next request", async (t) => {
+  const callId = "call-read";
+  const args = { path: "probe.txt" };
+  const requests = mockDevin(t, [
+    Buffer.concat([
+      encodeMessage(6, Buffer.concat([encodeString(1, callId), encodeString(2, "read"), encodeString(3, JSON.stringify(args))])),
+      encodeVarintField(5, 10),
+    ]),
+    Buffer.concat([encodeString(3, "file-value"), encodeVarintField(5, 0)]),
+  ]);
+  const context = normalizeContext({ systemPrompt: "Use tools; never guess.", tools: [readTool], messages: [user("Read probe.txt")] });
+  const first = await complete(context);
+  assert.equal(first.stopReason, "toolUse");
+  assert.deepEqual(first.content, [{ type: "toolCall", id: callId, name: "read", arguments: args }]);
+  context.messages.push(first, {
+    role: "toolResult", toolCallId: callId, toolName: "read",
+    content: [{ type: "text", text: "file-value" }], isError: false, timestamp: 3,
+  });
+  const second = await complete(context);
+  assert.equal(second.stopReason, "stop");
+  assert.equal(second.content[0].text, "file-value");
+  assert.equal(requests.length, 2);
+  for (const request of requests) {
+    assert.equal(request.filter((field) => field.num === 2).length, 1);
+    assert.equal(stringField(request, 2), "Use tools; never guess.");
+  }
+  const history = messages(requests[1]);
+  assert.deepEqual(history.map((message) => message.find((field) => field.num === 2).value), [1n, 2n, 4n]);
+  assert.equal(stringField(history[0], 3), "Read probe.txt");
+  const call = fields(history[1].find((field) => field.num === 6).value);
+  assert.equal(stringField(call, 1), callId);
+  assert.equal(stringField(call, 2), "read");
+  assert.deepEqual(JSON.parse(stringField(call, 3)), args);
+  assert.equal(stringField(history[2], 7), callId);
+  assert.equal(stringField(history[2], 3), "file-value");
+});