feat(agent): add bounded inspection and guarded submission

This commit is contained in:
Aamer Akhter committed 2026-10-24 22:37:06 -04:00
1 parent 3e06f3c141
commit a24b0ec7bb
30 files changed
+2462 -36

No files matched your search

+148
View File
@@ -0,0 +1,148 @@
/**
* @fileoverview Bounded Codex rollout attribution for agent evidence. Never uses
* newest-mtime, prompt timestamps or cwd alone as proof of session ownership.
*/
import { realpathSync } from 'node:fs';
import { open, opendir } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import { join, resolve } from 'node:path';
const UUID_AT_END = /-([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})\.jsonl$/i;
const HEAD_BYTES = 131072;
export interface CodexAttributionOptions {
codexHome: string;
sessionId: string;
workingDir: string;
resumeSessionId?: string;
}
export interface CodexAttributionBudget {
maxEntries: number;
maxBytes: number;
}
interface Metadata {
id?: string;
cwd?: string;
originator?: string;
}
function canonical(path: string): string | undefined {
try {
return realpathSync.native(path);
} catch {
return undefined;
}
}
/** Match the CLI's relative CODEX_HOME interpretation without exposing env credentials. */
export function resolveEffectiveCodexHome(
sessionHome: string | undefined,
workingDir: string,
env: Readonly<Record<string, string | undefined>> = process.env
): string {
const path = resolve(workingDir, sessionHome || env.CODEX_HOME || join(env.HOME || tmpdir(), '.codex'));
return canonical(path) ?? path;
}
/** Complete metadata only: a prefix or a partially flushed JSONL row proves nothing. */
function strictMetadata(bytes: Buffer): Metadata | undefined {
const rows = bytes.toString('utf8').split('\n');
rows.pop();
let found: Metadata | undefined;
for (const line of rows) {
try {
const row = JSON.parse(line) as { type?: string; payload?: Metadata };
if (row.type !== 'session_meta') continue;
if (
found ||
typeof row.payload?.id !== 'string' ||
typeof row.payload.cwd !== 'string' ||
(row.payload.originator !== undefined && typeof row.payload.originator !== 'string')
)
return undefined;
found = row.payload;
} catch {
/* Incomplete or malformed rows are not identity evidence. */
}
}
return found;
}
/**
* A configured resume and a newly owned rollout conflict rather than pinning an
* old launch forever after /new. TUI resumes of foreign-originator threads cannot
* be authoritatively tracked from this metadata; this returns attributed evidence.
*/
export async function findExactCodexConversation(
options: CodexAttributionOptions,
budget: CodexAttributionBudget = { maxEntries: 8192, maxBytes: 1048576 }
): Promise<{ id: string; path: string } | undefined> {
const directories = [{ path: join(options.codexHome, 'sessions'), depth: 0 }];
const candidates: Array<{ id: string; path: string; meta: Metadata }> = [];
const originator = `codeman_${options.sessionId}`;
const explicit = options.resumeSessionId;
let entries = 0;
let scanned = HEAD_BYTES; // Reserve an independent final verification.
while (directories.length) {
const dir = directories.pop()!;
const handle = await opendir(dir.path).catch(() => undefined);
if (!handle) return undefined;
for await (const entry of handle) {
if (++entries > budget.maxEntries) return undefined;
const path = join(dir.path, entry.name);
if (entry.isDirectory()) {
if (dir.depth >= 8) return undefined;
directories.push({ path, depth: dir.depth + 1 });
} else if (entry.isFile() && entry.name.endsWith('.jsonl')) {
const file = await open(path, 'r').catch(() => undefined);
if (!file) return undefined;
try {
const info = await file.stat();
if (!info.isFile()) return undefined;
const id = entry.name.match(UUID_AT_END)?.[1];
const prefixBytes = Math.min(info.size, 4096);
if (scanned + prefixBytes > budget.maxBytes) return undefined;
scanned += prefixBytes;
const prefix = Buffer.alloc(prefixBytes);
const first = await file.read(prefix, 0, prefixBytes, 0);
// Cheap negative filter only; a match still needs full parsed metadata.
if (
id !== explicit &&
!prefix.subarray(0, first.bytesRead).toString('utf8').includes(JSON.stringify(originator))
)
continue;
const size = Math.min(info.size, HEAD_BYTES);
if (scanned + size > budget.maxBytes) return undefined;
scanned += size;
const bytes = Buffer.alloc(size);
const read = await file.read(bytes, 0, size, 0);
const meta = strictMetadata(bytes.subarray(0, read.bytesRead));
if (!meta || !id || meta.id !== id) continue;
if (id === explicit || meta.originator === originator) candidates.push({ id, path, meta });
} finally {
await file.close();
}
}
}
}
if (candidates.length !== 1) return undefined;
const candidate = candidates[0];
if (
!candidate.meta.cwd ||
canonical(candidate.meta.cwd) !== canonical(options.workingDir) ||
!canonical(options.workingDir)
)
return undefined;
const handle = await open(candidate.path, 'r').catch(() => undefined);
if (!handle) return undefined;
try {
const bytes = Buffer.alloc(HEAD_BYTES);
const read = await handle.read(bytes, 0, bytes.length, 0);
const meta = strictMetadata(bytes.subarray(0, read.bytesRead));
if (
!meta ||
meta.id !== candidate.id ||
meta.cwd !== candidate.meta.cwd ||
meta.originator !== candidate.meta.originator
)
return undefined;
return { id: candidate.id, path: candidate.path };
} finally {
await handle.close();
}
}
+191
View File
@@ -0,0 +1,191 @@
/**
* @fileoverview Bounded append-only JSONL evidence reads for agent inspection.
* Opaque, expiring server-side cursors contain no public filesystem paths. A cursor
* binds the caller, pane generation, attributed conversation and opened file identity.
* Complete records only; an oversized row stalls explicitly rather than losing evidence.
*/
import { open, type FileHandle } from 'node:fs/promises';
import { createHash, randomUUID } from 'node:crypto';
import { isUtf8 } from 'node:buffer';
export const INSPECT_MAX_BYTES = 65536;
export const INSPECT_SCAN_BYTES = 262144;
export interface InspectionScope {
owner: string;
sessionId: string;
generation: string;
conversationId: string;
}
interface CursorState {
scope: string;
path: string;
identity: string;
offset: number;
observedSize: number;
modifiedAt: number;
changedAt: number;
anchor: string;
expiresAt: number;
}
export class InspectionConflict extends Error {}
/** UTF8-safe suffix, also used for explicitly non-incremental pane evidence. */
export function boundedUtf8Tail(text: string, maxBytes: number): string {
// A UTF8 suffix fits within this many UTF16 units. Include a pair of extra
// units so slicing cannot introduce a surrogate replacement inside the result.
const bytes = Buffer.from(text.slice(-maxBytes - 2));
let start = Math.max(0, bytes.length - maxBytes);
while (start < bytes.length && (bytes[start] & 0xc0) === 0x80) start++;
return bytes.subarray(start).toString('utf8');
}
async function anchor(handle: FileHandle, offset: number): Promise<string> {
const count = Math.min(offset, 4096);
const head = Buffer.alloc(count);
const tail = Buffer.alloc(count);
await handle.read(head, 0, count, 0);
await handle.read(tail, 0, count, Math.max(0, offset - count));
return createHash('sha256').update(head).update(tail).digest('hex');
}
export class InspectionReader {
private readonly cursors = new Map<string, CursorState>();
constructor(
private readonly now = Date.now,
private readonly capacity = 1024,
private readonly ttlMs = 900000
) {}
async read(path: string, scope: InspectionScope, maxBytes: number, since?: string) {
if (!Number.isInteger(maxBytes) || maxBytes < 1 || maxBytes > INSPECT_MAX_BYTES)
throw new Error('Invalid byte budget');
const scopeKey = JSON.stringify(scope);
const previous = since ? this.cursors.get(since) : undefined;
if (
since &&
(!previous || previous.expiresAt <= this.now() || previous.scope !== scopeKey || previous.path !== path)
) {
throw new InspectionConflict('Invalid or expired inspection cursor; inspect without since to start again');
}
const handle = await open(path, 'r');
try {
const before = await handle.stat();
if (!before.isFile()) throw new InspectionConflict('Inspection source is not a regular file');
const identity = `${before.dev}:${before.ino}:${before.birthtimeMs}`;
let offset = previous?.offset ?? Math.max(0, before.size - INSPECT_SCAN_BYTES + 1);
if (
previous &&
(identity !== previous.identity ||
before.size < previous.observedSize ||
(before.size === previous.observedSize &&
(before.mtimeMs !== previous.modifiedAt || before.ctimeMs !== previous.changedAt)) ||
(await anchor(handle, offset)) !== previous.anchor)
) {
throw new InspectionConflict('Inspection cursor source changed; inspect without since to start again');
}
let startsAtBoundary = offset === 0;
if (!previous && offset > 0) {
const preceding = Buffer.alloc(1);
await handle.read(preceding, 0, 1, offset - 1);
startsAtBoundary = preceding[0] === 10;
}
const scan = Buffer.alloc(
Math.min(INSPECT_SCAN_BYTES - (!previous && offset > 0 ? 1 : 0), Math.max(0, before.size - offset))
);
const { bytesRead } = await handle.read(scan, 0, scan.length, offset);
let selectedStart = 0;
let consumed = 0;
let blockedReason: string | undefined;
let omittedBeforeBytes = 0;
if (!previous) {
const leading = startsAtBoundary ? 0 : scan.indexOf(10) + 1;
const completeEnd = scan.lastIndexOf(10, bytesRead - 1) + 1;
if (!startsAtBoundary && leading === 0) {
// The tail lies wholly inside one unfinished oversized record. Its
// beginning is not known: retain the known file boundary rather than
// manufacture a cursor pointing into a JSON/UTF8 fragment.
offset = 0;
blockedReason = 'record-exceeds-scan-limit';
} else {
selectedStart = completeEnd;
consumed = completeEnd;
while (selectedStart > leading) {
const recordStart = Math.max(leading, scan.lastIndexOf(10, selectedStart - 2) + 1);
if (completeEnd - recordStart > maxBytes) {
if (selectedStart === completeEnd) {
selectedStart = recordStart;
consumed = recordStart;
blockedReason = 'record-exceeds-output-budget';
}
break;
}
selectedStart = recordStart;
}
if (completeEnd < bytesRead && !blockedReason) blockedReason = 'partial-record';
// If the only known boundary is the end of a skipped leading row,
// the remainder is a partial record and stays pending there.
if (completeEnd < leading) selectedStart = consumed = leading;
omittedBeforeBytes = offset + selectedStart;
}
} else {
while (consumed < bytesRead) {
const end = scan.indexOf(10, consumed);
if (end < 0) {
blockedReason = bytesRead === INSPECT_SCAN_BYTES ? 'record-exceeds-scan-limit' : 'partial-record';
break;
}
if (end + 1 > maxBytes) {
blockedReason = 'record-exceeds-output-budget';
break;
}
consumed = end + 1;
}
}
const evidenceBytes = scan.subarray(selectedStart, consumed);
if (!isUtf8(evidenceBytes)) throw new InspectionConflict('Inspection source contains invalid UTF8');
for (const row of evidenceBytes.toString('utf8').split('\n').slice(0, -1)) {
try {
JSON.parse(row);
} catch {
throw new InspectionConflict('Inspection source contains invalid JSONL');
}
}
const after = await handle.stat();
if (
after.size < before.size ||
after.ino !== before.ino ||
(after.size === before.size && (after.mtimeMs !== before.mtimeMs || after.ctimeMs !== before.ctimeMs))
)
throw new InspectionConflict('Inspection source changed during read');
const nextAnchor = await anchor(handle, offset + consumed);
const cursor = randomUUID();
// No await between capacity enforcement and insertion: concurrent reads
// must not all reserve the same remaining slot while hashing anchors.
for (const [key, value] of this.cursors) if (value.expiresAt <= this.now()) this.cursors.delete(key);
while (this.cursors.size >= this.capacity) this.cursors.delete(this.cursors.keys().next().value!);
this.cursors.set(cursor, {
scope: scopeKey,
path,
identity,
offset: offset + consumed,
observedSize: before.size,
modifiedAt: before.mtimeMs,
changedAt: before.ctimeMs,
anchor: nextAnchor,
expiresAt: this.now() + this.ttlMs,
});
return {
evidence: evidenceBytes.toString('utf8'),
cursor,
omittedBeforeBytes,
truncated: omittedBeforeBytes > 0 || offset + consumed < before.size,
blockedReason,
scannedBytes: bytesRead,
};
} finally {
await handle.close();
}
}
clear(): void {
this.cursors.clear();
}
}
+183
View File
@@ -0,0 +1,183 @@
/**
* @fileoverview Short-lived, bounded guarded prompt receipts. No prompt content is
* retained: only caller/session identity, lifecycle/input revisions and SHA-256.
* A receipt is consumed synchronously before verification/delivery, including failures.
*/
import { createHash, randomUUID } from 'node:crypto';
import { stripAnsi } from './utils/regex-patterns.js';
import { visibleWidth } from './tui/tui-ansi.js';
export const STAGE_MAX_BYTES = 8192;
export const promptSha256 = (text: string): string => createHash('sha256').update(text).digest('hex');
export interface GuardedInputSnapshot {
generation: string;
revision: number;
composerSha: string;
}
export interface GuardedInputTarget {
id: string;
snapshot(): GuardedInputSnapshot | null;
write(text: string): Promise<boolean>;
submit(expected: GuardedInputSnapshot): Promise<boolean>;
}
interface Receipt extends GuardedInputSnapshot {
owner: string;
sessionId: string;
expiresAt: number;
sha256: string;
}
export class GuardedInputConflict extends Error {}
/**
* Conservatively read a Codex composer: exactly one visible, unwrapped row and a
* known footer. Never remove internal spaces, infer an unknown shell prompt, or
* interpret dialog options as authority. An unsupported frame yields no proof.
*/
export function readGuardedCodexComposer(screen: string): string | null {
const lines = stripAnsi(screen)
.split('\n')
.map((line) => line.trimEnd());
const rows = lines.map((line, index) => ({ line, index })).filter((row) => /^\s*›(?:\s|$)/.test(row.line));
const last = rows.at(-1);
if (!last) return null;
const below = lines.slice(last.index + 1).filter((line) => line.trim());
if (
below.length === 0 ||
below.length > 3 ||
below.some(
(line) =>
!/^\s+(?:\d+% context left|\? for shortcuts|← for agents|\S+ (?:default|low|medium|high|xhigh|max) · )/.test(
line
)
)
)
return null;
// Preserve exact printable content; only SGR-dim placeholder/suggestion text is excluded.
const raw = screen.split('\n')[last.index];
let dim = false;
let visible = '';
// eslint-disable-next-line no-control-regex
for (const token of raw.split(/(\x1b\[[0-9;]*m)/)) {
// eslint-disable-next-line no-control-regex
if (/^\x1b\[/.test(token)) {
const codes = token.slice(2, -1).split(';').map(Number);
for (let i = 0; i < codes.length; i++) {
// Extended color payloads may contain 0, 2 or 22; those are not SGR modes.
if (codes[i] === 38 || codes[i] === 48 || codes[i] === 58) {
i += codes[i + 1] === 2 ? 4 : codes[i + 1] === 5 ? 2 : 0;
} else if (codes[i] === 0 || codes[i] === 22) dim = false;
else if (codes[i] === 2) dim = true;
}
} else if (!dim) visible += token;
}
return stripAnsi(visible).trimStart().slice(2).trimEnd();
}
/** Require the capture's caret at the exact end of the unwrapped composer. */
export function readGuardedCodexSnapshot(snapshot: string): string | null {
// formatPaneSnapshot emits one absolute row address per line, then the caret.
// Plain captures lack caret evidence and cannot prove invisible trailing edits.
// eslint-disable-next-line no-control-regex
const positions = [...snapshot.matchAll(/\x1b\[(\d+);(\d+)H/g)];
if (positions.length < 2) return null;
const caret = positions.at(-1)!;
if (caret.index! + caret[0].length !== snapshot.length) return null;
const rows: string[] = [];
for (let i = 0; i < positions.length - 1; i++) {
const position = positions[i];
if (Number(position[1]) !== i + 1 || position[2] !== '1') return null;
rows.push(snapshot.slice(position.index! + position[0].length, positions[i + 1].index));
}
const composer = readGuardedCodexComposer(rows.join('\n'));
if (composer === null) return null;
let composerRow = -1;
for (let i = rows.length - 1; i >= 0; i--)
if (/^\s*›(?:\s|$)/.test(stripAnsi(rows[i]))) {
composerRow = i;
break;
}
const prefix = stripAnsi(rows[composerRow]).match(/^(\s*›[ \u00a0])/);
if (!prefix || Number(caret[1]) !== composerRow + 1 || Number(caret[2]) !== visibleWidth(prefix[1] + composer) + 1)
return null;
return composer;
}
export class GuardedStageStore {
private readonly receipts = new Map<string, Receipt>();
private readonly busy = new Set<string>();
constructor(
private readonly now = Date.now,
private readonly capacity = 1024,
private readonly ttlMs = 120000
) {}
async stage(target: GuardedInputTarget, owner: string, message: string) {
// eslint-disable-next-line no-control-regex
const hasControls = /[\x00-\x1f\x7f-\x9f]/.test(message);
if (
!message ||
message !== message.trimEnd() ||
Buffer.byteLength(message) > STAGE_MAX_BYTES ||
Buffer.from(message).toString('utf8') !== message ||
hasControls
)
throw new GuardedInputConflict(
'Stage requires printable single-line input within 8192 UTF8 bytes, without trailing whitespace'
);
const key = target.id;
if (this.busy.has(key)) throw new GuardedInputConflict('Another guarded input operation is in progress');
this.busy.add(key);
try {
const before = target.snapshot();
if (!before || before.composerSha !== promptSha256(''))
throw new GuardedInputConflict('An empty, verifiable composer is required');
for (const [id, receipt] of this.receipts)
if (receipt.expiresAt <= this.now() || receipt.sessionId === target.id) this.receipts.delete(id);
// Refuse capacity before typing. Do not evict a caller's still-live receipt.
// Include in-flight stages (this operation already owns one busy slot),
// so different sessions cannot type concurrently into the same last slot.
if (this.receipts.size + this.busy.size > this.capacity)
throw new GuardedInputConflict('Guarded stage receipt capacity reached');
if (!(await target.write(message)))
throw new GuardedInputConflict('Staging delivery failed; inspect the pane before trying again');
const after = target.snapshot();
if (!after || after.generation !== before.generation || after.revision !== before.revision + 1)
throw new GuardedInputConflict('Input or pane changed while staging; no receipt was issued');
const stageId = randomUUID();
const expiresAt = this.now() + this.ttlMs;
const sha256 = promptSha256(message);
this.receipts.set(stageId, { ...after, owner, sessionId: target.id, expiresAt, sha256 });
return { stageId, sha256, expiresAt, sessionId: target.id, submitted: false as const };
} finally {
this.busy.delete(key);
}
}
async submit(target: GuardedInputTarget, owner: string, stageId: string, confirmSha: string) {
const receipt = this.receipts.get(stageId);
if (!receipt || receipt.owner !== owner || receipt.sessionId !== target.id)
throw new GuardedInputConflict('Unknown, used, or foreign stage receipt');
// One shot, even a wrong hash or unprovable composer. No retry of ambiguous Enter.
this.receipts.delete(stageId);
if (this.busy.has(target.id)) throw new GuardedInputConflict('Another guarded input operation is in progress');
this.busy.add(target.id);
try {
if (receipt.expiresAt <= this.now() || receipt.sha256 !== confirmSha)
throw new GuardedInputConflict('Expired stage receipt or confirmation hash mismatch');
const current = target.snapshot();
if (
!current ||
current.generation !== receipt.generation ||
current.revision !== receipt.revision ||
current.composerSha !== receipt.sha256
)
throw new GuardedInputConflict('Staged composer, input, or pane identity changed; nothing submitted');
if (!(await target.submit(current)))
throw new GuardedInputConflict('Submission delivery failed or ambiguous; receipt consumed, inspect the pane');
return { stageId, sessionId: target.id, submitted: true as const };
} finally {
this.busy.delete(target.id);
}
}
clear(): void {
this.receipts.clear();
this.busy.clear();
}
}
+138
View File
@@ -31,6 +31,10 @@
*/
import http from 'node:http';
import https from 'node:https';
import { open, stat } from 'node:fs/promises';
import { isUtf8 } from 'node:buffer';
import { STAGE_MAX_BYTES } from './agent-staging.js';
import { INSPECT_MAX_BYTES } from './agent-inspection.js';
import type { Command } from 'commander';
import {
basicAuthHeader,
@@ -757,6 +761,103 @@ export async function agentWait(deps: AgentDeps, options: WaitOptions): Promise<
return waitExitCode(data?.wait);
}
/** Bounded incremental evidence, with an opaque server-owned cursor. */
export async function agentInspect(
deps: AgentDeps,
options: { id: string; maxBytes: number; since?: string }
): Promise<number> {
if (!Number.isInteger(options.maxBytes) || options.maxBytes < 1 || options.maxBytes > INSPECT_MAX_BYTES)
return fail(deps, '--max-bytes must be between 1 and 65536', EXIT.refused);
const target = await resolveSessionId(deps, options.id);
if ('error' in target) return fail(deps, target.error, target.code);
const response = await deps.request(deps.ctx, {
method: 'GET',
path: `/api/v1/sessions/${encodeURIComponent(target.id)}/inspect`,
query: { maxBytes: options.maxBytes, since: options.since },
});
if (!response.json?.success) return fail(deps, describeFailure(response));
if (deps.json) emitJson(deps, response.json.data);
else {
const data = response.json.data as {
source?: string;
evidence?: string;
truncated?: boolean;
cursor?: string;
blockedReason?: string;
};
deps.io.out(
`[${data.source ?? 'unknown'}]${data.truncated ? ' truncated' : ''}${data.blockedReason ? ` (${data.blockedReason})` : ''}`
);
deps.io.out(data.evidence ?? '');
if (data.cursor) deps.io.out(`cursor: ${data.cursor}`);
}
return EXIT.ok;
}
/** Stage exactly the printable UTF8 file contents; never sends Enter. */
export async function agentStage(deps: AgentDeps, options: { id: string; messageFile: string }): Promise<number> {
if (isSelfSession(deps.ctx.selfId, options.id)) return fail(deps, `refusing: ${options.id} is me`, EXIT.refused);
let message: string;
try {
const info = await stat(options.messageFile);
if (!info.isFile() || info.size > STAGE_MAX_BYTES)
return fail(deps, 'message file must be a regular file within 8192 UTF8 bytes', EXIT.refused);
const handle = await open(options.messageFile, 'r');
try {
const buffer = Buffer.alloc(STAGE_MAX_BYTES + 1);
const { bytesRead } = await handle.read(buffer, 0, buffer.length, 0);
const bytes = buffer.subarray(0, bytesRead);
if (bytesRead > STAGE_MAX_BYTES || !isUtf8(bytes))
return fail(deps, 'message file must contain at most 8192 bytes of UTF8', EXIT.refused);
message = bytes.toString('utf8');
} finally {
await handle.close();
}
} catch (error) {
return fail(deps, `cannot read message file: ${getErrorMessage(error)}`, EXIT.refused);
}
const refusal = inputRefusal(message);
if (refusal || message !== message.trimEnd())
return fail(deps, refusal ?? 'stage refuses trailing whitespace', EXIT.refused);
const target = await resolveSessionId(deps, options.id);
if ('error' in target) return fail(deps, target.error, target.code);
const response = await deps.request(deps.ctx, {
method: 'POST',
path: `/api/v1/sessions/${encodeURIComponent(target.id)}/stage`,
body: { message },
});
if (!response.json?.success) return fail(deps, describeFailure(response));
if (deps.json) emitJson(deps, response.json.data);
else {
const data = response.json.data as { stageId: string; sha256: string; expiresAt: number };
deps.io.out(
`staged (no Enter): ${data.stageId}\nsha256: ${data.sha256}\nexpires: ${new Date(data.expiresAt).toISOString()}`
);
}
return EXIT.ok;
}
/** Submit one verified staged composer; failure never automatically replays Enter. */
export async function agentSubmit(
deps: AgentDeps,
options: { id: string; stageId: string; confirmSha: string }
): Promise<number> {
if (isSelfSession(deps.ctx.selfId, options.id)) return fail(deps, `refusing: ${options.id} is me`, EXIT.refused);
if (!/^[a-f0-9]{64}$/.test(options.confirmSha) || !/^[a-f0-9-]{36}$/.test(options.stageId))
return fail(deps, 'submit requires a stage UUID and its exact SHA-256', EXIT.refused);
const target = await resolveSessionId(deps, options.id);
if ('error' in target) return fail(deps, target.error, target.code);
const response = await deps.request(deps.ctx, {
method: 'POST',
path: `/api/v1/sessions/${encodeURIComponent(target.id)}/submit`,
body: { stageId: options.stageId, confirmSha: options.confirmSha },
});
if (!response.json?.success) return fail(deps, describeFailure(response));
if (deps.json) emitJson(deps, response.json.data);
else deps.io.out('Enter delivered once; inspect or wait for acceptance (no automatic retry).');
return EXIT.ok;
}
export interface ReadOptions {
id: string;
/** Bytes of raw terminal to fetch; ANSI is stripped for humans. */
@@ -1169,6 +1270,43 @@ export function registerAgentCommands(program: Command): Command {
)
);
agent
.command('inspect <id>')
.description('Bounded transcript evidence or a labeled pane snapshot; opaque incremental cursors')
.option('--max-bytes <n>', 'UTF8 evidence budget (1–65536)', '8192')
.option('--since <cursor>', 'Resume a cursor from the same owner, pane and transcript')
.option('--json', 'Machine-readable output')
.action((id: string, options: { maxBytes: string; since?: string; json?: boolean }) =>
run(Boolean(options.json), (deps) =>
agentInspect(deps, {
id,
maxBytes: parsePositiveInt(options.maxBytes, 8192, '--max-bytes'),
since: options.since,
})
)
);
agent
.command('stage <id>')
.description('Type a printable single-line UTF8 file without Enter; requires a verifiable local Codex composer')
.requiredOption('--message-file <path>', 'Regular UTF8 file, at most 8192 bytes, no trailing newline')
.option('--json', 'Machine-readable output')
.action((id: string, options: { messageFile: string; json?: boolean }) =>
run(Boolean(options.json), (deps) => agentStage(deps, { id, messageFile: options.messageFile }))
);
agent
.command('submit <id>')
.description('Send Enter once only when the staged composer, pane and hash still match')
.requiredOption('--stage-id <receipt>', 'One-shot stage receipt')
.requiredOption('--confirm-sha <hash>', 'Exact SHA-256 returned by stage')
.option('--json', 'Machine-readable output')
.action((id: string, options: { stageId: string; confirmSha: string; json?: boolean }) =>
run(Boolean(options.json), (deps) =>
agentSubmit(deps, { id, stageId: options.stageId, confirmSha: options.confirmSha })
)
);
agent
.command('read <id>')
.description("Print a session's last answer (as the server reads it for that mode) or, with --tail, its terminal")
+2 -1
View File
@@ -316,7 +316,8 @@ export interface TerminalMultiplexer extends EventEmitter {
/**
* Send input to a session via tmux send-keys.
*/
sendInput(sessionId: string, input: string): Promise<boolean>;
/** Internal conditional Enter guard, checked synchronously before transport dispatch. */
sendInput(sessionId: string, input: string, options?: { beforeDispatch?: () => boolean }): Promise<boolean>;
// ========== Metadata ==========
+147 -15
View File
@@ -129,6 +129,9 @@ import { DEFAULT_TMUX_HISTORY_LIMIT } from './config/terminal-history.js';
import { EXEC_TIMEOUT_MS } from './config/exec-timeout.js';
import { getCli } from './config/cli-registry/registry.js';
import { SubmitVerifier } from './session-submit-verifier.js';
import { promptSha256, readGuardedCodexSnapshot, type GuardedInputSnapshot } from './agent-staging.js';
import { isKeyboardTerminalInput } from './terminal-input-classifier.js';
import { stripAnsi } from './utils/regex-patterns.js';
import { KEEP_WARM_NUDGE } from './prompt-cache-keepwarm.js';
import { compileVersionRegex } from './config/cli-registry/patterns.js';
import { resolveSessionCliVersion } from './utils/cli-resolver.js';
@@ -1308,6 +1311,7 @@ export class Session extends EventEmitter {
/** Set the owning username (used by recovery to restore ownership). */
set owner(username: string | undefined) {
if (this._owner !== username) this._guardedPaneRevision++;
this._owner = username;
}
@@ -1412,11 +1416,13 @@ export class Session extends EventEmitter {
* tmux session whose record is about to be deleted (Ark0N/Codeman#446).
*/
markClosing(closing: boolean): void {
if (this._closing !== closing) this._guardedPaneRevision++;
this._closing = closing;
}
/** Run one pane start, attach or relaunch with {@link paneLifecycleInFlight} raised. */
private async _withPaneLifecycle<T>(op: () => Promise<T>): Promise<T> {
this._guardedPaneRevision++;
this._paneLifecycleOps++;
try {
return await op();
@@ -1443,6 +1449,7 @@ export class Session extends EventEmitter {
* Ark0N/Codeman#446 review; both now persist on their success path.
*/
private clearPaneExitForNewPane(): void {
this._guardedPaneRevision++;
this.setPaneExit(undefined);
if (this._muxSession) this._mux?.clearPaneExit?.(this._muxSession.muxName);
}
@@ -4206,6 +4213,11 @@ export class Session extends EventEmitter {
prompt: string,
options?: { model?: string; onProgress?: (info: { tokens?: number; status?: string }) => void }
): Promise<{ result: string; cost: number }> {
if (this._guardedSubmitInFlight) {
this._guardedInputRevision++;
throw new Error('Guarded submission in progress');
}
this._guardedPaneRevision++;
return new Promise((resolve, reject) => {
if (this.ptyProcess) {
reject(new Error('Session already has a running process'));
@@ -4733,7 +4745,109 @@ export class Session extends EventEmitter {
* discards the data, but it used to do so with no signal at all — which is how
* input could disappear while the caller believed it had been delivered.
*/
private readonly _inspectionInstance = uuidv4();
private _guardedPaneRevision = 0;
private _guardedInputRevision = 0;
private _guardedPendingWrites = 0;
private _guardedSubmitInFlight = false;
/** Ephemeral identity, never persisted or based on a potentially reused pid alone. */
get inspectionGeneration(): string {
return `${this._inspectionInstance}:${this._guardedPaneRevision}:${this.owner ?? ''}`;
}
get guardedInputPendingWrites(): number {
return this._guardedPendingWrites;
}
/** Expose only the path selection needed for attributed Codex evidence, never credentials. */
getCodexHomeOverride(): string | undefined {
return this._envOverrides?.CODEX_HOME || undefined;
}
private _observeGuardedKeyboardInput(data: string): void {
if (data && isKeyboardTerminalInput(data)) this._guardedInputRevision++;
}
guardedInputSnapshot(): GuardedInputSnapshot | null {
return this._readGuardedInputSnapshot(false);
}
private _readGuardedInputSnapshot(allowLease: boolean): GuardedInputSnapshot | null {
if (
this._isStopped ||
this._closing ||
this.paneLifecycleInFlight ||
this._guardedPendingWrites ||
(!allowLease && this._guardedSubmitInFlight) ||
this.remote ||
this.docker ||
this.paneExit ||
getCli(this.mode)?.capabilities.transcript !== 'codex-rollout' ||
!this._muxSession
)
return null;
const screen = this._mux?.capturePaneBuffer?.(this._muxSession.muxName);
if (!screen) return null;
// tmux's replay has CUP addresses rather than line breaks. Read the SAME
// fresh frame for dialog/work chrome and exact composer/caret proof.
// eslint-disable-next-line no-control-regex
const plain = stripAnsi(screen.replace(/\x1b\[\d+;\d+H/g, '\n'));
const working = this._workingLinePattern();
working.lastIndex = 0;
if (this._paneShowsBlocked(plain) || working.test(plain) || this._paneAwaitsWorkers(plain)) return null;
const composer = readGuardedCodexSnapshot(screen);
return composer === null
? null
: {
generation: this.inspectionGeneration,
revision: this._guardedInputRevision,
composerSha: promptSha256(composer),
};
}
/** Stage through ordinary delivery, after cancelling any older Enter retry timer. */
async stageGuardedInput(message: string): Promise<boolean> {
if (/[\r\n]/.test(message)) return false;
if (this._guardedPendingWrites || this._guardedSubmitInFlight) {
this._guardedInputRevision++;
return false;
}
this._submitVerifier?.cancel();
return this.writeViaMux(message, { fromUser: true });
}
/** One conditional Enter. An ambiguous acknowledgement is never replayed. */
async submitGuardedInput(expected: GuardedInputSnapshot): Promise<boolean> {
const matches = (current: GuardedInputSnapshot | null): boolean =>
!!current &&
current.generation === expected.generation &&
current.revision === expected.revision &&
current.composerSha === expected.composerSha;
if (!matches(this.guardedInputSnapshot()) || !this._mux || !this._muxSession) return false;
this._guardedSubmitInFlight = true;
let dispatched = false;
try {
const sent = await this._mux.sendInput(this.id, '\r', {
beforeDispatch: () => {
if (!matches(this._readGuardedInputSnapshot(true))) return false;
this._guardedInputRevision++;
dispatched = true;
return true;
},
});
if (sent && dispatched) this._emitSubmittedPrompt(this._trackSubmit('\r', { fromUser: true }));
return sent && dispatched;
} finally {
this._guardedSubmitInFlight = false;
}
}
/** Lease/pending boundary for named keys whose bytes bypass the ordinary write methods. */
async performAuxiliaryInput(operation: () => Promise<void>): Promise<void> {
this._guardedInputRevision++;
if (this._guardedSubmitInFlight) throw new Error('Guarded submission in progress');
this._guardedPendingWrites++;
try {
await operation();
} finally {
this._guardedPendingWrites--;
}
}
write(data: string, options: SessionWriteOptions = {}): boolean {
this._observeGuardedKeyboardInput(data);
if (this._guardedSubmitInFlight) return false;
const submittedPrompt = this._trackSubmit(data, options);
if (!this.ptyProcess) return false;
this.ptyProcess.write(data);
@@ -4782,6 +4896,7 @@ export class Session extends EventEmitter {
* separator. Reports submissions like a write would (a line feed never is one).
*/
trackUserInput(data: string): void {
this._observeGuardedKeyboardInput(data);
if (!this._acceptsPrompts) return;
this._emitSubmittedPrompt(this._submittedPromptTracker.feed(data));
}
@@ -4803,6 +4918,7 @@ export class Session extends EventEmitter {
* their prompt must count as real and alert normally.
*/
markPromptSubmitted(prompt?: string): void {
this._guardedInputRevision++;
if (prompt !== undefined && prompt.trim() !== KEEP_WARM_NUDGE) this._keepAliveTurnUntil = 0;
this._stampSubmit();
}
@@ -4898,22 +5014,29 @@ export class Session extends EventEmitter {
* ```
*/
async writeViaMux(data: string, options: SessionWriteOptions = {}): Promise<boolean> {
const submittedPrompt = this._trackSubmit(data, options);
if (this._mux && this._muxSession) {
const sent = await this._mux.sendInput(this.id, data);
if (sent) {
this._emitSubmittedPrompt(submittedPrompt);
this._verifySubmitted(data);
this._observeGuardedKeyboardInput(data);
if (this._guardedSubmitInFlight) return false;
this._guardedPendingWrites++;
try {
const submittedPrompt = this._trackSubmit(data, options);
if (this._mux && this._muxSession) {
const sent = await this._mux.sendInput(this.id, data);
if (sent) {
this._emitSubmittedPrompt(submittedPrompt);
this._verifySubmitted(data);
}
return sent;
}
return sent;
// Fallback to PTY write
if (this.ptyProcess) {
this.ptyProcess.write(data);
this._emitSubmittedPrompt(submittedPrompt);
return true;
}
return false;
} finally {
this._guardedPendingWrites--;
}
// Fallback to PTY write
if (this.ptyProcess) {
this.ptyProcess.write(data);
this._emitSubmittedPrompt(submittedPrompt);
return true;
}
return false;
}
/**
@@ -4941,7 +5064,10 @@ export class Session extends EventEmitter {
this._isStopped || !this._mux || !this._muxSession
? null
: this._mux.capturePaneText?.(this._muxSession.muxName),
sendEnter: () => this._mux?.sendInput(this.id, '\r'),
sendEnter: () => {
if (this._guardedSubmitInFlight) this._submitVerifier?.cancel();
return this.writeViaMux('\r');
},
// ⚠ NO fallback glyph here, unlike the screen-reading probe elsewhere in this file.
// Only claude, codex, pi, opencode, omp, gemini and antigravity declare a promptGlyph; the other modes would fall back
// to claude's `❯`, which is ALSO starship's default shell prompt (and pure's, and
@@ -4970,6 +5096,7 @@ export class Session extends EventEmitter {
* `resize()` skipped a real resize that happened to match them.
*/
private _notePtySpawnGeometry(cols: number, rows: number): void {
this._guardedPaneRevision++;
this._ptyCols = cols;
this._ptyRows = rows;
}
@@ -5092,6 +5219,7 @@ export class Session extends EventEmitter {
}
const dimsChanged = cols !== this._ptyCols || rows !== this._ptyRows;
if (this.ptyProcess && (dimsChanged || options.force)) {
this._guardedPaneRevision++;
this._ptyCols = cols;
this._ptyRows = rows;
if (!IS_TEST_MODE && this._mux && this._muxSession) {
@@ -5108,6 +5236,10 @@ export class Session extends EventEmitter {
// Legacy method for sending input - wraps runPrompt
async sendInput(input: string): Promise<void> {
if (this._guardedSubmitInFlight) {
this._guardedInputRevision++;
throw new Error('Guarded submission in progress');
}
this._status = 'busy';
this._markActivity(true);
this.runPrompt(input).catch((err) => {
+62
View File
@@ -0,0 +1,62 @@
/** @fileoverview Distinguish complete terminal focus/mouse reports from keyboard-bearing input. */
const FOCUS_REPORT_LENGTH = 3;
const X10_REPORT_LENGTH = 6;
function sgrMouseReportLength(rest: string): number {
if (!rest.startsWith('\x1b[<')) return 0;
let offset = 3;
for (let field = 0; field < 3; field += 1) {
const digitStart = offset;
while (offset < rest.length && rest.charCodeAt(offset) >= 48 && rest.charCodeAt(offset) <= 57) {
offset += 1;
}
if (offset === digitStart) return 0;
if (field < 2) {
if (rest[offset] !== ';') return 0;
offset += 1;
}
}
return rest[offset] === 'M' || rest[offset] === 'm' ? offset + 1 : 0;
}
function x10MouseReportLength(rest: string): number {
if (!rest.startsWith('\x1b[M') || rest.length < X10_REPORT_LENGTH) return 0;
const button = rest.charCodeAt(3);
const x = rest.charCodeAt(4);
const y = rest.charCodeAt(5);
if (button < 32 || button > 255 || x < 33 || x > 255 || y < 33 || y > 255) return 0;
return X10_REPORT_LENGTH;
}
function terminalReportLengthAt(data: string, offset: number): number {
const rest = data.slice(offset);
if (rest.startsWith('\x1b[I') || rest.startsWith('\x1b[O')) return FOCUS_REPORT_LENGTH;
const sgrLength = sgrMouseReportLength(rest);
if (sgrLength > 0) return sgrLength;
return x10MouseReportLength(rest);
}
/**
* Whether terminal input contains any keyboard-bearing bytes.
*
* Returns false only when every byte is consumed as an allowlisted mouse or
* focus report. Unknown, malformed, mixed, and empty input fails closed.
*/
export function isKeyboardTerminalInput(data: string): boolean {
if (data.length === 0) return true;
let offset = 0;
while (offset < data.length) {
const reportLength = terminalReportLengthAt(data, offset);
if (reportLength === 0) return true;
offset += reportLength;
}
return false;
}
+5 -3
View File
@@ -3983,7 +3983,8 @@ export class TmuxManager extends EventEmitter implements TerminalMultiplexer {
* - `Enter` key is sent as a SEPARATE tmux invocation after a small delay
* - Ink (Claude CLI) needs text and Enter split to avoid treating Enter as a newline
*/
async sendInput(sessionId: string, input: string): Promise<boolean> {
async sendInput(sessionId: string, input: string, options?: { beforeDispatch?: () => boolean }): Promise<boolean> {
if (options?.beforeDispatch && input !== '\r') return false;
const session = this.sessions.get(sessionId);
if (!session) {
console.error(
@@ -3996,7 +3997,7 @@ export class TmuxManager extends EventEmitter implements TerminalMultiplexer {
// inputRunner() so a test subclass can record what WOULD reach tmux.
const run = this.inputRunner();
if (!run) {
return true;
return options?.beforeDispatch?.() ?? true;
}
console.log(
@@ -4050,7 +4051,8 @@ export class TmuxManager extends EventEmitter implements TerminalMultiplexer {
`${this.tmux()} send-keys -t "${session.muxName}" -l -- ${shellescape(escapeTmuxTrailingSemicolon(textPart))}`
);
} else if (hasCarriageReturn) {
// Enter only
// Enter only: no await between the guard and dispatch.
if (options?.beforeDispatch && !options.beforeDispatch()) return false;
await run(`${this.tmux()} send-keys -t "${session.muxName}" Enter`);
}
+148
View File
@@ -0,0 +1,148 @@
/**
* @fileoverview Agent-facing bounded inspection and guarded stage/submit endpoints.
* Ownership is checked before any disk or pane read. Native evidence comes only
* from strict conversation attribution; all other modes get labeled snapshots.
*/
import type { FastifyInstance } from 'fastify';
import type { Session } from '../../session.js';
import type { SessionPort, InfraPort } from '../ports/index.js';
import { getCli } from '../../config/cli-registry/registry.js';
import { findExactCodexConversation, resolveEffectiveCodexHome } from '../../agent-codex-identity.js';
import { InspectionReader, InspectionConflict, boundedUtf8Tail } from '../../agent-inspection.js';
import { GuardedStageStore, GuardedInputConflict, type GuardedInputTarget } from '../../agent-staging.js';
import { ApiErrorCode, createErrorResponse } from '../../types.js';
import { findSessionOrFail, getAuthUser, parseBody } from '../route-helpers.js';
import { AgentInspectQuerySchema, AgentStageSchema, AgentSubmitSchema } from '../schemas.js';
export function registerAgentSessionRoutes(app: FastifyInstance, ctx: SessionPort & InfraPort): void {
const reader = new InspectionReader();
const stages = new GuardedStageStore();
app.addHook('onClose', async () => {
reader.clear();
stages.clear();
});
const generation = (session: Session) => session.inspectionGeneration;
const target = (session: Session): GuardedInputTarget => ({
id: session.id,
snapshot: () => {
if (ctx.sessions.get(session.id) !== session) return null;
const snapshot = session.guardedInputSnapshot?.();
return snapshot ? { ...snapshot, generation: generation(session) } : null;
},
write: (text) => session.stageGuardedInput(text),
submit: (expected) => session.submitGuardedInput(expected),
});
app.get('/api/sessions/:id/inspect', async (req, reply) => {
const { id } = req.params as { id: string };
const session = findSessionOrFail(ctx, id, req);
const { maxBytes, since } = parseBody(AgentInspectQuerySchema, req.query);
const initialGeneration = generation(session);
const identity = { sessionId: session.id, mode: session.mode };
const activity = { status: session.status, lastSubmitAt: session.lastSubmitAt || null, observedAt: Date.now() };
try {
if (getCli(session.mode)?.capabilities.transcript === 'codex-rollout' && !session.remote && !session.docker) {
const conversation = await findExactCodexConversation(
{
codexHome: resolveEffectiveCodexHome(session.getCodexHomeOverride(), session.workingDir),
sessionId: session.id,
workingDir: session.workingDir,
resumeSessionId: session.codexConfig?.resumeSessionId,
},
{ maxEntries: 8192, maxBytes: 1048576 }
);
if (generation(session) !== initialGeneration || ctx.sessions.get(id) !== session)
throw new InspectionConflict('Session changed during inspection');
if (conversation) {
const result = await reader.read(
conversation.path,
{
owner: getAuthUser(req).username,
sessionId: id,
generation: initialGeneration,
conversationId: conversation.id,
},
maxBytes,
since
);
if (generation(session) !== initialGeneration || ctx.sessions.get(id) !== session)
throw new InspectionConflict('Session changed during inspection');
return {
success: true,
data: {
source: 'codex-jsonl',
identity: { ...identity, conversationId: conversation.id },
activity,
incremental: true,
maxBytes,
...result,
},
};
}
}
if (since)
throw new InspectionConflict('Attributed transcript unavailable; this cursor cannot resume a pane snapshot');
const frame = session.muxName ? ctx.mux.capturePaneText?.(session.muxName) : null;
const evidence = frame ?? session.terminalBuffer;
return {
success: true,
data: {
source: 'pane-snapshot',
identity,
activity,
incremental: false,
cursor: null,
maxBytes,
evidence: boundedUtf8Tail(evidence, maxBytes),
truncated: evidence.length > maxBytes || Buffer.byteLength(evidence) > maxBytes,
},
};
} catch (error) {
if (error instanceof InspectionConflict)
return reply.code(409).send(createErrorResponse(ApiErrorCode.CONFLICT, error.message));
// Native paths and arbitrary IO errors must never escape into the response.
return reply
.code(409)
.send(createErrorResponse(ApiErrorCode.CONFLICT, 'Inspection source unavailable; retry without a cursor'));
}
});
app.post('/api/sessions/:id/stage', async (req, reply) => {
const session = findSessionOrFail(ctx, (req.params as { id: string }).id, req);
const { message } = parseBody(AgentStageSchema, req.body);
try {
const receipt = await stages.stage(target(session), getAuthUser(req).username, message);
return { success: true, data: receipt };
} catch (error) {
return reply
.code(409)
.send(
createErrorResponse(
ApiErrorCode.CONFLICT,
error instanceof GuardedInputConflict
? error.message
: 'Stage delivery failed; inspect the pane before retrying'
)
);
}
});
app.post('/api/sessions/:id/submit', async (req, reply) => {
const session = findSessionOrFail(ctx, (req.params as { id: string }).id, req);
const { stageId, confirmSha } = parseBody(AgentSubmitSchema, req.body);
try {
return {
success: true,
data: await stages.submit(target(session), getAuthUser(req).username, stageId, confirmSha),
};
} catch (error) {
return reply
.code(409)
.send(
createErrorResponse(
ApiErrorCode.CONFLICT,
error instanceof GuardedInputConflict
? error.message
: 'Submit delivery failed or ambiguous; receipt consumed, inspect the pane'
)
);
}
});
}
+17 -11
View File
@@ -5,6 +5,7 @@
*/
import { claudeProjectDirName } from '../../case-rename.js';
import { registerAgentSessionRoutes } from './agent-session-routes.js';
import { FastifyInstance, type FastifyRequest } from 'fastify';
import { z } from 'zod';
import { join, dirname, extname, basename, resolve as resolvePath } from 'node:path';
@@ -1072,6 +1073,8 @@ export function registerSessionRoutes(
// ========== Logout ==========
registerAgentSessionRoutes(app, ctx);
app.post('/api/logout', async (req, reply) => {
// Invalidate server-side session token (not just the browser cookie)
const sessionToken = req.cookies[AUTH_COOKIE_NAME];
@@ -2428,17 +2431,20 @@ export function registerSessionRoutes(
try {
// Route through the dedicated Codeman socket — bare `tmux` would target the
// user's default server and never find this session (same #80 regression class).
await new Promise<void>((resolve, reject) => {
execFile(
'tmux',
['-L', ctx.mux.muxSocket, 'send-keys', '-H', '-t', muxName, ...hex],
{ timeout: 5000 },
(err) => {
if (err) reject(err);
else resolve();
}
);
});
await session.performAuxiliaryInput(
() =>
new Promise<void>((resolve, reject) => {
execFile(
'tmux',
['-L', ctx.mux.muxSocket, 'send-keys', '-H', '-t', muxName, ...hex],
{ timeout: 5000 },
(err) => {
if (err) reject(err);
else resolve();
}
);
})
);
} catch (err) {
console.error('[Server] send-key failed:', err);
return createErrorResponse(ApiErrorCode.INTERNAL_ERROR, 'tmux send-keys failed');
+12
View File
@@ -2410,3 +2410,15 @@ export const CustomModelSelectionSchema = z.union([
}),
z.object({ clear: z.literal(true) }),
]);
/** Agent inspection bounds are bytes of evidence, independent of the JSON envelope. */
export const AgentInspectQuerySchema = z
.object({
maxBytes: z.coerce.number().int().min(1).max(65536).default(8192),
since: z.string().uuid().optional(),
})
.strict();
export const AgentStageSchema = z.object({ message: z.string().min(1).max(8192) }).strict();
export const AgentSubmitSchema = z
.object({ stageId: z.string().uuid(), confirmSha: z.string().regex(/^[a-f0-9]{64}$/) })
.strict();