Add revision-guarded PTY send

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`,

README.md

@@ -420,7 +420,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";
@@ -455,6 +455,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:

docs/client.md

@@ -306,6 +306,8 @@ each writable client's requested size and which min-wins axes it constrains.
 ```typescript
 interface StatsResult {
   name: string;
+  generation: string;
+  ioRevision: number;
   terminal: {
     cols: number; rows: number;
     cursorX: number; cursorY: number;
@@ -363,6 +365,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<GuardedSendResponse>`
+
+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<ActivityPublisher>`
 
 Claim the session's single live activity lease and publish harness-neutral
@@ -537,6 +573,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
   GEOMETRY: 10, // Effective shared rows/cols (server → client)
 };
 ```

src/cli.ts

@@ -2531,6 +2531,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}`);

src/client-api.ts

@@ -42,6 +42,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 {
@@ -73,6 +81,6 @@ export { resolveKey, parseSeqValue } from "./keys.ts";
 
 // Protocol (advanced)
 export {
-  PacketReader, MessageType, encodeActivity,
+  PacketReader, MessageType, encodeActivity, encodeGuardedData,
   type Packet,
 } from "./protocol.ts";

src/client.ts

@@ -295,6 +295,8 @@ export interface ProcessResources {
 
 export interface StatsResult {
   name: string;
+  generation: string;
+  ioRevision: number;
   terminal: {
     cols: number;
     rows: number;

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<GuardedSendResponse> {
+  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));
+    });
+  });
+}

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<string, unknown>;
+    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;
+  }
+}

src/protocol.ts

@@ -10,7 +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
-  // Value 9 is reserved for an independent protocol extension.
+  GUARDED_DATA: 9, // Bidirectional: generation/revision-conditional input
   GEOMETRY: 10, // Server → Client: effective shared rows/cols
 } as const;
 
@@ -108,6 +108,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 };