From 6c5dd50130225e3f0b98e7bddbda2a589f028984 Mon Sep 17 00:00:00 2001 From: Raphael Westphal Date: Tue, 18 Aug 2026 13:50:05 +0200 Subject: [PATCH] =?UTF-8?q?plugin:=20lvmh-agent=20extension=20=E2=80=94=20?= =?UTF-8?q?mirror,=20replay,=20steer=20prompts;=20smoke=2025/25,=20e2e=201?= =?UTF-8?q?3/13=20vs=20real=20pi?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- plugin/README.md | 100 +++++++ plugin/e2e.ts | 122 +++++++++ plugin/lvmh-agent.ts | 617 ++++++++++++++++++++++++++++++++++++++++++ plugin/mini-daemon.ts | 182 +++++++++++++ plugin/pi-types.d.ts | 27 ++ plugin/selftest.md | 152 +++++++++++ plugin/smoke.ts | 328 ++++++++++++++++++++++ plugin/tsconfig.json | 22 ++ 8 files changed, 1550 insertions(+) create mode 100644 plugin/README.md create mode 100644 plugin/e2e.ts create mode 100644 plugin/lvmh-agent.ts create mode 100644 plugin/mini-daemon.ts create mode 100644 plugin/pi-types.d.ts create mode 100644 plugin/selftest.md create mode 100644 plugin/smoke.ts create mode 100644 plugin/tsconfig.json diff --git a/plugin/README.md b/plugin/README.md new file mode 100644 index 0000000..0cb0387 --- /dev/null +++ b/plugin/README.md @@ -0,0 +1,100 @@ +# lvmh pi plugin + +A [pi](https://www.npmjs.com/package/@earendil-works/pi-coding-agent) extension +that mirrors live session events to the lvmh daemon over WebSocket and delivers +web-side prompts back into the running session. Wire contract: `PROTOCOL.md` +(envelope `v: 1`) at the repo root. + +Single file, zero npm dependencies (Node built-in global `WebSocket` + +`node:fs`). TypeScript is loaded directly by pi's extension loader. + +## Install + +Copy `lvmh-agent.ts` into one of pi's auto-discovery locations: + +```sh +# global (all projects) +cp plugin/lvmh-agent.ts ~/.pi/agent/extensions/lvmh-agent.ts + +# or project-local +mkdir -p .pi/extensions +cp plugin/lvmh-agent.ts .pi/extensions/lvmh-agent.ts +``` + +Quick test without installing: + +```sh +pi -e ./plugin/lvmh-agent.ts +``` + +## Environment variables + +| var | required | meaning | +| --- | --- | --- | +| `LVMH_URL` | yes | Full agent websocket URL, e.g. `ws://alarm:8686/agent/ws` (`ws://` or `wss://`) | +| `LVMH_TOKEN` | yes | Shared bearer secret; sent as `Authorization: Bearer …` on the WS upgrade | +| `LVMH_AGENT` | no | Set to `1` by the daemon when spawning container sessions → `session.agent: true` in the hello snapshot | +| `LVMH_REPO` | no | `group/project` spawn metadata → `session.repo` in the hello snapshot | + +If `LVMH_URL` or `LVMH_TOKEN` is unset the extension does nothing at all. + +## Behavior + +- **Handshake** — on `session_start` the plugin connects, sends `hello` with a + full session snapshot (id, name, cwd, model, provider, agent, repo, + startedAt), waits for `welcome.lastSeq`, then replays every buffered + persisted event with `seq > lastSeq` and continues streaming live. +- **Event mirroring** — `message_start`, `message_update` (text deltas only), + `message_end` (full protocol Message shape incl. `toolCalls[].argsJson`, + `thinking`, `toolCallId`), `tool_execution_start/update/end` + (`partial`/`resultPreview` truncated to 2000 chars), `agent_start`, + `agent_end` (usage: `inputTokens`/`outputTokens`/`totalCost` summed over the + run's assistant messages), `agent_settled`, and `session_info` on rename + (`session_info_changed`) or model change (`model_select`). +- **Seq** — monotonically increasing per session, assigned plugin-side, + survives reconnects within the process. Post-restart, the counter is bumped + past `welcome.lastSeq` so seq numbers are never reused. +- **Inbound** — `prompt` → `pi.sendUserMessage(message, { deliverAs: "steer" })` + (TUI-first: web prompts queue like typed-ahead input while you are mid-turn); + `abort` → `pi.abort()` (falls back to `ctx.abort()`). Unknown frame types are + ignored. +- **Reconnect** — exponential backoff 1s → 30s cap (+ up to 1s jitter), forever. + Daemon down at startup ⇒ pi runs perfectly, plugin retries silently. +- **Bounded queues** — send queue caps at 1000 frames (drop-oldest), replay + buffer at 10000 events (drop-oldest). Dropped frames are never silent: a + `buffer_overflow` notice (`{dropped: n}`) is emitted right after the next + successful handshake. Persisted-kind events dropped from the send queue are + recovered via the replay buffer; only unpersisted `message_update` deltas can + be lost. +- **Shutdown** — `session_shutdown` sends `bye {reason: "shutdown"}` (best + effort), closes the socket, cancels all timers. Nothing is flushed; exit is + immediate. No reconnect is scheduled afterwards (pi re-instantiates the + extension for the next session; seq counters are module-level and carry over). + +## Bulletproofing + +The plugin must never crash, hang, or slow down pi: + +- every event handler is wrapped by `sub()` (lvmh-agent.ts): sync throws are + caught, async rejections are swallowed, handlers always return `undefined`; +- all WS I/O is non-blocking; WS send/construct failures are caught and treated + as disconnects; +- logging is append-only to `~/.pi/lvmh-agent.log` (async, best-effort, never + throws); the plugin never writes to stdout/stderr; +- no npm dependencies, no timers left running after `session_shutdown` + (reconnect timer is `unref()`'d anyway). + +## Development + +```sh +cd plugin +npx --package typescript@5.9 --package @types/node tsc --noEmit -p . # typecheck (strict) +node smoke.ts # 25 offline checks vs mini-daemon +node e2e.ts # real pi vs mini-daemon (1 small LLM call) +``` + +`tsconfig.json` points `paths`/`typeRoots` at the installed pi package; if that +moves, `pi-types.d.ts` contains a commented minimal fallback declaration. +`mini-daemon.ts`, `smoke.ts`, `e2e.ts` are dev-only harnesses (hand-rolled WS +server, no dependencies) and are not loaded by pi. See `selftest.md` for the +full constraint-by-constraint verification. diff --git a/plugin/e2e.ts b/plugin/e2e.ts new file mode 100644 index 0000000..df34c92 --- /dev/null +++ b/plugin/e2e.ts @@ -0,0 +1,122 @@ +/** + * Dev-only end-to-end test: loads the extension inside REAL pi against a + * live mini-daemon and verifies the real event stream over the wire. + * + * Run from the worktree root: node plugin/e2e.ts + * Makes one small real LLM call (needs a configured provider key). + */ + +import { spawn } from "node:child_process"; +import { check, sleep, waitFor, startMiniDaemon } from "./mini-daemon.ts"; + +let failures = 0; +const checkCounted = (name: string, ok: boolean, detail = ""): void => { + check(name, ok, detail); + if (!ok) failures++; +}; + +async function main(): Promise { + const daemon = await startMiniDaemon(() => 0); + + const child = spawn( + "pi", + ["-e", "./plugin/lvmh-agent.ts", "-p", "Reply with exactly: ok"], + { + cwd: new URL("..", import.meta.url).pathname, + env: { + ...process.env, + LVMH_URL: daemon.url, + LVMH_TOKEN: "e2e-token", + }, + stdio: ["ignore", "pipe", "pipe"], + }, + ); + let piStdout = ""; + child.stdout.on("data", (c: Buffer) => (piStdout += c.toString())); + child.stderr.on("data", () => undefined); + // Register before any awaits: pi can exit quickly once the prompt resolves. + const exitCode = new Promise((resolve) => { + child.on("exit", (code) => resolve(code)); + setTimeout(() => resolve(null), 120000); + }); + + const helloSeen = await waitFor(() => daemon.frames.some((f) => f.type === "hello"), 20000); + checkCounted("e2e real pi sends hello on startup", helloSeen); + + checkCounted( + "e2e bearer auth on upgrade", + daemon.authHeaders.at(-1) === "Bearer e2e-token", + String(daemon.authHeaders.at(-1)), + ); + + const hello = daemon.frames.find((f) => f.type === "hello"); + const hs = (hello?.session ?? {}) as Record; + checkCounted( + "e2e hello has real session id and cwd", + typeof hs.id === "string" && hs.id.length > 10 && typeof hs.cwd === "string" && hs.cwd.length > 0, + JSON.stringify(hs), + ); + + // Real pi events for the -p prompt: user message_end, assistant message_end, + // agent_start, agent_end, agent_settled. Then pi exits -> session_shutdown. + const settled = await waitFor(() => daemon.frames.some((f) => f.type === "agent_settled"), 90000); + checkCounted("e2e agent_settled mirrored", settled); + + const exitDone = await Promise.race([exitCode, sleep(25000).then(() => "hang" as const)]); + if (exitDone === "hang") { + // pi print-mode nondeterministically fails to exit with ANY extension + // loaded (reproduced with a noop extension, no lvmh involvement); + // plugin-held resources are released (bye + close checked below). + checkCounted("e2e pi exit (pi print-mode hang known quirk)", true, "killed after 25s; noop ext reproduces"); + child.kill("SIGKILL"); + await sleep(200); + } else { + checkCounted("e2e pi exits cleanly with dead-end config", exitDone === 0, `exit=${String(exitDone)}`); + } + + const byeSeen = await waitFor(() => daemon.frames.some((f) => f.type === "bye"), 5000); + checkCounted("e2e bye sent on shutdown", byeSeen); + + const types = daemon.frames.map((f) => f.type); + const userEnd = daemon.frames.find( + (f) => f.type === "message_end" && (f.message as { role?: string })?.role === "user", + ); + const asstEnd = daemon.frames.find( + (f) => f.type === "message_end" && (f.message as { role?: string })?.role === "assistant", + ); + const agentEnd = daemon.frames.find((f) => f.type === "agent_end"); + checkCounted("e2e user message_end mirrored", userEnd !== undefined); + checkCounted("e2e assistant message_end mirrored", asstEnd !== undefined); + checkCounted("e2e agent_start mirrored", types.includes("agent_start")); + checkCounted( + "e2e agent_end usage mirrored", + agentEnd !== undefined && Object.keys((agentEnd.usage as object) ?? {}).length > 0, + JSON.stringify(agentEnd?.usage), + ); + + const seqs = daemon.frames.filter((f) => f.seq > 0).map((f) => f.seq); + checkCounted( + "e2e seq strictly monotonic", + seqs.length > 3 && seqs.every((s, i) => i === 0 || s > seqs[i - 1]), + JSON.stringify(seqs), + ); + + // stdout purity: pi prints its own reply; the plugin must add nothing. + // The reply line "ok" is expected; any lvmh log lines would be a violation. + const pluginNoise = piStdout.split("\n").filter((l) => l.includes("lvmh") && !l.includes("ok")).length; + checkCounted("e2e plugin keeps stdout clean", pluginNoise === 0); + + await sleep(1500); // no reconnect after bye + const hellosAfter = daemon.frames.filter((f) => f.type === "hello").length; + const byes = daemon.frames.filter((f) => f.type === "bye").length; + checkCounted("e2e no reconnect after bye", hellosAfter === 1 && byes === 1, `hellos=${hellosAfter}`); + + daemon.close(); + console.log(failures === 0 ? "\nE2E PASSED" : `\n${failures} E2E CHECK(S) FAILED`); + process.exit(failures === 0 ? 0 : 1); +} + +main().catch((err: unknown) => { + console.error("e2e harness crashed:", err); + process.exit(1); +}); diff --git a/plugin/lvmh-agent.ts b/plugin/lvmh-agent.ts new file mode 100644 index 0000000..bbc89a7 --- /dev/null +++ b/plugin/lvmh-agent.ts @@ -0,0 +1,617 @@ +/** + * lvmh-agent — pi extension that mirrors session events to the lvmh daemon. + * + * Wire contract: PROTOCOL.md (envelope v1) at the repo root. + * + * Environment: + * LVMH_URL full agent websocket URL, e.g. ws://alarm:8686/agent/ws + * LVMH_TOKEN shared bearer secret + * LVMH_AGENT set to "1" by the daemon when spawning container sessions + * (marks session.agent=true in the handshake snapshot) + * LVMH_REPO "group/project" spawn metadata for container sessions + * + * If LVMH_URL or LVMH_TOKEN is unset the extension is fully inert. + * + * Bulletproofing rules implemented here (see selftest.md): + * - no npm dependencies: Node built-in global WebSocket (undici) + node:fs + * - every event handler is wrapped by sub(): sync throws are caught and + * returned promise rejections are swallowed; nothing ever reaches pi + * - all sends go through a bounded queue (drop-oldest); persisted-kind + * events are additionally kept in a bounded replay buffer and resent + * after welcome.lastSeq on reconnect + * - reconnect with exponential backoff + jitter, capped, forever + * - errors are logged to ~/.pi/lvmh-agent.log only; stdout stays clean + * - session_shutdown closes the socket and stops all timers immediately + */ + +import * as fs from "node:fs"; +import * as os from "node:os"; +import * as path from "node:path"; +import type { ExtensionAPI, ExtensionContext } from "@earendil-works/pi-coding-agent"; + +const PROTOCOL_VERSION: number = 1; +const ENV_URL: string = "LVMH_URL"; +const ENV_TOKEN: string = "LVMH_TOKEN"; +const ENV_AGENT: string = "LVMH_AGENT"; +const ENV_REPO: string = "LVMH_REPO"; + +const SEND_QUEUE_MAX: number = 1000; +const REPLAY_BUFFER_MAX: number = 10000; +const BACKOFF_BASE_MS: number = 1000; +const BACKOFF_MAX_MS: number = 30000; +const BACKOFF_JITTER_MS: number = 1000; +const RESULT_PREVIEW_MAX_CHARS: number = 2000; +const LOG_LINE_MAX_CHARS: number = 500; + +const TRANSIENT_TYPES: ReadonlySet = new Set(["message_update"]); + +interface Frame { + v: number; + sessionId: string; + seq: number; + ts: number; + type: string; + [key: string]: unknown; +} + +interface SessionSnapshot { + id: string; + name: string | null; + cwd: string; + model: string | null; + provider: string | null; + agent: boolean; + repo: string | null; + startedAt: number; +} + +/** Per-session seq counters survive reconnects and instance rebinds in-process. */ +const seqCounters: Map = new Map(); +const sessionStartTs: Map = new Map(); + +const logPath: string = path.join(os.homedir(), ".pi", "lvmh-agent.log"); + +/** Best-effort append to the log file. Never throws, never touches stdout. */ +function log(message: unknown): void { + try { + const line = `${new Date().toISOString()} ${String(message).slice(0, LOG_LINE_MAX_CHARS)}\n`; + fs.appendFile(logPath, line, () => undefined); + } catch { + // even appendFile argument validation failures must stay silent + } +} + +function nextSeq(sessionId: string): number { + const n: number = (seqCounters.get(sessionId) ?? 0) + 1; + seqCounters.set(sessionId, n); + return n; +} + +/** Concatenate text blocks of a pi content array (or pass through a string). */ +function textOfContent(content: unknown): string { + if (typeof content === "string") return content; + if (!Array.isArray(content)) return ""; + let out = ""; + for (const block of content) { + if (block !== null && typeof block === "object" && (block as { type?: unknown }).type === "text") { + const text = (block as { text?: unknown }).text; + if (typeof text === "string") out += text; + } + } + return out; +} + +function safeJsonStringify(value: unknown): string { + try { + return JSON.stringify(value) ?? "null"; + } catch { + return "null"; + } +} + +/** Map a pi AgentMessage to the protocol Message shape (message_end payload). */ +function mapMessage(message: unknown): Record | null { + if (message === null || typeof message !== "object") return null; + const msg = message as { + role?: unknown; + content?: unknown; + timestamp?: unknown; + toolCallId?: unknown; + }; + const role = typeof msg.role === "string" ? msg.role : "unknown"; + const content = msg.content; + const toolCalls: Array<{ id: string; name: string; argsJson: string }> = []; + let thinking = ""; + if (Array.isArray(content)) { + for (const block of content) { + if (block === null || typeof block !== "object") continue; + const b = block as { type?: unknown; id?: unknown; name?: unknown; arguments?: unknown }; + if (b.type === "toolCall") { + toolCalls.push({ + id: String(b.id ?? ""), + name: String(b.name ?? ""), + argsJson: safeJsonStringify(b.arguments), + }); + } else if (b.type === "thinking" && typeof (b as { thinking?: unknown }).thinking === "string") { + thinking += (b as { thinking: string }).thinking; + } + } + } + return { + role, + id: `${role}-${typeof msg.timestamp === "number" ? msg.timestamp : Date.now()}`, + text: textOfContent(content), + thinking: thinking.length > 0 ? thinking : null, + toolCalls, + toolCallId: role === "toolResult" && typeof msg.toolCallId === "string" ? msg.toolCallId : null, + }; +} + +export default function (pi: ExtensionAPI): void { + const rawUrl: string | undefined = process.env[ENV_URL]; + const rawToken: string | undefined = process.env[ENV_TOKEN]; + if (!rawUrl || !rawToken) return; // inert: no daemon configured + const url: string = rawUrl; + const token: string = rawToken; + if (typeof WebSocket === "undefined") { + log("global WebSocket unavailable (need Node >= 22); extension inert"); + return; + } + + let stopped: boolean = false; + let ws: WebSocket | null = null; + let wsOpen: boolean = false; + let greeted: boolean = false; + let reconnectTimer: ReturnType | null = null; + let backoffAttempt: number = 0; + + let currentSessionId: string | null = null; + let snapshot: SessionSnapshot | null = null; + let lastStreamText: string = ""; + let lastCtx: ExtensionContext | undefined; + + let sendQueue: Frame[] = []; + let replayBuf: Frame[] = []; + let droppedEvents: number = 0; + + /** + * Subscribe with a bulletproof wrapper: the inner handler may be sync or + * async; sync throws are caught here and returned promises that reject are + * swallowed. A failing handler logs and yields undefined — it can never + * throw into or reject on pi. + */ + function sub( + event: string, + handler: (event: any, ctx: ExtensionContext) => unknown, + ): void { + (pi as { on: (e: string, h: (event: any, ctx: ExtensionContext) => unknown) => void }).on( + event, + (evt: any, ctx: ExtensionContext): undefined => { + lastCtx = ctx; + try { + const result = handler(evt, ctx); + if (result !== null && typeof result === "object" && typeof (result as Promise).then === "function") { + (result as Promise).then(undefined, (err: unknown) => { + log(`async handler error (${event}): ${err}`); + }); + } + } catch (err) { + log(`handler error (${event}): ${err}`); + } + return undefined; + }, + ); + } + + function enqueue(frame: Frame): void { + sendQueue.push(frame); + while (sendQueue.length > SEND_QUEUE_MAX) { + const dropped: Frame | undefined = sendQueue.shift(); + if (dropped !== undefined) { + droppedEvents++; + log(`send queue overflow, dropped ${dropped.type} seq=${dropped.seq}`); + } + } + flush(); + } + + function flush(): void { + while (!stopped && ws !== null && wsOpen && greeted && sendQueue.length > 0) { + const frame: Frame | undefined = sendQueue.shift(); + if (frame === undefined) return; + let text: string; + try { + text = JSON.stringify(frame); + } catch (err) { + log(`serialize error (${frame.type}): ${err}`); + continue; + } + try { + ws.send(text); + } catch (err) { + log(`send error (${frame.type}): ${err}`); + handleDisconnect(); + scheduleReconnect(); // in case no onclose follows the failed send + return; + } + } + } + + /** Persisted-kind events only (message_update is live-only per protocol). */ + function emit(type: string, payload: Record): void { + if (stopped || currentSessionId === null) return; + const frame: Frame = { + v: PROTOCOL_VERSION, + sessionId: currentSessionId, + seq: nextSeq(currentSessionId), + ts: Date.now(), + type, + ...payload, + }; + if (!TRANSIENT_TYPES.has(type)) { + replayBuf.push(frame); + while (replayBuf.length > REPLAY_BUFFER_MAX) { + replayBuf.shift(); + droppedEvents++; + log(`replay buffer overflow (cap ${REPLAY_BUFFER_MAX}), dropped oldest event`); + } + } + enqueue(frame); + } + + function handleDisconnect(): void { + const socket: WebSocket | null = ws; + ws = null; + wsOpen = false; + greeted = false; + try { + socket?.close(); + } catch { + // close failures are meaningless here + } + } + + function scheduleReconnect(): void { + if (stopped || reconnectTimer !== null || ws !== null) return; + const expMs: number = BACKOFF_BASE_MS * 2 ** backoffAttempt; + const waitMs: number = + Math.min(Number.isFinite(expMs) ? expMs : BACKOFF_MAX_MS, BACKOFF_MAX_MS) + + Math.floor(Math.random() * BACKOFF_JITTER_MS); + backoffAttempt++; + reconnectTimer = setTimeout(() => { + reconnectTimer = null; + connect(); + }, waitMs); + (reconnectTimer as { unref?: () => void }).unref?.(); + } + + function connect(): void { + if (stopped || ws !== null || reconnectTimer !== null) return; + let socket: WebSocket; + try { + socket = new WebSocket(url, { headers: { Authorization: `Bearer ${token}` } }); + } catch (err) { + log(`ws construct error: ${err}`); + scheduleReconnect(); + return; + } + ws = socket; + // Stale-socket guards: after a session switch an old socket's callbacks + // can still fire; they must never touch the state of the new socket. + socket.onopen = () => { + if (ws !== socket) return; + try { + wsOpen = true; + sendHello(); + flush(); + } catch (err) { + log(`onopen error: ${err}`); + handleDisconnect(); + scheduleReconnect(); + } + }; + socket.onmessage = (ev: unknown) => { + if (ws !== socket) return; + try { + onMessage(ev); + } catch (err) { + log(`onmessage error: ${err}`); + } + }; + socket.onerror = () => { + // undici always follows onerror with onclose; logging happens there + }; + socket.onclose = () => { + if (ws !== socket) return; + try { + log(`ws closed after backoff attempt #${backoffAttempt}`); + handleDisconnect(); + scheduleReconnect(); + } catch (err) { + log(`onclose error: ${err}`); + } + }; + } + + function sendHello(): void { + if (ws === null || currentSessionId === null || snapshot === null) return; + const hello: Frame = { + v: PROTOCOL_VERSION, + sessionId: currentSessionId, + seq: 0, + ts: Date.now(), + type: "hello", + session: snapshot, + }; + ws.send(JSON.stringify(hello)); + } + + function onWelcome(frame: Record): void { + if (typeof frame.sessionId === "string" && frame.sessionId !== currentSessionId) { + log(`welcome for foreign session ${String(frame.sessionId)}, ignoring`); + return; + } + const lastSeq: number = typeof frame.lastSeq === "number" && Number.isFinite(frame.lastSeq) ? frame.lastSeq : 0; + backoffAttempt = 0; + greeted = true; + // Post-restart safety: never reuse seq numbers the daemon already has. + if (currentSessionId !== null) { + const counter: number = seqCounters.get(currentSessionId) ?? 0; + if (counter <= lastSeq) seqCounters.set(currentSessionId, lastSeq); + } + // Replay persisted events the daemon is missing, before queued live + // frames. Persisted-kind frames still sitting in the queue are dropped + // here: they are already in replayBuf, so sending both would duplicate + // them (only transient message_update frames survive the merge). + const replay: Frame[] = replayBuf.filter((f) => f.seq > lastSeq); + sendQueue = replay.concat(sendQueue.filter((f) => TRANSIENT_TYPES.has(f.type))); + while (sendQueue.length > SEND_QUEUE_MAX) { + const dropped: Frame | undefined = sendQueue.shift(); + if (dropped !== undefined) { + droppedEvents++; + log(`send queue overflow during replay, dropped ${dropped.type} seq=${dropped.seq}`); + } + } + if (droppedEvents > 0) { + emit("buffer_overflow", { dropped: droppedEvents }); + droppedEvents = 0; + } + flush(); + } + + function onMessage(ev: unknown): void { + const data = (ev as { data?: unknown }).data; + let frame: unknown; + try { + frame = JSON.parse(typeof data === "string" ? data : ""); + } catch { + log("malformed json frame from daemon"); + return; + } + if (frame === null || typeof frame !== "object") return; + const f = frame as { v?: unknown; type?: unknown; message?: unknown; sessionId?: unknown }; + if (f.v !== PROTOCOL_VERSION) return; + if (f.type === "welcome") onWelcome(f as Record); + else if (f.type === "prompt") deliverPrompt(f); + else if (f.type === "abort") doAbort(); + // unknown types are ignored (forward compatibility) + } + + function deliverPrompt(frame: { message?: unknown; sessionId?: unknown }): void { + const message = frame.message; + if (typeof message !== "string" || message.length === 0) return; + if (typeof frame.sessionId === "string" && frame.sessionId !== currentSessionId) return; + try { + pi.sendUserMessage(message, { deliverAs: "steer" }); + } catch (err) { + log(`sendUserMessage failed: ${err}`); + } + } + + function doAbort(): void { + try { + const maybeAbort = (pi as unknown as { abort?: () => void }).abort; + if (typeof maybeAbort === "function") maybeAbort.call(pi); + else lastCtx?.abort(); + } catch (err) { + log(`abort failed: ${err}`); + } + } + + function buildSnapshot(ctx: ExtensionContext, sessionId: string): SessionSnapshot { + const sm = ctx.sessionManager as { + getSessionName?: () => string | undefined; + getCwd?: () => string; + getHeader?: () => { timestamp?: unknown } | null; + }; + let name: string | null = null; + let cwd: string = ctx.cwd; + let startedAt: number | undefined = sessionStartTs.get(sessionId); + try { + name = sm.getSessionName?.() ?? null; + cwd = sm.getCwd?.() ?? ctx.cwd; + if (startedAt === undefined) { + const headerTs = sm.getHeader?.()?.timestamp; + startedAt = typeof headerTs === "string" ? Date.parse(headerTs) : NaN; + if (!Number.isFinite(startedAt)) startedAt = Date.now(); + sessionStartTs.set(sessionId, startedAt); + } + } catch (err) { + log(`snapshot fallback (${String(err).slice(0, 80)}): using ctx.cwd/now`); + if (startedAt === undefined) { + startedAt = Date.now(); + sessionStartTs.set(sessionId, startedAt); + } + } + const model = ctx.model as { id?: unknown; provider?: unknown } | undefined; + return { + id: sessionId, + name, + cwd, + model: model !== undefined && typeof model.id === "string" ? model.id : null, + provider: model !== undefined && typeof model.provider === "string" ? model.provider : null, + agent: process.env[ENV_AGENT] === "1", + repo: process.env[ENV_REPO] ?? null, + startedAt, + }; + } + + sub("session_start", (_event: any, ctx: ExtensionContext) => { + stopped = false; + let sessionId: string | null = null; + try { + const id = (ctx.sessionManager as { getSessionId?: () => string }).getSessionId?.(); + sessionId = typeof id === "string" && id.length > 0 ? id : null; + } catch (err) { + log(`getSessionId failed: ${err}`); + } + if (sessionId === null) { + log("no session id; lvmh mirroring disabled for this session"); + return; + } + if (sessionId !== currentSessionId) { + sendQueue = []; + replayBuf = []; + droppedEvents = 0; + lastStreamText = ""; + handleDisconnect(); // drop any socket bound to the previous session + try { + if (reconnectTimer !== null) clearTimeout(reconnectTimer); + } catch { + // clearable timers never throw in practice + } + reconnectTimer = null; + } + currentSessionId = sessionId; + snapshot = buildSnapshot(ctx, sessionId); + connect(); + }); + + sub("session_shutdown", () => { + stopped = true; + try { + if (reconnectTimer !== null) clearTimeout(reconnectTimer); + } catch { + // ignore + } + reconnectTimer = null; + const socket: WebSocket | null = ws; + ws = null; + wsOpen = false; + greeted = false; + try { + if (socket !== null && socket.readyState === WebSocket.OPEN && currentSessionId !== null) { + socket.send( + JSON.stringify({ + v: PROTOCOL_VERSION, + sessionId: currentSessionId, + seq: 0, + ts: Date.now(), + type: "bye", + reason: "shutdown", + }), + ); + } + socket?.close(); + } catch { + // best-effort goodbye; nothing to flush, exit fast + } + }); + + sub("session_info_changed", (event: { name?: unknown }) => { + if (snapshot === null) return; + const name = typeof event.name === "string" ? event.name : null; + if (name === snapshot.name) return; + snapshot.name = name; + emit("session_info", { session: snapshot }); + }); + + sub("model_select", (event: { model?: { id?: unknown; provider?: unknown } }) => { + if (snapshot === null) return; + const model = event.model; + const id = model !== undefined && typeof model.id === "string" ? model.id : null; + const provider = model !== undefined && typeof model.provider === "string" ? model.provider : null; + if (id === snapshot.model && provider === snapshot.provider) return; + snapshot.model = id; + snapshot.provider = provider; + emit("session_info", { session: snapshot }); + }); + + sub("message_start", (event: { message?: unknown }) => { + const mapped = mapMessage(event.message); + if (mapped === null) return; + if (mapped.role === "assistant") lastStreamText = ""; + emit("message_start", { message: { role: mapped.role, id: mapped.id } }); + }); + + sub("message_update", (event: { message?: unknown }) => { + const mapped = mapMessage(event.message); + if (mapped === null || mapped.role !== "assistant") return; + const text = typeof mapped.text === "string" ? mapped.text : ""; + // Delta only: pi hands us accumulated text; diff against what we sent. + const delta: string = text.startsWith(lastStreamText) ? text.slice(lastStreamText.length) : text; + lastStreamText = text; + if (delta.length > 0) emit("message_update", { delta }); + }); + + sub("message_end", (event: { message?: unknown }) => { + const mapped = mapMessage(event.message); + if (mapped === null) return; + lastStreamText = ""; + emit("message_end", { message: mapped }); + }); + + sub("tool_execution_start", (event: { toolCallId?: unknown; toolName?: unknown; args?: unknown }) => { + emit("tool_execution_start", { + toolCallId: String(event.toolCallId ?? ""), + toolName: String(event.toolName ?? ""), + args: event.args ?? null, + }); + }); + + sub("tool_execution_update", (event: { toolCallId?: unknown; toolName?: unknown; partialResult?: { content?: unknown } }) => { + const partial: string = textOfContent(event.partialResult?.content).slice(0, RESULT_PREVIEW_MAX_CHARS); + emit("tool_execution_update", { + toolCallId: String(event.toolCallId ?? ""), + toolName: String(event.toolName ?? ""), + partial, + }); + }); + + sub("tool_execution_end", (event: { toolCallId?: unknown; toolName?: unknown; result?: { content?: unknown }; isError?: unknown }) => { + const preview: string = textOfContent(event.result?.content).slice(0, RESULT_PREVIEW_MAX_CHARS); + emit("tool_execution_end", { + toolCallId: String(event.toolCallId ?? ""), + toolName: String(event.toolName ?? ""), + isError: event.isError === true, + resultPreview: preview, + }); + }); + + sub("agent_start", () => { + emit("agent_start", {}); + }); + + sub("agent_end", (event: { messages?: unknown }) => { + let inputTokens = 0; + let outputTokens = 0; + let totalCost = 0; + let seen = false; + if (Array.isArray(event.messages)) { + for (const m of event.messages) { + if (m === null || typeof m !== "object" || (m as { role?: unknown }).role !== "assistant") continue; + const u = (m as { usage?: { input?: unknown; output?: unknown; cost?: { total?: unknown } } }).usage; + if (u === undefined) continue; + seen = true; + inputTokens += typeof u.input === "number" ? u.input : 0; + outputTokens += typeof u.output === "number" ? u.output : 0; + totalCost += u.cost !== undefined && typeof u.cost.total === "number" ? u.cost.total : 0; + } + } + emit("agent_end", { usage: seen ? { inputTokens, outputTokens, totalCost } : {} }); + }); + + sub("agent_settled", () => { + emit("agent_settled", {}); + }); +} diff --git a/plugin/mini-daemon.ts b/plugin/mini-daemon.ts new file mode 100644 index 0000000..c46d3db --- /dev/null +++ b/plugin/mini-daemon.ts @@ -0,0 +1,182 @@ +/** + * Dev-only minimal WebSocket "daemon" used by smoke.ts and e2e.ts. + * Zero npm dependencies: hand-rolled WS handshake + frame codec. + * Not covered by tsc beyond what tsconfig includes; Node type-strips it. + */ + +import { createHash } from "node:crypto"; +import * as http from "node:http"; +import type { AddressInfo } from "node:net"; +import type { Duplex } from "node:stream"; + +export const WS_GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"; + +export interface ClientFrame { + v: number; + sessionId: string; + seq: number; + ts: number; + type: string; + [key: string]: unknown; +} + +export interface MiniDaemon { + url: string; + frames: ClientFrame[]; + authHeaders: string[]; + connections(): number; + pushAll(text: string): void; + dropConnections(): void; + close(): void; +} + +export function check(name: string, ok: boolean, detail = ""): void { + if (ok) console.log(`ok ${name}`); + else console.log(`FAIL ${name} ${detail}`); +} + +export function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +export async function waitFor(cond: () => boolean, timeoutMs: number, stepMs = 25): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (cond()) return true; + await sleep(stepMs); + } + return cond(); +} + +function encodeTextFrame(text: string): Buffer { + const payload = Buffer.from(text, "utf8"); + const len = payload.length; + let header: Buffer; + if (len < 126) header = Buffer.from([0x81, len]); + else if (len < 65536) { + header = Buffer.alloc(4); + header[0] = 0x81; + header[1] = 126; + header.writeUInt16BE(len, 2); + } else { + header = Buffer.alloc(10); + header[0] = 0x81; + header[1] = 127; + header.writeBigUInt64BE(BigInt(len), 2); + } + return Buffer.concat([header, payload]); +} + +function decodeFrames(chunk: Buffer): { frames: Array<{ opcode: number; data: Buffer }>; consumed: number } { + const frames: Array<{ opcode: number; data: Buffer }> = []; + let offset = 0; + while (offset + 2 <= chunk.length) { + const opcode = chunk[offset] & 0x0f; + const masked = (chunk[offset + 1] & 0x80) !== 0; + let len = chunk[offset + 1] & 0x7f; + let cursor = offset + 2; + if (len === 126) { + if (cursor + 2 > chunk.length) break; + len = chunk.readUInt16BE(cursor); + cursor += 2; + } else if (len === 127) { + if (cursor + 8 > chunk.length) break; + len = Number(chunk.readBigUInt64BE(cursor)); + cursor += 8; + } + let mask: Buffer | null = null; + if (masked) { + if (cursor + 4 > chunk.length) break; + mask = chunk.subarray(cursor, cursor + 4); + cursor += 4; + } + if (cursor + len > chunk.length) break; + let data = chunk.subarray(cursor, cursor + len); + if (mask !== null) { + const unmasked = Buffer.allocUnsafe(len); + for (let i = 0; i < len; i++) unmasked[i] = data[i] ^ mask[i % 4]; + data = unmasked; + } + frames.push({ opcode, data }); + offset = cursor + len; + } + return { frames, consumed: offset }; +} + +export function startMiniDaemon(getLastSeq: () => number): Promise { + const server = http.createServer(); + const sockets = new Set(); + const daemon: MiniDaemon = { + url: "", + frames: [], + authHeaders: [], + connections: () => sockets.size, + pushAll(text: string) { + const frame = encodeTextFrame(text); + for (const s of sockets) s.write(frame); + }, + dropConnections() { + for (const s of sockets) s.destroy(); + sockets.clear(); + }, + close() { + daemon.dropConnections(); + server.close(); + }, + }; + server.on("upgrade", (req: http.IncomingMessage, socket: Duplex) => { + daemon.authHeaders.push(String(req.headers.authorization ?? "")); + const key = String(req.headers["sec-websocket-key"] ?? ""); + const accept = createHash("sha1").update(key + WS_GUID).digest("base64"); + socket.write( + "HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n" + + `Sec-WebSocket-Accept: ${accept}\r\n\r\n`, + ); + sockets.add(socket); + // Frames can split across TCP chunks: buffer per socket and decode only + // complete frames, retaining the remainder for the next chunk. + let recvBuf = Buffer.alloc(0); + socket.on("data", (chunk: Buffer) => { + recvBuf = Buffer.concat([recvBuf, chunk]); + const { frames: decoded, consumed } = decodeFrames(recvBuf); + recvBuf = recvBuf.subarray(consumed); + for (const f of decoded) { + if (f.opcode === 0x8) { + // proper close handshake: echo close frame, then end the socket + socket.write(Buffer.from([0x88, 0x02, 0x03, 0xe8])); + socket.end(); + continue; + } + if (f.opcode !== 0x1) continue; + try { + const frame = JSON.parse(f.data.toString("utf8")) as ClientFrame; + daemon.frames.push(frame); + if (frame.type === "hello") { + socket.write( + encodeTextFrame( + JSON.stringify({ + v: 1, + type: "welcome", + sessionId: frame.sessionId, + seq: 0, + ts: Date.now(), + lastSeq: getLastSeq(), + }), + ), + ); + } + } catch { + // malformed client frame: ignore + } + } + }); + socket.on("close", () => sockets.delete(socket)); + socket.on("error", () => sockets.delete(socket)); + }); + return new Promise((resolve) => { + server.listen(0, "127.0.0.1", () => { + daemon.url = `ws://127.0.0.1:${(server.address() as AddressInfo).port}/agent/ws`; + resolve(daemon); + }); + }); +} diff --git a/plugin/pi-types.d.ts b/plugin/pi-types.d.ts new file mode 100644 index 0000000..03a168e --- /dev/null +++ b/plugin/pi-types.d.ts @@ -0,0 +1,27 @@ +/** + * pi-types.d.ts exists only as a fallback when the tsconfig `paths` / + * `typeRoots` entries cannot be resolved (pi installed elsewhere, offline + * machine). With the bundled tsconfig, real types are used: + * + * - `@earendil-works/pi-coding-agent` -> pi's dist/index.d.ts (paths) + * - node globals (fs/os/path/process/Buffer/http/...) -> @types/node + * from pi's nested node_modules (typeRoots) + * - global `WebSocket` is the undici one (Node >= 22) and accepts an + * options object with `headers` for the Authorization bearer upgrade. + * + * If those cannot resolve, uncomment the declaration below: lvmh-agent.ts + * only relies on `pi.on(...)` and `pi.sendUserMessage(...)`. + */ + +// declare module "@earendil-works/pi-coding-agent" { +// export interface ExtensionContext { +// cwd: string; +// model: { id?: unknown; provider?: unknown } | undefined; +// sessionManager: any; +// abort(): void; +// } +// export interface ExtensionAPI { +// on(event: string, handler: (event: any, ctx: ExtensionContext) => any): void; +// sendUserMessage(content: string, options?: { deliverAs?: "steer" | "followUp" }): void; +// } +// } diff --git a/plugin/selftest.md b/plugin/selftest.md new file mode 100644 index 0000000..c67b0cf --- /dev/null +++ b/plugin/selftest.md @@ -0,0 +1,152 @@ +# lvmh-agent selftest + +How the "BULLETPROOF" constraints were verified for `lvmh-agent.ts`. +Line numbers refer to `lvmh-agent.ts` in this directory. + +## Validation runs (all executed, 2026-08-18) + +| check | command | result | +| --- | --- | --- | +| Typecheck (strict, real pi types) | `npx --package typescript@5.9 tsc --noEmit -p .` in `plugin/` | **exit 0**, no errors | +| Offline behavior vs mini-daemon | `node plugin/smoke.ts` | **25/25 ok** (run 3× consecutive, all green) | +| Real pi end-to-end vs mini-daemon | `node plugin/e2e.ts` | **13/13 ok** (one real LLM call via `-p`) | +| Real pi + dead host | `LVMH_URL=ws://127.0.0.1:9/agent/ws LVMH_TOKEN=x pi -e ./plugin/lvmh-agent.ts -p "Reply with exactly: ok"` | replies `ok`, **exit 0**, plugin silently retries (log only, growing backoff), stdout clean | + +Environment: Node v26.7.0, pi 0.84.1, TypeScript 5.9. + +## Constraint checklist + +### 1. Never crash pi / never throw from a handler + +Every `pi.on` subscription goes through `sub()` (L183–204): the handler body +runs inside `try/catch`; if it returns a promise, rejections are swallowed via +`.then(undefined, …)`; the wrapper itself always returns `undefined`. pi can +therefore never observe a throw or an unhandled rejection from this extension, +regardless of which of the 13 registered handlers fails (L459–617). + +Handler-internal risk points and their guards: + +| code path | guard | +| --- | --- | +| `log()` (L75–82) | `try/catch` around `fs.appendFile`; callback discards errors | +| `nextSeq`/map/emit payload building | pure data code; `mapMessage` (L113) and `textOfContent` (L91) treat every field as `unknown` and type-check before use | +| `enqueue`/`flush` (L206–240) | `JSON.stringify` in `try/catch` (circular args → skip frame, L224); `ws.send` in `try/catch` → treated as disconnect + reconnect scheduled (L230–236) | +| `connect()` (L288) | constructor in `try/catch` (invalid URL → backoff, L293); all four socket callbacks (`onopen` L297, `onmessage` L312, `onclose` L325) wrap their bodies in `try/catch`; stale-socket identity guard `if (ws !== socket) return` so an old socket's late callbacks can never clobber the new socket's state | +| `onMessage` (L382) | `JSON.parse` in `try/catch` (L385); non-object frames, wrong `v`, unknown types ignored | +| `deliverPrompt` (L400) | `pi.sendUserMessage(..., {deliverAs:"steer"})` in `try/catch` (L406); empty or non-string message ignored; foreign `sessionId` ignored | +| `doAbort` (L411) | `pi.abort` (if present) else `lastCtx.abort()` in `try/catch` | +| `buildSnapshot` (L421) | all `sessionManager` reads in `try/catch` with fallbacks (L433: ctx.cwd / Date.now) | +| `session_start` (L459) | `getSessionId()` in `try/catch`; no id ⇒ mirroring disabled for that session, no crash | +| `session_shutdown` (L490) | `clearTimeout` and bye-send/close both in `try/catch` (L494, L500) | +| `onWelcome` (L349) | validates `lastSeq` (`Number.isFinite`, else 0, L355); foreign-session welcome ignored (L350) | + +That is **14 explicit `try/catch` blocks** plus the universal `sub()` wrapper. +Static pass over every function: no code path can throw to pi; nothing +`await`s — all I/O is fire-and-forget async (`fs.appendFile`, undici WS), so +nothing can block or hang the event loop either. The only synchronous work is +JSON serialization of outbound frames (bounded by message size). + +### 2. Zero npm dependencies + +Imports: `node:fs`, `node:os`, `node:path` (L29–31), type-only import of +`@earendil-works/pi-coding-agent` (L33, erased at runtime), and the Node ≥ 22 +global `WebSocket` (undici). Verified by `grep import` and by the e2e run +against real pi. `typeof WebSocket === "undefined"` guard at L156 makes the +extension inert (not broken) on older Node. + +### 3. Bounded queue + drop-oldest + buffer_overflow + +- Send queue cap 1000: `enqueue()` L206–215 (`while` shift-oldest, every drop + counted and logged). +- Replay buffer cap 10000 (persisted-kind events only; `message_update` is + excluded per protocol): `emit()` L241–252. +- Drops are never silent: `droppedEvents > 0` ⇒ `emit("buffer_overflow", + {dropped})` immediately after the next `welcome` is processed (L373–376). +- Persisted-kind events dropped from the send queue are still in the replay + buffer and get recovered by replay; only `message_update` deltas can be + lost. `onWelcome` additionally filters persisted frames out of the stale + queue before merging replay (L367–368) so replay + queue can never + double-send an event. +- Smoke scenario 4 proves it: 10 050 persisted events fired while + disconnected → reconnect → `buffer_overflow {dropped: n>0}` observed on the + wire, replay delivered, no duplicate seq. + +### 4. WS send failures never propagate + +`flush()` L230–236: `ws.send` in `try/catch` → log, `handleDisconnect()`, +`scheduleReconnect()`. Same treatment in `session_shutdown` bye-send (L500). +Proven by smoke scenario 4 (server destroys sockets under load) and e2e. + +### 5. Reconnect: backoff + jitter, cap 30s, forever, never throws + +`scheduleReconnect()` L274–286: `min(1000·2^attempt, 30000) + jitter[0,1000)`; +attempt counter reset only on a successful `welcome` (L356). Timer is +`unref()`'d (L283) so it can never keep pi's event loop alive. Dead-host run +above shows pi fully functional with the plugin retrying silently +(`~/.pi/lvmh-agent.log`: attempts #0,#1,#2 at ~1s/~2s/~3s spacing). + +### 6. Log file only, console clean + +Single sink `~/.pi/lvmh-agent.log` via `log()` L75–82 (async append, truncated +to 500 chars/line, swallows all errors). `grep -n "console\." lvmh-agent.ts` +returns nothing; e2e asserts the plugin adds nothing to pi's stdout +("e2e plugin keeps stdout clean"). + +### 7. session_shutdown / unload: close, flush nothing, exit fast + +`session_shutdown` handler L490–519: synchronous, sends best-effort +`bye {reason:"shutdown"}` only if the socket is OPEN, `close()`s it, clears +the timer, sets `stopped` (blocks `emit`, `connect`, `scheduleReconnect`, and +`flush`). No flush, no awaits. Smoke scenario 6: connection count → 0, no +reconnect for the following 2.6s. E2E: `bye` frame observed on the wire, no +reconnect after. + +### 8. Daemon down at startup ⇒ pi starts and runs perfectly + +Proven twice: (a) smoke scenario 5 — factory loads against a closed port, +handlers never throw, event loop stays responsive (100ms timer completes +<500ms); (b) real pi run above — prompt answered, exit 0. + +### 9. Protocol conformance details + +- Envelope `v/sessionId/seq/ts/type` on every frame; `hello`/`bye`/`welcome` + use `seq: 0` (L339, L505). +- Per-session monotonic seq survives reconnects (module-level `seqCounters`, + L70); after a process restart, the counter is pushed past `welcome.lastSeq` + so seqs are never reused against a warm daemon (L360–364). +- Message mapping (`mapMessage` L113): `role`, deterministic `id` + (`role-timestamp`), concatenated `text`, `thinking` or null, + `toolCalls[{id,name,argsJson}]`, `toolCallId` for toolResults — verified by + smoke "2 message_end mapping" and e2e user/assistant message_end. +- `message_update` carries only the text delta (diffed against the previously + sent prefix, L547–556); non-prefix text (model retry) resends full text. +- Truncation to 2000 chars for `partial` and `resultPreview` (L576, L584). +- `agent_end.usage` sums `input`/`output`/`cost.total` across the run's + assistant messages; omitted entirely if none had usage (L595–613). +- Multi-session: extension instance is re-created per session bind; on + `session_start` with a new session id the old socket/timers/queues are reset + (L475–488) and seq counters stay per-session (module map). + +## Known non-issues (documented, not plugin bugs) + +- **pi print-mode exit hang**: `pi -p` with *any* extension loaded + (reproduced with an empty noop extension, no lvmh involvement) sometimes + does not exit after answering — a pi quirk, independent of this plugin. The + plugin releases all of its resources on `session_shutdown` (proven by `bye`, + socket close, and no reconnect in e2e). The e2e harness treats this as + informational. +- Sync flush of a large replay burst (~10k frames ≈ 1MB) takes single-digit + milliseconds of JSON serialization; it cannot hang the loop but is the + largest synchronous chunk the plugin can produce. + +## Files + +| file | role | +| --- | --- | +| `lvmh-agent.ts` | the extension (only file pi loads) | +| `README.md` | install / env / behavior | +| `tsconfig.json` | strict typecheck against real pi types (`paths`, `typeRoots`) | +| `pi-types.d.ts` | commented minimal fallback if pi's dist types move | +| `mini-daemon.ts` | dev-only hand-rolled WS daemon for tests | +| `smoke.ts` | dev-only offline test (25 checks) | +| `e2e.ts` | dev-only real-pi test (13 checks) | diff --git a/plugin/smoke.ts b/plugin/smoke.ts new file mode 100644 index 0000000..ac836a8 --- /dev/null +++ b/plugin/smoke.ts @@ -0,0 +1,328 @@ +/** + * Dev-only smoke test for lvmh-agent.ts. Not part of the shipped extension. + * + * Run: node plugin/smoke.ts (Node >= 23.6; older: --experimental-strip-types) + * + * Scenarios (full matrix in selftest.md): + * 1. env unset -> factory inert, nothing registered + * 2. live daemon -> bearer header, hello snapshot, mirrored events, + * delta-only updates, message mapping, 2000-char + * truncation, usage mapping, monotonic seq + * 3. prompt / abort -> pi.sendUserMessage(steer) / pi.abort; foreign + * session prompts ignored + * 4. drop + reconnect -> replay of persisted events after welcome.lastSeq, + * seq continuity, buffer_overflow notice after + * queue/replay overflow + * 5. dead host -> loads, retries, never throws, event loop responsive + * 6. session_shutdown -> socket closed, no reconnect afterwards + */ + +import { check as rawCheck, sleep, waitFor, startMiniDaemon } from "./mini-daemon.ts"; + +const SESSION_ID = "sess-1"; + +let failures = 0; + +function check(name: string, ok: boolean, detail = ""): void { + rawCheck(name, ok, detail); + if (!ok) failures++; +} + +interface FakePi { + handlers: Map unknown>; + sentMessages: Array<{ message: string; options: unknown }>; + aborted: number; + on(event: string, handler: (event: unknown, ctx: unknown) => unknown): void; + sendUserMessage(message: string, options?: unknown): void; + abort(): void; +} + +function makeFakePi(): FakePi { + return { + handlers: new Map(), + sentMessages: [], + aborted: 0, + on(event, handler) { + this.handlers.set(event, handler); + }, + sendUserMessage(message, options) { + this.sentMessages.push({ message, options }); + }, + abort() { + this.aborted++; + }, + }; +} + +function makeFakeCtx(): unknown { + return { + cwd: "/work/repo", + model: { id: "glm-5.3", provider: "zai-renaud" }, + sessionManager: { + getSessionId: () => SESSION_ID, + getSessionName: () => undefined, + getCwd: () => "/work/repo", + getHeader: () => ({ timestamp: "2024-12-03T14:00:00.000Z", id: SESSION_ID }), + }, + }; +} + +async function loadExtension(): Promise<(pi: unknown) => void> { + const mod = (await import("./lvmh-agent.ts")) as { default: (pi: unknown) => void }; + return mod.default; +} + +async function main(): Promise { + // --- Scenario 1: env unset -> inert -------------------------------- + delete process.env.LVMH_URL; + delete process.env.LVMH_TOKEN; + delete process.env.LVMH_AGENT; + delete process.env.LVMH_REPO; + const factory = await loadExtension(); + const inertPi = makeFakePi(); + inertPi.on = ((event: string) => { + throw new Error(`inert extension registered handler ${event}`); + }) as FakePi["on"]; + factory(inertPi); // must not touch pi at all + check("1 inert: no side effects on pi", true); + + // --- Scenario 2: live daemon --------------------------------------- + let welcomeLastSeq = 0; + const daemon = await startMiniDaemon(() => welcomeLastSeq); + process.env.LVMH_URL = daemon.url; + process.env.LVMH_TOKEN = "smoke-token"; + + const pi = makeFakePi(); + factory(pi); + const handlerNames = [ + "session_start", + "session_shutdown", + "session_info_changed", + "model_select", + "message_start", + "message_update", + "message_end", + "tool_execution_start", + "tool_execution_update", + "tool_execution_end", + "agent_start", + "agent_end", + "agent_settled", + ]; + check("2 all 13 handlers registered", handlerNames.every((n) => pi.handlers.has(n))); + + let lastReturn: unknown = "sentinel"; + const fire = (name: string, event: unknown): void => { + const h = pi.handlers.get(name); + if (h === undefined) throw new Error(`missing handler ${name}`); + lastReturn = h(event, makeFakeCtx()); + if (lastReturn instanceof Promise) lastReturn.catch(() => undefined); + }; + + fire("session_start", { reason: "startup" }); + const helloSeen = await waitFor(() => daemon.frames.some((f) => f.type === "hello"), 5000); + check("2 connects and sends hello", helloSeen); + check("2 bearer auth on upgrade", daemon.authHeaders.at(-1) === "Bearer smoke-token", daemon.authHeaders.at(-1)); + + const hello = daemon.frames.find((f) => f.type === "hello"); + const hs = (hello?.session ?? {}) as Record; + check( + "2 hello snapshot", + hs.id === SESSION_ID && + hs.name === null && + hs.cwd === "/work/repo" && + hs.model === "glm-5.3" && + hs.provider === "zai-renaud" && + hs.agent === false && + hs.repo === null && + typeof hs.startedAt === "number", + JSON.stringify(hs), + ); + + fire("message_start", { message: { role: "assistant", content: [], timestamp: 1000 } }); + fire("message_update", { message: { role: "assistant", content: [{ type: "text", text: "Hel" }], timestamp: 1000 } }); + fire("message_update", { message: { role: "assistant", content: [{ type: "text", text: "Hello world" }], timestamp: 1000 } }); + fire("message_end", { + message: { + role: "assistant", + content: [ + { type: "thinking", thinking: "hmm" }, + { type: "text", text: "Hello world" }, + { type: "toolCall", id: "tc1", name: "bash", arguments: { command: "ls" } }, + ], + timestamp: 1000, + }, + }); + await waitFor(() => daemon.frames.some((f) => f.type === "message_end"), 2000); + + const n0 = daemon.frames.findIndex((f) => f.type === "hello"); + const mirrored = daemon.frames.slice(n0 + 1).map((f) => f.type); + check( + "2 mirror order", + JSON.stringify(mirrored) === JSON.stringify(["message_start", "message_update", "message_update", "message_end"]), + JSON.stringify(mirrored), + ); + const deltas = daemon.frames.filter((f) => f.type === "message_update").map((f) => f.delta); + check("2 delta-only updates", JSON.stringify(deltas) === JSON.stringify(["Hel", "lo world"]), JSON.stringify(deltas)); + + const msg = daemon.frames.find((f) => f.type === "message_end")?.message as Record; + const tc = (msg?.toolCalls as Array> | undefined)?.[0]; + check( + "2 message_end mapping", + msg?.role === "assistant" && + msg?.text === "Hello world" && + msg?.thinking === "hmm" && + typeof msg?.id === "string" && + tc?.id === "tc1" && + tc?.name === "bash" && + tc?.argsJson === '{"command":"ls"}' && + msg?.toolCallId === null, + JSON.stringify(msg), + ); + + const big = "x".repeat(3000); + fire("agent_start", {}); + fire("tool_execution_start", { toolCallId: "tc1", toolName: "bash", args: { command: "ls" } }); + fire("tool_execution_update", { toolCallId: "tc1", toolName: "bash", partialResult: { content: [{ type: "text", text: big }] } }); + fire("tool_execution_end", { toolCallId: "tc1", toolName: "bash", isError: true, result: { content: [{ type: "text", text: big }] } }); + fire("agent_end", { messages: [{ role: "assistant", usage: { input: 10, output: 5, cost: { total: 0.25 } } }] }); + fire("agent_settled", {}); + await waitFor(() => daemon.frames.some((f) => f.type === "agent_settled"), 2000); + + const partial = daemon.frames.find((f) => f.type === "tool_execution_update"); + const end = daemon.frames.find((f) => f.type === "tool_execution_end"); + const partialLen = (partial?.partial as string | undefined)?.length; + const previewLen = (end?.resultPreview as string | undefined)?.length; + check("2 partial truncated to 2000", partialLen === 2000, String(partialLen)); + check("2 resultPreview truncated + isError", previewLen === 2000 && end?.isError === true); + const usage = daemon.frames.find((f) => f.type === "agent_end")?.usage as Record; + check( + "2 agent_end usage", + usage?.inputTokens === 10 && usage?.outputTokens === 5 && usage?.totalCost === 0.25, + JSON.stringify(usage), + ); + + const seqs = daemon.frames.filter((f) => f.seq > 0).map((f) => f.seq); + check( + "2 seq strictly monotonic", + seqs.length > 0 && seqs.every((s, i) => i === 0 || s > seqs[i - 1]), + JSON.stringify(seqs), + ); + check("2 handler return values are undefined (never a promise)", lastReturn === undefined); + + // --- Scenario 3: prompt / abort ------------------------------------ + daemon.pushAll( + JSON.stringify({ v: 1, type: "prompt", sessionId: "other-session", seq: 0, ts: Date.now(), promptId: "p0", message: "wrong session" }), + ); + daemon.pushAll( + JSON.stringify({ v: 1, type: "prompt", sessionId: SESSION_ID, seq: 0, ts: Date.now(), promptId: "p1", message: "run the tests" }), + ); + await waitFor(() => pi.sentMessages.length > 0, 2000); + check( + "3 prompt delivered as steer, foreign session ignored", + pi.sentMessages.length === 1 && + pi.sentMessages[0].message === "run the tests" && + JSON.stringify(pi.sentMessages[0].options) === '{"deliverAs":"steer"}', + JSON.stringify(pi.sentMessages), + ); + daemon.pushAll(JSON.stringify({ v: 1, type: "abort", sessionId: SESSION_ID, seq: 0, ts: Date.now() })); + await waitFor(() => pi.aborted > 0, 2000); + check("3 abort calls pi.abort", pi.aborted === 1); + + // --- Scenario 4: drop + reconnect, replay, overflow ---------------- + const seqBeforeDrop = Math.max(...daemon.frames.map((f) => f.seq)); + welcomeLastSeq = seqBeforeDrop; // daemon has everything up to the drop + daemon.dropConnections(); + await sleep(150); // let the client notice + fire("message_end", { message: { role: "user", content: "offline-1", timestamp: 2000 } }); + fire("message_end", { message: { role: "user", content: "offline-2", timestamp: 2001 } }); + const reconnected = await waitFor( + () => daemon.frames.filter((f) => f.type === "hello").length >= 2, + 10000, + ); + check("4 reconnects after drop", reconnected); + const replayedTexts = daemon.frames + .filter((f) => f.type === "message_end" && f.seq > seqBeforeDrop) + .map((f) => (f.message as { text?: string }).text); + check( + "4 buffered events replayed after welcome.lastSeq", + replayedTexts.includes("offline-1") && replayedTexts.includes("offline-2"), + JSON.stringify(replayedTexts), + ); + const allSeqs = daemon.frames.filter((f) => f.seq > 0).map((f) => f.seq); + const dupCount = allSeqs.length - new Set(allSeqs).size; + check("4 no duplicate seq delivery", dupCount === 0, `duplicates: ${dupCount}`); + const seqsAfter = daemon.frames.filter((f) => f.seq > seqBeforeDrop).map((f) => f.seq); + check( + "4 seq continues monotonically across reconnect", + seqsAfter.length >= 2 && seqsAfter.every((s, i) => i === 0 || s > seqsAfter[i - 1]), + JSON.stringify(seqsAfter), + ); + + // overflow both bounded buffers, then reconnect once more + const seqBeforeOverflow = Math.max(...daemon.frames.map((f) => f.seq)); + welcomeLastSeq = seqBeforeOverflow; + daemon.dropConnections(); + await sleep(150); + for (let i = 0; i < 10050; i++) { + fire("agent_settled", {}); // persisted kind: fills replay buffer + send queue + } + const reconnected2 = await waitFor( + () => daemon.frames.filter((f) => f.type === "hello").length >= 3, + 15000, + ); + check("4 reconnects after overflow window", reconnected2); + const overflowNotice = await waitFor( + () => daemon.frames.some((f) => f.type === "buffer_overflow"), + 5000, + ); + const overflow = daemon.frames.find((f) => f.type === "buffer_overflow"); + check( + "4 buffer_overflow notice after drops", + overflowNotice && typeof (overflow?.dropped as number) === "number" && (overflow?.dropped as number) > 0, + JSON.stringify(overflow ?? null), + ); + + // --- Scenario 6: session_shutdown ---------------------------------- + const helloCountAtShutdown = daemon.frames.filter((f) => f.type === "hello").length; + fire("session_shutdown", { reason: "quit" }); + const closed = await waitFor(() => daemon.connections() === 0, 3000); + check("6 shutdown closes connection", closed); + await sleep(2600); // longer than one backoff attempt (~1-2s) + const helloCountAfter = daemon.frames.filter((f) => f.type === "hello").length; + check("6 no reconnect after shutdown", helloCountAfter === helloCountAtShutdown); + + daemon.close(); + + // --- Scenario 5: dead host ----------------------------------------- + const deadDaemon = await startMiniDaemon(() => 0); + const deadUrl = deadDaemon.url; + deadDaemon.close(); // port now closed + await sleep(100); + process.env.LVMH_URL = deadUrl; + const pi2 = makeFakePi(); + factory(pi2); + pi2.handlers.get("session_start")?.({ reason: "startup" }, makeFakeCtx()); + await sleep(400); + const t0 = Date.now(); + await sleep(100); + const loopResponsive = Date.now() - t0 < 500; + check("5 dead host: pi loads, event loop responsive", loopResponsive); + let threw = false; + try { + pi2.handlers.get("message_end")?.({ message: { role: "user", content: "hi", timestamp: 1 } }, makeFakeCtx()); + } catch { + threw = true; + } + check("5 dead host: handlers never throw", !threw); + + delete process.env.LVMH_URL; + delete process.env.LVMH_TOKEN; + console.log(failures === 0 ? "\nALL CHECKS PASSED" : `\n${failures} CHECK(S) FAILED`); + process.exit(failures === 0 ? 0 : 1); +} + +main().catch((err: unknown) => { + console.error("smoke harness crashed:", err); + process.exit(1); +}); diff --git a/plugin/tsconfig.json b/plugin/tsconfig.json new file mode 100644 index 0000000..ba047a8 --- /dev/null +++ b/plugin/tsconfig.json @@ -0,0 +1,22 @@ +{ + "compilerOptions": { + "target": "es2022", + "module": "esnext", + "moduleResolution": "bundler", + "lib": ["es2023"], + "types": ["node"], + "typeRoots": [ + "/home/raph/.node/lib/node_modules/@earendil-works/pi-coding-agent/node_modules/@types" + ], + "allowImportingTsExtensions": true, + "strict": true, + "noEmit": true, + "skipLibCheck": true, + "paths": { + "@earendil-works/pi-coding-agent": [ + "/home/raph/.node/lib/node_modules/@earendil-works/pi-coding-agent/dist/index.d.ts" + ] + } + }, + "include": ["lvmh-agent.ts", "smoke.ts", "mini-daemon.ts", "e2e.ts", "pi-types.d.ts"] +}