From cd2871241ae651d050ac85b45940e6f07858bf28 Mon Sep 17 00:00:00 2001 From: Yuto Nishida Date: Sat, 24 Oct 2026 17:20:52 -0700 Subject: [PATCH] papyrus: consume streaming creation and own provider process lifetime --- experimental/papyrus/README.md | 18 ++ experimental/papyrus/TESTING.md | 13 +- .../client/src/workstreamState.test.ts | 12 - .../papyrus/client/src/workstreamState.ts | 5 +- experimental/papyrus/flake.nix | 10 +- .../papyrus/server-tests/api.e2e.test.ts | 53 ++++- .../server-tests/creation-shutdown.test.ts | 26 +++ .../papyrus/server-tests/creation.test.ts | 85 +++++++ .../server-tests/helpers/creation-shutdown.ts | 42 ++++ .../papyrus/server-tests/helpers/forest.ts | 38 +++- .../server-tests/silverwood-contract.test.ts | 39 ++-- experimental/papyrus/server/index.ts | 15 +- experimental/papyrus/server/routes/api.ts | 28 +-- .../papyrus/server/services/creation.ts | 209 ++++++++++++++++++ .../papyrus/server/services/silverwood.ts | 49 ++-- 15 files changed, 517 insertions(+), 125 deletions(-) create mode 100644 experimental/papyrus/server-tests/creation-shutdown.test.ts create mode 100644 experimental/papyrus/server-tests/creation.test.ts create mode 100644 experimental/papyrus/server-tests/helpers/creation-shutdown.ts create mode 100644 experimental/papyrus/server/services/creation.ts diff --git a/experimental/papyrus/README.md b/experimental/papyrus/README.md index 5ad991d..e98e4ce 100644 --- a/experimental/papyrus/README.md +++ b/experimental/papyrus/README.md @@ -167,3 +167,21 @@ Ralph includes rate limiting, circuit breakers, and intelligent exit detection t ## License MIT + + +### Creation process lifetime + +Basic workstreams use one `silverwood --json-events new basic` process. Papyrus +returns the authoritative registered ID after saving canvas position, while its +creation registry owns that process through provisioning. Browser polling reads +the saved ready/failed result. Disconnecting the HTTP caller does not cancel +accepted creation; errors after registration are logged with the workstream ID. +A metadata-write error response includes `workstreamId`, so callers can inspect +that accepted workstream before retrying. + +SIGINT/SIGTERM stops new requests, allows active creations five seconds to finish, +then terminates their process groups (including provider descendants) and reaps +them. A hard crash may leave pending state and partial files; inspect/archive that +workstream and create a fresh one. Papyrus does not automatically restart unknown +operations. The packaged launcher execs the server directly so signals reach its +shutdown handler. diff --git a/experimental/papyrus/TESTING.md b/experimental/papyrus/TESTING.md index 554d8ae..ef32a12 100644 --- a/experimental/papyrus/TESTING.md +++ b/experimental/papyrus/TESTING.md @@ -84,12 +84,13 @@ SILVERWOOD_BIN=../silverwood/target/debug/silverwood OPENUI_QUIET=1 bun test ser the package root (the "tests" substring would also match `client/tests/`, whose happy-dom preload only loads under `client/`). -**Scope / constraints.** Basic workstreams use `--checkout-extent skip`; session launch -journeys use `local-blank` with mocked PTYs, so no agent runs. The suite is network-free -and runs in the `nix flake check` sandbox via the -`server-tests` derivation (`flake.nix`), which sets `SILVERWOOD_BIN` to the flake's wrapped -silverwood. The full create→checkout→live-terminal journey needs network + jj and is **out of -scope** here, matching silverwood's own `#[ignore]`d e2e split. +**Scope / constraints.** Basic creation uses the real streaming CLI with a gated +fake `jj` on PATH. The Nix target supplies the **unwrapped** Silverwood binary, +so its production wrapper cannot replace that fake tool with real jj. Tests cover +early HTTP registration, persisted position, concurrent metadata, ready/failed +outcomes, parser chunking and malformed events, and operation cleanup. Session +journeys use mocked PTYs; no agent runs. Teardown awaits creation processes before +removing forests. The suite is network-free and runs in `nix flake check`. **Ownership.** These are developer-owned CUJ invariants (like behavioral tests): the coding agent may add/update them when it changes the server↔silverwood contract, but the developer diff --git a/experimental/papyrus/client/src/workstreamState.test.ts b/experimental/papyrus/client/src/workstreamState.test.ts index 1aca24b..d0eb0cf 100644 --- a/experimental/papyrus/client/src/workstreamState.test.ts +++ b/experimental/papyrus/client/src/workstreamState.test.ts @@ -28,18 +28,6 @@ describe("workstreamStateLabel", () => { expect(color).toBe("#6B7280"); }); - test("initialized-without-checkout: base label only, amber (provisioning-adjacent)", () => { - const { label, color } = workstreamStateLabel({ - overallState: "active - basic.initialized-without-checkout", - kind: "basic", - checkoutState: "initialized-without-checkout", - connected: false, - tabs: [], - }); - expect(label).toBe("active - basic.initialized-without-checkout"); - expect(color).toBe("#FBBF24"); - }); - test("pending: base label only, amber", () => { const { label, color } = workstreamStateLabel({ overallState: "active - basic.pending", diff --git a/experimental/papyrus/client/src/workstreamState.ts b/experimental/papyrus/client/src/workstreamState.ts index 9b49ce8..7783b23 100644 --- a/experimental/papyrus/client/src/workstreamState.ts +++ b/experimental/papyrus/client/src/workstreamState.ts @@ -27,10 +27,7 @@ export function workstreamStateLabel(input: WorkstreamStateInput): { label = `${n}/${tabs.length} Connected`; } - // `initialized-without-checkout` (deferred checkout, about to provision) is grouped - // with `pending` as a not-ready-yet amber. - const provisioning = - checkoutState === "pending" || checkoutState === "initialized-without-checkout"; + const provisioning = checkoutState === "pending"; const color = checkoutState === "failed" ? "#EF4444" diff --git a/experimental/papyrus/flake.nix b/experimental/papyrus/flake.nix index 7fb17fd..680c4a4 100644 --- a/experimental/papyrus/flake.nix +++ b/experimental/papyrus/flake.nix @@ -93,17 +93,15 @@ # Server-side e2e: drive the real Hono routes in-process (`apiRoutes.request`) # against a REAL silverwood binary + a per-test temp forest, asserting canvas # CUJs round-trip through silverwood (the delegation the server exists to do). - # `bun-pty` is stubbed and every journey is `--checkout-extent skip`, so nothing - # clones — it runs in the network-isolated sandbox and a red test fails - # `nix flake check`. `bun test server-tests` scopes to the server-tests/ dir (the - # substring never matches client/tests/). See server-tests/ and TESTING.md. + # Native PTYs are stubbed for route tests. Basic streaming tests use an + # unwrapped CLI and a gated fake jj, with no network access. server-tests = b2n.mkDerivation { pname = "papyrus-server-tests"; version = "1.2.1"; src = ./.; bunDeps = b2n.fetchBunDeps { bunNix = ./bun.nix; }; buildPhase = '' - export SILVERWOOD_BIN=${silverwood-bin}/bin/silverwood + export SILVERWOOD_BIN=${silverwood.packages.${system}.unwrapped}/bin/silverwood export OPENUI_QUIET=1 bun test server-tests ''; @@ -131,7 +129,7 @@ "--backend=symlink" ]; - startScript = "bun run server/index.ts"; + startScript = "exec bun server/index.ts"; # writeBunApplication's wrapper --chdir's into $out/share/papyrus at # runtime, so server/index.ts's `serveStatic({ root: "./client/dist" })` diff --git a/experimental/papyrus/server-tests/api.e2e.test.ts b/experimental/papyrus/server-tests/api.e2e.test.ts index f9bcc1e..595029f 100644 --- a/experimental/papyrus/server-tests/api.e2e.test.ts +++ b/experimental/papyrus/server-tests/api.e2e.test.ts @@ -2,7 +2,7 @@ // client against a real silverwood binary + a temp forest. Asserts that canvas CUJs // (create / edit / delete a node, session metadata) round-trip through silverwood — // the delegation the server exists to do. Ground truth is re-read from silverwood, not -// server memory. Skip-mode only (no clone), so it runs in the sandbox. +// server memory. A fake provider keeps the tests offline. import { describe, test, expect, beforeEach, afterEach } from "bun:test"; import { api, jsonInit, spawnCalls } from "./helpers/app"; // stubs bun-pty, then loads the routes @@ -12,7 +12,7 @@ import { newForest, cleanupForest, cli, - SKIP_SOURCE, + TEST_SOURCE, SILVERWOOD_BIN, } from "./helpers/forest"; @@ -22,7 +22,7 @@ function createNode(name: string, extra: Record = {}) { jsonInit("POST", { name, path: ["basic", "jj-colocated"], - args: [SKIP_SOURCE], + args: [TEST_SOURCE], ...extra, }), ); @@ -33,7 +33,7 @@ async function nodes(): Promise { return body.nodes; } -describe("papyrus routes → silverwood (in-process, skip-mode)", () => { +describe("papyrus routes → silverwood (in-process, fake provider)", () => { let dir: string; beforeEach(() => { dir = newForest(); @@ -63,7 +63,7 @@ describe("papyrus routes → silverwood (in-process, skip-mode)", () => { test("create node registers a workstream and stores its position", async () => { const { status, body } = await createNode("web", { position: { x: 10, y: 20 } }); expect(status).toBe(200); - expect(body.checkoutState).toBe("initialized-without-checkout"); + expect(body.checkoutState).toBe("pending"); const ns = await nodes(); expect(ns).toHaveLength(1); @@ -74,7 +74,7 @@ describe("papyrus routes → silverwood (in-process, skip-mode)", () => { }); // A checkout-less `local-blank` node: the route creates it with no checkout mode and - // skips the basic-only background `checkout` (which would error on a non-basic kind). + // requires no background provisioning. test("create a local-blank node (no checkout, no background provisioning)", async () => { const { status, body } = await api( "/sessions", @@ -306,3 +306,44 @@ describe("papyrus routes → silverwood (in-process, skip-mode)", () => { expect(res.body.error).toBeTruthy(); }); }); + +describe("streaming creation responsiveness", () => { + let dir: string; + let gate: string; + beforeEach(() => { + dir = newForest(); + gate = `${dir}/provider-release`; + process.env.SILVERWOOD_TEST_GATE = gate; + }); + afterEach(async () => { + await Bun.write(gate, "release even after assertion failure"); + await cleanupForest(dir); + }); + test.each([false, true])("HTTP returns and position persists while provider is gated (fail=%s)", async (fail) => { + if (fail) process.env.SILVERWOOD_TEST_FAIL = "1"; + const { status, body } = await createNode("streamed", { position: { x: 9, y: 11 } }); + expect(status).toBe(200); + expect(body.checkoutState).toBe("pending"); + expect((await nodes())[0].position).toEqual({ x: 9, y: 11 }); + expect((await sw.get(body.nodeId)).mode?.state).toBe("pending"); + await sw.rename(body.nodeId, "concurrent rename"); + await sw.archive(body.nodeId); + await Bun.write(gate, "go"); + const { settleCreations } = await import("../server/services/creation"); + await settleCreations(); + const final = await sw.get(body.nodeId); + expect(final.mode?.state).toBe(fail ? "failed" : "ready"); + expect(final.name).toBe("concurrent rename"); + expect(final.status).toBe("archived"); + expect(sw.decodeKv((await sw.getPapyrusKv(body.nodeId)), "position")).toEqual({ x: 9, y: 11 }); + }); + test("an aborted caller leaves accepted creation owned", async () => { + const controller = new AbortController(); + const { body } = await api("/sessions", { ...jsonInit("POST", { name: "accepted", path: ["basic", "jj-colocated"], args: [TEST_SOURCE] }), signal: controller.signal }); + controller.abort(); + await Bun.write(gate, "go"); + const { settleCreations } = await import("../server/services/creation"); + await settleCreations(); + expect((await sw.get(body.nodeId)).mode?.state).toBe("ready"); + }); +}); diff --git a/experimental/papyrus/server-tests/creation-shutdown.test.ts b/experimental/papyrus/server-tests/creation-shutdown.test.ts new file mode 100644 index 0000000..96f9131 --- /dev/null +++ b/experimental/papyrus/server-tests/creation-shutdown.test.ts @@ -0,0 +1,26 @@ +import { expect, test } from "bun:test"; +import { mkdtempSync, readFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +test.each(["yes", "no", "parent-exits", "parent-crashes"])("shutdown reaps creation and descendants (registered=%s)", async (registered) => { + const dir = mkdtempSync(join(tmpdir(), "papyrus-shutdown-")); + const proc = Bun.spawn([process.execPath, `${import.meta.dir}/helpers/creation-shutdown.ts`, dir, registered], { stdout: "pipe", stderr: "pipe" }); + const stderr = new Response(proc.stderr).text(); + const stdout = new Response(proc.stdout).text(); + try { + const exit = await proc.exited; + expect({ exit, stderr: exit === 0 ? "" : await stderr }).toEqual({ exit: 0, stderr: "" }); + expect(readFileSync(`${dir}/complete`, "utf8")).toBe("reaped"); + } finally { + proc.kill("SIGKILL"); + // On assertion failure, clean up only this fixture's recorded process IDs. + try { + for (const pid of JSON.parse(readFileSync(`${dir}/pids`, "utf8"))) { + try { process.kill(pid, "SIGKILL"); } catch {} + } + } catch {} + await Promise.all([proc.exited, stdout, stderr]); + rmSync(dir, { recursive: true, force: true }); + } +}, 10_000); diff --git a/experimental/papyrus/server-tests/creation.test.ts b/experimental/papyrus/server-tests/creation.test.ts new file mode 100644 index 0000000..14ffb3f --- /dev/null +++ b/experimental/papyrus/server-tests/creation.test.ts @@ -0,0 +1,85 @@ +import { describe, expect, test } from "bun:test"; +import { CreationProtocol, startCreation } from "../server/services/creation"; + +const id = "uuidv7_019f9d07-c030-7f22-bd1b-46088aa1a707"; +const ws = (state: string) => ({ + id, name: "café", status: "active", kind: "basic", created_at: "2026-10-24T00:00:00Z", + overall_state: `active - basic.${state}`, + mode: { checkout_mode: "jj-colocated", initial_source: "https://example.invalid/x", state }, + location: { forest_id: "test-forest", within: { forest_kind: "basic-forest", path: "/tmp/test" } }, +}); +const registered = { schemaVersion: 1, event: "registered", workstreamId: id, workstream: ws("pending") }; +const started = { schemaVersion: 1, event: "checkout-started", workstreamId: id }; +const ready = { schemaVersion: 1, event: "ready", workstreamId: id, workstream: ws("ready") }; +const failed = { schemaVersion: 1, event: "failed", workstreamId: id, error: { code: "provisioning", message: "boom" } }; +const line = (event: unknown) => JSON.stringify(event) + "\n"; +const fake = (code: string) => startCreation([process.execPath, "-e", code], true); + +// Pure state-machine coverage stays independent of subprocess chunk boundaries. +describe("creation protocol", () => { + test("allows extra fields and requires a terminal event with matching exit status", () => { + let accepted: unknown; + const parser = new CreationProtocol((value) => { accepted = value; }); + parser.accept(JSON.stringify({ ...registered, extra: true })); + expect(accepted).toEqual(ws("pending")); + parser.accept(JSON.stringify(started)); + expect(() => parser.finish(0)).toThrow(); + parser.accept(JSON.stringify(ready)); + expect(parser.finish(0)).toEqual(ws("ready")); + expect(() => parser.finish(1)).toThrow(); + expect(() => parser.accept(JSON.stringify(ready))).toThrow(); + }); + test.each([ + [started], [ready], [{ ...registered, schemaVersion: 2 }], + [{ ...registered, workstream: { id } }], [{ ...registered, event: "unknown" }], + [registered, registered], [registered, ready], [registered, { ...started, workstreamId: "other" }], + [registered, started, { ...ready, workstream: ws("pending") }], + [registered, { ...failed, error: { code: "unknown", message: "x" } }], + ])("rejects invalid events %#", (...events) => { + const parser = new CreationProtocol(() => {}); + expect(() => { for (const event of events) parser.accept(JSON.stringify(event)); }).toThrow(); + }); + test("preregistration failure carries no ID", () => { + const parser = new CreationProtocol(() => {}); + parser.accept(JSON.stringify({ schemaVersion: 1, event: "failed", error: { code: "validation", message: "bad input" } })); + expect(() => parser.finish(1)).toThrow("bad input"); + expect(() => parser.finish(0)).toThrow("exit zero"); + }); +}); + +describe("incremental process reader", () => { + test("split UTF8, partial lines, batched lines and stderr volume", async () => { + const output = line(registered) + line(started) + line(ready); + const op = fake(` + const bytes = new TextEncoder().encode(${JSON.stringify(output)}); + for (let i = 0; i < bytes.length; i++) { process.stdout.write(bytes.slice(i, i + 1)); await Bun.sleep(1); } + process.stderr.write('x'.repeat(200000)); + `); + expect((await op.registered).name).toBe("café"); + expect((await op.completed).mode?.state).toBe("ready"); + }); + test("fast failure preserves already accepted registration", async () => { + const op = fake(`process.stdout.write(${JSON.stringify(line(registered) + line(started) + line(failed))}); process.exitCode = 1;`); + expect((await op.registered).id).toBe(id); + await expect(op.completed).rejects.toThrow("boom"); + }); + test.each([ + ["", 0], [line(registered), 0], [line(registered) + line(started) + line(ready), 1], + [JSON.stringify(registered), 0], ["garbage\n", 0], [line({ schemaVersion: 99, event: "registered" }), 0], + ])("rejects interrupted/malformed streams %#", async (output, exitCode) => { + const op = fake(`process.stdout.write(${JSON.stringify(output)}); process.exitCode = ${exitCode};`); + await expect(op.completed).rejects.toThrow(); + }); + test("preregistration failure rejects both promises", async () => { + const op = fake(`process.stdout.write(${JSON.stringify(line({ schemaVersion: 1, event: "failed", error: { code: "validation", message: "bad source" } }))}); process.exitCode = 1;`); + await expect(op.registered).rejects.toThrow("bad source"); + await expect(op.completed).rejects.toThrow("bad source"); + }); +}); + + +test("malformed child ignoring SIGTERM is killed and reaped", async () => { + const op = fake(`process.on('SIGTERM', () => {}); process.stdout.write('garbage\\n'); setInterval(() => {}, 1000);`); + await expect(op.registered).rejects.toThrow(); + await expect(op.completed).rejects.toThrow(); +}); diff --git a/experimental/papyrus/server-tests/helpers/creation-shutdown.ts b/experimental/papyrus/server-tests/helpers/creation-shutdown.ts new file mode 100644 index 0000000..f5ac3e3 --- /dev/null +++ b/experimental/papyrus/server-tests/helpers/creation-shutdown.ts @@ -0,0 +1,42 @@ +// Isolated process: shutting down its registry must not close the test runner's. +import { startCreation, shutdownCreations } from "../../server/services/creation"; +import { existsSync, readFileSync, writeFileSync } from "node:fs"; + +const [dir, registeredFirst] = process.argv.slice(2); +const id = "uuidv7_019f9d07-c030-7f22-bd1b-46088aa1a707"; +const ws = { + id, name: "shutdown fixture", status: "active", kind: "basic", created_at: "now", + overall_state: "active - basic.pending", + mode: { checkout_mode: "jj-colocated", initial_source: "https://example.invalid/x", state: "pending" }, + location: { forest_id: "fixture", within: { forest_kind: "basic-forest", path: dir } }, +}; +const descendantCode = ` + process.on('SIGTERM', () => {}); + setInterval(() => require('node:fs').writeFileSync(${JSON.stringify(`${dir}/ticks`)}, String(Date.now())), 10); + setTimeout(() => process.exit(0), 15000); +`; +const code = ` + const fs = require('node:fs'); + ${registeredFirst === "parent-exits" ? "" : "process.on('SIGTERM', () => {});"} + const descendant = Bun.spawn([process.execPath, '-e', ${JSON.stringify(descendantCode)}], { stdout: 'ignore', stderr: 'ignore' }); + fs.writeFileSync(${JSON.stringify(`${dir}/pids`)}, JSON.stringify([process.pid, descendant.pid])); + ${registeredFirst !== "no" ? `process.stdout.write(${JSON.stringify(JSON.stringify({ schemaVersion: 1, event: "registered", workstreamId: id, workstream: ws }) + "\n")});` : ""} + setTimeout(() => process.exit(${registeredFirst === "parent-crashes" ? "1" : "0"}), ${registeredFirst === "parent-crashes" ? "250" : "15000"}); +`; +const operation = startCreation([process.execPath, "-e", code], true); +if (registeredFirst !== "no") await operation.registered; +const deadline = Date.now() + 3000; +while (!existsSync(`${dir}/ticks`)) { + if (Date.now() > deadline) throw new Error("descendant never started"); + await Bun.sleep(10); +} +if (registeredFirst === "parent-crashes") await Promise.allSettled([operation.completed]); +else await shutdownCreations(25); +const result = await Promise.allSettled([operation.registered, operation.completed]); +if (result[1].status !== "rejected" || (registeredFirst === "no" && result[0].status !== "rejected")) { + throw new Error("shutdown did not settle operation promises"); +} +const stoppedAt = readFileSync(`${dir}/ticks`, "utf8"); +await Bun.sleep(100); +if (readFileSync(`${dir}/ticks`, "utf8") !== stoppedAt) throw new Error("provider descendant survived shutdown"); +writeFileSync(`${dir}/complete`, "reaped"); diff --git a/experimental/papyrus/server-tests/helpers/forest.ts b/experimental/papyrus/server-tests/helpers/forest.ts index e97e7cf..c4b3f17 100644 --- a/experimental/papyrus/server-tests/helpers/forest.ts +++ b/experimental/papyrus/server-tests/helpers/forest.ts @@ -1,8 +1,9 @@ +import { settleCreations } from "../../server/services/creation"; // Server e2e harness: point silverwood at a throwaway forest per test, and run the // real binary directly for seeding fixtures + asserting ground truth (independent of // papyrus's own wrapper). The TS analog of silverwood's `tests/common/mod.rs`. -import { mkdtempSync, rmSync } from "fs"; +import { mkdtempSync, rmSync, mkdirSync, writeFileSync, chmodSync } from "fs"; import { tmpdir } from "os"; import { join } from "path"; @@ -17,25 +18,40 @@ if (!SILVERWOOD_BIN) { ); } -// A well-formed https source that never resolves (RFC 6761 `.invalid`), so the -// non-awaited background checkout the create route kicks off fails fast everywhere -// (and leaves no real clone) while `--checkout-extent skip` registration still works. -export const SKIP_SOURCE = "https://example.invalid/papyrus-e2e.git"; +// Controlled fake jj consumes this source without network access. +export const TEST_SOURCE = "https://example.invalid/papyrus-e2e.git"; // Create a fresh forest and point every subsequent silverwood invocation at it (the // server wrapper and `cli()` both read this env at spawn time). Call in `beforeEach`. +const originalPath = process.env.PATH; + export function newForest(): string { const dir = mkdtempSync(join(tmpdir(), "papyrus-e2e-")); process.env.SILVERWOOD_FOREST_PATH = dir; + const tools = join(dir, "test-tools"); + mkdirSync(tools); + const jj = join(tools, "jj"); + writeFileSync(jj, `#!${Bun.which("sh") ?? "/bin/sh"} +set -eu +for dest do :; done +if [ -n "\${SILVERWOOD_TEST_GATE:-}" ]; then + while [ ! -e "$SILVERWOOD_TEST_GATE" ]; do sleep 0.01; done +fi +if [ "\${SILVERWOOD_TEST_FAIL:-}" = 1 ]; then echo 'fixture failure' >&2; exit 1; fi +mkdir -p "$dest" +echo 'provider stdout is captured' +`); + chmodSync(jj, 0o755); + process.env.PATH = `${tools}:${originalPath}`; return dir; } -export function cleanupForest(dir: string): void { - try { - rmSync(dir, { recursive: true, force: true }); - } catch { - // A detached background clone may still hold the dir; best-effort. - } +export async function cleanupForest(dir: string): Promise { + await settleCreations(); + delete process.env.SILVERWOOD_TEST_GATE; + delete process.env.SILVERWOOD_TEST_FAIL; + process.env.PATH = originalPath; + rmSync(dir, { recursive: true, force: true }); } export interface CliResult { diff --git a/experimental/papyrus/server-tests/silverwood-contract.test.ts b/experimental/papyrus/server-tests/silverwood-contract.test.ts index bba929e..53aef09 100644 --- a/experimental/papyrus/server-tests/silverwood-contract.test.ts +++ b/experimental/papyrus/server-tests/silverwood-contract.test.ts @@ -1,21 +1,19 @@ // The silverwood CLI contract, driven through papyrus's own wrapper against a real -// binary + a temp forest. This is the direct regression guard for "silverwood changed -// its CLI": every argv the wrapper builds is exercised end-to-end. Skip-mode only, so -// nothing clones — runs in the network-isolated sandbox. Imports no native module. +// binary + a temp forest and a controlled provider. Imports no native module. import { describe, test, expect, beforeEach, afterEach } from "bun:test"; import { mkdtempSync, rmSync } from "fs"; import { tmpdir } from "os"; import { join } from "path"; import * as sw from "../server/services/silverwood"; -import { newForest, cleanupForest, cli, SKIP_SOURCE } from "./helpers/forest"; +import { newForest, cleanupForest, cli, TEST_SOURCE } from "./helpers/forest"; -function newSkipWs(name: string) { +function newBasicWs(name: string) { // Mirrors what POST /api/sessions passes: variant path + positional source. - return sw.create({ name, path: ["basic", "jj-colocated"], args: [SKIP_SOURCE] }); + return sw.create({ name, path: ["basic", "jj-colocated"], args: [TEST_SOURCE] }).completed; } -describe("silverwood wrapper contract (skip-mode, no network)", () => { +describe("silverwood wrapper contract (fake provider, no network)", () => { let dir: string; beforeEach(() => { dir = newForest(); @@ -23,10 +21,10 @@ describe("silverwood wrapper contract (skip-mode, no network)", () => { afterEach(() => cleanupForest(dir)); test("create → get → list round-trip", async () => { - const ws = await newSkipWs("alpha"); + const ws = await newBasicWs("alpha"); expect(ws.name).toBe("alpha"); expect(ws.status).toBe("active"); - expect(sw.basicCheckoutState(ws)).toBe("initialized-without-checkout"); + expect(sw.basicCheckoutState(ws)).toBe("ready"); const got = await sw.get(ws.id); expect(got.id).toBe(ws.id); @@ -37,13 +35,13 @@ describe("silverwood wrapper contract (skip-mode, no network)", () => { }); test("rename persists", async () => { - const ws = await newSkipWs("before"); + const ws = await newBasicWs("before"); await sw.rename(ws.id, "after"); expect((await sw.get(ws.id)).name).toBe("after"); }); test("archive hides from list but survives as archived", async () => { - const ws = await newSkipWs("doomed"); + const ws = await newBasicWs("doomed"); await sw.archive(ws.id); expect((await sw.list()).map((w) => w.id)).not.toContain(ws.id); // Ground truth straight from silverwood: still present, tombstoned. @@ -52,7 +50,7 @@ describe("silverwood wrapper contract (skip-mode, no network)", () => { }); test("kv set/get/unset round-trip in the papyrus namespace", async () => { - const ws = await newSkipWs("kv"); + const ws = await newBasicWs("kv"); await sw.setKv(ws.id, "color", "#abcdef"); await sw.setKv(ws.id, "position", { x: 1, y: 2 }); @@ -66,7 +64,7 @@ describe("silverwood wrapper contract (skip-mode, no network)", () => { }); test("session create/ls/rename/rm lifecycle (kind preserved)", async () => { - const ws = await newSkipWs("sessions"); + const ws = await newBasicWs("sessions"); const sid = crypto.randomUUID(); await sw.sessionCreate("plain-shell", ws.id, sid, "shell"); @@ -104,7 +102,7 @@ describe("silverwood wrapper contract (skip-mode, no network)", () => { }); test("session create records disk-space + claude-code-noninteractiveshell (options preserved)", async () => { - const ws = await newSkipWs("kinds"); + const ws = await newBasicWs("kinds"); const diskId = crypto.randomUUID(); await sw.sessionCreate("disk-space", ws.id, diskId, "disk"); @@ -119,7 +117,7 @@ describe("silverwood wrapper contract (skip-mode, no network)", () => { }); test("session lock/unlock advisory lifecycle", async () => { - const ws = await newSkipWs("lock"); + const ws = await newBasicWs("lock"); const sid = crypto.randomUUID(); await sw.sessionCreate("claude-code", ws.id, sid, "claude"); @@ -130,11 +128,8 @@ describe("silverwood wrapper contract (skip-mode, no network)", () => { expect((await sw.sessionLs(ws.id))[sid]?.lock).toBeUndefined(); }); - // A checkout-less kind: `create` must NOT pass the basic-only `--checkout-extent` - // flag (proven by this succeeding), the workstream has no `mode`, and an empty - // `local-blank` removes without --force. - test("local-blank: create without a checkout-extent flag, no mode, empty remove", async () => { - const ws = await sw.create({ name: "blank", path: ["local-blank"], args: [] }); + test("local-blank: create without a checkout mode, empty remove", async () => { + const ws = await sw.create({ name: "blank", path: ["local-blank"], args: [] }).completed; expect(ws.kind).toBe("local-blank"); expect(ws.mode).toBeUndefined(); // Checkout state is Basic-only; a checkout-less kind has none (not a fabricated "none"). @@ -151,7 +146,7 @@ describe("silverwood wrapper contract (skip-mode, no network)", () => { // spawnable: a local-blank session resolves to a runnable plan in the workstream's dir, // and silverwood — not papyrus — reports it not-spawnable once that dir is gone. test("spawnPlan: local-blank sessions are spawnable; silverwood owns the not-spawnable case", async () => { - const ws = await sw.create({ name: "blank", path: ["local-blank"], args: [] }); + const ws = await sw.create({ name: "blank", path: ["local-blank"], args: [] }).completed; const sid = crypto.randomUUID(); await sw.sessionCreate("plain-shell", ws.id, sid, "shell"); @@ -171,7 +166,7 @@ describe("silverwood wrapper contract (skip-mode, no network)", () => { name: "adopt", path: ["local-unmanaged-existing-path"], args: [adopted], - }); + }).completed; expect(ws.kind).toBe("local-unmanaged-existing-path"); expect(sw.checkoutLocation(ws)).toBe(adopted); diff --git a/experimental/papyrus/server/index.ts b/experimental/papyrus/server/index.ts index 03ea88a..e89a690 100644 --- a/experimental/papyrus/server/index.ts +++ b/experimental/papyrus/server/index.ts @@ -6,6 +6,7 @@ import { apiRoutes } from "./routes/api"; import { sessions, resolveRuntime, HOLDER } from "./services/sessionManager"; import { stripQueries } from "./services/ansiQueries"; import * as sw from "./services/silverwood"; +import { shutdownCreations } from "./services/creation"; import { PORT } from "./config"; import type { WebSocketData } from "./types"; @@ -25,7 +26,7 @@ app.route("/api", apiRoutes); app.use("/*", serveStatic({ root: "./client/dist" })); // WebSocket server -Bun.serve({ +const server = Bun.serve({ port: PORT, fetch(req, server) { const url = new URL(req.url); @@ -113,11 +114,19 @@ log(`\x1b[38;5;245m[server]\x1b[0m Forest: ${process.env.SILVERWOOD_FOREST_PATH // Cleanup on exit: release advisory locks this instance holds (best-effort), then // kill terminals. A hard crash skips this; a stuck lock is recovered via a force-steal. -process.on("SIGINT", async () => { +let shuttingDown = false; +async function shutdown() { + if (shuttingDown) return; + shuttingDown = true; + // Stop new HTTP work while accepted creation processes receive a grace period. + void server.stop(); + await shutdownCreations(); const held = [...sessions.entries()].filter(([, s]) => s.holdsLock); await Promise.allSettled( held.map(([key, s]) => sw.sessionUnlock(s.workstreamId, key, HOLDER)), ); for (const [, session] of sessions) session.pty.kill(); process.exit(0); -}); +} +process.on("SIGINT", shutdown); +process.on("SIGTERM", shutdown); diff --git a/experimental/papyrus/server/routes/api.ts b/experimental/papyrus/server/routes/api.ts index 35ea31d..8877218 100644 --- a/experimental/papyrus/server/routes/api.ts +++ b/experimental/papyrus/server/routes/api.ts @@ -302,33 +302,23 @@ apiRoutes.post("/sessions", async (c) => { let ws: sw.Workstream; try { - ws = await sw.create({ name, path, args: argList }); + ws = await sw.create({ name, path, args: argList }).registered; } catch (e: any) { logError(`\x1b[38;5;141m[create]\x1b[0m ${e.message}`); return c.json({ error: e.message }, 400); } - if (position) await sw.setKv(ws.id, "position", position); + try { + if (position) await sw.setKv(ws.id, "position", position); + } catch (e: any) { + logError(`[create] ${ws.id}: metadata write failed: ${e.message}`); + return c.json({ error: e.message, workstreamId: ws.id }, 400); + } const cwd = sw.checkoutLocation(ws) || ""; - // Provision the checkout in the background — deliberately NOT awaited, so the response - // returns now (modal closes). The node appears pending → ready/failed on the canvas via - // reconcile. No session is attached on create: a fresh workstream starts with zero - // sessions; the user adds one from the sidebar ("+" / "Start a session"). A provisioning - // failure surfaces as `basic.failed` on the canvas via reconcile, not here. - // - // Only a workstream awaiting a checkout needs provisioning — i.e. a `basic` kind created - // with `--checkout-extent skip` (`mode.state === "initialized-without-checkout"`). The - // `local-*` kinds have no checkout mode and are ready on create, so skip the (basic-only) - // `checkout` call, which would otherwise error on their non-basic kind. - const wsId = ws.id; - if (ws.mode?.state === "initialized-without-checkout") { - sw.checkout(wsId).catch((e: any) => { - logError(`\x1b[38;5;141m[checkout]\x1b[0m ${wsId}: ${e.message}`); - }); - } - + // The creation registry owns provisioning after this response. Polling reads + // its saved ready/failed outcome; no second CLI process is needed. return c.json({ sessionId: ws.id, nodeId: ws.id, diff --git a/experimental/papyrus/server/services/creation.ts b/experimental/papyrus/server/services/creation.ts new file mode 100644 index 0000000..b12f2d6 --- /dev/null +++ b/experimental/papyrus/server/services/creation.ts @@ -0,0 +1,209 @@ +// Version 1 Silverwood creation protocol and ownership of accepted operations. +import type { Workstream } from "./silverwood"; + +export interface CreationOperation { + registered: Promise; + completed: Promise; +} + +function deferred() { + let resolve!: (value: T) => void; + let reject!: (reason: Error) => void; + const promise = new Promise((yes, no) => { resolve = yes; reject = no; }); + // Registration can fail after its HTTP request has disconnected. + void promise.catch(() => {}); + return { promise, resolve, reject }; +} + +function object(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +export class CreationProtocol { + private stage: "initial" | "registered" | "started" | "terminal" = "initial"; + id?: string; + result?: Workstream; + failure?: Error; + + constructor(private registered: (ws: Workstream) => void) {} + + accept(line: string) { + const event: unknown = JSON.parse(line); + if (!object(event) || event.schemaVersion !== 1 || this.stage === "terminal") { + throw new Error("Invalid creation event version or order"); + } + if (event.event === "failed") { + if (!object(event.error) || !["validation", "provisioning", "persistence", "internal"].includes(event.error.code) + || typeof event.error.message !== "string" || event.workstreamId !== this.id) { + throw new Error("Invalid creation failure"); + } + this.failure = new Error(`${event.error.code}: ${event.error.message}`); + this.stage = "terminal"; + return; + } + if (typeof event.workstreamId !== "string" || !/^uuidv7_[0-9a-f]{8}-[0-9a-f]{4}-7[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/.test(event.workstreamId) + || (this.id && this.id !== event.workstreamId)) { + throw new Error("Conflicting or invalid creation identity"); + } + if (event.event === "checkout-started" && this.stage === "registered") { + this.stage = "started"; + return; + } + if ((event.event === "registered" && this.stage === "initial") || (event.event === "ready" && this.stage === "started")) { + const ws = event.workstream; + const state = event.event === "registered" ? "pending" : "ready"; + if (!object(ws) || ws.id !== event.workstreamId || ws.kind !== "basic" + || !["active", "archived"].includes(ws.status) || typeof ws.name !== "string" + || typeof ws.created_at !== "string" || typeof ws.overall_state !== "string" + || !object(ws.mode) || ws.mode.state !== state || typeof ws.mode.initial_source !== "string" + || typeof ws.mode.checkout_mode !== "string" || !object(ws.location) + || typeof ws.location.forest_id !== "string" || !object(ws.location.within) + || ws.location.within.forest_kind !== "basic-forest" || typeof ws.location.within.path !== "string" + || !ws.location.within.path.startsWith("/")) { + throw new Error("Invalid creation workstream payload"); + } + this.id = event.workstreamId; + if (event.event === "registered") { + this.stage = "registered"; + this.registered(ws as Workstream); + } else { + this.stage = "terminal"; + this.result = ws as Workstream; + } + return; + } + throw new Error(`Unknown or out-of-order creation event: ${event.event}`); + } + + finish(exitCode: number): Workstream { + if (this.failure) { + if (exitCode === 0) throw new Error("Creation failed event followed by exit zero"); + throw this.failure; + } + if (exitCode !== 0 || !this.result) throw new Error(`Creation ended without a successful ready event (exit ${exitCode})`); + return this.result; + } +} + +interface ActiveCreation { + id?: string; + stop: (signal: "SIGTERM" | "SIGKILL") => void; + completed: Promise; +} +const active = new Set(); +let stopping = false; + +export function startCreation(command: string[], streaming: boolean): CreationOperation { + if (stopping) throw new Error("Papyrus is shutting down; creation is closed"); + const registered = deferred(); + // A separate process group lets shutdown stop jj/cp/direnv descendants too. + const proc = Bun.spawn(command, { stdout: "pipe", stderr: "pipe", env: process.env, detached: true }); + const parser = new CreationProtocol((ws) => { + operation.id = ws.id; + registered.resolve(ws); + }); + let stderr = ""; + let groupStopped: Promise | undefined; + let stderrError: Error | undefined; + const drainStderr = (async () => { + const decoder = new TextDecoder(); + for await (const chunk of proc.stderr) { + stderr = (stderr + decoder.decode(chunk, { stream: true })).slice(-16_384); + } + stderr = (stderr + decoder.decode()).slice(-16_384); + })().catch((error: Error) => { + stderrError = error; + operation.stop("SIGTERM"); + }); + const operation: ActiveCreation = { + stop(signal) { + try { process.kill(-proc.pid, signal); } + catch { try { proc.kill(signal); } catch { /* Already exited. */ } } + if (signal === "SIGTERM" && !groupStopped) { + // A provider can outlive the CLI without inheriting these pipes. Finish + // group cleanup even if the direct child exits immediately on SIGTERM. + groupStopped = new Promise((resolve) => { + setTimeout(() => { operation.stop("SIGKILL"); resolve(); }, 1_000); + }); + } + }, + completed: undefined!, + }; + const completed = (async () => { + let error: Error | undefined; + let buffer = ""; + const decoder = new TextDecoder("utf-8", { fatal: true }); + try { + for await (const chunk of proc.stdout) { + if (error) continue; // Keep draining even after malformed output. + try { + buffer += decoder.decode(chunk, { stream: true }); + if (buffer.length > 1_048_576) throw new Error("Creation output exceeds protocol buffer limit"); + if (streaming) { + let newline: number; + while ((newline = buffer.indexOf("\n")) !== -1) { + const line = buffer.slice(0, newline); + buffer = buffer.slice(newline + 1); + parser.accept(line); + } + } + } catch (e) { + error = e instanceof Error ? e : new Error(String(e)); + operation.stop("SIGTERM"); + } + } + if (!error) { + buffer += decoder.decode(); + if (streaming && buffer.length) throw new Error("Truncated creation event (missing newline)"); + } + } catch (e) { + error = e instanceof Error ? e : new Error(String(e)); + operation.stop("SIGTERM"); + } + const exitCode = await proc.exited; + await drainStderr; + await groupStopped; + if (error || stderrError) throw error ?? stderrError; + if (streaming) return parser.finish(exitCode); + if (exitCode !== 0) throw new Error(stderr || `silverwood exited ${exitCode}`); + const ws = JSON.parse(buffer) as Workstream; + operation.id = ws.id; + registered.resolve(ws); + return ws; + })().catch(async (error: Error) => { + // Unexpected EOF/exit can leave a captured-output provider alive after the + // CLI has died. Retain ownership until that group has also been stopped. + operation.stop("SIGTERM"); + await groupStopped; + registered.reject(error); + throw new Error(`${error.message}${stderr ? `\n${stderr}` : ""}`); + }); + operation.completed = completed; + active.add(operation); + void completed.then( + () => active.delete(operation), + (error) => { + active.delete(operation); + console.error(`[creation] ${operation.id ?? "before registration"}: ${error.message}`); + }, + ); + return { registered: registered.promise, completed }; +} + +export async function settleCreations(): Promise { + await Promise.allSettled([...active].map((operation) => operation.completed)); +} + +export async function shutdownCreations(graceMs = 5_000): Promise { + stopping = true; + let timer: ReturnType | undefined; + await Promise.race([settleCreations(), new Promise((resolve) => { timer = setTimeout(resolve, graceMs); })]); + clearTimeout(timer); + const remaining = [...active]; + for (const operation of remaining) { + console.error(`[creation] interrupted ${operation.id ?? "before registration"}; inspect/archive pending records before creating a replacement`); + operation.stop("SIGTERM"); + } + // Each stop escalates to SIGKILL after one second if its pipes/child remain. + if (remaining.length) await settleCreations(); +} diff --git a/experimental/papyrus/server/services/silverwood.ts b/experimental/papyrus/server/services/silverwood.ts index a6d5fe4..58b82a0 100644 --- a/experimental/papyrus/server/services/silverwood.ts +++ b/experimental/papyrus/server/services/silverwood.ts @@ -1,3 +1,5 @@ +import { startCreation, type CreationOperation } from "./creation"; + // The single persistence boundary: papyrus stores *all* durable state in // silverwood by shelling out to the `silverwood --json` CLI. There is no local // disk state — no `.openui/`. A canvas node IS a silverwood workstream; its @@ -20,9 +22,7 @@ export const PAPYRUS_NS = "app.andref.papyrus"; export interface CheckoutMode { checkout_mode: string; initial_source: string; - // `initialized-without-checkout` = registered with the checkout deferred (papyrus - // creates with `--checkout-extent skip`, then provisions in the background). - state: "initialized-without-checkout" | "pending" | "ready" | "failed"; + state: "pending" | "ready" | "failed"; } // Where the checkout physically lives: which forest, and (per forest kind) where. @@ -164,36 +164,14 @@ export function spawnPlan(id: string, sessionId: string): Promise { export function create(params: { name: string; - path: string[]; // chosen subcommand names, e.g. ["basic", "apfs-cow"] - args: string[]; // positional values along that path, in order -}): Promise { - // For `basic`, register WITHOUT provisioning (`--checkout-extent skip`): the fast, - // validating phase — precheck + write the doc — returning the workstream in the - // `initialized-without-checkout` state; the caller provisions via `checkout()` in the - // background. `--checkout-extent` is a flag on the `basic` variant ONLY. The `local-*` - // kinds have no slow provisioning (they just create/adopt a directory), so they take no - // such flag and complete synchronously here. Either way a synchronous validation/precheck - // error throws (non-zero exit). `path`/`args` come from `new-schema` and become argv - // verbatim (no shell) — silverwood is the single validator. - const [variant, ...rest] = params.path; - const extentFlag = variant === "basic" ? ["--checkout-extent", "skip"] : []; - return run([ - "new", - variant, - ...extentFlag, - ...rest, - ...params.args, - "--name", - params.name, - ]); -} - -/// Provision the checkout of a workstream created with `create` (which defers it). -/// Blocks on provisioning (e.g. `jj git clone --colocate`); the returned workstream is -/// `ready` on success. Throws if provisioning fails (the workstream is left `failed`, -/// recoverable) or the workstream is not awaiting a checkout. Serialized per id. -export function checkout(id: string): Promise { - return serialize(id, () => run(["workstream", "--workstream-id", id, "basic", "checkout"])); + path: string[]; + args: string[]; +}): CreationOperation { + const streaming = params.path[0] === "basic"; + return startCreation([ + SILVERWOOD_BIN, streaming ? "--json-events" : "--json", + "new", ...params.path, ...params.args, "--name", params.name, + ], streaming); } export function archive(id: string): Promise { @@ -277,9 +255,8 @@ export function sessionUnlock( ); } -// silverwood does read-modify-overwrite of the whole document with no file -// locking, so two concurrent writers to the SAME workstream can lose an update. -// Serialize every mutation per workstream id through an in-process promise chain. +// Core serializes document writes with cross-process locks. Keep local request +// ordering deterministic with a per-workstream in-process promise chain. // (Different workstreams are independent files and run concurrently.) const writeQueues = new Map>(); -- 2.54.0