docs(spikes): Deno 2.9.7 + TS 7 compatibility spike and ADR-004

Co-Authored-By: Claude Fable 5.1 <[email protected]>

spikes/deno/apps/server/src/db.ts

@@ -0,0 +1,50 @@
+// Kysely 0.29 on Deno: DIALECT=pg (node-postgres Pool) or DIALECT=postgresjs (kysely-postgres-js).
+import { Kysely, PostgresDialect, sql } from "kysely";
+import { FileMigrationProvider, Migrator } from "kysely/migration";
+import { PostgresJSDialect } from "kysely-postgres-js";
+import pg from "pg";
+import postgres from "postgres";
+import { promises as fs } from "node:fs";
+import * as path from "node:path";
+
+const url = Deno.env.get("PG_URL") ?? "postgresql://postgres:[email protected]:55435/postgres";
+const dialectName = Deno.env.get("DIALECT") ?? "pg";
+const schema = Deno.env.get("SCHEMA") ?? `kysely_${dialectName}`;
+
+export function createDb(): Kysely<any> {
+  const dialect = dialectName === "postgresjs"
+    ? new PostgresJSDialect({ postgres: postgres(url, { max: 4, onnotice: () => {} }) })
+    : new PostgresDialect({ pool: new pg.Pool({ connectionString: url, max: 4 }) });
+  return new Kysely<any>({ dialect });
+}
+
+if (import.meta.main) {
+  const root = createDb();
+  await sql`DROP SCHEMA IF EXISTS ${sql.id(schema)} CASCADE`.execute(root);
+  await sql`CREATE SCHEMA ${sql.id(schema)}`.execute(root);
+  const db = root.withSchema(schema);
+  const migrator = new Migrator({
+    db: root,
+    migrationTableSchema: schema,
+    provider: new FileMigrationProvider({ fs, path, migrationFolder: Deno.env.get("MIGRATIONS_DIR") ?? new URL("./migrations", import.meta.url).pathname }),
+  });
+  // Migrations use unqualified names: run them with search_path pointing at the dialect's schema.
+  await sql`SET search_path TO ${sql.id(schema)}`.execute(root);
+  const r = await migrator.migrateToLatest();
+  console.log(dialectName, "migrate:", JSON.stringify(r.results?.map((x) => `${x.migrationName}:${x.status}`)), "error:", r.error ?? null);
+  if (Deno.env.get("MIGRATIONS_DIR")) { await root.destroy(); Deno.exit(r.error ? 1 : 0); }
+  await db.insertInto("items").values([{ title: "Dune", meta: JSON.stringify({ q: "1080p" }) }, { title: "Arrival", meta: null }]).execute();
+  const rows = await db.selectFrom("items").select(["id", "title", "meta", "created_at"]).orderBy("id").execute();
+  console.log(dialectName, "rows:", rows.map((r) => `${typeof r.id}:${r.id}/${r.title}/${JSON.stringify(r.meta)}/${r.created_at instanceof Date}`).join(" "));
+  // transaction + rollback
+  await db.transaction().execute(async (tx) => { await tx.insertInto("items").values({ title: "tx-ok" }).execute(); }).catch(() => {});
+  await db.transaction().execute(async (tx) => { await tx.insertInto("items").values({ title: "tx-rollback" }).execute(); throw new Error("rollback"); }).catch((e) => console.log(dialectName, "tx error ->", e.message));
+  const titles = (await db.selectFrom("items").select("title").execute()).map((r) => r.title);
+  console.log(dialectName, "titles after tx:", titles.join(","));
+  // concurrency: 50 parallel queries on pool max=4
+  const t0 = performance.now();
+  await Promise.all(Array.from({ length: 50 }, (_, i) => sql`SELECT pg_sleep(0.01), ${i}::int AS i`.execute(root)));
+  console.log(dialectName, `50 parallel queries on pool=4: ${(performance.now() - t0).toFixed(0)}ms`);
+  await root.destroy();
+  console.log(dialectName, "destroyed cleanly");
+}

spikes/deno/apps/server/src/http.ts

@@ -0,0 +1,52 @@
+// Hono 4 on Deno.serve with an SSE stream + graceful shutdown.
+import { Hono } from "hono";
+import { streamSSE } from "hono/streaming";
+import { serveStatic } from "hono/deno";
+import { redact } from "@spike/x";
+
+export function createApp() {
+  const app = new Hono();
+  app.get("/api/v1/health", (c) => c.json({ ok: true, runtime: `deno ${Deno.version.deno}`, ts: Deno.version.typescript }));
+  app.get("/api/v1/redact", (c) => c.json(redact({ url: c.req.url, apiKey: "x" }) as Record<string, unknown>));
+  app.get("/api/v1/events", (c) =>
+    streamSSE(c, async (stream) => {
+      const after = Number(c.req.query("after") ?? 0);
+      let aborted = false;
+      stream.onAbort(() => { aborted = true; });
+      for (let i = after + 1; i <= after + 3 && !aborted; i++) {
+        await stream.writeSSE({ id: String(i), event: "tick", data: JSON.stringify({ i }) });
+        await stream.sleep(100);
+      }
+    }));
+  // Endless stream: used to check how Deno.serve shutdown treats open SSE connections.
+  app.get("/api/v1/forever", (c) =>
+    streamSSE(c, async (stream) => {
+      let i = 0;
+      while (!stream.aborted && !shutdownSignal.aborted) {
+        await stream.writeSSE({ id: String(++i), data: "ping" });
+        await stream.sleep(200);
+      }
+    }));
+  // Single-container mode: serve the built SPA (react-router ssr:false) from WEB_ROOT with index.html fallback.
+  const webRoot = Deno.env.get("WEB_ROOT");
+  if (webRoot) {
+    app.use("/*", serveStatic({ root: webRoot }));
+    app.get("*", serveStatic({ root: webRoot, path: "index.html" }));
+  }
+  return app;
+}
+
+/** App-level shutdown signal: SSE loops must observe it, otherwise graceful shutdown waits forever. */
+export const shutdownCtl = new AbortController();
+const shutdownSignal = shutdownCtl.signal;
+
+if (import.meta.main) {
+  const port = Number(Deno.env.get("PORT") ?? 5199);
+  const ac = new AbortController();
+  const server = Deno.serve({ port, hostname: Deno.env.get("HOST") ?? "127.0.0.1", signal: ac.signal, onListen: ({ port }) => console.log(`listening ${port}`) }, createApp().fetch);
+  const stop = () => { console.log("shutdown"); if (Deno.env.get("CLOSE_SSE") !== "0") shutdownCtl.abort(); ac.abort(); };
+  Deno.addSignalListener("SIGTERM", stop);
+  Deno.addSignalListener("SIGINT", stop);
+  await server.finished;
+  console.log("server finished");
+}

spikes/deno/apps/server/src/log-logtape.ts

@@ -0,0 +1,12 @@
+// LogTape alternative: JSON lines + redaction sink wrapper.
+import { configure, getJsonLinesFormatter, getLogger, getStreamSink, type LogRecord, type Sink } from "@logtape/logtape";
+import { redact, redactString } from "@spike/x";
+
+const base = getStreamSink(Deno.stdout.writable, { formatter: getJsonLinesFormatter() });
+const redacting = (inner: Sink): Sink => (r: LogRecord) =>
+  inner({ ...r, message: r.message.map((m: unknown) => (typeof m === "string" ? redactString(m) : redact(m))), rawMessage: typeof r.rawMessage === "string" ? redactString(r.rawMessage) : r.rawMessage, properties: redact(r.properties) as Record<string, unknown> });
+await configure({ sinks: { out: redacting(base) }, loggers: [{ category: ["umt"], lowestLevel: "info", sinks: ["out"] }, { category: ["logtape", "meta"], lowestLevel: "warning", sinks: ["out"] }] });
+const log = getLogger(["umt", "worker"]);
+log.info("fetch {url}", { url: "https://x/rss?passkey=SECRET", token: "abc", n: 3 });
+log.with({ runId: "r1", apiKey: "k" }).warn("child context");
+await new Promise((r) => setTimeout(r, 100));

spikes/deno/apps/server/src/log-pino.ts

@@ -0,0 +1,32 @@
+// pino 10 on Deno: sync stdout, custom destination, formatters/hooks redaction, child loggers, pino.transport (worker_threads).
+import pino from "pino";
+import { redact, redactString } from "@spike/x";
+import { Writable } from "node:stream";
+
+const opts: pino.LoggerOptions = {
+  level: "info",
+  formatters: { log: (obj) => redact(obj) as Record<string, unknown> },
+  hooks: {
+    logMethod(args, method) {
+      return method.apply(this, args.map((a) => (typeof a === "string" ? redactString(a) : a)) as typeof args);
+    },
+  },
+};
+const lines: string[] = [];
+const dest = new Writable({ write(chunk, _enc, cb) { lines.push(String(chunk)); cb(); } });
+const log = pino(opts, dest);
+log.info({ token: "abc", url: "https://x/rss?passkey=SECRET&a=1", n: 1 }, "fetch https://x?api_key=K");
+log.child({ component: "worker", password: "p" }).warn({ err: new Error("boom token=zzz") }, "child");
+console.log("custom-dest:", lines.join("").trim());
+
+pino(opts).info({ cookie: "c=1" }, "stdout default destination");
+const t = pino.transport({ target: "pino/file", options: { destination: 1 } });
+const lt = pino(opts, t);
+lt.info({ secret: "s" }, "via pino.transport (worker_threads)");
+await new Promise((r) => setTimeout(r, 300));
+t.end();
+if (Deno.env.get("PRETTY")) {
+  const pp = pino({ ...opts, transport: { target: "pino-pretty", options: { translateTime: "SYS:HH:MM:ss" } } });
+  pp.info({ token: "x" }, "via pino-pretty transport");
+  await new Promise((r) => setTimeout(r, 500));
+}

spikes/deno/apps/server/src/migrations/0001_init.ts

@@ -0,0 +1,10 @@
+import { type Kysely, sql } from "kysely";
+export async function up(db: Kysely<any>) {
+  await db.schema.createTable("items").ifNotExists()
+    .addColumn("id", "bigserial", (c) => c.primaryKey())
+    .addColumn("title", "text", (c) => c.notNull())
+    .addColumn("meta", "jsonb")
+    .addColumn("created_at", "timestamptz", (c) => c.notNull().defaultTo(sql`now()`))
+    .execute();
+}
+export async function down(db: Kysely<any>) { await db.schema.dropTable("items").execute(); }

spikes/deno/apps/server/src/migrations/0002_index.ts

@@ -0,0 +1,5 @@
+import type { Kysely } from "kysely";
+export async function up(db: Kysely<any>) {
+  await db.schema.createIndex("items_title_idx").ifNotExists().on("items").column("title").execute();
+}
+export async function down(db: Kysely<any>) { await db.schema.dropIndex("items_title_idx").execute(); }

spikes/deno/apps/server/src/node-compat.ts

@@ -0,0 +1,32 @@
+// Node-only API probe under Deno (things the current server/tasks code uses or is likely to use).
+import process from "node:process";
+import { Worker, isMainThread, parentPort } from "node:worker_threads";
+import { statfs } from "node:fs/promises";
+import { spawn } from "node:child_process";
+import { parseArgs } from "node:util";
+import os from "node:os";
+import { AsyncLocalStorage } from "node:async_hooks";
+import { createHash, randomUUID } from "node:crypto";
+
+if (!isMainThread) { parentPort!.postMessage({ echo: "from worker", pid: process.pid }); } else {
+  const r: Record<string, unknown> = {};
+  r.parseArgs = parseArgs({ args: ["--role", "api"], options: { role: { type: "string" } } }).values;
+  const s = await statfs("/");
+  r.statfs = { bsize: s.bsize, bavailGB: Math.round((s.bavail * s.bsize) / 1e9) };
+  r.os = { cpus: os.cpus().length, freememMB: Math.round(os.freemem() / 1e6), loadavg: os.loadavg().length };
+  r.memoryUsage = Object.keys(process.memoryUsage()).join(",");
+  const als = new AsyncLocalStorage<string>();
+  r.als = await als.run("req-1", async () => { await new Promise((x) => setTimeout(x, 5)); return als.getStore(); });
+  r.crypto = createHash("sha256").update("x").digest("hex").slice(0, 8) + " " + randomUUID().length;
+  r.worker = await new Promise((res) => { const w = new Worker(new URL(import.meta.url), { type: "module" } as any); w.once("message", (m) => { res(m); w.terminate(); }); });
+  const child = spawn("sh", ["-c", "trap 'echo got-TERM; exit 0' TERM; echo ready; sleep 5 & wait"]);
+  let out = ""; child.stdout.on("data", (d) => { out += d; });
+  await new Promise((x) => setTimeout(x, 200));
+  child.kill("SIGTERM");
+  r.childKill = await new Promise((res) => child.on("exit", (code, sig) => res({ code, sig, out: out.trim().replace(/\n/g, "|") })));
+  r.execPath = process.execPath;
+  console.log(JSON.stringify(r, null, 1));
+  process.on("SIGTERM", () => { console.log("process.on(SIGTERM) handler fired"); process.exit(0); });
+  setTimeout(() => process.kill(process.pid, "SIGTERM"), 100);
+  setTimeout(() => { console.log("handler did NOT fire"); process.exit(1); }, 2000);
+}

spikes/deno/apps/server/src/ow/cli.ts

@@ -0,0 +1,65 @@
+import process from "node:process";
+import { makeClient, initBiz, db } from "./common.ts";
+import { sql } from "kysely";
+await initBiz();
+const { ow, backend, acquire, flaky, slow, burst } = await makeClient();
+const [cmd, ...args] = process.argv.slice(2) as [string, ...string[]];
+const out = (x: unknown) => console.log(JSON.stringify(x, null, 0));
+switch (cmd) {
+  case "start-acquire": {
+    const [downloadId, key, copyMs, retention] = args;
+    const h = await acquire.run({ downloadId, copyMs: Number(copyMs ?? 0), retention: retention ?? "3s" },
+      key ? { idempotencyKey: key } : undefined);
+    out({ runId: h.workflowRun.id, status: h.workflowRun.status, version: h.workflowRun.version });
+    break;
+  }
+  case "start-flaky": {
+    const h = await flaky.run({ failTimes: Number(args[0]), maxAttempts: Number(args[1]) });
+    out({ runId: h.workflowRun.id });
+    break;
+  }
+  case "start-slow": { const h = await slow.run({}); out({ runId: h.workflowRun.id }); break; }
+  case "start-burst": {
+    const ids = [];
+    for (let i = 0; i < Number(args[0]); i++) ids.push((await burst.run({ i })).workflowRun.id);
+    out({ count: ids.length });
+    break;
+  }
+  case "signal": {
+    const r = await ow.sendSignal({ signal: `download-done:${args[0]}`, data: { ok: true }, idempotencyKey: args[1] });
+    out(r);
+    break;
+  }
+  case "cancel": { await ow.cancelWorkflowRun(args[0]!); out({ canceled: args[0] }); break; }
+  case "get": {
+    const r = await backend.getWorkflowRun({ workflowRunId: args[0]! });
+    out({ status: r?.status, version: r?.version, attempts: r?.attempts, error: r?.error, output: r?.output });
+    break;
+  }
+  case "steps": {
+    const r = await backend.listStepAttempts({ workflowRunId: args[0], limit: 100 } as any);
+    for (const s of [...r.data].sort((a,b)=>+a.createdAt-+b.createdAt)) console.log(s.stepName.padEnd(18), s.kind.padEnd(12), s.status, s.error ? JSON.stringify((s.error as any).message) : "");
+    break;
+  }
+  case "ledger": {
+    const rows = await db.selectFrom("biz.side_effect_ledger").selectAll()
+      .$if(!!args[0], (q) => q.where("run_id", "=", args[0])).orderBy("id").execute();
+    for (const r of rows) console.log(r.run_id.slice(0, 8), r.step.padEnd(16), r.worker.padEnd(6), r.note);
+    break;
+  }
+  case "burst-dups": {
+    const rows = await sql<any>`SELECT count(*) total, count(DISTINCT run_id) runs,
+      count(*) FILTER (WHERE worker='A') a, count(*) FILTER (WHERE worker='B') b
+      FROM biz.side_effect_ledger WHERE step='work'`.execute(db);
+    out(rows.rows[0]);
+    break;
+  }
+  case "tables": {
+    const r = await sql<any>`SELECT table_schema, table_name FROM information_schema.tables
+      WHERE table_schema IN ('openworkflow','biz') ORDER BY 1,2`.execute(db);
+    for (const t of r.rows) console.log(`${t.table_schema}.${t.table_name}`);
+    break;
+  }
+}
+await backend.stop();
+await db.destroy();

spikes/deno/apps/server/src/ow/common.ts

@@ -0,0 +1,111 @@
+// Shared: DB URLs, OpenWorkflow client, Kysely business DB (schema "biz"), workflow definitions.
+import { OpenWorkflow } from "openworkflow";
+import { BackendPostgres } from "openworkflow/postgres";
+import { Kysely, PostgresDialect, sql } from "kysely";
+import pg from "pg";
+import process from "node:process";
+
+export const PG_URL = process.env.PG_URL ?? "postgresql://postgres:[email protected]:55435/postgres";
+export const WF_VERSION = process.env.WF_VERSION ?? "v1";
+export const TAG = process.env.TAG ?? `pid${process.pid}`;
+
+export const db = new Kysely<any>({
+  dialect: new PostgresDialect({ pool: new pg.Pool({ connectionString: PG_URL, max: 4 }) }),
+});
+
+export async function initBiz() {
+  await sql`CREATE SCHEMA IF NOT EXISTS biz`.execute(db);
+  await sql`CREATE TABLE IF NOT EXISTS biz.side_effect_ledger (
+    id bigserial PRIMARY KEY, run_id text, step text, worker text, pid int, note text,
+    at timestamptz DEFAULT now())`.execute(db);
+}
+
+export async function ledger(runId: string, step: string, note = "") {
+  await db.insertInto("biz.side_effect_ledger")
+    .values({ run_id: runId, step, worker: TAG, pid: process.pid, note }).execute();
+  console.log(`[${TAG}] run=${runId.slice(0, 8)} step=${step} ${note}`);
+}
+
+const wait = (ms: number) => new Promise((r) => setTimeout(r, ms));
+
+export async function makeClient() {
+  // connect() runs migrations by default (runMigrations: true), schema default "openworkflow".
+  const backend = await BackendPostgres.connect(PG_URL, { schema: "openworkflow" });
+  const ow = new OpenWorkflow({ backend });
+
+  // Main vertical slice: submit download -> wait signal -> copy -> register -> retention sleep -> cleanup
+  const acquire = ow.defineWorkflow(
+    { name: "acquire", version: WF_VERSION },
+    async ({ input, step, run, version }) => {
+      const { downloadId, copyMs = 0, retention = "3s" } = input as any;
+      const sub = await step.run({ name: "submit-download" }, async () => {
+        await ledger(run.id, "submit-download", `version=${version}`);
+        return { torrentHash: `hash-${downloadId}` };
+      });
+      const done = await step.waitForSignal<{ ok: boolean }>({
+        name: "wait-download", signal: `download-done:${downloadId}`, timeout: "10m",
+      });
+      if (!done) throw new Error("download wait timed out");
+      const copied = await step.run({ name: "copy-file" }, async () => {
+        await ledger(run.id, "copy-file", "start");
+        await wait(copyMs);
+        await ledger(run.id, "copy-file", "end");
+        return { dest: `/lib/${downloadId}.mkv` };
+      });
+      if (process.env.EXTRA_STEP === "1") {
+        // EXTRA_STEP=1 simulates a code change between copy and register (tests code upgrade semantics)
+        await step.run({ name: "v2-extra-verify" }, async () => { await ledger(run.id, "v2-extra-verify"); return 1; });
+      }
+      await step.run({ name: "register" }, async () => { await ledger(run.id, "register", `version=${version}`); return 1; });
+      await step.sleep("retention", retention);
+      await step.run({ name: "cleanup" }, async () => { await ledger(run.id, "cleanup"); return 1; });
+      return { sub, copied, signal: done.data };
+    },
+  );
+
+  // Retry test: step fails `failTimes` times then succeeds; step maximumAttempts = maxAttempts
+  const flaky = ow.defineWorkflow(
+    { name: "flaky", version: WF_VERSION },
+    async ({ input, step, run }) => {
+      const { failTimes, maxAttempts } = input as any;
+      await step.run({ name: "before" }, async () => { await ledger(run.id, "before"); return 1; });
+      return await step.run(
+        { name: "flaky-step", retryPolicy: { maximumAttempts: maxAttempts, initialInterval: "200ms", maximumInterval: "1s" } },
+        async () => {
+          const n = await db.selectFrom("biz.side_effect_ledger").select(db.fn.countAll().as("c"))
+            .where("run_id", "=", run.id).where("step", "=", "flaky-step").executeTakeFirst();
+          const attempt = Number(n?.c ?? 0) + 1;
+          await ledger(run.id, "flaky-step", `attempt=${attempt}`);
+          if (attempt <= failTimes) throw new Error(`boom attempt ${attempt}`);
+          return { attempt };
+        },
+      );
+    },
+  );
+
+  // Cancel test: long step, then a follow-up step
+  const slow = ow.defineWorkflow(
+    { name: "slow", version: WF_VERSION },
+    async ({ step, run }) => {
+      await step.run({ name: "long" }, async () => {
+        await ledger(run.id, "long", "start");
+        await wait(8000);
+        await ledger(run.id, "long", "end (still ran after cancel?)");
+        return 1;
+      });
+      await step.run({ name: "after-long" }, async () => { await ledger(run.id, "after-long"); return 1; });
+      return "done";
+    },
+  );
+
+  // Concurrency test: many short runs
+  const burst = ow.defineWorkflow(
+    { name: "burst", version: WF_VERSION },
+    async ({ step, run }) => {
+      await step.run({ name: "work" }, async () => { await wait(300); await ledger(run.id, "work"); return 1; });
+      return 1;
+    },
+  );
+
+  return { ow, backend, acquire, flaky, slow, burst };
+}

spikes/deno/apps/server/src/ow/worker.ts

@@ -0,0 +1,14 @@
+// Deno port of spikes/openworkflow/worker.ts: node:fs/process.on -> Deno APIs
+import { initBiz, makeClient, TAG, WF_VERSION } from "./common.ts";
+await initBiz();
+const { ow } = await makeClient();
+const worker = ow.newWorker({ concurrency: Number(Deno.env.get("CONCURRENCY") ?? 4) });
+await worker.start();
+const pidfile = Deno.env.get("PIDFILE");
+if (pidfile) Deno.writeTextFileSync(pidfile, String(Deno.pid));
+console.log(`[${TAG}] worker started pid=${Deno.pid} version=${WF_VERSION} runtime=deno ${Deno.version.deno}`);
+Deno.addSignalListener("SIGTERM", async () => {
+  await worker.stop();
+  console.log(`[${TAG}] worker stopped gracefully`);
+  Deno.exit(0);
+});

spikes/deno/apps/server/src/pi/abort-resume.ts

@@ -0,0 +1,61 @@
+import process from "node:process";
+// Modes: abort | crash | resume. Uses a file-backed JSONL session so a second process can recover.
+import { createModels, fauxProvider } from "@earendil-works/pi-ai";
+import { AgentHarness, BACKGROUND_CONTEXT, JsonlSessionRepo } from "@earendil-works/pi-agent-core";
+import { NodeExecutionEnv } from "@earendil-works/pi-agent-core/harness/env/nodejs";
+import { downloadSubmit, inventoryQuery, script, submitted } from "./common.ts";
+
+const mode = process.argv[2];
+const ctx = BACKGROUND_CONTEXT;
+const faux = fauxProvider({ provider: "umt-fake" });
+const models = createModels();
+models.setProvider(faux.provider);
+faux.setResponses(mode === "resume" ? [script[2]!] : script);
+
+const root = new URL("./.sessions", import.meta.url).pathname;
+const fs = new NodeExecutionEnv({ cwd: root }); // used only as FileSystem for session storage; no shell tool is exposed
+const repo = new JsonlSessionRepo({ fileSystem: fs, sessionsRoot: root });
+let session;
+if (mode === "resume") {
+  const list = await repo.list({ cwd: root }, ctx);
+  const meta = list.find((m) => m.id === "crash-demo");
+  console.log("found session", meta?.id, meta?.path);
+  session = await repo.open(meta!, ctx);
+} else {
+  session = await repo.create({ id: mode === "crash" ? "crash-demo" : `abort-${Date.now()}`, cwd: root }, ctx);
+}
+
+const { harness, open } = await AgentHarness.create({
+  session, models, model: faux.getModel(), systemPrompt: "媒体库助手",
+  tools: [inventoryQuery, downloadSubmit], toolContext: { actorId: "user:1", grants: ["download.submit"] },
+  toolExecution: "sequential",
+}, ctx);
+console.log("open operations at attach:", JSON.stringify(open));
+
+let lane = await harness.lane("main", ctx);
+for (const t of ["tool_start", "tool_end", "run_end", "operation_abort", "run_resume"] as const)
+  harness.events.on(t as any, async (e: any) => {
+    console.log(`[${e.type}]${e.recovery ? " (recovery)" : ""} ${e.toolName ?? ""} ${e.status ?? ""} ${e.result ? JSON.stringify(e.result.content[0]?.text).slice(0, 220) : ""}`);
+    if (e.type === "tool_start" && e.toolName === "download__submit") {
+      if (mode === "abort") setTimeout(async () => { console.log(">> lane.abort()"); console.log("abort result:", JSON.stringify(await lane.abort(ctx))); }, 300);
+      if (mode === "crash") setTimeout(() => { console.log(">> SIGKILL self during download.submit"); process.kill(process.pid, "SIGKILL"); }, 300);
+    }
+  });
+
+if (mode === "resume") {
+  const r = await lane.resume(ctx);
+  console.log("RESUME RESULT", JSON.stringify(r));
+} else {
+  const r = await lane.prompt("把《Dune》1080p 收进来", undefined, ctx);
+  console.log("PROMPT RESULT", JSON.stringify(r));
+}
+console.log("SUBMITTED in this process", JSON.stringify(submitted), "fake calls:", faux.state.callCount);
+const entries = (await lane.findEntries(undefined, ctx)).reverse();
+for (const e of entries as any[]) {
+  if (e.type !== "message") { console.log(" entry", e.type); continue; }
+  const m = e.message;
+  const text = m.role === "assistant" ? m.content.map((c: any) => c.type === "toolCall" ? `toolCall(${c.name})` : c.type === "text" ? c.text : `<${c.type}>`).join(" | ")
+    : Array.isArray(m.content) ? m.content.map((c: any) => c.text).join("") : m.content;
+  console.log(` ${m.role.padEnd(10)} ${m.toolName ?? ""} ${m.isError ? "[isError]" : ""} ${String(text).slice(0, 200)}`);
+}
+await harness.close(ctx);

spikes/deno/apps/server/src/pi/common.ts

@@ -0,0 +1,84 @@
+import process from "node:process";
+// Shared pieces for the Pi spike: whitelisted business tools + scripted fake model.
+import { Type } from "typebox";
+import {
+  fauxAssistantMessage, fauxText, fauxThinking, fauxToolCall,
+  type FauxResponseFactory,
+} from "@earendil-works/pi-ai";
+import type { AgentHarnessTool } from "@earendil-works/pi-agent-core";
+
+// Capability id -> provider-safe tool name. Anthropic/OpenAI tool names must match ^[a-zA-Z0-9_-]{1,64}$,
+// so "inventory.query" cannot be sent verbatim; the port maps ids <-> names.
+export const toolName = (capabilityId: string) => capabilityId.replace(/\./g, "__");
+
+export const submitted: Array<{ key: string; title: string }> = [];
+
+type Ctx = { actorId: string; grants: string[] };
+
+export const inventoryQuery: AgentHarnessTool<Ctx> = {
+  name: toolName("inventory.query"),
+  label: "inventory.query",
+  description: "Query local library inventory for a title. Returns whether it exists and which versions.",
+  parameters: Type.Object({ title: Type.String(), year: Type.Optional(Type.Number()) }),
+  replay: "safe", // read-only: harness may re-execute after a crash
+  async execute(_id, params: any, onUpdate, toolCtx, invocation, context) {
+    onUpdate({ content: [{ type: "text", text: "searching…" }], details: {} });
+    const found = params.title === "Arrival";
+    return {
+      content: [{ type: "text", text: JSON.stringify({ title: params.title, inLibrary: found, versions: found ? ["2160p"] : [] }) }],
+      details: { actor: toolCtx.actorId, invocationId: invocation.invocationId, aborted: context.abortSignal?.aborted ?? false },
+    };
+  },
+};
+
+export const downloadSubmit: AgentHarnessTool<Ctx> = {
+  name: toolName("download.submit"),
+  label: "download.submit",
+  description: "Submit a download plan. Server validates authorization and generates the idempotency key.",
+  parameters: Type.Object({ title: Type.String(), quality: Type.String() }),
+  replay: "never", // external side effect: after crash the harness synthesizes an 'interrupted' result instead of re-running
+  executionMode: "sequential",
+  async execute(_id, params: any, _onUpdate, toolCtx, invocation, context) {
+    if (!toolCtx.grants.includes("download.submit")) throw new Error("not authorized: download.submit");
+    // Server-side idempotency key derived from operation semantics, not from the model.
+    const key = `dl:${params.title}:${params.quality}`;
+    const delayMs = Number(process.env.SUBMIT_DELAY_MS ?? 0);
+    if (delayMs) {
+      await new Promise<void>((resolve, reject) => {
+        const t = setTimeout(resolve, delayMs);
+        context.abortSignal?.addEventListener("abort", () => { clearTimeout(t); reject(new Error("aborted by signal")); });
+      });
+    }
+    if (!submitted.find((s) => s.key === key)) submitted.push({ key, title: params.title });
+    await invocation.setMemo("submitted", key);
+    return { content: [{ type: "text", text: JSON.stringify({ accepted: true, idempotencyKey: key, status: "submitted" }) }], details: { key } };
+  },
+};
+
+const lastToolResult = (ctx: any, name: string) =>
+  [...ctx.messages].reverse().find((m: any) => m.role === "toolResult" && m.toolName === name);
+
+// Scripted fake model: each step is a factory that looks at the transcript (tool results) and decides.
+export const script: FauxResponseFactory[] = [
+  () => fauxAssistantMessage([
+    fauxThinking("先查本地库存"),
+    fauxText("我先查一下库里有没有《Dune》。"),
+    fauxToolCall(toolName("inventory.query"), { title: "Dune", year: 2021 }),
+  ], { stopReason: "toolUse" }),
+  (ctx) => {
+    const inv = JSON.parse(lastToolResult(ctx, toolName("inventory.query")).content[0].text);
+    if (inv.inLibrary) return fauxAssistantMessage("库里已有,无需下载。");
+    return fauxAssistantMessage([
+      fauxText("库里没有,提交 1080p 下载;顺便试一下未授权的 bash。"),
+      fauxToolCall(toolName("download.submit"), { title: "Dune", quality: "1080p" }),
+      fauxToolCall("bash", { command: "rm -rf /" }),
+    ], { stopReason: "toolUse" });
+  },
+  (ctx) => {
+    const dl = lastToolResult(ctx, toolName("download.submit"));
+    const bash = lastToolResult(ctx, "bash");
+    return fauxAssistantMessage(
+      `下载已提交(仅代表已提交,不代表已入库):${dl.content[0].text};bash 结果 isError=${bash?.isError}。`,
+    );
+  },
+];

spikes/deno/apps/server/src/pi/fake-run.ts

@@ -0,0 +1,51 @@
+import process from "node:process";
+// Fake model drives AgentHarness through 2 whitelisted tools; streams events.
+import { createModels, fauxProvider } from "@earendil-works/pi-ai";
+import { AgentHarness, BACKGROUND_CONTEXT, MemorySessionRepo } from "@earendil-works/pi-agent-core";
+import { downloadSubmit, inventoryQuery, script, submitted } from "./common.ts";
+
+const faux = fauxProvider({ provider: "umt-fake" });
+const models = createModels();
+models.setProvider(faux.provider);
+faux.setResponses(script);
+
+const repo = new MemorySessionRepo();
+const session = await repo.create({ id: "spike-1" }, BACKGROUND_CONTEXT);
+const { harness } = await AgentHarness.create({
+  session, models, model: faux.getModel(),
+  systemPrompt: "你是媒体库自动化助手,只能使用提供的工具。",
+  tools: [inventoryQuery, downloadSubmit],           // explicit whitelist; no discovery
+  toolContext: { actorId: "user:1", grants: ["download.submit"] },
+  thinkingLevel: "off",
+  compaction: { enabled: true, reserveTokens: 4096, keepRecentTokens: 8000 },
+  retry: { enabled: true, maxRetries: 2, baseDelayMs: 100 },
+  toolExecution: "sequential",
+}, BACKGROUND_CONTEXT);
+
+// Server-side authorization gate in addition to the whitelist.
+harness.hooks.on("before_tool", (ev) => {
+  console.log(`  [hook before_tool] ${ev.toolName} args=${JSON.stringify(ev.args)}`);
+  return undefined;
+});
+
+const types = ["run_start", "turn_start", "message_update", "tool_start", "tool_update", "tool_end", "turn_end", "run_end", "usage", "fault", "handler_error"] as const;
+for (const t of types) harness.events.on(t as any, (e: any) => {
+  if (e.type === "message_update") {
+    if (e.event.type === "text_delta") process.stdout.write(`  [text_delta] ${JSON.stringify(e.event.delta)}\n`);
+    else if (e.event.type === "toolcall_end") console.log(`  [toolcall_end] ${e.event.toolCall.name}`);
+    return;
+  }
+  if (e.type === "tool_end") return console.log(`[tool_end] ${e.toolName} isError=${e.isError} -> ${JSON.stringify(e.result.content[0]?.text)}`);
+  if (e.type === "tool_start") return console.log(`[tool_start] ${e.toolName} ${JSON.stringify(e.args)}`);
+  if (e.type === "usage") return console.log(`[usage] in=${e.row?.usage?.input ?? "?"} totals.input=${e.totals.input} output=${e.totals.output}`);
+  if (e.type === "run_end") return console.log(`[run_end] status=${e.status}${e.error ? " error=" + JSON.stringify(e.error) : ""}`);
+  console.log(`[${e.type}]${e.lane ? " lane=" + e.lane : ""}`);
+});
+
+const lane = await harness.lane("main", BACKGROUND_CONTEXT);
+const result = await lane.prompt("把《Dune》1080p 收进来", undefined, BACKGROUND_CONTEXT);
+console.log("RESULT", JSON.stringify(result));
+console.log("SUBMITTED", JSON.stringify(submitted), "fake calls:", faux.state.callCount);
+const entries = await lane.findEntries(undefined, BACKGROUND_CONTEXT);
+console.log("TRANSCRIPT roles:", entries.map((e: any) => e.type === "message" ? e.message.role : e.type).join(","));
+await harness.close(BACKGROUND_CONTEXT);