feat(codex): inspect helper activity from an existing session (#3962)

* feat(codex): inspect helper activity from an existing session

* fix(codex): keep the selected helper visible in small terminals

* test(codex): address activity viewer review feedback
This commit is contained in:
Devesh Maheshwari authored and GitHub committed 2026-10-19 21:50:20 -04:00
1 parent e86134e45b
commit f6aacd57ea
17 files changed
+1270

No files matched your search

+45
View File
@@ -0,0 +1,45 @@
name: Codex activity
on:
push:
branches: [main]
paths:
- 'v3/@claude-flow/codex/**'
- 'scripts/smoke-codex-activity.mjs'
- '.github/workflows/codex-activity.yml'
pull_request:
paths:
- 'v3/@claude-flow/codex/**'
- 'scripts/smoke-codex-activity.mjs'
- '.github/workflows/codex-activity.yml'
permissions:
contents: read
concurrency:
group: codex-activity-${{ github.event.pull_request.number || github.ref }}
cancel-in-progress: true
jobs:
activity:
name: Codex activity (${{ matrix.os }})
runs-on: ${{ matrix.os }}
timeout-minutes: 10
strategy:
fail-fast: false
matrix:
os: [ubuntu-latest, macos-latest, windows-latest]
steps:
- uses: actions/checkout@11bd71901bbe5b1630ceea73d27597364c9af683
with:
persist-credentials: false
- uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020
with:
node-version: '22'
- name: Install Codex package dependencies
run: npm install --prefix v3/@claude-flow/codex --workspaces=false --ignore-scripts --legacy-peer-deps --package-lock=false --no-audit --no-fund
- run: npm run build --prefix v3/@claude-flow/codex
- name: Test activity reader and viewer
working-directory: v3/@claude-flow/codex
run: npm test -- --run tests/activity-files.test.ts tests/activity-parser.test.ts tests/activity-viewer.test.ts
- run: node scripts/smoke-codex-activity.mjs
+62
View File
@@ -0,0 +1,62 @@
#!/usr/bin/env node
// Runs the built viewer in child processes with synthetic logs; no model calls.
import assert from 'node:assert/strict';
import { mkdtemp, readFile, realpath, rm, writeFile } from 'node:fs/promises';
import os from 'node:os';
import path from 'node:path';
import { fileURLToPath } from 'node:url';
import { spawnSync } from 'node:child_process';
const cli = fileURLToPath(new URL('../v3/@claude-flow/codex/dist/cli.js', import.meta.url));
const directory = await realpath(await mkdtemp(path.join(os.tmpdir(), 'ruflo-activity-smoke-')));
const parent = path.join(directory, 'parent.jsonl');
const child = path.join(directory, 'child.jsonl');
const row = (type, payload) => JSON.stringify({ type, timestamp: '2026-10-09T00:00:00Z', payload });
const args = ['activity', '--session-file', parent, '--sessions-dir', directory];
function run(extra) {
return spawnSync(process.execPath, [cli, ...args, ...extra], {
encoding: 'utf8', timeout: 15000, env: { ...process.env, NO_COLOR: '1' },
});
}
try {
await writeFile(parent, row('session_meta', { id: 'root', source: 'cli' }) + '\n');
await writeFile(child, [
row('session_meta', { id: 'child', agent_path: '/root/parser-review', source: {
subagent: { thread_spawn: { parent_thread_id: 'root' } },
} }),
row('response_item', { type: 'message', role: 'user', content: 'INHERITED_PRIVATE_SENTINEL' }),
row('event_msg', { type: 'thread_settings_applied', thread_id: 'child' }),
row('response_item', { type: 'message', role: 'user', content: 'Check parser fixtures' }),
row('response_item', { type: 'message', role: 'assistant', phase: 'analysis', content: 'REASONING_SENTINEL' }),
row('response_item', { type: 'function_call', name: 'exec_command', call_id: 'one', arguments: '{"cmd":"npm test"}' }),
row('response_item', { type: 'function_call_output', call_id: 'one', output: { exit_code: 0, output: 'SYNTHETIC_TEST_PASS' } }),
row('event_msg', { type: 'task_complete' }),
].join('\n') + '\n');
const originals = await Promise.all([readFile(parent), readFile(child)]);
const list = run(['--once']);
assert.equal(list.status, 0, list.stderr);
assert.match(list.stdout, /parser-review/);
const detail = run(['--once', '--agent', 'child']);
assert.equal(detail.status, 0, detail.stderr);
assert.match(detail.stdout, /Check parser fixtures/);
assert.match(detail.stdout, /npm test/);
assert.match(detail.stdout, /SYNTHETIC_TEST_PASS/);
assert.doesNotMatch(detail.stdout, /INHERITED_PRIVATE_SENTINEL|REASONING_SENTINEL/);
assert.equal(run(['--once', '--agent', 'unrelated']).status, 1);
assert.equal(run(['--once', '--interval', '0']).status, 1);
const nonTTY = run([]);
assert.equal(nonTTY.status, 1);
assert.match(nonTTY.stderr, /Use --once/);
assert.deepEqual(await readFile(parent), originals[0]);
assert.deepEqual(await readFile(child), originals[1]);
const oversized = row('session_meta', { id: 'HEADER_SENTINEL', source: 'cli' })
.padEnd(64 * 1024 + 1, ' ') + '\n';
await writeFile(parent, oversized);
const rejected = run(['--once']);
assert.equal(rejected.status, 1);
assert.doesNotMatch(rejected.stdout + rejected.stderr, /HEADER_SENTINEL|SYNTHETIC_TEST_PASS/);
assert.equal(await readFile(parent, 'utf8'), oversized);
console.log('Codex activity smoke passed: picker, detail, isolation, invalid input, non-TTY, read-only files, oversized metadata.');
} finally {
await rm(directory, { recursive: true, force: true });
}
+78
View File
@@ -1042,6 +1042,84 @@ npx claude-flow memory optimize --rebuild-index
---
## Inspect an existing Codex session
The `activity` command lets you pick a helper and read its assignment, recorded
state, public progress messages, and tool calls/results. It reads local rollout
files without starting model sessions or sending commands to agents.
From a checkout containing this change:
```bash
npm install --prefix v3/@claude-flow/codex --workspaces=false --ignore-scripts --legacy-peer-deps --package-lock=false
npm run build --prefix v3/@claude-flow/codex
node v3/@claude-flow/codex/dist/cli.js activity \
--session-file /path/to/sessions/parent-rollout.jsonl \
--sessions-dir /path/to/sessions
```
Choose the parent rollout explicitly. Only helpers whose recorded parent chain
leads back to that session appear. The selected directory must contain both the
parent and helper files; use the smallest directory that contains them all.
Use **j/k** or the arrow keys to select a helper, then **Enter** to open it.
In the detail view, **j/k** scroll, **Space** moves down a page, **g** goes to the
top, **G** follows the latest activity, and **b** or **Escape** returns to the list.
Press **q** to quit; the Codex session keeps running.
Use `--agent <thread-id>` to open a specific helper, `--once` to print a snapshot
without an interactive terminal, or `--interval <seconds>` to change the refresh
interval (default: 2 seconds; range: 0.2–60).
These screenshots use made-up session records:
![Agent picker with a parser-review helper](docs/images/activity-picker.png)
![Helper assignment, public progress, command and result](docs/images/activity-detail.png)
### What the viewer can read
The parser supports CLI rollout files with a `session_meta` header, helper ancestry
in `source.subagent.thread_spawn.parent_thread_id`, and an own-thread marker
(`payload.thread_id`) before helper activity. Public message, function/custom tool
call and result records, and supported task/item events are shown. Formats without
these attribution fields may produce an empty list or unavailable activity.
Copied parent history, reasoning, and system/developer messages are excluded.
Ambiguous identities and paths through symlinks are rejected. The viewer strips
terminal controls and masks common credential patterns before shortening previews.
Masking cannot catch every secret in arbitrary command output: check the visible
content before sharing a screenshot.
State is the last **recorded** state, not a heartbeat. Discovery stops after 4,000
entries or eight directory levels; ancestry is limited to 64 links. Each detail
refresh reads at most 1 MiB, lines are limited to 64 KiB, previews to 2,400 characters,
and history to 80 events. The metadata header must include its terminating newline
within the first 64 KiB; an oversized or incomplete header excludes that file.
The screen labels incomplete scans, omitted records and logs that are still being
read. Missing data stays unavailable.
### Check the viewer locally
```bash
cd v3/@claude-flow/codex
npm test -- --run tests/activity-files.test.ts tests/activity-parser.test.ts tests/activity-viewer.test.ts
cd ../../..
node scripts/smoke-codex-activity.mjs
```
The Node smoke check runs the built CLI against synthetic files and checks the
picker, detail output, history isolation, invalid input and read-only access.
The screenshots above were captured from the interactive viewer using synthetic
records; screenshot generation is not part of CI.
### Release integration
Ship the viewer in `@claude-flow/codex` with the next coordinated Ruflo release.
At release, bump the package version and exported `VERSION` together, update the
bundled dependency pins and lockfile, and include the built package in the release
train.
## Related Packages
| Package | Description |
Binary file not shown.

After

Width:  |  Height:  |  Size: 79 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 29 KiB

+95
View File
@@ -0,0 +1,95 @@
import { Command, InvalidArgumentError } from 'commander';
import { realpath } from 'node:fs/promises';
import path from 'node:path';
import { emitKeypressEvents, type Key } from 'node:readline';
import { ActivityViewer } from './viewer.js';
import { sanitize } from './sanitize.js';
export function createActivityCommand(): Command {
return new Command('activity')
.description('Inspect helpers from an existing Codex session (read-only)')
.requiredOption('--session-file <file>', 'Parent CLI rollout JSONL file')
.requiredOption('--sessions-dir <directory>', 'Directory containing parent and helper rollouts')
.option('--agent <id>', 'Open a particular helper thread ID')
.option('--once', 'Print one bounded snapshot and exit (no interactive terminal needed)')
.option('--interval <seconds>', 'Refresh interval, 0.2 to 60 seconds', interval, 2)
.action(async (options: { sessionFile: string; sessionsDir: string; agent?: string; once?: boolean; interval: number }) => {
try {
const root = await realpath(path.resolve(options.sessionsDir));
// Canonicalize the directory only. A selected file through a symlink is rejected.
const file = path.resolve(root, path.relative(path.resolve(options.sessionsDir), path.resolve(options.sessionFile)));
const viewer = new ActivityViewer(root, file);
await viewer.refresh();
if (viewer.list.error) throw new Error(viewer.list.error);
if (options.agent) {
if (!viewer.list.helpers.some(item => item.id === options.agent)) throw new Error('Agent is not an attributable helper of this session.');
viewer.selected = options.agent;
viewer.detail = true;
await viewer.refresh();
}
if (options.once) console.log(viewer.render(100, 200));
else await runTerminal(viewer, options.interval * 1000);
} catch (error) {
console.error(`Activity viewer: ${sanitize(error instanceof Error ? error.message : 'Unable to read session.', 300)}`);
process.exitCode = 1;
}
});
}
function interval(value: string): number {
const parsed = Number(value);
if (!Number.isFinite(parsed) || parsed < 0.2 || parsed > 60) throw new InvalidArgumentError('Choose 0.2 to 60 seconds.');
return parsed;
}
async function runTerminal(viewer: ActivityViewer, intervalMs: number): Promise<void> {
const input = process.stdin;
const output = process.stdout;
if (!input.isTTY || !output.isTTY) throw new Error('Use --once when input or output is not a terminal.');
const wasRaw = input.isRaw;
let stopped = false;
let busy = false;
let finish: () => void = () => {};
const done = new Promise<void>(resolve => { finish = resolve; });
const draw = (): void => {
if (!stopped) output.write(`\x1b[H\x1b[2J${viewer.render(output.columns || 80, output.rows || 24)}`);
};
const close = (): void => { stopped = true; finish(); };
const update = async (): Promise<void> => {
if (busy || stopped) return;
busy = true;
try { await viewer.refresh(); draw(); }
catch { close(); process.exitCode = 1; }
finally { busy = false; }
};
const keypress = (text: string, key: Key): void => {
if (key?.name === 'q' || key?.ctrl && key.name === 'c') { close(); return; }
viewer.key(text === 'G' ? 'G' : key?.name ?? text);
draw();
if (key?.name === 'return') void update();
};
emitKeypressEvents(input);
const timer = setInterval(() => { void update(); }, intervalMs);
try {
input.setRawMode(true);
input.resume();
input.on('keypress', keypress);
output.on('resize', draw);
process.on('SIGINT', close);
process.on('SIGTERM', close);
output.write('\x1b[?1049h\x1b[?25l');
draw();
await done;
} finally {
stopped = true;
clearInterval(timer);
input.off('keypress', keypress);
output.off('resize', draw);
process.off('SIGINT', close);
process.off('SIGTERM', close);
input.setRawMode(wasRaw);
// This command owns stdin. A resumed stream would keep Node alive after q.
input.pause();
output.write('\x1b[?25h\x1b[?1049l');
}
}
+121
View File
@@ -0,0 +1,121 @@
import { constants } from 'node:fs';
import { lstat, open, opendir, realpath } from 'node:fs/promises';
import type { FileHandle } from 'node:fs/promises';
import path from 'node:path';
import { object, type SessionIdentity } from './types.js';
import { sanitize } from './sanitize.js';
export const MAX_LINE = 64 * 1024;
const MAX_ENTRIES = 4000;
/** Only regular files whose resolved path stays inside the selected directory. */
export async function openSession(root: string, file: string): Promise<FileHandle> {
const absolute = path.resolve(file);
const relative = path.relative(root, absolute);
if (!relative || relative === '..' || relative.startsWith(`..${path.sep}`)
|| path.isAbsolute(relative) || await realpath(absolute) !== absolute) {
throw new Error('Session file is outside the selected directory or uses a symlink.');
}
const before = await lstat(absolute);
if (!before.isFile()) throw new Error('Session is not a regular file.');
const handle = await open(absolute, constants.O_RDONLY | (constants.O_NOFOLLOW ?? 0));
try {
const after = await handle.stat();
if (!after.isFile() || after.ino !== before.ino || after.dev !== before.dev
|| await realpath(absolute) !== absolute) throw new Error('Session changed while opening.');
return handle;
} catch (error) {
await handle.close();
throw error;
}
}
export async function readIdentity(root: string, file: string): Promise<SessionIdentity | null> {
const handle = await openSession(root, file);
try {
const buffer = Buffer.alloc(MAX_LINE);
const { bytesRead } = await handle.read(buffer, 0, buffer.length, 0);
const newline = buffer.subarray(0, bytesRead).indexOf(10);
// A valid JSON prefix is not enough: require the complete metadata line.
if (newline < 0) return null;
const record = object(JSON.parse(buffer.subarray(0, newline).toString('utf8')));
const meta = object(record.payload);
if (record.type !== 'session_meta' || typeof meta.id !== 'string' || !meta.id
|| meta.id.length > 300) return null;
const spawn = object(object(object(meta.source).subagent).thread_spawn);
const parent = typeof spawn.parent_thread_id === 'string' ? spawn.parent_thread_id : null;
const rootSession = meta.source === 'cli' && !meta.forked_from_id
&& !meta.parent_thread_id && (!meta.thread_source || meta.thread_source === 'user');
if (!parent && !rootSession) return null; // Excludes guardians and unknown sources.
return {
id: meta.id, parent, file,
name: sanitize(meta.agent_nickname ?? spawn.agent_nickname ?? '', 120),
path: sanitize(meta.agent_path ?? spawn.agent_path ?? meta.id, 200),
};
} finally {
await handle.close();
}
}
export interface SessionList {
root: SessionIdentity | null;
helpers: SessionIdentity[];
partial: boolean;
error: string;
}
export async function listSessions(root: string, rootFile: string, expectedId?: string): Promise<SessionList> {
const empty: SessionList = { root: null, helpers: [], partial: false, error: '' };
let selected: SessionIdentity | null;
try { selected = await readIdentity(root, rootFile); }
catch { return { ...empty, error: 'The selected session file is unavailable.' }; }
if (!selected || selected.parent || (expectedId && selected.id !== expectedId)) {
return { ...empty, error: 'Select a root CLI session with an unchanged identity.' };
}
const sessions = new Map<string, SessionIdentity>([[selected.id, selected]]);
const duplicates = new Set<string>();
const queue = [{ directory: root, depth: 0 }];
let entries = 0;
let partial = false;
while (queue.length && entries < MAX_ENTRIES) {
const current = queue.shift()!;
try {
// Revalidate directories too; never recurse through a symlink.
if (await realpath(current.directory) !== current.directory) { partial = true; continue; }
const directory = await opendir(current.directory);
for await (const entry of directory) {
if (++entries > MAX_ENTRIES) { partial = true; break; }
const file = path.join(current.directory, entry.name);
if (entry.isDirectory()) {
if (current.depth < 8) queue.push({ directory: file, depth: current.depth + 1 });
else partial = true;
} else if (entry.isFile() && entry.name.endsWith('.jsonl') && file !== rootFile) {
try {
const identity = await readIdentity(root, file);
if (!identity) continue;
if (sessions.has(identity.id)) duplicates.add(identity.id);
else sessions.set(identity.id, identity);
} catch { partial = true; }
}
}
} catch { partial = true; }
}
partial ||= queue.length > 0;
if (duplicates.has(selected.id)) return { ...empty, partial: true, error: 'Root session identity is ambiguous.' };
const belongs = (identity: SessionIdentity): boolean => {
const seen = new Set<string>();
let node: SessionIdentity | undefined = identity;
while (node?.parent && seen.size < 64) {
if (seen.has(node.id) || duplicates.has(node.id)) return false;
seen.add(node.id);
if (node.parent === selected.id) return true;
node = sessions.get(node.parent);
}
return false;
};
return {
root: selected,
helpers: [...sessions.values()].filter(belongs).sort((a, b) => a.path.localeCompare(b.path)),
partial: partial || duplicates.size > 0, error: '',
};
}
@@ -0,0 +1,148 @@
import { object, type ActivityEvent, type ActivitySnapshot, type RecordObject } from './types.js';
import { publicText, sanitize } from './sanitize.js';
const MAX_EVENTS = 80;
const CALL_ID = /^[\w.:/-]{1,300}$/;
/** An allowlist parser. Helper logs may begin with a complete copy of parent history. */
export class ActivityParser {
readonly snapshot: ActivitySnapshot = {
available: false, status: 'unknown', assignment: '', events: [], updatedAt: '',
bytesRead: 0, fileSize: 0, partial: false, limited: false, attributed: false, error: '',
};
private first = true;
private verified = false;
private own = false;
private calls = new Map<string, ActivityEvent>();
constructor(private readonly identity: string, private readonly agentPath: string) {}
invalidate(): void {
this.own = false;
this.calls.clear();
this.snapshot.limited = true;
if (this.first) { this.first = false; this.snapshot.error = 'Invalid first session record.'; }
}
consume(value: unknown): void {
const record = object(value);
if (!Object.keys(object(record.payload)).length) { this.invalidate(); return; }
const payload = object(record.payload);
if (this.first) {
this.first = false;
if (record.type !== 'session_meta' || payload.id !== this.identity) {
this.snapshot.error = 'Session identity does not match the selected agent.';
return;
}
this.verified = true;
this.snapshot.available = true;
return; // A helper's first header never opens the attribution boundary.
}
if (!this.verified) return;
if (record.type === 'session_meta') { this.own = false; this.calls.clear(); return; }
if (payload.thread_id !== undefined) {
this.own = payload.thread_id === this.identity;
if (!this.own) this.calls.clear();
}
if (!this.own) return;
this.snapshot.attributed = true;
const time = sanitize(record.timestamp ?? '', 80);
if (record.type === 'event_msg') this.event(payload, time);
if (record.type === 'response_item') this.response(payload, time);
}
private append(event: ActivityEvent): void {
this.snapshot.events.push(event);
if (this.snapshot.events.length > MAX_EVENTS) {
const removed = this.snapshot.events.shift();
for (const [id, item] of this.calls) if (item === removed) this.calls.delete(id);
this.snapshot.limited = true;
}
this.snapshot.updatedAt = event.time;
}
private message(text: string, label: string, time: string): void {
if (!text) return;
const preview = sanitize(text);
if (label === 'Assignment') this.snapshot.assignment = preview;
if (this.snapshot.events.at(-1)?.text === preview) return;
this.append({ kind: 'message', time, label, text: preview, output: '', status: '' });
}
private tool(id: unknown, name: unknown, input: unknown, time: string): void {
if (typeof id !== 'string' || !CALL_ID.test(id) || this.calls.has(id)) return;
const event: ActivityEvent = {
kind: 'tool', time, label: sanitize(name ?? 'tool', 200), text: sanitize(input), output: '', status: 'running',
};
this.append(event);
this.calls.set(id, event);
this.snapshot.status = 'running';
}
private result(id: unknown, output: unknown, time: string, status?: string): void {
const call = typeof id === 'string' ? this.calls.get(id) : undefined;
if (!call) return;
call.output = sanitize(output);
call.status = status ?? resultStatus(output);
this.snapshot.updatedAt = time;
}
private event(payload: RecordObject, time: string): void {
const states: Record<string, string> = {
task_started: 'running', task_complete: 'completed', task_failed: 'failed',
turn_failed: 'failed', turn_aborted: 'interrupted', task_aborted: 'interrupted',
};
const state = typeof payload.type === 'string' ? states[payload.type] : undefined;
if (state) { this.snapshot.status = state; this.snapshot.updatedAt = time; }
if (payload.type === 'user_message') this.message(publicText(payload.message), 'Assignment', time);
if (payload.type === 'agent_message') this.message(publicText(payload.message), 'Assistant', time);
if (payload.type !== 'item_started' && payload.type !== 'item_completed') return;
const item = object(payload.item);
if (item.type === 'AgentMessage' && ['commentary', 'final', 'final_answer'].includes(String(item.phase))) {
this.message(publicText(item.content), 'Assistant', time);
this.snapshot.status = item.phase === 'commentary' ? 'running' : 'completed';
return;
}
if (item.type !== 'CommandExecution' && item.type !== 'McpToolCall') return;
this.tool(item.id, item.type === 'CommandExecution' ? 'command' : `${String(item.server)}.${String(item.tool)}`,
item.command ?? item.arguments ?? '', time);
if (payload.type === 'item_completed') {
const status = ['failed', 'error', 'declined'].includes(String(item.status))
|| (item.exit_code !== undefined && item.exit_code !== null && item.exit_code !== 0) ? 'failed' : 'completed';
this.result(item.id, item.aggregated_output ?? item.result ?? '', time, status);
}
}
private response(payload: RecordObject, time: string): void {
if (payload.type === 'message') {
const phase = payload.channel ?? payload.phase;
if (payload.role === 'user') this.message(publicText(payload.content), 'Assignment', time);
else if (payload.role === 'assistant' && ['commentary', 'final', 'final_answer'].includes(String(phase))) {
this.message(publicText(payload.content), 'Assistant', time);
this.snapshot.status = phase === 'commentary' ? 'running' : 'completed';
}
} else if (payload.type === 'agent_message') {
const incoming = !payload.recipient || payload.recipient === this.agentPath || payload.recipient === this.identity;
this.message(publicText(payload.content), incoming ? 'Assignment' : 'Assistant', time);
} else if (payload.type === 'function_call' || payload.type === 'custom_tool_call') {
this.tool(payload.call_id, payload.name, payload.arguments ?? payload.input ?? '', time);
} else if (payload.type === 'function_call_output' || payload.type === 'custom_tool_call_output') {
this.result(payload.call_id, payload.output ?? '', time);
}
}
}
function resultStatus(output: unknown): string {
if (typeof output === 'string') {
try { return resultStatus(JSON.parse(output)); }
catch {
if (/(?:Script running with cell ID|Process running with session ID)/.test(output)) return 'running';
const code = /(?:Process exited with code|exit_code["']?\s*:)\s*(-?\d+)/.exec(output);
return code && Number(code[1]) !== 0 ? 'failed' : 'returned';
}
}
const value = object(output);
if (value.isError === true || (value.exit_code != null && value.exit_code !== 0)) return 'failed';
if (value.session_id != null && value.exit_code == null) return 'running';
return value.exit_code === 0 ? 'completed' : 'returned';
}
@@ -0,0 +1,95 @@
import { MAX_LINE, openSession } from './files.js';
import { ActivityParser } from './parser.js';
import type { ActivitySnapshot, SessionIdentity } from './types.js';
const READ_BUDGET = 1024 * 1024;
/** Incremental reads keep UI refreshes bounded even when a helper inherits a large log. */
export class ActivityReader {
private parser: ActivityParser;
private offset = 0;
private signature = '';
private mtime = 0;
private head = Buffer.alloc(0);
private anchor = Buffer.alloc(0);
private pending = Buffer.alloc(0);
private skipping = false;
constructor(private readonly root: string, private readonly identity: SessionIdentity,
private readonly budget = READ_BUDGET) {
this.parser = new ActivityParser(identity.id, identity.path);
}
private reset(): void {
this.parser = new ActivityParser(this.identity.id, this.identity.path);
this.offset = 0;
this.pending = Buffer.alloc(0);
this.skipping = false;
this.head = Buffer.alloc(0);
this.anchor = Buffer.alloc(0);
}
async read(): Promise<ActivitySnapshot> {
try {
const handle = await openSession(this.root, this.identity.file);
try {
const stat = await handle.stat();
const signature = `${stat.dev}:${stat.ino}`;
const unchanged = async (expected: Buffer, position: number): Promise<boolean> => {
const buffer = Buffer.alloc(expected.length);
await handle.read(buffer, 0, buffer.length, position);
return buffer.equals(expected);
};
if (signature !== this.signature || stat.size < this.offset
|| (this.mtime !== stat.mtimeMs && stat.size === this.offset)
|| !await unchanged(this.head, 0)
|| !await unchanged(this.anchor, Math.max(0, this.offset - this.anchor.length))) this.reset();
this.signature = signature;
this.mtime = stat.mtimeMs;
const buffer = Buffer.alloc(Math.min(this.budget, Math.max(0, stat.size - this.offset)));
const { bytesRead } = await handle.read(buffer, 0, buffer.length, this.offset);
const data = buffer.subarray(0, bytesRead);
this.offset += bytesRead;
this.feed(data);
this.head = Buffer.alloc(Math.min(256, this.offset));
await handle.read(this.head, 0, this.head.length, 0);
this.anchor = Buffer.alloc(Math.min(128, this.offset));
await handle.read(this.anchor, 0, this.anchor.length, this.offset - this.anchor.length);
const state = this.parser.snapshot;
state.bytesRead = this.offset;
state.fileSize = stat.size;
state.partial = this.offset < stat.size || this.pending.length > 0 || this.skipping;
// Do not expose stale previews after an identity failure.
return structuredClone(state);
} finally { await handle.close(); }
} catch {
this.reset();
this.signature = '';
return { ...this.parser.snapshot, error: 'Agent log unavailable; no cached activity is shown.' };
}
}
private feed(data: Buffer): void {
let start = 0;
while (start < data.length) {
const newline = data.indexOf(10, start);
const end = newline < 0 ? data.length : newline;
const part = data.subarray(start, end);
if (!this.skipping) {
if (this.pending.length + part.length > MAX_LINE) {
this.pending = Buffer.alloc(0);
this.skipping = true;
this.parser.invalidate();
} else this.pending = Buffer.concat([this.pending, part]);
}
if (newline < 0) break;
if (!this.skipping) {
try { this.parser.consume(JSON.parse(this.pending.toString('utf8'))); }
catch { this.parser.invalidate(); }
}
this.pending = Buffer.alloc(0);
this.skipping = false;
start = newline + 1;
}
}
}
@@ -0,0 +1,29 @@
import { stripVTControlCharacters } from 'node:util';
/** Best-effort masking for public previews, not a guarantee that logs are shareable. */
export function sanitize(value: unknown, limit = 2400): string {
let text = typeof value === 'string' ? value : JSON.stringify(value) ?? '';
text = stripVTControlCharacters(text).replace(/[\p{Cc}\p{Cf}\p{Cs}]/gu,
character => character === '\n' || character === '\t' ? character : '');
text = text
.replace(/-----BEGIN [^-]*PRIVATE KEY-----[\s\S]*?(?:-----END [^-]*PRIVATE KEY-----|$)/g,
'[REDACTED PRIVATE KEY]')
.replace(/(?<![\w-])(["']?[\w-]{0,128}(?:api[_-]?key|password|passwd|secret|access[_-]?key|token)[\w-]{0,128}["']?\s*[:=]\s*)(?:"(?:\\.|[^"\\])*"|'(?:\\.|[^'\\])*'|[^\s,;}]+)/gi,
'$1[REDACTED]')
.replace(/\b(?:sk-(?:proj-|ant-)?[\w-]{12,}|(?:ghp|gho|github_pat)_[\w]{12,}|AKIA[A-Z0-9]{16})\b/g,
'[REDACTED KEY]')
.replace(/\bBearer\s+[^\s"',;]+/gi, 'Bearer [REDACTED]')
.replace(/\beyJ[\w-]+\.[\w-]+\.[\w-]+\b/g, '[REDACTED JWT]')
.replace(/(\b[a-z][a-z0-9+.-]{0,31}:\/\/[^\s/:@]+:)[^\s/@]+@/gi, '$1[REDACTED]@');
return text.length > limit ? `${text.slice(0, limit)} ... [truncated]` : text;
}
export function publicText(value: unknown): string {
if (typeof value === 'string') return value;
if (!Array.isArray(value)) return '';
return value.flatMap(block => {
if (block && ['text', 'Text', 'input_text', 'output_text'].includes(block.type)
&& typeof block.text === 'string') return [block.text];
return [];
}).join('\n');
}
@@ -0,0 +1,37 @@
export type RecordObject = Record<string, unknown>;
export function object(value: unknown): RecordObject {
return value !== null && typeof value === 'object' && !Array.isArray(value)
? value as RecordObject : {};
}
export interface SessionIdentity {
id: string;
parent: string | null;
name: string;
path: string;
file: string;
}
export interface ActivityEvent {
kind: 'message' | 'tool';
time: string;
label: string;
text: string;
output: string;
status: string;
}
export interface ActivitySnapshot {
available: boolean;
status: string;
assignment: string;
events: ActivityEvent[];
updatedAt: string;
bytesRead: number;
fileSize: number;
partial: boolean;
limited: boolean;
attributed: boolean;
error: string;
}
@@ -0,0 +1,129 @@
import { listSessions, type SessionList } from './files.js';
import { ActivityReader } from './reader.js';
import { sanitize } from './sanitize.js';
import type { ActivitySnapshot } from './types.js';
export class ActivityViewer {
list: SessionList = { root: null, helpers: [], partial: false, error: '' };
selected: string | null = null;
detail = false;
offset = 0;
follow = false;
activity: ActivitySnapshot | null = null;
private rootId: string | undefined;
private reader: ActivityReader | null = null;
private readerKey = '';
constructor(private readonly directory: string, private readonly sessionFile: string) {}
async refresh(): Promise<void> {
this.list = await listSessions(this.directory, this.sessionFile, this.rootId);
if (this.list.root) this.rootId = this.list.root.id;
if (!this.selected) this.selected = this.list.helpers[0]?.id ?? null;
const helper = this.list.helpers.find(item => item.id === this.selected);
if (!this.detail || !helper || this.list.error) { this.activity = null; return; }
const key = `${helper.file}\0${helper.id}`;
if (key !== this.readerKey) {
this.reader = new ActivityReader(this.directory, helper);
this.readerKey = key;
}
const activity = await this.reader!.read();
if (this.detail && this.selected === helper.id && this.readerKey === key) this.activity = activity;
}
key(name: string): void {
const movement = name === 'down' || name === 'j' ? 1 : name === 'up' || name === 'k' ? -1 : 0;
if (name === 'escape' || name === 'b') {
this.detail = false; this.offset = 0; this.follow = false; this.activity = null;
}
else if (name === 'return' && this.list.helpers.some(item => item.id === this.selected)) {
this.detail = true; this.offset = 0; this.activity = null;
} else if (this.detail) {
if (name === 'G') this.follow = true;
else if (name === 'g') { this.follow = false; this.offset = 0; }
else if (movement || name === 'space') {
this.follow = false;
this.offset = Math.max(0, this.offset + (name === 'space' ? 10 : movement));
}
} else if (movement) {
const current = this.list.helpers.findIndex(item => item.id === this.selected);
const index = Math.max(0, Math.min(this.list.helpers.length - 1, current + movement));
this.selected = this.list.helpers[index]?.id ?? null;
this.activity = null;
}
}
render(width = 100, height = 30, now = Date.now()): string {
const columns = Math.max(10, Math.min(240, width - 1));
const lines: string[] = ['CODEX AGENT ACTIVITY', 'Read-only | state comes from local records', ''];
if (this.list.error) lines.push(this.list.error);
else {
lines.push(`Session: ${sanitize(this.list.root?.id ?? '', 100)}`);
if (this.list.partial) lines.push('Discovery incomplete: scan limit, duplicate identity, or unreadable entries.');
if (!this.detail) {
lines.push('', 'AGENTS - choose one and press Enter');
if (!this.list.helpers.length) lines.push('No attributable helpers found in the selected directory.');
const selected = this.list.helpers.findIndex(item => item.id === this.selected);
const count = Math.max(1, height - 10);
const start = Math.max(0, selected - count + 1);
for (const helper of this.list.helpers.slice(start, start + count)) {
const label = `${helper.id === this.selected ? '>' : ' '} ${helper.path}${helper.name ? ` (${helper.name})` : ''}`;
lines.push(wrap(label, columns).length > 1 ? `${wrap(label, columns - 3)[0]}...` : label);
}
if (this.list.helpers.length > count) lines.push(`Showing ${start + 1}-${Math.min(start + count, this.list.helpers.length)} of ${this.list.helpers.length}`);
} else this.detailLines(lines, now);
}
const wrapped = lines.flatMap(line => wrap(line, columns));
const room = Math.max(1, height - 2);
const maximum = Math.max(0, wrapped.length - room);
// Wrapped headings can fill a short terminal too. Keep the selected row visible.
const pickerOffset = Math.max(0, wrapped.findIndex(line => line.startsWith('> ')) - room + 1);
this.offset = this.detail ? (this.follow ? maximum : Math.min(this.offset, maximum)) : pickerOffset;
const footer = this.detail ? 'b back | j/k scroll | Space page | g top | G latest | q quit'
: 'j/k or arrows select | Enter details | q quit';
return [...wrapped.slice(this.offset, this.offset + room), '', ...wrap(footer, Math.max(10, width - 1)).slice(0, 1)].join('\n');
}
private detailLines(lines: string[], now: number): void {
const helper = this.list.helpers.find(item => item.id === this.selected);
if (!helper) { lines.push('', 'Selected agent is no longer available. Press b to return.'); return; }
lines.push('', helper.path, helper.name, `Agent: ${sanitize(helper.id, 100)}`);
const activity = this.activity;
if (!activity) { lines.push('Reading agent log...'); return; }
if (activity.error) { lines.push(activity.error); return; }
lines.push(`State: ${activity.status} (recorded)`);
const timestamp = Date.parse(activity.updatedAt);
lines.push(`Last recorded activity: ${Number.isFinite(timestamp) ? `${Math.max(0, Math.floor((now - timestamp) / 1000))}s ago` : 'unavailable'}`);
if (activity.partial) lines.push(`Reading: ${activity.bytesRead}/${activity.fileSize} bytes; incomplete records may follow.`);
if (!activity.attributed) lines.push('Waiting for a record attributed to this helper; inherited history is hidden.');
if (activity.limited) lines.push('History is limited: old, oversized, or malformed records were omitted.');
lines.push('', 'ASSIGNMENT', activity.assignment || 'Not available in attributable records.', '', 'ACTIVITY');
for (const event of activity.events) {
lines.push(`${event.time} ${event.label}${event.status ? ` [${event.status}]` : ''}`);
lines.push(event.kind === 'tool' ? `Command / input: ${event.text}` : event.text);
if (event.output) lines.push(`Result: ${event.output}`);
lines.push('');
}
}
}
/** Keep wide characters inside the terminal; unknown emoji sequences may use extra space. */
function wrap(text: string, width: number): string[] {
const result: string[] = [];
for (const line of sanitize(text, 16000).replace(/\t/g, ' ').split('\n')) {
let row = '';
let columns = 0;
for (const character of line) {
const code = character.codePointAt(0)!;
const size = /\p{Mark}/u.test(character) ? 0 :
code >= 0x1100 && (code <= 0x115f || code >= 0x2329 && code <= 0xa4cf
|| code >= 0xac00 && code <= 0xd7a3 || code >= 0xf900 && code <= 0xfaff
|| code >= 0xfe10 && code <= 0xfe6f || code >= 0xff01 && code <= 0xff60
|| code >= 0x1f000) ? 2 : 1;
if (columns + size > width) { result.push(row); row = ''; columns = 0; }
row += character; columns += size;
}
result.push(row);
}
return result;
}
+4
View File
@@ -704,6 +704,10 @@ program.addCommand(createDualModeCommand());
import { createLoopCommand } from './loop/cli.js';
program.addCommand(createLoopCommand());
// Read-only activity from an already-running Codex session.
import { createActivityCommand } from './activity/cli.js';
program.addCommand(createActivityCommand());
// Policy-governed worktree isolation for concurrent Codex writers (ADR-324)
import { CodexWorktreeCoordinator } from './worktrees/index.js';
const worktree = program.command('worktree').description('Prepare, inspect, integrate, or clean registry-owned Codex swarm worktrees');
@@ -0,0 +1,174 @@
import { afterEach, beforeEach, describe, expect, it } from 'vitest';
import { appendFile, rename, rm, symlink, writeFile } from 'node:fs/promises';
import path from 'node:path';
import { listSessions, readIdentity, MAX_LINE } from '../src/activity/files.js';
import { ActivityReader } from '../src/activity/reader.js';
import { fixture, jsonl, message, meta, own, record } from './activity-fixtures.js';
let files: Awaited<ReturnType<typeof fixture>>;
beforeEach(async () => { files = await fixture(); });
afterEach(async () => { await rm(files.directory, { recursive: true, force: true }); });
describe('session discovery', () => {
it('accepts a complete metadata line ending at the prefix boundary', async () => {
const header = JSON.stringify(meta()).padEnd(MAX_LINE - 1, ' ');
await writeFile(files.helper, header + '\n');
expect(await readIdentity(files.directory, files.helper)).toMatchObject({ id: 'helper' });
});
it.each([MAX_LINE, MAX_LINE + 1])('rejects a %i-byte metadata line even with a valid JSON prefix', async (size) => {
const header = JSON.stringify(meta('HEADER_SENTINEL')).padEnd(size, ' ');
expect(() => JSON.parse(header.slice(0, MAX_LINE))).not.toThrow();
await writeFile(files.helper, header + '\n' + jsonl([own(), message('ACTIVITY_SENTINEL')]));
expect(await readIdentity(files.directory, files.helper)).toBeNull();
const listed = await listSessions(files.directory, files.parent);
expect(listed.helpers).toEqual([]);
expect(JSON.stringify(listed)).not.toMatch(/HEADER_SENTINEL|ACTIVITY_SENTINEL/);
});
it('rejects oversized JSON metadata without exposing a truncated string', async () => {
const header = JSON.stringify(record('session_meta', {
id: 'root', source: 'cli', agent_nickname: 'HEADER_SENTINEL', padding: '界'.repeat(MAX_LINE),
}));
await writeFile(files.parent, header + '\n' + jsonl([message('ACTIVITY_SENTINEL')]));
expect(await readIdentity(files.directory, files.parent)).toBeNull();
const listed = await listSessions(files.directory, files.parent);
expect(listed.root).toBeNull();
expect(listed.helpers).toEqual([]);
expect(JSON.stringify(listed)).not.toMatch(/HEADER_SENTINEL|ACTIVITY_SENTINEL/);
});
it('waits for the metadata newline before discovering a helper', async () => {
await writeFile(files.helper, JSON.stringify(meta()));
expect(await readIdentity(files.directory, files.helper)).toBeNull();
await appendFile(files.helper, '\n');
expect(await readIdentity(files.directory, files.helper)).toMatchObject({ id: 'helper' });
});
it('includes nested descendants across date folders and excludes unrelated/guardian logs', async () => {
await writeFile(path.join(files.nested, 'nested.jsonl'), jsonl([meta('nested', 'helper')]));
await writeFile(path.join(files.directory, 'other.jsonl'), jsonl([meta('other', 'unrelated')]));
await writeFile(path.join(files.directory, 'guardian.jsonl'), jsonl([record('session_meta', {
id: 'guardian', source: { subagent: 'guardian' }, parent_thread_id: 'root',
})]));
const result = await listSessions(files.directory, files.parent);
expect(result.helpers.map(item => item.id).sort()).toEqual(['helper', 'nested']);
expect(result.error).toBe('');
});
it('rejects duplicate identities rather than choosing a file', async () => {
await writeFile(path.join(files.nested, 'copy.jsonl'), jsonl([meta()]));
await writeFile(path.join(files.nested, 'child.jsonl'), jsonl([meta('child', 'helper')]));
const result = await listSessions(files.directory, files.parent);
expect(result.helpers).toEqual([]);
expect(result.partial).toBe(true);
});
it('rejects a duplicate root identity', async () => {
await writeFile(path.join(files.nested, 'copy.jsonl'), jsonl([meta('root', null)]));
expect((await listSessions(files.directory, files.parent)).error).toContain('ambiguous');
});
it('does not treat a helper as a root or switch to a replacement root', async () => {
expect((await listSessions(files.directory, files.helper)).error).toContain('root CLI');
expect((await listSessions(files.directory, files.parent, 'different')).helpers).toEqual([]);
});
it('excludes cycles and unsupported metadata', async () => {
await writeFile(path.join(files.nested, 'a.jsonl'), jsonl([meta('a', 'b')]));
await writeFile(path.join(files.nested, 'b.jsonl'), jsonl([meta('b', 'a')]));
await writeFile(path.join(files.nested, 'wrong.jsonl'), jsonl([record('event_msg', { id: 'wrong' })]));
expect((await listSessions(files.directory, files.parent)).helpers.map(item => item.id)).toEqual(['helper']);
});
it('rejects outside paths and symlinked session files', async () => {
await expect(readIdentity(files.nested, files.parent)).rejects.toThrow();
const linked = path.join(files.directory, 'linked.jsonl');
await symlink(files.helper, linked);
await expect(readIdentity(files.directory, linked)).rejects.toThrow();
expect((await listSessions(files.directory, files.parent)).helpers).toHaveLength(1);
});
it('does not recurse through linked directories', async () => {
await symlink(files.directory, path.join(files.nested, 'loop'), 'junction');
expect((await listSessions(files.directory, files.parent)).helpers).toHaveLength(1);
});
it('treats a deleted root as unavailable', async () => {
await rm(files.parent);
expect((await listSessions(files.directory, files.parent)).error).toContain('unavailable');
});
});
async function reader(budget?: number) {
const identity = await readIdentity(files.directory, files.helper);
if (!identity) throw new Error('Invalid fixture');
return new ActivityReader(files.directory, identity, budget);
}
describe('incremental activity reader', () => {
it('exposes no activity when a discovered header becomes oversized', async () => {
const view = await reader(MAX_LINE);
expect((await view.read()).events.length).toBeGreaterThan(0);
await writeFile(files.helper, JSON.stringify(meta()).padEnd(MAX_LINE + 1, ' ') + '\n' +
jsonl([meta(), own(), message('ACTIVITY_SENTINEL', 'user'), message('ACTIVITY_SENTINEL')]));
const prefix = await view.read();
expect(prefix.partial).toBe(true);
expect(prefix.available).toBe(false);
expect(prefix.events).toEqual([]);
const complete = await view.read();
expect(complete.limited).toBe(true);
expect(complete.available).toBe(false);
expect(complete.assignment).toBe('');
expect(complete.events).toEqual([]);
expect(JSON.stringify(complete)).not.toContain('ACTIVITY_SENTINEL');
});
it('reads bounded chunks and joins split UTF-8 and JSON without showing partial records', async () => {
await writeFile(files.helper, jsonl([meta(), own(), message('Hello 世界')]).trimEnd());
const view = await reader(13);
let result = await view.read();
expect(result.bytesRead).toBe(13);
for (let i = 0; i < 100 && result.bytesRead < result.fileSize; i++) result = await view.read();
expect(result.events).toEqual([]);
expect(result.partial).toBe(true);
await appendFile(files.helper, '\n');
result = await view.read();
expect(result.events[0]?.text).toBe('Hello 世界');
expect(result.partial).toBe(false);
});
it('resets after truncation and rejects replacement identity', async () => {
const view = await reader();
expect((await view.read()).events.length).toBeGreaterThan(0);
await writeFile(files.helper, jsonl([meta('wrong'), own(), message('HIDDEN')]));
const result = await view.read();
expect(result.available).toBe(false);
expect(result.events).toEqual([]);
});
it('resets after rename/replacement and does not retain old activity', async () => {
const view = await reader();
await view.read();
await rename(files.helper, path.join(files.directory, 'old.txt'));
await writeFile(files.helper, jsonl([meta(), own(), message('New file')]));
expect((await view.read()).events.map(event => event.text)).toEqual(['New file']);
});
it('detects a truncate-and-regrow rewrite of the same inode', async () => {
const view = await reader();
await view.read();
await writeFile(files.helper, jsonl([meta('wrong'), own(), message('HIDDEN'.repeat(500))]));
expect((await view.read()).events).toEqual([]);
});
it('drops stale data on a read error and recovers when the file returns', async () => {
const view = await reader();
await view.read();
await rm(files.helper);
expect((await view.read()).events).toEqual([]);
await writeFile(files.helper, jsonl([meta(), own(), message('Recovered')]));
expect((await view.read()).events[0]?.text).toBe('Recovered');
});
it('skips oversized lines and closes attribution until a new own-thread record', async () => {
await writeFile(files.helper, jsonl([meta(), own()]) + 'x'.repeat(MAX_LINE + 20) + '\n' +
jsonl([message('HIDDEN'), own(), message('visible')]));
const result = await (await reader()).read();
expect(result.events.map(event => event.text)).toEqual(['visible']);
expect(result.limited).toBe(true);
});
it('closes attribution on malformed complete lines', async () => {
await writeFile(files.helper, jsonl([meta(), own()]) + '{bad}\n' + jsonl([message('HIDDEN'), own(), message('visible')]));
expect((await (await reader()).read()).events.map(event => event.text)).toEqual(['visible']);
});
it('returns detached snapshots', async () => {
const view = await reader();
const first = await view.read();
first.events.length = 0;
expect((await view.read()).events.length).toBeGreaterThan(0);
});
});
@@ -0,0 +1,29 @@
import { mkdir, mkdtemp, realpath, writeFile } from 'node:fs/promises';
import os from 'node:os';
import path from 'node:path';
export const record = (type: string, payload: unknown) => ({ type, timestamp: '2026-10-09T00:00:00Z', payload });
export const meta = (id = 'helper', parent: string | null = 'root') => record('session_meta', {
id, agent_path: `/root/${id}`, agent_nickname: `Agent ${id}`,
source: parent ? { subagent: { thread_spawn: { parent_thread_id: parent } } } : 'cli',
});
export const own = (id = 'helper') => record('event_msg', { type: 'thread_settings_applied', thread_id: id });
export const message = (text: string, role = 'assistant', phase: string | null = 'commentary') =>
record('response_item', { type: 'message', role, phase, content: [{ type: 'output_text', text }] });
export const call = (id = 'call', command = 'npm test') => record('response_item', {
type: 'function_call', call_id: id, name: 'exec_command', arguments: JSON.stringify({ cmd: command }),
});
export const result = (id = 'call', output: unknown = { exit_code: 0, output: 'Tests passed' }) =>
record('response_item', { type: 'function_call_output', call_id: id, output });
export const jsonl = (rows: unknown[]) => rows.map(row => JSON.stringify(row)).join('\n') + '\n';
export async function fixture() {
const directory = await realpath(await mkdtemp(path.join(os.tmpdir(), 'ruflo-activity-')));
const parent = path.join(directory, 'parent.jsonl');
const helper = path.join(directory, 'helper.jsonl');
await writeFile(parent, jsonl([meta('root', null)]));
await writeFile(helper, jsonl([meta(), own(), message('Check the parser fixtures.', 'user'), call(), result(),
record('event_msg', { type: 'task_complete' })]));
const nested = path.join(directory, 'next-day');
await mkdir(nested);
return { directory, parent, helper, nested };
}
@@ -0,0 +1,120 @@
import { describe, expect, it } from 'vitest';
import { ActivityParser } from '../src/activity/parser.js';
import { sanitize } from '../src/activity/sanitize.js';
import { call, message, meta, own, record, result } from './activity-fixtures.js';
function parse(rows: unknown[]) {
const parser = new ActivityParser('helper', '/root/helper');
for (const row of rows) parser.consume(row);
return parser;
}
describe('helper activity attribution', () => {
it('hides copied parent history until an explicit own-thread record', () => {
const p = parse([meta(), meta('root', null), message('PARENT SECRET', 'user'), call(), result(),
own(), message('Check parser', 'user'), message('Working'), call('own'), result('own')]);
expect(JSON.stringify(p.snapshot)).not.toContain('PARENT SECRET');
expect(p.snapshot.assignment).toBe('Check parser');
expect(p.snapshot.events.filter(e => e.kind === 'tool')).toHaveLength(1);
});
it('requires first-record identity and never trusts later matching metadata', () => {
for (const first of [meta('wrong'), {}, null, []]) {
const p = parse([first, meta(), own(), message('HIDDEN')]);
expect(p.snapshot.available).toBe(false);
expect(p.snapshot.events).toEqual([]);
}
});
it('closes attribution on foreign metadata, wrong thread and corrupt records', () => {
for (const boundary of [meta('foreign'), own('foreign'), null]) {
const p = parse([meta(), own(), message('before'), boundary, message('HIDDEN'), own(), message('after')]);
expect(p.snapshot.events.map(e => e.text)).toEqual(['before', 'after']);
}
});
it('does not pair results across a foreign boundary', () => {
const p = parse([meta(), own(), call(), own('foreign'), result(), own(), result()]);
expect(p.snapshot.events[0]?.output).toBe('');
});
it('excludes reasoning, private roles and assistant messages without a public phase', () => {
const p = parse([meta(), own(), message('HIDDEN', 'system'), message('HIDDEN', 'developer'),
message('HIDDEN', 'assistant', 'analysis'), message('HIDDEN', 'assistant', null),
record('response_item', { type: 'reasoning', text: 'HIDDEN' }),
record('event_msg', { type: 'item_completed', item: { type: 'Reasoning', text: 'HIDDEN' } }),
message('Public answer', 'assistant', 'final')]);
expect(JSON.stringify(p.snapshot)).not.toContain('HIDDEN');
expect(p.snapshot.events[0]?.text).toBe('Public answer');
expect(p.snapshot.status).toBe('completed');
});
it('ignores encrypted content and outgoing messages do not change the assignment', () => {
const p = parse([meta(), own(), record('response_item', { type: 'agent_message', content: [
{ type: 'text', text: 'Initial assignment' }, { type: 'encrypted_text', text: 'HIDDEN' }], recipient: '/root/helper' }),
record('response_item', { type: 'agent_message', content: 'Reply to parent', recipient: '/root' })]);
expect(p.snapshot.assignment).toBe('Initial assignment');
expect(JSON.stringify(p.snapshot)).not.toContain('HIDDEN');
});
});
describe('public tool records', () => {
it.each([
[{ exit_code: 0, output: '0 errors' }, 'completed'],
[{ exit_code: 2, output: 'failed' }, 'failed'],
[{ session_id: 42, exit_code: null }, 'running'],
['Script running with cell ID 7', 'running'],
['Process exited with code 1', 'failed'],
['0 errors', 'returned'],
[{ isError: true }, 'failed'],
])('pairs a result and interprets explicit completion fields: %j', (output, status) => {
const p = parse([meta(), own(), call(), result('unrelated', 'HIDDEN'), result('call', output)]);
expect(p.snapshot.events).toHaveLength(1);
expect(p.snapshot.events[0]?.status).toBe(status);
expect(JSON.stringify(p.snapshot)).not.toContain('HIDDEN');
});
it('supports custom calls and command execution item events', () => {
const p = parse([meta(), own(),
record('response_item', { type: 'custom_tool_call', name: 'exec', call_id: 'x', input: 'echo safe' }),
record('response_item', { type: 'custom_tool_call_output', call_id: 'x', output: 'safe' }),
record('event_msg', { type: 'item_started', item: { type: 'CommandExecution', id: 'y', command: 'false' } }),
record('event_msg', { type: 'item_completed', item: { type: 'CommandExecution', id: 'y', exit_code: 1, aggregated_output: 'failed' } }),
]);
expect(p.snapshot.events).toHaveLength(2);
expect(p.snapshot.events[1]?.status).toBe('failed');
});
it('shows native AgentMessage final answers with Text blocks and completed state', () => {
const p = parse([meta(), own(), call(), record('event_msg', {
type: 'item_completed', item: { type: 'AgentMessage', phase: 'final_answer', content: [
{ type: 'Text', text: 'Finished' }, { type: 'encrypted_text', text: 'HIDDEN' },
] },
})]);
expect(p.snapshot.events.at(-1)?.text).toBe('Finished');
expect(p.snapshot.status).toBe('completed');
expect(JSON.stringify(p.snapshot)).not.toContain('HIDDEN');
});
it('bounds event history and previews, and rejects invalid call IDs', () => {
const p = parse([meta(), own(), call('\x1b[2J'), ...Array.from({ length: 100 }, (_, i) => message(`${i}: ${'x'.repeat(4000)}`))]);
expect(p.snapshot.events).toHaveLength(80);
expect(p.snapshot.limited).toBe(true);
expect(p.snapshot.events.every(e => e.text.length < 2500 && e.kind === 'message')).toBe(true);
});
});
describe('preview sanitization', () => {
it('removes terminal escapes and bidi controls while retaining ordinary text', () => {
expect(sanitize('\x1b[2Jhello\x1b]0;bad\x07\u202eworld\nnext')).toBe('helloworld\nnext');
});
it.each([
'OPENAI_API_KEY=sk-proj-1234567890123456789',
'{"password":"sensitive-value"}', 'Authorization: Bearer sensitive-value',
'https://user:sensitive-value@example.test',
'-----BEGIN PRIVATE KEY-----\nsensitive-value\n-----END PRIVATE KEY-----',
])('masks credentials before clipping: %s', input => {
expect(sanitize(input)).toContain('REDACTED');
expect(sanitize(input)).not.toContain('sensitive-value');
expect(sanitize(input)).not.toContain('sk-proj-1234567890123456789');
});
it('keeps benign command output readable', () => {
expect(sanitize('npm test\n42 passed, 0 errors')).toBe('npm test\n42 passed, 0 errors');
});
it('handles long ordinary words and hyphenated output without backtracking over every prefix', () => {
expect(sanitize('x'.repeat(64000))).toBe('x'.repeat(2400) + ' ... [truncated]');
expect(sanitize('a-'.repeat(32000))).toBe('a-'.repeat(1200) + ' ... [truncated]');
}, 1000);
});
@@ -0,0 +1,104 @@
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import { rm, writeFile } from 'node:fs/promises';
import path from 'node:path';
import { ActivityViewer } from '../src/activity/viewer.js';
import { ActivityReader } from '../src/activity/reader.js';
import { fixture, jsonl, meta } from './activity-fixtures.js';
let files: Awaited<ReturnType<typeof fixture>>;
beforeEach(async () => { files = await fixture(); });
afterEach(async () => { vi.restoreAllMocks(); await rm(files.directory, { recursive: true, force: true }); });
describe('agent picker', () => {
it('opens the selected helper and shows its assignment, command and result', async () => {
const viewer = new ActivityViewer(files.directory, files.parent);
await viewer.refresh();
expect(viewer.render()).toContain('AGENTS - choose one');
viewer.key('return');
await viewer.refresh();
const output = viewer.render(100, 100);
expect(output).toContain('Check the parser fixtures.');
expect(output).toContain('npm test');
expect(output).toContain('Tests passed');
expect(output).toContain('completed (recorded)');
viewer.key('b');
expect(viewer.render()).toContain('AGENTS - choose one');
});
it('keeps selection when the list order changes and does not substitute a vanished agent', async () => {
const viewer = new ActivityViewer(files.directory, files.parent);
await viewer.refresh();
viewer.key('return');
await writeFile(path.join(files.nested, 'aaa.jsonl'), jsonl([meta('aaa')]));
await viewer.refresh();
expect(viewer.selected).toBe('helper');
await rm(files.helper);
await viewer.refresh();
expect(viewer.render()).toContain('no longer available');
expect(viewer.selected).toBe('helper');
viewer.key('b'); viewer.key('down');
expect(viewer.selected).toBe('aaa');
});
it('rejects a changed parent identity and clears the visible helper activity', async () => {
const viewer = new ActivityViewer(files.directory, files.parent);
await viewer.refresh(); viewer.key('return'); await viewer.refresh();
await writeFile(files.parent, jsonl([meta('other', null)]));
await viewer.refresh();
expect(viewer.render()).toContain('unchanged identity');
expect(viewer.render()).not.toContain('npm test');
});
it('clears the prior helper immediately, before the new helper is read', async () => {
await writeFile(path.join(files.nested, 'zzz.jsonl'), jsonl([meta('zzz')]));
const viewer = new ActivityViewer(files.directory, files.parent);
await viewer.refresh(); viewer.key('return'); await viewer.refresh();
expect(viewer.render(100, 100)).toContain('npm test');
viewer.key('b'); viewer.key('down'); viewer.key('return');
expect(viewer.render(100, 100)).toContain('/root/zzz');
expect(viewer.render(100, 100)).not.toContain('npm test');
});
it('does not publish a previous helper read that finishes after selection changes', async () => {
await writeFile(path.join(files.nested, 'zzz.jsonl'), jsonl([meta('zzz')]));
const viewer = new ActivityViewer(files.directory, files.parent);
await viewer.refresh(); viewer.key('return'); await viewer.refresh();
const prior = viewer.activity!;
let complete: (value: typeof prior) => void = () => {};
let started: () => void = () => {};
const reading = new Promise<void>(resolve => { started = resolve; });
vi.spyOn(ActivityReader.prototype, 'read').mockImplementation(() => {
started(); return new Promise(resolve => { complete = resolve; });
});
const refresh = viewer.refresh();
await reading;
viewer.key('b'); viewer.key('down'); viewer.key('return');
complete(prior); await refresh;
expect(viewer.activity).toBeNull();
expect(viewer.render(100, 100)).not.toContain('npm test');
});
it.each([[25, 12], [40, 10], [80, 8]])('keeps the selected picker row visible at %ix%i', async (width, height) => {
for (const id of ['aaa', 'bbb', 'ccc']) {
const header = meta(id);
header.payload.agent_path = `/root/${id}-${'long-name-'.repeat(10)}`;
await writeFile(path.join(files.nested, `${id}.jsonl`), jsonl([header]));
}
const viewer = new ActivityViewer(files.directory, files.parent);
await viewer.refresh();
viewer.selected = 'ccc';
for (const partial of [false, true]) {
viewer.list.partial = partial;
const output = viewer.render(width, height);
expect(output).toContain('> /root/ccc-');
expect(output.split('\n').length).toBeLessThanOrEqual(height);
expect(output.split('\n').every(line => line.length < width)).toBe(true);
}
viewer.key('up');
expect(viewer.render(width, height)).toContain('> /root/bbb-');
});
it('supports narrow windows, scroll, follow-latest and top navigation', async () => {
const viewer = new ActivityViewer(files.directory, files.parent);
await viewer.refresh(); viewer.key('return'); await viewer.refresh();
expect(viewer.render(25, 12).split('\n').every(line => line.length <= 24)).toBe(true);
viewer.key('G'); viewer.render(25, 12);
expect(viewer.offset).toBeGreaterThan(0);
viewer.key('g'); viewer.render(25, 12);
expect(viewer.offset).toBe(0);
});
});