From 46c71d31c0d6daee43adf568061b2b84a65ae8c0 Mon Sep 17 00:00:00 2001 From: Nathan Herald Date: Fri, 31 Jul 2026 00:11:53 +0200 Subject: [PATCH] Expose generation-bound PTY activity status --- CHANGELOG.md | 12 ++ README.md | 8 ++ docs/client.md | 40 +++++++ src/activity-client.ts | 152 ++++++++++++++++++++++++ src/activity.ts | 166 ++++++++++++++++++++++++++ src/cli.ts | 5 + src/client-api.ts | 13 +- src/client.ts | 3 + src/protocol.ts | 5 + src/server.ts | 22 ++++ tests/activity.test.ts | 184 +++++++++++++++++++++++++++++ tests/fixtures/activity/cases.json | 84 +++++++++++++ tests/integration.test.ts | 141 ++++++++++++++++++++++ tests/protocol.test.ts | 18 +++ tests/stats-cli.test.ts | 9 +- 15 files changed, 860 insertions(+), 2 deletions(-) create mode 100644 src/activity-client.ts create mode 100644 src/activity.ts create mode 100644 tests/activity.test.ts create mode 100644 tests/fixtures/activity/cases.json diff --git a/CHANGELOG.md b/CHANGELOG.md index 37225b6..986ae5d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,18 @@ ## Unreleased +### Harness-neutral live activity status + +- Add a single generation-bound activity lease with ordered `unknown`, + `active`, `child_command`, and `idle` transitions. Live status exposes the + accepted state and resets it to `unknown` when the publisher disconnects or + the daemon generation changes; stale epochs, skipped updates, malformed + commands, and competing publishers are rejected. +- Export `connectActivityPublisher()` for harness-specific adapters while + keeping hook and log interpretation outside pty. `pty stats` and + `queryStats()` now also expose `modes.alternateScreen` as a diagnostic only; + terminal modes do not imply semantic activity or idleness. + ### Read-only session listing - `listSessions()` and `pty list` are now strictly observational: they no diff --git a/README.md b/README.md index 66f81fb..37c11d8 100644 --- a/README.md +++ b/README.md @@ -373,6 +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, EventFollower, readRecentEvents, extractFilterTags, matchesAllTags, } from "@compoundingtech/pty/client"; @@ -400,6 +401,13 @@ const sessions = await listSessions(); const stats = await queryStats("myserver"); ``` +`queryStats()` includes generation-bound live activity +(`unknown`, `active`, `child_command`, or `idle`) and terminal diagnostics such +as `modes.alternateScreen`. Activity is explicit publisher state; terminal +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. + ### 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 4798baf..c40d649 100644 --- a/docs/client.md +++ b/docs/client.md @@ -254,9 +254,18 @@ interface StatsResult { }; clients: { total: number; attached: number; readOnly: number }; modes: { + alternateScreen: boolean; sgrMouse: boolean; cursorHidden: boolean; kittyKeyboard: boolean; kittyKeyboardFlags: number[]; }; + activity: { + state: "unknown" | "active" | "child_command" | "idle"; + generation: string; + producerEpoch: string | null; + sequence: number; + turnId?: string; + source?: string; + }; uptimeSeconds: number | null; createdAt: string | null; } @@ -267,6 +276,36 @@ interface ProcessResources { } ``` +`activity` is an explicitly published, generation-bound live fact. It starts +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. + +### `connectActivityPublisher(name, options?): Promise` + +Claim the session's single live activity lease and publish harness-neutral +state transitions. Harness-specific hook or log parsing belongs in the caller. + +```typescript +import { connectActivityPublisher } from "@compoundingtech/pty/client"; + +const activity = await connectActivityPublisher("myserver", { + source: "codex", +}); + +await activity.publish("active", { turnId: "turn-42" }); +await activity.publish("child_command", { turnId: "turn-42" }); +await activity.publish("idle", { turnId: "turn-42" }); + +activity.close(); // STATUS immediately falls back to unknown +``` + +Updates are strictly sequenced within a random producer epoch. A competing +publisher is rejected while the lease is held. Publisher failure, socket loss, +malformed updates, skipped sequences, stale epochs, and daemon replacement all +fail closed to `unknown`. Rejected competing sockets do not disturb the current +publisher. + ## Session Interaction (CLI-oriented) These functions use `process.stdin`/`process.stdout` directly and may call `process.exit()`. They are re-exported for tools that want CLI-like behavior. @@ -384,6 +423,7 @@ const MessageType = { SCREEN: 5, // Screen replay PEEK: 6, // Read-only peek request STATUS: 7, // Stats query/response + ACTIVITY: 8, // Generic activity lease commands/responses }; ``` diff --git a/src/activity-client.ts b/src/activity-client.ts new file mode 100644 index 0000000..749c653 --- /dev/null +++ b/src/activity-client.ts @@ -0,0 +1,152 @@ +import * as net from "node:net"; +import { randomUUID } from "node:crypto"; +import { + type ActivityClaim, + type ActivityResponse, + type ActivityState, + type ActivityStatus, + type ActivityUpdate, +} from "./activity.ts"; +import { MessageType, PacketReader, encodeActivity } from "./protocol.ts"; +import { getSocketPath } from "./sessions.ts"; + +export interface ActivityPublisherOptions { + producerEpoch?: string; + source?: string; + timeoutMs?: number; +} + +export interface ActivityPublishOptions { + turnId?: string; +} + +export class ActivityPublisher { + private readonly reader = new PacketReader(); + private pending: { + resolve: (value: ActivityStatus) => void; + reject: (error: Error) => void; + timer: NodeJS.Timeout; + } | null = null; + private sequence = 0; + private closed = false; + + private constructor( + private readonly socket: net.Socket, + readonly producerEpoch: string, + private readonly timeoutMs: number, + ) { + socket.on("data", (data) => { + let packets; + try { + packets = this.reader.feed(Buffer.isBuffer(data) ? data : Buffer.from(data)); + } catch (error) { + this.failPending(error instanceof Error ? error : new Error(String(error))); + socket.destroy(); + return; + } + for (const packet of packets) { + if (packet.type !== MessageType.ACTIVITY || this.pending === null) continue; + let response: ActivityResponse; + try { + response = JSON.parse(packet.payload.toString("utf8")) as ActivityResponse; + } catch { + this.failPending(new Error("invalid activity response")); + socket.destroy(); + return; + } + const pending = this.pending; + this.pending = null; + clearTimeout(pending.timer); + if (!response.ok) { + pending.reject(new Error(response.error ?? "activity update rejected")); + } else { + this.sequence = response.activity.sequence; + pending.resolve(response.activity); + } + } + }); + socket.on("error", (error) => this.failPending(error)); + socket.on("close", () => { + this.closed = true; + this.failPending(new Error("activity publisher connection closed")); + }); + } + + static async connect( + name: string, + options: ActivityPublisherOptions = {}, + ): Promise { + const socket = net.createConnection(getSocketPath(name)); + await new Promise((resolve, reject) => { + socket.once("connect", resolve); + socket.once("error", reject); + }); + const publisher = new ActivityPublisher( + socket, + options.producerEpoch ?? randomUUID(), + options.timeoutMs ?? 2000, + ); + const claim: ActivityClaim = { + op: "claim", + producerEpoch: publisher.producerEpoch, + ...(options.source === undefined ? {} : { source: options.source }), + }; + try { + await publisher.request(claim); + return publisher; + } catch (error) { + publisher.close(); + throw error; + } + } + + publish( + state: ActivityState, + options: ActivityPublishOptions = {}, + ): Promise { + const update: ActivityUpdate = { + op: "set", + producerEpoch: this.producerEpoch, + sequence: this.sequence + 1, + state, + ...(options.turnId === undefined ? {} : { turnId: options.turnId }), + }; + return this.request(update); + } + + close(): void { + this.closed = true; + this.socket.destroy(); + } + + private request(command: ActivityClaim | ActivityUpdate): Promise { + if (this.closed) return Promise.reject(new Error("activity publisher is closed")); + if (this.pending !== null) { + return Promise.reject(new Error("activity update already in flight")); + } + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + this.pending = null; + reject(new Error("timed out waiting for activity response")); + this.socket.destroy(); + }, this.timeoutMs); + this.pending = { resolve, reject, timer }; + this.socket.write(encodeActivity(command)); + }); + } + + private failPending(error: Error): void { + if (this.pending === null) return; + const pending = this.pending; + this.pending = null; + clearTimeout(pending.timer); + pending.reject(error); + } +} + +export function connectActivityPublisher( + name: string, + options: ActivityPublisherOptions = {}, +): Promise { + return ActivityPublisher.connect(name, options); +} diff --git a/src/activity.ts b/src/activity.ts new file mode 100644 index 0000000..aefea43 --- /dev/null +++ b/src/activity.ts @@ -0,0 +1,166 @@ +export const ACTIVITY_STATES = [ + "unknown", + "active", + "child_command", + "idle", +] as const; + +export type ActivityState = (typeof ACTIVITY_STATES)[number]; +/** Alias used by stable model-free fixture schemas. */ +export type ActivityFixtureState = ActivityState; + +export interface ActivityStatus { + state: ActivityState; + generation: string; + producerEpoch: string | null; + sequence: number; + turnId?: string; + source?: string; +} + +export interface ActivityClaim { + op: "claim"; + producerEpoch: string; + source?: string; +} + +export interface ActivityUpdate { + op: "set"; + producerEpoch: string; + sequence: number; + state: ActivityState; + turnId?: string; +} + +export type ActivityCommand = ActivityClaim | ActivityUpdate; + +export interface ActivityResponse { + ok: boolean; + activity: ActivityStatus; + error?: string; +} + +const MAX_ACTIVITY_PAYLOAD_BYTES = 4096; +const MAX_EPOCH_LENGTH = 128; +const MAX_SOURCE_LENGTH = 64; +const MAX_TURN_ID_LENGTH = 256; + +function boundedString( + value: unknown, + maxLength: number, + optional = false, +): boolean { + if (value === undefined && optional) return true; + return typeof value === "string" && value.length > 0 && value.length <= maxLength; +} + +export function parseActivityCommand(payload: Buffer): ActivityCommand | null { + if (payload.length === 0 || payload.length > MAX_ACTIVITY_PAYLOAD_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; + const producerEpoch = value.producerEpoch; + if ( + typeof producerEpoch !== "string" || + !boundedString(producerEpoch, MAX_EPOCH_LENGTH) + ) return null; + if (value.op === "claim") { + if (!boundedString(value.source, MAX_SOURCE_LENGTH, true)) return null; + return { + op: "claim", + producerEpoch, + ...(value.source === undefined ? {} : { source: value.source as string }), + }; + } + if (value.op !== "set") return null; + if ( + !Number.isSafeInteger(value.sequence) || + Number(value.sequence) <= 0 || + !ACTIVITY_STATES.includes(value.state as ActivityState) || + !boundedString(value.turnId, MAX_TURN_ID_LENGTH, true) + ) { + return null; + } + return { + op: "set", + producerEpoch, + sequence: Number(value.sequence), + state: value.state as ActivityState, + ...(value.turnId === undefined ? {} : { turnId: value.turnId as string }), + }; + } catch { + return null; + } +} + +export class ActivityLease { + private owner: Owner | null = null; + private status: ActivityStatus; + + constructor(private readonly generation: string) { + this.status = this.unknownStatus(); + } + + snapshot(): ActivityStatus { + return { ...this.status }; + } + + apply(owner: Owner, command: ActivityCommand): ActivityResponse { + if (command.op === "claim") { + if (this.owner !== null && this.owner !== owner) { + return this.failure("activity lease already held"); + } + this.owner = owner; + this.status = { + state: "unknown", + generation: this.generation, + producerEpoch: command.producerEpoch, + sequence: 0, + ...(command.source === undefined ? {} : { source: command.source }), + }; + return { ok: true, activity: this.snapshot() }; + } + + if (this.owner !== owner) { + return this.failure("activity lease identity mismatch"); + } + if (this.status.producerEpoch !== command.producerEpoch) { + this.release(owner); + return this.failure("activity lease identity mismatch"); + } + const expectedSequence = this.status.sequence + 1; + if (command.sequence !== expectedSequence) { + this.release(owner); + return this.failure(`activity sequence must be ${expectedSequence}`); + } + this.status = { + state: command.state, + generation: this.generation, + producerEpoch: command.producerEpoch, + sequence: command.sequence, + ...(command.turnId === undefined ? {} : { turnId: command.turnId }), + ...(this.status.source === undefined ? {} : { source: this.status.source }), + }; + return { ok: true, activity: this.snapshot() }; + } + + release(owner: Owner): void { + if (this.owner !== owner) return; + this.owner = null; + this.status = this.unknownStatus(); + } + + private failure(error: string): ActivityResponse { + return { ok: false, error, activity: this.snapshot() }; + } + + private unknownStatus(): ActivityStatus { + return { + state: "unknown", + generation: this.generation, + producerEpoch: null, + sequence: 0, + }; + } +} diff --git a/src/cli.ts b/src/cli.ts index ae6684a..d8cc617 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -2395,10 +2395,15 @@ function printStats(stats: StatsResult, meta: SessionInfo["metadata"]): void { console.log(` Clients: ${stats.clients.total} (${stats.clients.attached} attached, ${stats.clients.readOnly} readonly)`); const modes: string[] = []; + if (stats.modes.alternateScreen) modes.push("alternate screen"); if (stats.modes.sgrMouse) modes.push("SGR mouse"); if (stats.modes.cursorHidden) modes.push("cursor hidden"); if (stats.modes.kittyKeyboard) modes.push(`kitty keyboard (flags: ${stats.modes.kittyKeyboardFlags.join(",")})`); console.log(` Modes: ${modes.length > 0 ? modes.join(", ") : "none"}`); + console.log( + ` Activity: ${stats.activity.state}` + + (stats.activity.source ? ` (${stats.activity.source})` : ""), + ); } function formatMemory(rssKb: number): string { diff --git a/src/client-api.ts b/src/client-api.ts index 5bcefa4..4ce01aa 100644 --- a/src/client-api.ts +++ b/src/client-api.ts @@ -31,6 +31,17 @@ export { type StatsResult, type ProcessResources, } from "./client.ts"; +// Harness-neutral activity lease. Codex/Claude-specific adapters live outside +// pty and publish only these generic states. +export { + ActivityPublisher, connectActivityPublisher, + type ActivityPublisherOptions, type ActivityPublishOptions, +} from "./activity-client.ts"; +export { + ACTIVITY_STATES, + type ActivityState, type ActivityStatus, +} from "./activity.ts"; + // Events export { EventType, @@ -61,6 +72,6 @@ export { resolveKey, parseSeqValue } from "./keys.ts"; // Protocol (advanced) export { - PacketReader, MessageType, + PacketReader, MessageType, encodeActivity, type Packet, } from "./protocol.ts"; diff --git a/src/client.ts b/src/client.ts index 00acafa..3682ea4 100644 --- a/src/client.ts +++ b/src/client.ts @@ -11,6 +11,7 @@ import { encodeStatus, decodeExit, } from "./protocol.ts"; +import type { ActivityStatus } from "./activity.ts"; import { getSocketPath } from "./sessions.ts"; import { stripAnsi } from "./tui/colors.ts"; import { BRACKETED_PASTE_START, BRACKETED_PASTE_END } from "./paste.ts"; @@ -316,11 +317,13 @@ export interface StatsResult { readOnly: number; }; modes: { + alternateScreen: boolean; sgrMouse: boolean; cursorHidden: boolean; kittyKeyboard: boolean; kittyKeyboardFlags: number[]; }; + activity: ActivityStatus; uptimeSeconds: number | null; createdAt: string | null; } diff --git a/src/protocol.ts b/src/protocol.ts index a04a6fd..923e8b1 100644 --- a/src/protocol.ts +++ b/src/protocol.ts @@ -9,6 +9,7 @@ export const MessageType = { SCREEN: 5, // Server → Client: screen buffer replay on attach 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 } as const; export type MessageType = (typeof MessageType)[keyof typeof MessageType]; @@ -94,6 +95,10 @@ export function encodeStatusResponse(json: string): Buffer { return encodePacket(MessageType.STATUS, Buffer.from(json)); } +export function encodeActivity(value: unknown): Buffer { + return encodePacket(MessageType.ACTIVITY, 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 762597f..4349842 100644 --- a/src/server.ts +++ b/src/server.ts @@ -16,8 +16,10 @@ import { encodeExit, encodeScreen, encodeStatusResponse, + encodeActivity, decodeSize, } from "./protocol.ts"; +import { ActivityLease, parseActivityCommand } from "./activity.ts"; import { getSocketPath, getPidPath, @@ -253,6 +255,7 @@ export class PtyServer { private eventWriter: EventWriter; private generation: string; private lastTitle = ""; + private activity: ActivityLease; readonly ready: Promise; // Resolves when the child process's onExit has fired — used by close() to // make sure session_exit has been queued to the event chain before we @@ -264,6 +267,7 @@ export class PtyServer { this.name = options.name; this.options = options; this.generation = options.generation ?? randomBytes(16).toString("hex"); + this.activity = new ActivityLease(this.generation); this.eventWriter = new EventWriter(options.name); this.childExited = new Promise((resolve) => { this.resolveChildExited = resolve; @@ -733,16 +737,32 @@ export class PtyServer { socket.write(encodeStatusResponse(JSON.stringify(stats))); break; } + + case MessageType.ACTIVITY: { + const command = parseActivityCommand(packet.payload); + if (command === null) this.activity.release(socket); + const response = command === null + ? { + ok: false, + error: "invalid activity command", + activity: this.activity.snapshot(), + } + : this.activity.apply(socket, command); + socket.write(encodeActivity(response)); + break; + } } } }); socket.on("close", () => { + this.activity.release(socket); this.clients.delete(socket); this.negotiateSize(); }); socket.on("error", () => { + this.activity.release(socket); this.clients.delete(socket); this.negotiateSize(); }); @@ -812,11 +832,13 @@ export class PtyServer { readOnly, }, modes: { + alternateScreen: this.altScreenActive, sgrMouse: this.sgrMouseMode, cursorHidden: this.cursorHidden, kittyKeyboard: this.kittyKeyboardStack.length > 0, kittyKeyboardFlags: [...this.kittyKeyboardStack], }, + activity: this.activity.snapshot(), uptimeSeconds, createdAt, }; diff --git a/tests/activity.test.ts b/tests/activity.test.ts new file mode 100644 index 0000000..273e819 --- /dev/null +++ b/tests/activity.test.ts @@ -0,0 +1,184 @@ +import * as fs from "node:fs"; +import * as path from "node:path"; +import { describe, expect, it } from "vitest"; +import { + ActivityLease, + parseActivityCommand, + type ActivityFixtureState, +} from "../src/activity.ts"; + +describe("activity fixtures", () => { + it("make explicit activity the only configured-adapter eligibility fact", () => { + const fixturePath = path.join( + import.meta.dirname, + "fixtures", + "activity", + "cases.json", + ); + const fixture = JSON.parse(fs.readFileSync(fixturePath, "utf8")) as { + cases: Array<{ + name: string; + screen: string; + activity: ActivityFixtureState; + alternateScreen: boolean; + inputMode: string; + eligible: boolean; + }>; + }; + + expect(fixture.cases.map((entry) => entry.name)).toEqual([ + "active-turn", + "idle-prompt", + "long-child-command", + "alternate-screen", + "terminal-restored", + "compaction", + "clear", + "adapter-crash", + "stale-log-idle", + "daemon-restart", + ]); + for (const entry of fixture.cases) { + expect(entry.eligible, entry.name).toBe(entry.activity === "idle"); + } + expect(fixture.cases.find((entry) => entry.name === "idle-prompt")) + .toMatchObject({ alternateScreen: true, inputMode: "raw", eligible: true }); + expect(fixture.cases.find((entry) => entry.name === "stale-log-idle")) + .toMatchObject({ screen: "> ", activity: "unknown", eligible: false }); + }); +}); + +describe("ActivityLease", () => { + it("starts unknown and generation-bound", () => { + const lease = new ActivityLease("generation-a"); + expect(lease.snapshot()).toEqual({ + state: "unknown", + generation: "generation-a", + producerEpoch: null, + sequence: 0, + }); + }); + + it("orders active, child-command, and idle transitions from one live owner", () => { + const lease = new ActivityLease("generation-a"); + const owner = {}; + + expect(lease.apply(owner, { + op: "claim", + producerEpoch: "epoch-a", + source: "codex", + })).toMatchObject({ ok: true, activity: { state: "unknown", sequence: 0 } }); + expect(lease.apply(owner, { + op: "set", + producerEpoch: "epoch-a", + sequence: 1, + state: "active", + turnId: "turn-1", + })).toMatchObject({ ok: true, activity: { state: "active", sequence: 1 } }); + expect(lease.apply(owner, { + op: "set", + producerEpoch: "epoch-a", + sequence: 2, + state: "child_command", + turnId: "turn-1", + })).toMatchObject({ + ok: true, + activity: { state: "child_command", sequence: 2 }, + }); + expect(lease.apply(owner, { + op: "set", + producerEpoch: "epoch-a", + sequence: 3, + state: "idle", + turnId: "turn-1", + })).toMatchObject({ ok: true, activity: { state: "idle", sequence: 3 } }); + }); + + it("rejects a competing owner and stale, skipped, or wrong-epoch updates", () => { + const lease = new ActivityLease("generation-a"); + const owner = {}; + const other = {}; + expect(lease.apply(owner, { + op: "claim", + producerEpoch: "epoch-a", + }).ok).toBe(true); + expect(lease.apply(other, { + op: "claim", + producerEpoch: "epoch-b", + })).toMatchObject({ ok: false, error: "activity lease already held" }); + expect(lease.apply(owner, { + op: "set", + producerEpoch: "epoch-a", + sequence: 1, + state: "idle", + }).ok).toBe(true); + expect(lease.apply(owner, { + op: "set", + producerEpoch: "epoch-a", + sequence: 3, + state: "active", + })).toMatchObject({ + ok: false, + error: "activity sequence must be 2", + activity: { state: "unknown", producerEpoch: null, sequence: 0 }, + }); + expect(lease.apply(owner, { + op: "claim", + producerEpoch: "epoch-c", + }).ok).toBe(true); + expect(lease.apply(owner, { + op: "set", + producerEpoch: "epoch-b", + sequence: 1, + state: "idle", + })).toMatchObject({ + ok: false, + error: "activity lease identity mismatch", + activity: { state: "unknown", producerEpoch: null, sequence: 0 }, + }); + }); + + it("resets to unknown on adapter crash and daemon restart", () => { + const owner = {}; + const lease = new ActivityLease("generation-a"); + lease.apply(owner, { op: "claim", producerEpoch: "epoch-a" }); + lease.apply(owner, { + op: "set", + producerEpoch: "epoch-a", + sequence: 1, + state: "idle", + }); + lease.release(owner); + expect(lease.snapshot()).toEqual({ + state: "unknown", + generation: "generation-a", + producerEpoch: null, + sequence: 0, + }); + + const restarted = new ActivityLease("generation-b"); + expect(restarted.snapshot()).toEqual({ + state: "unknown", + generation: "generation-b", + producerEpoch: null, + sequence: 0, + }); + }); + + it("rejects malformed and oversized adapter commands", () => { + expect(parseActivityCommand(Buffer.from("{bad"))).toBeNull(); + expect(parseActivityCommand(Buffer.from(JSON.stringify({ + op: "set", + producerEpoch: "epoch-a", + sequence: 1, + state: "idle", + turnId: "x".repeat(257), + })))).toBeNull(); + expect(parseActivityCommand(Buffer.from(JSON.stringify({ + op: "set", + producerEpoch: "epoch-a", + sequence: 1, + state: "screen-looked-idle", + })))).toBeNull(); + }); +}); diff --git a/tests/fixtures/activity/cases.json b/tests/fixtures/activity/cases.json new file mode 100644 index 0000000..64cffdd --- /dev/null +++ b/tests/fixtures/activity/cases.json @@ -0,0 +1,84 @@ +{ + "cases": [ + { + "name": "active-turn", + "screen": "Working\n", + "activity": "active", + "alternateScreen": true, + "inputMode": "raw", + "eligible": false + }, + { + "name": "idle-prompt", + "screen": "> ", + "activity": "idle", + "alternateScreen": true, + "inputMode": "raw", + "eligible": true + }, + { + "name": "long-child-command", + "screen": "Running command\n", + "activity": "child_command", + "alternateScreen": false, + "inputMode": "canonical", + "eligible": false + }, + { + "name": "alternate-screen", + "screen": "-- INSERT --\n", + "activity": "unknown", + "alternateScreen": true, + "inputMode": "raw", + "eligible": false + }, + { + "name": "terminal-restored", + "screen": "$ ", + "activity": "idle", + "alternateScreen": false, + "inputMode": "canonical", + "eligible": true + }, + { + "name": "compaction", + "screen": "Conversation compacted\n", + "activity": "unknown", + "alternateScreen": true, + "inputMode": "raw", + "eligible": false + }, + { + "name": "clear", + "screen": "", + "activity": "unknown", + "alternateScreen": true, + "inputMode": "raw", + "eligible": false + }, + { + "name": "adapter-crash", + "screen": "> ", + "activity": "unknown", + "alternateScreen": true, + "inputMode": "raw", + "eligible": false + }, + { + "name": "stale-log-idle", + "screen": "> ", + "activity": "unknown", + "alternateScreen": false, + "inputMode": "canonical", + "eligible": false + }, + { + "name": "daemon-restart", + "screen": "> ", + "activity": "unknown", + "alternateScreen": false, + "inputMode": "unknown", + "eligible": false + } + ] +} diff --git a/tests/integration.test.ts b/tests/integration.test.ts index 379f5c2..9c9a925 100644 --- a/tests/integration.test.ts +++ b/tests/integration.test.ts @@ -15,6 +15,7 @@ import { encodePeek, encodeResize, encodeStatus, + encodeActivity, decodeExit, } from "../src/protocol.ts"; import { @@ -26,6 +27,8 @@ import { acquireLock, releaseLock, } from "../src/sessions.ts"; +import { queryStats } from "../src/client.ts"; +import { connectActivityPublisher } from "../src/activity-client.ts"; // All tests run in a tmp directory to avoid polluting the project const testCwd = fs.mkdtempSync(path.join(os.tmpdir(), "pty-int-")); @@ -75,6 +78,19 @@ function connect(name: string): Promise { }); } +async function waitForActivityState( + name: string, + state: "unknown" | "active" | "child_command" | "idle", +): Promise>["activity"]> { + const deadline = Date.now() + 2000; + while (Date.now() < deadline) { + const activity = (await queryStats(name)).activity; + if (activity.state === state) return activity; + await new Promise((resolve) => setTimeout(resolve, 20)); + } + throw new Error(`timed out waiting for activity state ${state}`); +} + /** Collect packets from a socket until we have at least `count`, or timeout. */ function collectPackets( socket: net.Socket, @@ -1422,4 +1438,129 @@ describe("STATUS message", () => { client.destroy(); }); + + it("reports alternate-screen without treating it as semantic activity", async () => { + const name = uniqueName(); + await startServer(name, "sh", [ + "-c", + "printf '\\033[?1049hALT'; sleep 30", + ]); + await new Promise((r) => setTimeout(r, 300)); + + const client = await connect(name); + const reader = new PacketReader(); + client.write(encodeStatus()); + const packet = await waitForType(client, reader, MessageType.STATUS); + const stats = JSON.parse(packet.payload.toString()); + + expect(stats.modes.alternateScreen).toBe(true); + expect(stats.activity.state).toBe("unknown"); + client.destroy(); + }); + + it("holds a live activity lease and resets unknown when its socket closes", async () => { + const name = uniqueName(); + await startServer(name, "cat"); + const publisher = await connect(name); + const publisherReader = new PacketReader(); + + publisher.write(encodeActivity({ + op: "claim", + producerEpoch: "epoch-a", + source: "codex", + })); + const claimed = JSON.parse( + (await waitForType(publisher, publisherReader, MessageType.ACTIVITY)) + .payload.toString(), + ); + expect(claimed).toMatchObject({ + ok: true, + activity: { + state: "unknown", + producerEpoch: "epoch-a", + sequence: 0, + }, + }); + + publisher.write(encodeActivity({ + op: "set", + producerEpoch: "epoch-a", + sequence: 1, + state: "active", + turnId: "turn-1", + })); + const active = JSON.parse( + (await waitForType(publisher, publisherReader, MessageType.ACTIVITY)) + .payload.toString(), + ); + expect(active).toMatchObject({ + ok: true, + activity: { state: "active", sequence: 1, turnId: "turn-1" }, + }); + + const statsClient = await connect(name); + const statsReader = new PacketReader(); + statsClient.write(encodeStatus()); + const live = JSON.parse( + (await waitForType(statsClient, statsReader, MessageType.STATUS)) + .payload.toString(), + ); + expect(live.activity).toEqual(active.activity); + statsClient.destroy(); + + publisher.write(encodeActivity({ + op: "set", + producerEpoch: "epoch-a", + sequence: 3, + state: "idle", + })); + const rejected = JSON.parse( + (await waitForType(publisher, publisherReader, MessageType.ACTIVITY)) + .payload.toString(), + ); + expect(rejected).toMatchObject({ + ok: false, + activity: { state: "unknown", producerEpoch: null, sequence: 0 }, + }); + + publisher.destroy(); + expect(await waitForActivityState(name, "unknown")).toMatchObject({ + state: "unknown", + producerEpoch: null, + sequence: 0, + }); + }); + + it("publishes ordered activity through the public API and rejects competition", async () => { + const name = uniqueName(); + await startServer(name, "cat"); + const publisher = await connectActivityPublisher(name, { + producerEpoch: "epoch-public", + source: "codex", + }); + + await publisher.publish("active", { turnId: "turn-public" }); + await publisher.publish("child_command", { turnId: "turn-public" }); + const idle = await publisher.publish("idle", { turnId: "turn-public" }); + expect(idle).toMatchObject({ + state: "idle", + producerEpoch: "epoch-public", + sequence: 3, + turnId: "turn-public", + source: "codex", + }); + + await expect(connectActivityPublisher(name, { + producerEpoch: "epoch-competing", + source: "claude", + })).rejects.toThrow("activity lease already held"); + expect((await queryStats(name)).activity).toEqual(idle); + + publisher.close(); + expect(await waitForActivityState(name, "unknown")).toMatchObject({ + state: "unknown", + producerEpoch: null, + sequence: 0, + }); + }); }); diff --git a/tests/protocol.test.ts b/tests/protocol.test.ts index 56db597..81b1e02 100644 --- a/tests/protocol.test.ts +++ b/tests/protocol.test.ts @@ -13,6 +13,7 @@ import { encodeScreen, encodeStatus, encodeStatusResponse, + encodeActivity, decodeSize, decodeExit, } from "../src/protocol.ts"; @@ -269,4 +270,21 @@ describe("protocol", () => { }); }); }); + + describe("ACTIVITY", () => { + it("round-trips bounded JSON commands and responses", () => { + const reader = new PacketReader(); + const encoded = encodeActivity({ + op: "claim", + producerEpoch: "epoch-a", + }); + const packets = reader.feed(encoded); + expect(packets).toHaveLength(1); + expect(packets[0].type).toBe(MessageType.ACTIVITY); + expect(JSON.parse(packets[0].payload.toString())).toEqual({ + op: "claim", + producerEpoch: "epoch-a", + }); + }); + }); }); diff --git a/tests/stats-cli.test.ts b/tests/stats-cli.test.ts index 5d9d1e7..4dd774c 100644 --- a/tests/stats-cli.test.ts +++ b/tests/stats-cli.test.ts @@ -118,6 +118,7 @@ describe("pty stats CLI", () => { expect(output).toContain("Clients:"); expect(output).toContain("Process:"); expect(output).toContain("Modes:"); + expect(output).toContain("Activity: unknown"); expect(output).toContain("running"); expect(output).toContain("CPU:"); expect(output).toContain("Memory:"); @@ -147,7 +148,13 @@ describe("pty stats CLI", () => { expect(stats.daemon.resources).toBeDefined(); expect(stats.daemon.resources.rssKb).toBeTypeOf("number"); expect(stats.clients).toBeDefined(); - expect(stats.modes).toBeDefined(); + expect(stats.modes.alternateScreen).toBe(false); + expect(stats.activity).toMatchObject({ + state: "unknown", + producerEpoch: null, + sequence: 0, + }); + expect(stats.activity.generation).toBeTypeOf("string"); }, 15000); it("queries all running sessions when no name given", async () => {