From 5aac2430a25a5da574dad05658054ecf35b0cc15 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Fri, 31 Jul 2026 00:24:58 +0200 Subject: [PATCH 1/2] Add revision-guarded PTY send --- CHANGELOG.md | 11 + README.md | 8 +- docs/client.md | 37 +++ src/cli.ts | 2 + src/client-api.ts | 10 +- src/client.ts | 2 + src/guarded-send-client.ts | 72 ++++++ src/guarded-send.ts | 56 ++++ src/protocol.ts | 8 + src/server.ts | 100 +++++++- tests/guarded-send-integration.test.ts | 341 +++++++++++++++++++++++++ tests/guarded-send.test.ts | 50 ++++ tests/protocol.test.ts | 20 ++ tests/stats-cli.test.ts | 4 + 14 files changed, 705 insertions(+), 16 deletions(-) create mode 100644 src/guarded-send-client.ts create mode 100644 src/guarded-send.ts create mode 100644 tests/guarded-send-integration.test.ts create mode 100644 tests/guarded-send.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 986ae5d..40f661d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,17 @@ ## Unreleased +### Atomic generation/revision-guarded input + +- Expose the live daemon `generation` and monotonic `ioRevision` through + STATUS, `queryStats()`, and `pty stats`. Accepted input, child output, + activity-state changes, and actual PTY resizes advance the revision; passive + attach, peek, and status requests do not. +- Add `compareAndSend()` for one bounded, harness-neutral conditional write. + Exact generation/revision validation and the PTY write occur in one daemon + event-loop turn. Mismatch, replay, malformed or oversized input, exited + sessions, and read-only clients reject with zero guarded bytes. + ### Harness-neutral live activity status - Add a single generation-bound activity lease with ordered `unknown`, diff --git a/README.md b/README.md index 37c11d8..9c5e9c9 100644 --- a/README.md +++ b/README.md @@ -373,7 +373,7 @@ Like `git`, `pty` supports extensions: if you run `pty foo` and there's a `pty-f import { spawnDaemon, listSessions, getSession, SessionConnection, sendData, peekScreen, queryStats, - connectActivityPublisher, + connectActivityPublisher, compareAndSend, EventFollower, readRecentEvents, extractFilterTags, matchesAllTags, } from "@compoundingtech/pty/client"; @@ -408,6 +408,12 @@ modes never imply idleness. Harness adapters can hold the single live lease with `connectActivityPublisher()` and publish ordered transitions. The state resets to `unknown` when that connection or daemon generation ends. +For race-free delivery, pass a fresh `queryStats()` generation and +`ioRevision` to `compareAndSend()`. The daemon writes once only if both still +match; input, child output, activity changes, resize, restart, malformed +commands, and replay reject without writing guarded bytes. PTY does not assign +meaning to the supplied bytes or inspect provider composer state. + ### Connecting to a session `SessionConnection` provides a bidirectional, event-driven connection without taking over stdin/stdout — ideal for GUI apps, multiplexers, or web interfaces: diff --git a/docs/client.md b/docs/client.md index c40d649..1730bb4 100644 --- a/docs/client.md +++ b/docs/client.md @@ -238,6 +238,8 @@ Query live metrics from a running session. ```typescript interface StatsResult { name: string; + generation: string; + ioRevision: number; terminal: { cols: number; rows: number; cursorX: number; cursorY: number; @@ -281,6 +283,40 @@ as `unknown` and returns to `unknown` when the publisher disconnects or the daemon generation changes. Terminal modes such as `alternateScreen` are diagnostics only and do not imply activity or idleness. +### `compareAndSend(name, options): Promise` + +Conditionally write one non-empty string only while the daemon generation and +delivery-relevant I/O revision remain exact. + +```typescript +import { + compareAndSend, + queryStats, +} from "@compoundingtech/pty/client"; + +const observed = await queryStats("myserver"); +const result = await compareAndSend("myserver", { + generation: observed.generation, + ioRevision: observed.ioRevision, + data: "continue\r", +}); + +if (!result.ok) { + // Re-check external authority and session state; no guarded bytes were sent. +} +``` + +`ioRevision` advances on accepted input, child output, actual PTY size changes, +and accepted or released activity-lease state. Guard comparison and the single +PTY write run in one daemon event-loop turn. A mismatch, replay, malformed or +oversized command, exited session, or read-only connection rejects with zero +guarded bytes. Attach, peek, and status operations that cause no input or +resize leave the revision unchanged. + +The guard does not interpret the data. Provider-specific idle authority, +composer checks, key meanings, and delivery ownership remain the caller's +responsibility. + ### `connectActivityPublisher(name, options?): Promise` Claim the session's single live activity lease and publish harness-neutral @@ -424,6 +460,7 @@ const MessageType = { PEEK: 6, // Read-only peek request STATUS: 7, // Stats query/response ACTIVITY: 8, // Generic activity lease commands/responses + GUARDED_DATA: 9, // Generation/revision-conditional input }; ``` diff --git a/src/cli.ts b/src/cli.ts index d8cc617..0f93bcd 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -2379,6 +2379,8 @@ function printStats(stats: StatsResult, meta: SessionInfo["metadata"]): void { console.log(`Session: ${stats.name}`); console.log(` Command: ${cmd}`); console.log(` CWD: ${cwd}`); + console.log(` Generation: ${stats.generation}`); + console.log(` I/O rev: ${stats.ioRevision}`); console.log(` Uptime: ${formatUptime(stats.uptimeSeconds)}`); const pidSuffix = stats.process?.pid ? ` (pid ${stats.process.pid})` : ""; console.log(` Process: ${stats.process.alive ? "running" : `exited (code ${stats.process.exitCode})`}${pidSuffix}`); diff --git a/src/client-api.ts b/src/client-api.ts index 4ce01aa..4878adb 100644 --- a/src/client-api.ts +++ b/src/client-api.ts @@ -41,6 +41,14 @@ export { ACTIVITY_STATES, type ActivityState, type ActivityStatus, } from "./activity.ts"; +export { + compareAndSend, + type CompareAndSendOptions, +} from "./guarded-send-client.ts"; +export { + MAX_GUARDED_DATA_BYTES, + type GuardedSendCommand, type GuardedSendResponse, +} from "./guarded-send.ts"; // Events export { @@ -72,6 +80,6 @@ export { resolveKey, parseSeqValue } from "./keys.ts"; // Protocol (advanced) export { - PacketReader, MessageType, encodeActivity, + PacketReader, MessageType, encodeActivity, encodeGuardedData, type Packet, } from "./protocol.ts"; diff --git a/src/client.ts b/src/client.ts index 3682ea4..3421436 100644 --- a/src/client.ts +++ b/src/client.ts @@ -293,6 +293,8 @@ export interface ProcessResources { export interface StatsResult { name: string; + generation: string; + ioRevision: number; terminal: { cols: number; rows: number; diff --git a/src/guarded-send-client.ts b/src/guarded-send-client.ts new file mode 100644 index 0000000..7a402ca --- /dev/null +++ b/src/guarded-send-client.ts @@ -0,0 +1,72 @@ +import * as net from "node:net"; +import { + type GuardedSendCommand, + type GuardedSendResponse, +} from "./guarded-send.ts"; +import { + MessageType, + PacketReader, + encodeGuardedData, +} from "./protocol.ts"; +import { getSocketPath } from "./sessions.ts"; + +export interface CompareAndSendOptions extends GuardedSendCommand { + timeoutMs?: number; +} + +export function compareAndSend( + name: string, + options: CompareAndSendOptions, +): Promise { + return new Promise((resolve, reject) => { + const socket = net.createConnection(getSocketPath(name)); + const reader = new PacketReader(); + const timeoutMs = options.timeoutMs ?? 2000; + const timer = setTimeout(() => { + socket.destroy(); + reject(new Error(`Timeout sending guarded data to "${name}"`)); + }, timeoutMs); + + const finish = ( + callback: () => void, + ): void => { + clearTimeout(timer); + socket.destroy(); + callback(); + }; + + socket.once("connect", () => { + socket.write(encodeGuardedData({ + generation: options.generation, + ioRevision: options.ioRevision, + data: options.data, + })); + }); + socket.on("data", (data) => { + let packets; + try { + packets = reader.feed(Buffer.isBuffer(data) ? data : Buffer.from(data)); + } catch (error) { + finish(() => reject( + error instanceof Error ? error : new Error(String(error)), + )); + return; + } + for (const packet of packets) { + if (packet.type !== MessageType.GUARDED_DATA) continue; + try { + const response = JSON.parse( + packet.payload.toString("utf8"), + ) as GuardedSendResponse; + finish(() => resolve(response)); + } catch { + finish(() => reject(new Error("invalid guarded send response"))); + } + return; + } + }); + socket.once("error", (error) => { + finish(() => reject(error)); + }); + }); +} diff --git a/src/guarded-send.ts b/src/guarded-send.ts new file mode 100644 index 0000000..b7989dc --- /dev/null +++ b/src/guarded-send.ts @@ -0,0 +1,56 @@ +export const MAX_GUARDED_DATA_BYTES = 64 * 1024; + +export interface GuardedSendCommand { + generation: string; + ioRevision: number; + data: string; +} + +export interface GuardedSendResponse { + ok: boolean; + generation: string; + ioRevision: number; + error?: string; +} + +// JSON control-character escaping can expand one input byte to six wire bytes +// (for example NUL becomes "\\u0000"). Keep the wire packet bounded while +// accepting every payload whose decoded UTF-8 data is within the public cap. +const MAX_GUARDED_COMMAND_BYTES = MAX_GUARDED_DATA_BYTES * 6 + 512; +const MAX_GENERATION_LENGTH = 128; +const GUARDED_SEND_KEYS = new Set(["generation", "ioRevision", "data"]); + +export function parseGuardedSendCommand( + payload: Buffer, +): GuardedSendCommand | null { + if (payload.length === 0 || payload.length > MAX_GUARDED_COMMAND_BYTES) { + return null; + } + try { + const parsed: unknown = JSON.parse(payload.toString("utf8")); + if (typeof parsed !== "object" || parsed === null || Array.isArray(parsed)) { + return null; + } + const value = parsed as Record; + if ( + Object.keys(value).some((key) => !GUARDED_SEND_KEYS.has(key)) || + typeof value.generation !== "string" || + value.generation.length === 0 || + value.generation.length > MAX_GENERATION_LENGTH || + !Number.isSafeInteger(value.ioRevision) || + Number(value.ioRevision) < 0 || + typeof value.data !== "string" || + value.data.length === 0 || + Buffer.byteLength(value.data, "utf8") > MAX_GUARDED_DATA_BYTES + ) { + return null; + } + return { + generation: value.generation, + ioRevision: Number(value.ioRevision), + data: value.data, + }; + } catch { + return null; + } +} diff --git a/src/protocol.ts b/src/protocol.ts index 923e8b1..c12d1b2 100644 --- a/src/protocol.ts +++ b/src/protocol.ts @@ -10,6 +10,7 @@ export const MessageType = { PEEK: 6, // Client → Server: read-only attach (no input, no resize) STATUS: 7, // Client → Server: request stats; Server → Client: JSON stats response ACTIVITY: 8, // Bidirectional: generic activity lease commands/responses + GUARDED_DATA: 9, // Bidirectional: generation/revision-conditional input } as const; export type MessageType = (typeof MessageType)[keyof typeof MessageType]; @@ -99,6 +100,13 @@ export function encodeActivity(value: unknown): Buffer { return encodePacket(MessageType.ACTIVITY, Buffer.from(JSON.stringify(value))); } +export function encodeGuardedData(value: unknown): Buffer { + return encodePacket( + MessageType.GUARDED_DATA, + Buffer.from(JSON.stringify(value)), + ); +} + export function decodeSize(payload: Buffer): { rows: number; cols: number } { if (payload.length < 4) { return { rows: 24, cols: 80 }; diff --git a/src/server.ts b/src/server.ts index 4349842..7151eb9 100644 --- a/src/server.ts +++ b/src/server.ts @@ -17,9 +17,14 @@ import { encodeScreen, encodeStatusResponse, encodeActivity, + encodeGuardedData, decodeSize, } from "./protocol.ts"; import { ActivityLease, parseActivityCommand } from "./activity.ts"; +import { + parseGuardedSendCommand, + type GuardedSendResponse, +} from "./guarded-send.ts"; import { getSocketPath, getPidPath, @@ -254,6 +259,7 @@ export class PtyServer { private lastResizeTime = 0; private eventWriter: EventWriter; private generation: string; + private ioRevision = 0; private lastTitle = ""; private activity: ActivityLease; readonly ready: Promise; @@ -342,7 +348,7 @@ export class PtyServer { { final: "c" }, (params) => { if (params.length === 0 || params[0] === 0) { - this.ptyProcess.write("\x1b[?62;22c"); + this.writeToPty("\x1b[?62;22c"); } return false; } @@ -407,7 +413,7 @@ export class PtyServer { // Return true to consume the sequence so it doesn't leak to clients. this.terminal.parser.registerOscHandler(10, (data: string) => { if (data === "?") { - this.ptyProcess.write("\x1b]10;rgb:c0c0/c0c0/c0c0\x1b\\"); + this.writeToPty("\x1b]10;rgb:c0c0/c0c0/c0c0\x1b\\"); return true; // consume — don't pass to client } return false; @@ -415,7 +421,7 @@ export class PtyServer { // OSC 11: background color query (less, vim) this.terminal.parser.registerOscHandler(11, (data: string) => { if (data === "?") { - this.ptyProcess.write("\x1b]11;rgb:0000/0000/0000\x1b\\"); + this.writeToPty("\x1b]11;rgb:0000/0000/0000\x1b\\"); return true; } return false; @@ -425,7 +431,7 @@ export class PtyServer { if (data.includes("?")) { const idx = parseInt(data, 10); if (!isNaN(idx)) { - this.ptyProcess.write(`\x1b]4;${idx};rgb:0000/0000/0000\x1b\\`); + this.writeToPty(`\x1b]4;${idx};rgb:0000/0000/0000\x1b\\`); } return true; } @@ -436,7 +442,7 @@ export class PtyServer { { prefix: ">", final: "c" }, (_params) => { // Respond as xterm version 382 - this.ptyProcess.write("\x1b[>0;382;0c"); + this.writeToPty("\x1b[>0;382;0c"); return false; } ); @@ -446,7 +452,7 @@ export class PtyServer { (params) => { if (params.length === 1 && params[0] === 6) { const buf = this.terminal.buffer.active; - this.ptyProcess.write(`\x1b[${buf.cursorY + 1};${buf.cursorX + 1}R`); + this.writeToPty(`\x1b[${buf.cursorY + 1};${buf.cursorX + 1}R`); } return false; } @@ -455,7 +461,7 @@ export class PtyServer { this.terminal.parser.registerCsiHandler( { prefix: ">", final: "q" }, (_params) => { - this.ptyProcess.write("\x1bP>|pty(0.8)\x1b\\"); + this.writeToPty("\x1bP>|pty(0.8)\x1b\\"); return false; } ); @@ -505,6 +511,7 @@ export class PtyServer { // handlers above and must NOT be forwarded to clients — otherwise the // client's terminal responds and its response appears as garbage input. this.ptyProcess.onData((data: string) => { + this.bumpIoRevision(); this.terminal.write(data); const cleaned = stripTerminalQueries(data); if (cleaned.length > 0) { @@ -514,6 +521,7 @@ export class PtyServer { this.ptyProcess.onExit(({ exitCode, signal }) => { this.exited = true; + this.bumpIoRevision(); // A signal death (e.g. an OS OOM SIGKILL) arrives from node-pty with a // nonzero `signal` and often exitCode 0 — if we recorded only the raw // exitCode, a killed process would look like a clean finish and any @@ -711,7 +719,7 @@ export class PtyServer { case MessageType.DATA: { if (!this.exited && !client.readonly) { - this.ptyProcess.write(packet.payload.toString()); + this.writeToPty(packet.payload.toString()); } break; } @@ -740,6 +748,7 @@ export class PtyServer { case MessageType.ACTIVITY: { const command = parseActivityCommand(packet.payload); + const before = this.activity.snapshot(); if (command === null) this.activity.release(socket); const response = command === null ? { @@ -748,21 +757,48 @@ export class PtyServer { activity: this.activity.snapshot(), } : this.activity.apply(socket, command); + if (JSON.stringify(before) !== JSON.stringify(response.activity)) { + this.bumpIoRevision(); + } socket.write(encodeActivity(response)); break; } + + case MessageType.GUARDED_DATA: { + const command = parseGuardedSendCommand(packet.payload); + let response: GuardedSendResponse; + if (command === null) { + response = this.guardedFailure("invalid guarded send"); + } else if (client.readonly || this.exited) { + response = this.guardedFailure("session is not writable"); + } else if (command.generation !== this.generation) { + response = this.guardedFailure("daemon generation mismatch"); + } else if (command.ioRevision !== this.ioRevision) { + response = this.guardedFailure("I/O revision mismatch"); + } else if (!this.writeToPty(command.data)) { + response = this.guardedFailure("PTY write failed"); + } else { + response = { + ok: true, + generation: this.generation, + ioRevision: this.ioRevision, + }; + } + socket.write(encodeGuardedData(response)); + break; + } } } }); socket.on("close", () => { - this.activity.release(socket); + this.releaseActivity(socket); this.clients.delete(socket); this.negotiateSize(); }); socket.on("error", () => { - this.activity.release(socket); + this.releaseActivity(socket); this.clients.delete(socket); this.negotiateSize(); }); @@ -808,6 +844,8 @@ export class PtyServer { return { name: this.name, + generation: this.generation, + ioRevision: this.ioRevision, terminal: { cols: this.terminal.cols, rows: this.terminal.rows, @@ -859,8 +897,7 @@ export class PtyServer { if (rows > 0 && cols > 0) { if (rows !== this.terminal.rows || cols !== this.terminal.cols) { - this.ptyProcess.resize(cols, rows); - this.terminal.resize(cols, rows); + this.resizePty(cols, rows); this.lastResizeTime = Date.now(); return true; } @@ -874,10 +911,45 @@ export class PtyServer { private nudgeRedraw(): void { const cols = this.terminal.cols; const rows = this.terminal.rows; - this.ptyProcess.resize(cols - 1, rows); - this.terminal.resize(cols - 1, rows); + this.resizePty(cols - 1, rows); + this.resizePty(cols, rows); + } + + private bumpIoRevision(): void { + this.ioRevision += 1; + } + + private writeToPty(data: string): boolean { + try { + this.ptyProcess.write(data); + this.bumpIoRevision(); + return true; + } catch { + return false; + } + } + + private resizePty(cols: number, rows: number): void { this.ptyProcess.resize(cols, rows); this.terminal.resize(cols, rows); + this.bumpIoRevision(); + } + + private releaseActivity(socket: net.Socket): void { + const before = this.activity.snapshot(); + this.activity.release(socket); + if (JSON.stringify(before) !== JSON.stringify(this.activity.snapshot())) { + this.bumpIoRevision(); + } + } + + private guardedFailure(error: string): GuardedSendResponse { + return { + ok: false, + generation: this.generation, + ioRevision: this.ioRevision, + error, + }; } private emitEvent(type: EventType, fields?: Record): void { diff --git a/tests/guarded-send-integration.test.ts b/tests/guarded-send-integration.test.ts new file mode 100644 index 0000000..b17291a --- /dev/null +++ b/tests/guarded-send-integration.test.ts @@ -0,0 +1,341 @@ +import { afterAll, afterEach, describe, expect, it } from "vitest"; +import * as fs from "node:fs"; +import * as net from "node:net"; +import * as os from "node:os"; +import * as path from "node:path"; +import { PtyServer, type ServerOptions } from "../src/server.ts"; +import { + MessageType, + PacketReader, + encodeAttach, + encodeData, + encodeGuardedData, + encodePeek, +} from "../src/protocol.ts"; +import { compareAndSend } from "../src/guarded-send-client.ts"; +import { connectActivityPublisher } from "../src/activity-client.ts"; +import { queryStats, type StatsResult } from "../src/client.ts"; +import { cleanupAll, getSocketPath } from "../src/sessions.ts"; + +const testRoot = fs.mkdtempSync(path.join(os.tmpdir(), "pty-guarded-")); +const sessionDir = fs.mkdtempSync(path.join(testRoot, "sessions-")); +process.env.PTY_SESSION_DIR = sessionDir; + +let servers: PtyServer[] = []; +let names: string[] = []; + +afterAll(() => { + fs.rmSync(testRoot, { recursive: true, force: true }); +}); + +afterEach(async () => { + for (const server of servers) await server.close(); + for (const name of names) cleanupAll(name); + servers = []; + names = []; +}); + +function uniqueName(): string { + const name = `guarded-${process.pid}-${Date.now()}-${Math.random().toString(36).slice(2, 7)}`; + names.push(name); + return name; +} + +async function startServer( + name: string, + command = "sleep", + args = ["30"], + options: Partial = {}, +): Promise { + const server = new PtyServer({ + name, + command, + args, + displayCommand: command, + cwd: testRoot, + rows: 24, + cols: 80, + ...options, + }); + servers.push(server); + await server.ready; + return server; +} + +function connect(name: string): Promise { + return new Promise((resolve, reject) => { + const socket = net.createConnection(getSocketPath(name)); + socket.once("connect", () => resolve(socket)); + socket.once("error", reject); + }); +} + +async function waitForRevisionChange( + name: string, + revision: number, +): Promise { + const deadline = Date.now() + 3000; + while (Date.now() < deadline) { + const stats = await queryStats(name); + if (stats.ioRevision !== revision) return stats; + await new Promise((resolve) => setTimeout(resolve, 20)); + } + throw new Error(`timed out waiting for revision after ${revision}`); +} + +async function waitForFile(pathname: string, pattern: string): Promise { + const deadline = Date.now() + 3000; + while (Date.now() < deadline) { + const text = fs.existsSync(pathname) ? fs.readFileSync(pathname, "utf8") : ""; + if (text.includes(pattern)) return text; + await new Promise((resolve) => setTimeout(resolve, 20)); + } + return fs.existsSync(pathname) ? fs.readFileSync(pathname, "utf8") : ""; +} + +describe("guarded compare-and-send", () => { + it("succeeds once for an unchanged token and rejects replay with zero bytes", async () => { + const name = uniqueName(); + const captured = path.join(testRoot, `${name}.txt`); + await startServer(name, "sh", ["-c", 'cat > "$1"', "sh", captured]); + const before = await queryStats(name); + + const sent = await compareAndSend(name, { + generation: before.generation, + ioRevision: before.ioRevision, + data: "ONCE\n", + }); + expect(sent.ok).toBe(true); + expect(sent.ioRevision).toBeGreaterThan(before.ioRevision); + expect(await waitForFile(captured, "ONCE")).toContain("ONCE"); + + const replay = await compareAndSend(name, { + generation: before.generation, + ioRevision: before.ioRevision, + data: "TWICE\n", + }); + expect(replay.ok).toBe(false); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(fs.readFileSync(captured, "utf8")).not.toContain("TWICE"); + }); + + for (const [label, bytes] of [ + ["key-shaped", "x"], + ["newline-shaped", "\n"], + ["paste-shaped", "\x1b[200~draft\x1b[201~"], + ["escape-shaped", "\x1b"], + ] as const) { + it(`rejects after ${label} ordinary input without interpreting it`, async () => { + const name = uniqueName(); + const captured = path.join(testRoot, `${name}.txt`); + await startServer(name, "sh", ["-c", 'cat > "$1"', "sh", captured]); + const before = await queryStats(name); + const ordinary = await connect(name); + ordinary.write(encodeData(bytes)); + await waitForRevisionChange(name, before.ioRevision); + + const result = await compareAndSend(name, { + generation: before.generation, + ioRevision: before.ioRevision, + data: "GUARDED\n", + }); + expect(result.ok).toBe(false); + await new Promise((resolve) => setTimeout(resolve, 50)); + const text = fs.existsSync(captured) ? fs.readFileSync(captured, "utf8") : ""; + expect(text).not.toContain("GUARDED"); + ordinary.destroy(); + }); + } + + it("rejects after child output and writes zero guarded bytes", async () => { + const name = uniqueName(); + await startServer(name, process.execPath, [ + "-e", + "process.on('SIGUSR1',()=>process.stdout.write('RACE'));setInterval(()=>{},1000)", + ]); + const before = await queryStats(name); + process.kill(before.process.pid!, "SIGUSR1"); + await waitForRevisionChange(name, before.ioRevision); + + const result = await compareAndSend(name, { + generation: before.generation, + ioRevision: before.ioRevision, + data: "GUARDED", + }); + expect(result.ok).toBe(false); + }); + + it("rejects after an actual resize but allows an attached idle viewer", async () => { + const unchangedName = uniqueName(); + await startServer(unchangedName); + const unchanged = await queryStats(unchangedName); + const viewer = await connect(unchangedName); + viewer.write(encodeAttach(24, 80)); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect((await queryStats(unchangedName)).ioRevision).toBe(unchanged.ioRevision); + expect((await compareAndSend(unchangedName, { + generation: unchanged.generation, + ioRevision: unchanged.ioRevision, + data: "x", + })).ok).toBe(true); + viewer.destroy(); + + const resizedName = uniqueName(); + await startServer(resizedName); + const beforeResize = await queryStats(resizedName); + const resizer = await connect(resizedName); + resizer.write(encodeAttach(23, 79)); + await waitForRevisionChange(resizedName, beforeResize.ioRevision); + expect((await compareAndSend(resizedName, { + generation: beforeResize.generation, + ioRevision: beforeResize.ioRevision, + data: "x", + })).ok).toBe(false); + resizer.destroy(); + }); + + it("rejects a token from a replaced daemon generation", async () => { + const name = uniqueName(); + const first = await startServer(name, "sleep", ["30"], { + generation: "generation-a", + }); + const before = await queryStats(name); + await first.close(); + servers = servers.filter((server) => server !== first); + cleanupAll(name); + await startServer(name, "sleep", ["30"], { + generation: "generation-b", + }); + + const result = await compareAndSend(name, { + generation: before.generation, + ioRevision: before.ioRevision, + data: "x", + }); + expect(result).toMatchObject({ + ok: false, + generation: "generation-b", + }); + }); + + it("keeps the activity lease and publisher usable after a failed guard", async () => { + const name = uniqueName(); + await startServer(name); + const activity = await connectActivityPublisher(name, { + producerEpoch: "activity-a", + source: "adapter", + }); + await activity.publish("idle"); + const before = await queryStats(name); + const ordinary = await connect(name); + ordinary.write(encodeData("race")); + await waitForRevisionChange(name, before.ioRevision); + + expect((await compareAndSend(name, { + generation: before.generation, + ioRevision: before.ioRevision, + data: "x", + })).ok).toBe(false); + expect((await queryStats(name)).activity.state).toBe("idle"); + await expect(activity.publish("active")).resolves.toMatchObject({ + state: "active", + }); + ordinary.destroy(); + activity.close(); + }); + + it("rejects when activity changes after the guarded snapshot", async () => { + const name = uniqueName(); + await startServer(name); + const activity = await connectActivityPublisher(name, { + producerEpoch: "activity-race", + source: "adapter", + }); + await activity.publish("idle"); + const before = await queryStats(name); + await activity.publish("active"); + + expect((await compareAndSend(name, { + generation: before.generation, + ioRevision: before.ioRevision, + data: "x", + })).ok).toBe(false); + activity.close(); + }); + + it("rejects a valid guard on a read-only socket without closing it", async () => { + const name = uniqueName(); + await startServer(name); + const before = await queryStats(name); + const socket = await connect(name); + const reader = new PacketReader(); + socket.write(encodePeek()); + socket.write(encodeGuardedData({ + generation: before.generation, + ioRevision: before.ioRevision, + data: "x", + })); + + const response = await new Promise>((resolve, reject) => { + const timer = setTimeout(() => reject(new Error("guard response timeout")), 3000); + socket.on("data", (data) => { + for (const packet of reader.feed( + Buffer.isBuffer(data) ? data : Buffer.from(data), + )) { + if (packet.type === MessageType.GUARDED_DATA) { + clearTimeout(timer); + resolve(JSON.parse(packet.payload.toString())); + } + } + }); + }); + expect(response.ok).toBe(false); + expect(socket.destroyed).toBe(false); + socket.destroy(); + }); + + it("rejects live malformed and oversized commands without writing data", async () => { + const name = uniqueName(); + const captured = path.join(testRoot, `${name}.txt`); + await startServer(name, "sh", ["-c", 'cat > "$1"', "sh", captured]); + const before = await queryStats(name); + + for (const command of [ + { + generation: before.generation, + ioRevision: before.ioRevision, + data: "MALFORMED\n", + semanticKey: "enter", + }, + { + generation: before.generation, + ioRevision: before.ioRevision, + data: "OVERSIZED".repeat(8193), + }, + ]) { + const socket = await connect(name); + const reader = new PacketReader(); + socket.write(encodeGuardedData(command)); + const response = await new Promise>((resolve, reject) => { + const timer = setTimeout(() => reject(new Error("guard response timeout")), 3000); + socket.on("data", (data) => { + for (const packet of reader.feed( + Buffer.isBuffer(data) ? data : Buffer.from(data), + )) { + if (packet.type === MessageType.GUARDED_DATA) { + clearTimeout(timer); + resolve(JSON.parse(packet.payload.toString())); + } + } + }); + }); + expect(response.ok).toBe(false); + socket.destroy(); + } + + await new Promise((resolve) => setTimeout(resolve, 50)); + const text = fs.existsSync(captured) ? fs.readFileSync(captured, "utf8") : ""; + expect(text).not.toContain("MALFORMED"); + expect(text).not.toContain("OVERSIZED"); + }); +}); diff --git a/tests/guarded-send.test.ts b/tests/guarded-send.test.ts new file mode 100644 index 0000000..6571fab --- /dev/null +++ b/tests/guarded-send.test.ts @@ -0,0 +1,50 @@ +import { describe, expect, it } from "vitest"; +import { + MAX_GUARDED_DATA_BYTES, + parseGuardedSendCommand, +} from "../src/guarded-send.ts"; + +describe("parseGuardedSendCommand", () => { + it("accepts an exact generation, revision, and bounded non-empty payload", () => { + expect(parseGuardedSendCommand(Buffer.from(JSON.stringify({ + generation: "generation-a", + ioRevision: 42, + data: "submit\n", + })))).toEqual({ + generation: "generation-a", + ioRevision: 42, + data: "submit\n", + }); + const escapedAtLimit = "\0".repeat(MAX_GUARDED_DATA_BYTES); + expect(parseGuardedSendCommand(Buffer.from(JSON.stringify({ + generation: "generation-a", + ioRevision: 43, + data: escapedAtLimit, + })))?.data).toBe(escapedAtLimit); + }); + + it("rejects malformed, ambiguous, empty, and oversized guards", () => { + expect(parseGuardedSendCommand(Buffer.from("{bad"))).toBeNull(); + expect(parseGuardedSendCommand(Buffer.from(JSON.stringify({ + generation: "generation-a", + ioRevision: -1, + data: "x", + })))).toBeNull(); + expect(parseGuardedSendCommand(Buffer.from(JSON.stringify({ + generation: "generation-a", + ioRevision: 0, + data: "", + })))).toBeNull(); + expect(parseGuardedSendCommand(Buffer.from(JSON.stringify({ + generation: "generation-a", + ioRevision: 0, + data: "x", + semanticKey: "enter", + })))).toBeNull(); + expect(parseGuardedSendCommand(Buffer.from(JSON.stringify({ + generation: "generation-a", + ioRevision: 0, + data: "x".repeat(MAX_GUARDED_DATA_BYTES + 1), + })))).toBeNull(); + }); +}); diff --git a/tests/protocol.test.ts b/tests/protocol.test.ts index 81b1e02..cb7f512 100644 --- a/tests/protocol.test.ts +++ b/tests/protocol.test.ts @@ -14,6 +14,7 @@ import { encodeStatus, encodeStatusResponse, encodeActivity, + encodeGuardedData, decodeSize, decodeExit, } from "../src/protocol.ts"; @@ -287,4 +288,23 @@ describe("protocol", () => { }); }); }); + + describe("GUARDED_DATA", () => { + it("round-trips compare-and-send commands and responses", () => { + const reader = new PacketReader(); + const encoded = encodeGuardedData({ + generation: "generation-a", + ioRevision: 7, + data: "x", + }); + const packets = reader.feed(encoded); + expect(packets).toHaveLength(1); + expect(packets[0].type).toBe(MessageType.GUARDED_DATA); + expect(JSON.parse(packets[0].payload.toString())).toEqual({ + generation: "generation-a", + ioRevision: 7, + data: "x", + }); + }); + }); }); diff --git a/tests/stats-cli.test.ts b/tests/stats-cli.test.ts index 4dd774c..03133c7 100644 --- a/tests/stats-cli.test.ts +++ b/tests/stats-cli.test.ts @@ -113,6 +113,8 @@ describe("pty stats CLI", () => { const output = runStats(dir, name); expect(output).toContain(`Session: ${name}`); + expect(output).toContain("Generation:"); + expect(output).toContain("I/O rev:"); expect(output).toContain("Terminal:"); expect(output).toContain("Scrollback:"); expect(output).toContain("Clients:"); @@ -134,6 +136,8 @@ describe("pty stats CLI", () => { const stats = JSON.parse(output); expect(stats.name).toBe(name); + expect(stats.generation).toBeTypeOf("string"); + expect(stats.ioRevision).toBeTypeOf("number"); expect(stats.terminal).toBeDefined(); expect(stats.terminal.cols).toBe(80); expect(stats.terminal.rows).toBe(24); From 743ceb796a41a3282e31382575bff0d0e3826d59 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Fri, 31 Jul 2026 10:13:25 +0200 Subject: [PATCH 2/2] Cover combined activity delivery fixtures --- tests/fixtures/guarded-send/cases.json | 116 +++++++++++++++++++++++++ tests/guarded-send-integration.test.ts | 22 +++++ tests/guarded-send.test.ts | 61 +++++++++++++ 3 files changed, 199 insertions(+) create mode 100644 tests/fixtures/guarded-send/cases.json diff --git a/tests/fixtures/guarded-send/cases.json b/tests/fixtures/guarded-send/cases.json new file mode 100644 index 0000000..6926c88 --- /dev/null +++ b/tests/fixtures/guarded-send/cases.json @@ -0,0 +1,116 @@ +{ + "cases": [ + { + "name": "idle", + "screen": "> ", + "activity": "idle", + "alternateScreen": false, + "observed": { "generation": "generation-a", "ioRevision": 10 }, + "current": { "generation": "generation-a", "ioRevision": 10 }, + "change": "none", + "expected": { + "adapterEligible": true, + "guardAccepted": true, + "deliveryAllowed": true + } + }, + { + "name": "active", + "screen": "Working", + "activity": "active", + "alternateScreen": false, + "observed": { "generation": "generation-a", "ioRevision": 10 }, + "current": { "generation": "generation-a", "ioRevision": 10 }, + "change": "none", + "expected": { + "adapterEligible": false, + "guardAccepted": true, + "deliveryAllowed": false + } + }, + { + "name": "alternate-screen-vim", + "screen": "~\n~\nNORMAL", + "activity": "unknown", + "alternateScreen": true, + "observed": { "generation": "generation-a", "ioRevision": 10 }, + "current": { "generation": "generation-a", "ioRevision": 10 }, + "change": "none", + "expected": { + "adapterEligible": false, + "guardAccepted": true, + "deliveryAllowed": false + } + }, + { + "name": "user-input", + "screen": "> draft", + "activity": "idle", + "alternateScreen": false, + "observed": { "generation": "generation-a", "ioRevision": 10 }, + "current": { "generation": "generation-a", "ioRevision": 11 }, + "change": "user_input", + "expected": { + "adapterEligible": true, + "guardAccepted": false, + "deliveryAllowed": false + } + }, + { + "name": "child-output", + "screen": "background output", + "activity": "idle", + "alternateScreen": false, + "observed": { "generation": "generation-a", "ioRevision": 10 }, + "current": { "generation": "generation-a", "ioRevision": 11 }, + "change": "child_output", + "expected": { + "adapterEligible": true, + "guardAccepted": false, + "deliveryAllowed": false + } + }, + { + "name": "resize", + "screen": "> ", + "activity": "idle", + "alternateScreen": false, + "observed": { "generation": "generation-a", "ioRevision": 10 }, + "current": { "generation": "generation-a", "ioRevision": 11 }, + "change": "resize", + "expected": { + "adapterEligible": true, + "guardAccepted": false, + "deliveryAllowed": false + } + }, + { + "name": "stale-token", + "screen": "> ", + "activity": "idle", + "alternateScreen": false, + "observed": { "generation": "generation-a", "ioRevision": 9 }, + "current": { "generation": "generation-a", "ioRevision": 10 }, + "change": "stale_token", + "expected": { + "adapterEligible": true, + "guardAccepted": false, + "deliveryAllowed": false + } + }, + { + "name": "daemon-replacement", + "screen": "> ", + "activity": "idle", + "alternateScreen": false, + "observed": { "generation": "generation-a", "ioRevision": 10 }, + "current": { "generation": "generation-b", "ioRevision": 0 }, + "change": "daemon_replacement", + "expected": { + "adapterEligible": true, + "guardAccepted": false, + "deliveryAllowed": false + } + } + ] +} diff --git a/tests/guarded-send-integration.test.ts b/tests/guarded-send-integration.test.ts index b17291a..2959716 100644 --- a/tests/guarded-send-integration.test.ts +++ b/tests/guarded-send-integration.test.ts @@ -94,6 +94,28 @@ async function waitForFile(pathname: string, pattern: string): Promise { } describe("guarded compare-and-send", () => { + it("succeeds once with explicit idle authority and an unchanged token", async () => { + const name = uniqueName(); + await startServer(name); + const activity = await connectActivityPublisher(name, { + producerEpoch: "idle-authority", + source: "adapter", + }); + await activity.publish("idle"); + const observed = await queryStats(name); + expect(observed.activity.state).toBe("idle"); + + const result = await compareAndSend(name, { + generation: observed.generation, + ioRevision: observed.ioRevision, + data: "x", + }); + expect(result.ok).toBe(true); + expect(result.ioRevision).toBeGreaterThan(observed.ioRevision); + expect((await queryStats(name)).activity.state).toBe("idle"); + activity.close(); + }); + it("succeeds once for an unchanged token and rejects replay with zero bytes", async () => { const name = uniqueName(); const captured = path.join(testRoot, `${name}.txt`); diff --git a/tests/guarded-send.test.ts b/tests/guarded-send.test.ts index 6571fab..68ecef6 100644 --- a/tests/guarded-send.test.ts +++ b/tests/guarded-send.test.ts @@ -1,9 +1,70 @@ +import * as fs from "node:fs"; +import * as path from "node:path"; import { describe, expect, it } from "vitest"; +import type { ActivityFixtureState } from "../src/activity.ts"; import { MAX_GUARDED_DATA_BYTES, parseGuardedSendCommand, } from "../src/guarded-send.ts"; +describe("combined activity and guarded-send fixtures", () => { + it("separates explicit idle authority from exact-token acceptance", () => { + const fixturePath = path.join( + import.meta.dirname, + "fixtures", + "guarded-send", + "cases.json", + ); + const fixture = JSON.parse(fs.readFileSync(fixturePath, "utf8")) as { + cases: Array<{ + name: string; + screen: string; + activity: ActivityFixtureState; + alternateScreen: boolean; + observed: { generation: string; ioRevision: number }; + current: { generation: string; ioRevision: number }; + change: string; + expected: { + adapterEligible: boolean; + guardAccepted: boolean; + deliveryAllowed: boolean; + }; + }>; + }; + + expect(fixture.cases.map((entry) => entry.name)).toEqual([ + "idle", + "active", + "alternate-screen-vim", + "user-input", + "child-output", + "resize", + "stale-token", + "daemon-replacement", + ]); + for (const entry of fixture.cases) { + const adapterEligible = entry.activity === "idle"; + const guardAccepted = + entry.observed.generation === entry.current.generation && + entry.observed.ioRevision === entry.current.ioRevision; + expect(entry.expected.adapterEligible, entry.name).toBe(adapterEligible); + expect(entry.expected.guardAccepted, entry.name).toBe(guardAccepted); + expect(entry.expected.deliveryAllowed, entry.name).toBe( + adapterEligible && guardAccepted, + ); + } + expect(fixture.cases.find((entry) => entry.name === "alternate-screen-vim")) + .toMatchObject({ + alternateScreen: true, + activity: "unknown", + expected: { + guardAccepted: true, + deliveryAllowed: false, + }, + }); + }); +}); + describe("parseGuardedSendCommand", () => { it("accepts an exact generation, revision, and bounded non-empty payload", () => { expect(parseGuardedSendCommand(Buffer.from(JSON.stringify({