feat(agent): per-session agent inbox (discussion #558)

A coordinator can leave text for a worker without typing into its pane.

- Store (src/web/agent-inbox.ts): read until ack by id, seen only for the
  receiver, 16 KiB/message, 512 KiB + 200 messages per inbox, full = 409
  CONFLICT (never drops the oldest), 12 h TTL from each message's createdAt.
- Routes (routes/inbox-routes.ts): POST/GET /api/sessions/:id/inbox
  (?wait= parks while nothing is unseen), POST .../inbox/ack (receiver
  only, 403 otherwise), GET /api/agent-inbox/summary (owner-scoped counts,
  not in toState()). Caller = resolveParentSessionId().
- Inbox waiters are the wait registry's third kind: same caps, ended by
  cancelAll/cancelEverything.
- Opt-in snapshot (agentInboxPersistEnabled, synced, default off):
  agent-inbox.json via dataPath(), 0600 tmp+rename, off deletes it, boot
  prune only after a successful restore. Dropped on delete, kept on detach.
- CLI: codeman agent post | inbox [--wait] | ack, INBOX column in ls.
- waitCapacityResponse/abortOnClientHangUp moved to route-helpers.ts
  unchanged.
- Docs: api-reference, README, CLAUDE.md, architecture-invariants, skill +
  plugin mirror.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
RandalixandClaude Opus 5.5 committed 2026-10-23 12:45:48 +02:00
1 parent 002de32a08
commit c6dfaa2cac
34 files changed
+2308 -83

No files matched your search

+234 -13
View File
@@ -3,7 +3,8 @@
* inside a Codeman session, in every CLI mode.
*
* A thin HTTP client over endpoints that already exist (`quick-start`, `input`,
* `wait`, `wait-output`, `last-response`, `terminal`, `DELETE sessions/:id`). It
* `wait`, `wait-output`, `last-response`, `terminal`, `DELETE sessions/:id`, and the
* agent inbox: `inbox`, `inbox/ack`, `agent-inbox/summary`). It
* invents no route and no transport: everything goes through `CODEMAN_API_URL`, so
* auth, ownership and the per-session waiter cap apply unchanged. The behaviour is
* the packaged agent skill's (`skills/codeman`), ported from shell prose into code
@@ -416,15 +417,55 @@ export async function resolveSessionId(
return { error: `"${id}" is ambiguous: ${matches.map((s) => s.id.slice(0, 13)).join(', ')}`, code: EXIT.error };
}
/** `agent ls` — every session the caller can see, self marked. */
/** One session's row in `GET /api/agent-inbox/summary`. */
export interface InboxSummaryRow {
pending: number;
unseen: number;
oldestUnseenAt: number | null;
parked: boolean;
}
/**
* The inbox summary, or undefined when it cannot be had (a server without the route,
* a transient error). `ls` must still list sessions then, so this never fails the verb.
*/
async function fetchInboxSummary(deps: AgentDeps): Promise<Record<string, InboxSummaryRow> | undefined> {
try {
const res = await deps.request(deps.ctx, { method: 'GET', path: '/api/v1/agent-inbox/summary' });
if (!res.json?.success) return undefined;
return (res.json.data as { sessions?: Record<string, InboxSummaryRow> } | undefined)?.sessions ?? {};
} catch {
return undefined;
}
}
/** The INBOX cell of `ls`: `3 (1 new)`, `parked`, both, or empty. */
export function inboxCell(row: InboxSummaryRow | undefined): string {
if (!row) return '';
const parts: string[] = [];
if (row.pending > 0) parts.push(row.unseen > 0 ? `${row.pending} (${row.unseen} new)` : String(row.pending));
if (row.parked) parts.push('parked');
return parts.join(', ');
}
/** `agent ls` — every session the caller can see, self marked, with its inbox counts. */
export async function agentLs(deps: AgentDeps): Promise<number> {
const res = await deps.request(deps.ctx, { method: 'GET', path: '/api/v1/sessions' });
const [res, inbox] = await Promise.all([
deps.request(deps.ctx, { method: 'GET', path: '/api/v1/sessions' }),
fetchInboxSummary(deps),
]);
if (!res.json?.success) return fail(deps, describeFailure(res));
const sessions = (res.json.data as SessionRow[] | undefined) ?? [];
if (deps.json) {
emitJson(
deps,
sessions.map((s) => ({ ...s, self: isSelfSession(deps.ctx.selfId, s.id) }))
sessions.map((s) => ({
...s,
self: isSelfSession(deps.ctx.selfId, s.id),
// null = empty inbox and not parked. The key is absent when the server did
// not answer the summary, so "no data" never reads as "no mail".
...(inbox ? { inbox: inbox[s.id] ?? null } : {}),
}))
);
return EXIT.ok;
}
@@ -432,17 +473,22 @@ export async function agentLs(deps: AgentDeps): Promise<number> {
deps.io.out(palette.muted('(no sessions)'));
return EXIT.ok;
}
const rows = sessions.map((s) => [
isSelfSession(deps.ctx.selfId, s.id) ? '*' : ' ',
s.id.slice(0, 8),
s.mode ?? '?',
s.status ?? '?',
s.name || s.workingDir || '',
]);
deps.io.out(table([[' ', 'ID', 'MODE', 'STATUS', 'NAME'], ...rows], { gap: 2 }));
const rows = sessions.map((s) => {
const row = [isSelfSession(deps.ctx.selfId, s.id) ? '*' : ' ', s.id.slice(0, 8), s.mode ?? '?', s.status ?? '?'];
if (inbox) row.push(inboxCell(inbox[s.id]));
row.push(s.name || s.workingDir || '');
return row;
});
const header = inbox ? [' ', 'ID', 'MODE', 'STATUS', 'INBOX', 'NAME'] : [' ', 'ID', 'MODE', 'STATUS', 'NAME'];
deps.io.out(table([header, ...rows], { gap: 2 }));
deps.io.out(
palette.muted(`* = this session (${deps.ctx.selfId.slice(0, 8)}). status is a UI hint, never a sync signal.`)
);
if (inbox) {
deps.io.out(
palette.muted('INBOX: pending (unseen), parked = the session is waiting on `agent inbox --wait` right now.')
);
}
return EXIT.ok;
}
@@ -790,6 +836,140 @@ export async function agentRm(deps: AgentDeps, options: { id: string }): Promise
return EXIT.ok;
}
// ─────────────────────────────────────────────────────────────────────────────
// Agent inbox
// ─────────────────────────────────────────────────────────────────────────────
/** A message as `GET …/inbox` hands it out. */
export interface InboxMessageRow {
id: string;
from: string | null;
text: string;
createdAt: number;
seen: boolean;
}
/**
* The message `post` was given, which must be ONE argument, as with `send`: several
* would be an unquoted `$(cat notes.md)` split by the shell, and joining the pieces
* with spaces loses its newlines. Unlike `send`, newlines inside the one argument are
* fine: the inbox stores text as data and never types it.
*/
export function postTextFromArgs(words: readonly string[]): { text: string } | { error: string } {
if (words.length === 1) return { text: words[0] };
return { error: `refusing: the message must be ONE argument, got ${words.length} — quote it` };
}
/** `agent post` — store a message in another session's inbox. Never types into its pane. */
export async function agentPost(deps: AgentDeps, options: { id: string; text: string }): Promise<number> {
if (options.text.length === 0) return fail(deps, 'refusing: empty message', EXIT.refused);
const target = await resolveSessionId(deps, options.id);
if ('error' in target) return fail(deps, target.error, target.code);
const res = await deps.request(deps.ctx, {
method: 'POST',
path: `/api/v1/sessions/${encodeURIComponent(target.id)}/inbox`,
body: { text: options.text },
});
// 409 = the receiver's inbox is full; nothing was stored. Passed through as an error.
if (!res.json?.success) return fail(deps, describeFailure(res));
const data = res.json.data as { message?: InboxMessageRow; pending?: number; unseen?: number } | undefined;
if (deps.json) emitJson(deps, data ?? {});
else
deps.io.out(
palette.ok(
`${GLYPH.ok} posted to ${target.id.slice(0, 8)} (${data?.pending ?? '?'} pending, ${data?.unseen ?? '?'} unseen)`
)
);
return EXIT.ok;
}
export interface InboxReadOptions {
/** Another session's inbox (a peek); omitted = this session's own. */
id?: string;
/** Long-poll budget in ms; the read parks while nothing is unseen. */
waitMs?: number;
}
/** Relative age for humans: `12s`, `4m`, `3h`. */
function age(createdAt: number, now: number): string {
const s = Math.max(0, Math.round((now - createdAt) / 1000));
if (s < 60) return `${s}s`;
if (s < 3600) return `${Math.round(s / 60)}m`;
return `${Math.round(s / 3600)}h`;
}
/**
* `agent inbox` — read this session's inbox (or peek at another's). Removes nothing:
* a message stays until `agent ack <id>`. Only reading your OWN inbox marks messages
* seen. Exit 0 with mail, 2 when a `--wait` ran out with nothing unseen, 3 when the
* session went away during the wait.
*/
export async function agentInboxRead(deps: AgentDeps, options: InboxReadOptions): Promise<number> {
const target = await resolveSessionId(deps, options.id ?? deps.ctx.selfId);
if ('error' in target) return fail(deps, target.error, target.code);
const res = await deps.request(deps.ctx, {
method: 'GET',
path: `/api/v1/sessions/${encodeURIComponent(target.id)}/inbox`,
query: { wait: options.waitMs },
timeoutMs: options.waitMs !== undefined ? options.waitMs + 10_000 : undefined,
});
if (!res.json?.success) return fail(deps, describeFailure(res));
const data = res.json.data as
| {
receiver?: boolean;
messages?: InboxMessageRow[];
pending?: number;
unseen?: number;
wait?: { timedOut?: boolean; ended?: boolean };
}
| undefined;
const messages = data?.messages ?? [];
const code = data?.wait?.ended ? EXIT.dead : data?.wait?.timedOut ? EXIT.timeout : EXIT.ok;
if (deps.json) {
emitJson(deps, data ?? {});
return code;
}
const now = (deps.now ?? Date.now)();
for (const m of messages) {
const sender = m.from ? m.from.slice(0, 8) : 'user';
const tag = m.seen ? palette.muted('seen') : palette.emph('new');
deps.io.out(palette.muted(`── ${m.id} from ${sender}, ${age(m.createdAt, now)} ago, `) + tag);
deps.io.out(m.text);
}
if (data?.wait?.ended) {
deps.io.err(palette.err(`${GLYPH.fail} the session went away during the wait`));
} else if (messages.length === 0) {
deps.io.err(palette.muted(data?.wait?.timedOut ? '(no mail; wait timed out, exit 2)' : '(no mail)'));
} else if (data?.receiver) {
deps.io.err(
palette.muted(
`${messages.length} pending. Nothing is removed by reading: when done, \`codeman agent ack <id>…\` (full ids above).`
)
);
} else {
deps.io.err(palette.muted(`${messages.length} pending (peek: nothing was marked seen).`));
}
return code;
}
/** `agent ack` — remove messages from THIS session's inbox by id. Idempotent. */
export async function agentAck(deps: AgentDeps, options: { ids: string[] }): Promise<number> {
const ids = options.ids.map((id) => id.trim()).filter(Boolean);
if (ids.length === 0) return fail(deps, 'refusing: no message ids', EXIT.refused);
const self = await resolveSessionId(deps, deps.ctx.selfId);
if ('error' in self) return fail(deps, self.error, self.code);
const res = await deps.request(deps.ctx, {
method: 'POST',
path: `/api/v1/sessions/${encodeURIComponent(self.id)}/inbox/ack`,
body: { ids },
});
if (!res.json?.success) return fail(deps, describeFailure(res));
const data = res.json.data as { removed?: number; pending?: number } | undefined;
if (deps.json) emitJson(deps, data ?? {});
else deps.io.out(palette.ok(`${GLYPH.ok} acked ${data?.removed ?? 0} (${data?.pending ?? '?'} still pending)`));
return EXIT.ok;
}
// ─────────────────────────────────────────────────────────────────────────────
// Commander wiring
// ─────────────────────────────────────────────────────────────────────────────
@@ -833,7 +1013,9 @@ async function run(json: boolean, verb: (deps: AgentDeps) => Promise<number>): P
export function registerAgentCommands(program: Command): Command {
const agent = program
.command('agent')
.description('Talk to other sessions from inside one (any CLI mode): list, spawn, send, wait, read, interrupt, rm');
.description(
'Talk to other sessions from inside one (any CLI mode): list, spawn, send, wait, read, interrupt, rm, and the inbox: post, inbox, ack'
);
agent
.command('ls')
@@ -977,6 +1159,45 @@ export function registerAgentCommands(program: Command): Command {
run(Boolean(options.json), (deps) => agentInterrupt(deps, { id }))
);
agent
.command('post <id> <text...>')
.description(
"Leave a message in another session's inbox (ONE quoted argument; multi-line is fine). Nothing is typed into its pane: it reads with `agent inbox`"
)
.option('--json', 'Machine-readable output')
.action((id: string, words: string[], options: { json?: boolean }) =>
run(Boolean(options.json), (deps) => {
const message = postTextFromArgs(words);
if ('error' in message) return Promise.resolve(fail(deps, message.error, EXIT.refused));
return agentPost(deps, { id, text: message.text });
})
);
agent
.command('inbox [id]')
.description(
"Read this session's inbox (or peek at another's with [id]). Reading removes nothing; `agent ack` does. --wait blocks while nothing is new: exit 2 on timeout"
)
.option('-w, --wait', 'Block until a new (unseen) message arrives')
.option('-t, --timeout <ms>', 'Wait budget in ms (with --wait)', String(DEFAULT_WAIT_MS))
.option('--json', 'Machine-readable output')
.action((id: string | undefined, options: { wait?: boolean; timeout?: string; json?: boolean }) =>
run(Boolean(options.json), (deps) =>
agentInboxRead(deps, {
id,
waitMs: options.wait ? parsePositiveInt(options.timeout, DEFAULT_WAIT_MS) : undefined,
})
)
);
agent
.command('ack <ids...>')
.description("Remove messages from this session's inbox by id, once handled (unknown ids are not an error)")
.option('--json', 'Machine-readable output')
.action((ids: string[], options: { json?: boolean }) =>
run(Boolean(options.json), (deps) => agentAck(deps, { ids }))
);
agent
.command('rm <id>')
.description('Delete any session except this one (refuses your own id)')
+138
View File
@@ -0,0 +1,138 @@
/**
* @fileoverview Opt-in snapshot of the agent inbox, so pending mail between agents
* survives a server restart (the agents themselves keep running in tmux across it).
*
* Off by default: an install that never turns on `agentInboxPersistEnabled` writes
* nothing new to disk. The rules (discussion #558):
* - Off means no file. Switching it off deletes the snapshot; the in-memory inboxes
* stay as they are.
* - Its own file under `dataPath()`, written 0600 through tmp + rename like
* `intents.json`, since messages carry the same kind of text prompts do. It is never
* read by `/api/search` and `sensitive-path.ts` keeps the file viewer off it.
* - Writes are debounced (a burst of posts is one write) and flushed on shutdown.
* - Boot pruning is NOT done here: the server calls `agentInbox.pruneExcept()` only
* after the session restore succeeded, so a failed restore leaves the mail alone.
*
* @module web/agent-inbox-persistence
*/
import { existsSync, mkdirSync, readFileSync, renameSync, rmSync, writeFileSync } from 'node:fs';
import { dirname } from 'node:path';
import { dataPath } from '../config/instance.js';
import { agentInbox, type AgentInbox } from './agent-inbox.js';
/** A burst of posts (a coordinator fanning out to five workers) becomes one write. */
const WRITE_DEBOUNCE_MS = 250;
export class AgentInboxPersistence {
private enabled = false;
private timer: NodeJS.Timeout | null = null;
constructor(
private readonly inbox: AgentInbox = agentInbox,
/** Resolved per call: test HOMEs change between files. */
private readonly filePath: () => string = () => dataPath('agent-inbox.json'),
private readonly debounceMs: number = WRITE_DEBOUNCE_MS
) {
this.inbox.onChange = () => this.schedule();
}
get isEnabled(): boolean {
return this.enabled;
}
/**
* Boot: apply the setting and, when it is on, load the snapshot into the store.
* Returns how many messages were restored. A missing or unreadable file is an
* empty inbox, never a failed boot.
*/
load(enabled: boolean): number {
this.enabled = enabled;
if (!enabled) {
// A file left from an earlier "on" is removed, so off really means no file.
this.removeFile();
return 0;
}
const path = this.filePath();
if (!existsSync(path)) return 0;
try {
return this.inbox.restore(JSON.parse(readFileSync(path, 'utf-8')));
} catch (err) {
console.warn(`[AgentInbox] Failed to load ${path}, starting empty:`, err);
return 0;
}
}
/**
* The setting changed. On: write the current inboxes now. Off: drop any pending
* write and delete the snapshot; memory is untouched.
*/
setEnabled(enabled: boolean): void {
if (enabled === this.enabled) return;
this.enabled = enabled;
if (enabled) {
this.writeNow();
} else {
this.cancel();
this.removeFile();
}
}
/** Shutdown: write a pending change now instead of losing it with the timer. */
flush(): void {
if (!this.timer) return;
this.cancel();
this.writeNow();
}
/** Test hook: drop a pending write and detach from the store. */
dispose(): void {
this.cancel();
if (this.inbox.onChange) this.inbox.onChange = undefined;
}
private schedule(): void {
if (!this.enabled || this.timer) return;
this.timer = setTimeout(() => {
this.timer = null;
this.writeNow();
}, this.debounceMs);
// A pending write must not hold the process open; shutdown flushes it explicitly.
this.timer.unref();
}
private cancel(): void {
if (this.timer) clearTimeout(this.timer);
this.timer = null;
}
private writeNow(): void {
if (!this.enabled) return;
const path = this.filePath();
const tmpPath = `${path}.tmp`;
try {
mkdirSync(dirname(path), { recursive: true });
// 0600: agent messages can carry anything a prompt can, secrets included.
// `mode` applies only when the file is created, and the rename makes every
// write a fresh file, so the final file is always 0600.
rmSync(tmpPath, { force: true });
writeFileSync(tmpPath, JSON.stringify(this.inbox.snapshot()), { mode: 0o600 });
renameSync(tmpPath, path);
} catch (err) {
console.warn(`[AgentInbox] Failed to persist ${path}:`, err);
}
}
private removeFile(): void {
const path = this.filePath();
try {
rmSync(path, { force: true });
rmSync(`${path}.tmp`, { force: true });
} catch (err) {
console.warn(`[AgentInbox] Failed to remove ${path}:`, err);
}
}
}
/** Process-wide instance wired to `agentInbox`; the server and the settings route use it. */
export const agentInboxPersistence = new AgentInboxPersistence();
+302
View File
@@ -0,0 +1,302 @@
/**
* @fileoverview Agent inbox: a per-session mailbox other agents (or the user) post to,
* and the receiving agent reads when it chooses to.
*
* The other agent-to-agent channel, `POST /api/sessions/:id/input`, types into the
* receiver's composer. That is right for a prompt and wrong for a message: every
* delivery starts or interrupts a turn, the receiver cannot defer it, and it is one
* line only. The inbox stores instead of typing. Nothing here ever reaches a pane.
*
* Invariants (each pinned in `test/agent-inbox.test.ts`):
* - **Read until ack.** A read hands out every pending message and removes nothing.
* A message goes only when the receiver acknowledges its id. "Read and dropped"
* and "done" must not look the same to the sender (discussion #558, the 09-23 case).
* - **Seen is the receiver's alone.** `read(…, { receiver: true })` marks what it
* handed out as seen; any other read is a peek and changes nothing. The route
* decides who the receiver is; this module only takes the flag.
* - **Bounded, and full refuses.** `MAX_MESSAGES_PER_INBOX`, `MAX_MESSAGE_BYTES` and
* `MAX_INBOX_BYTES`. A post that would cross any of them is refused so the sender
* learns. The oldest message is never dropped to make room.
* - **TTL from the message's own timestamp** (`MESSAGE_TTL_MS`). A restore keeps
* `createdAt`, so a restart does not restart the clock. Expiry is lazy: every
* accessor sweeps the inbox it touches, `summary()` and `snapshot()` sweep all.
* - **No IO, no `Session` import.** Persistence (`agent-inbox-persistence.ts`) hooks
* `onChange`; long-poll waiters live in `session-wait-registry.ts`.
*
* @consumedby web/routes/inbox-routes, web/agent-inbox-persistence, web/server
* @module web/agent-inbox
*/
import { randomUUID } from 'node:crypto';
// ─── Bounds ──────────────────────────────────────────────────────────────────
/** Messages one inbox holds before posts are refused. */
export const MAX_MESSAGES_PER_INBOX = 200;
/** UTF-8 bytes per message. A bigger payload belongs in the workspace; post its path. */
export const MAX_MESSAGE_BYTES = 16_384;
/** UTF-8 bytes of message text one inbox holds before posts are refused. */
export const MAX_INBOX_BYTES = 512 * 1024;
/** A message older than this is gone, read or not (same span as the approvals inbox). */
export const MESSAGE_TTL_MS = 12 * 60 * 60 * 1000;
// ─── Types ───────────────────────────────────────────────────────────────────
/** A message as it is stored (and persisted). */
interface StoredMessage {
id: string;
/** The posting session's full id, or null when no session could be resolved (the user). */
from: string | null;
text: string;
/** Epoch ms. The TTL counts from here. */
createdAt: number;
/** Epoch ms of the first receiver read that handed this message out; null while unseen. */
seenAt: number | null;
}
/** A message as a read hands it out. */
export interface InboxMessage {
id: string;
from: string | null;
text: string;
createdAt: number;
/** True when a receiver read had handed this message out BEFORE this read. */
seen: boolean;
}
export interface InboxReadResult {
/** Every pending message, oldest first: unseen and seen-but-unacked alike. */
messages: InboxMessage[];
pending: number;
/** Unseen after this read (0 after a receiver read; unchanged by a peek). */
unseen: number;
}
export type PostResult =
| { ok: true; message: InboxMessage; pending: number; unseen: number }
| { ok: false; reason: 'empty' | 'message-too-large' | 'inbox-full-count' | 'inbox-full-bytes' };
/** What `summary()` reports for one session. Counts and times only, never text. */
export interface InboxCounts {
pending: number;
unseen: number;
/** `createdAt` of the oldest unseen message, null when nothing is unseen. */
oldestUnseenAt: number | null;
}
/** On-disk shape. `version` lets a later format refuse an older file instead of misreading it. */
export interface InboxSnapshot {
version: 1;
inboxes: Record<string, StoredMessage[]>;
}
// ─── Store ───────────────────────────────────────────────────────────────────
export class AgentInbox {
private readonly inboxes = new Map<string, StoredMessage[]>();
/** Any mutation (post, seen, ack, expiry, drop, prune, restore excluded). Persistence hooks this. */
onChange: (() => void) | undefined;
constructor(private readonly now: () => number = Date.now) {}
/**
* Store a message for `sessionId`. `from` is the RESOLVED sender (the route's job),
* never a client-supplied label.
*/
post(sessionId: string, from: string | null, text: string): PostResult {
if (text.length === 0) return { ok: false, reason: 'empty' };
const bytes = Buffer.byteLength(text, 'utf8');
if (bytes > MAX_MESSAGE_BYTES) return { ok: false, reason: 'message-too-large' };
const queue = this.live(sessionId);
if (queue.length >= MAX_MESSAGES_PER_INBOX) return { ok: false, reason: 'inbox-full-count' };
if (inboxBytes(queue) + bytes > MAX_INBOX_BYTES) return { ok: false, reason: 'inbox-full-bytes' };
const stored: StoredMessage = { id: randomUUID(), from, text, createdAt: this.now(), seenAt: null };
queue.push(stored);
this.inboxes.set(sessionId, queue);
this.onChange?.();
return { ok: true, message: view(stored), pending: queue.length, unseen: countUnseen(queue) };
}
/**
* Every pending message, oldest first. Removes nothing. With `receiver`, the
* messages handed out are marked seen (each keeps its first `seenAt`); without it
* this is a peek and changes nothing at all.
*/
read(sessionId: string, options: { receiver: boolean }): InboxReadResult {
const queue = this.live(sessionId);
const messages = queue.map(view);
if (options.receiver) {
let marked = false;
const at = this.now();
for (const m of queue) {
if (m.seenAt !== null) continue;
m.seenAt = at;
marked = true;
}
if (marked) this.onChange?.();
}
return { messages, pending: queue.length, unseen: countUnseen(queue) };
}
/** Messages no receiver read has handed out yet. A `?wait=` read parks while this is 0. */
unseenCount(sessionId: string): number {
return countUnseen(this.live(sessionId));
}
pendingCount(sessionId: string): number {
return this.live(sessionId).length;
}
/**
* Remove the given messages. Idempotent: an id that is unknown or already gone is
* not an error, so a retry after a lost response succeeds. Returns how many went.
*/
ack(sessionId: string, ids: readonly string[]): number {
const queue = this.live(sessionId);
if (queue.length === 0 || ids.length === 0) return 0;
const drop = new Set(ids);
const kept = queue.filter((m) => !drop.has(m.id));
const removed = queue.length - kept.length;
if (removed === 0) return 0;
this.store(sessionId, kept);
this.onChange?.();
return removed;
}
/** Per-session counts for every inbox that holds anything. Never blocks. */
summary(): Map<string, InboxCounts> {
const out = new Map<string, InboxCounts>();
for (const sessionId of [...this.inboxes.keys()]) {
const queue = this.live(sessionId);
if (queue.length === 0) continue;
const unseen = queue.filter((m) => m.seenAt === null);
out.set(sessionId, {
pending: queue.length,
unseen: unseen.length,
oldestUnseenAt: unseen.length > 0 ? unseen[0].createdAt : null,
});
}
return out;
}
/** The session was DELETED: its mail goes with it. Not called on detach or shutdown. */
drop(sessionId: string): void {
if (this.inboxes.delete(sessionId)) this.onChange?.();
}
/**
* Boot-time sweep, run only after the session restore succeeded: discard inboxes
* whose session did not come back. Every route needs a live session, so an orphan
* could otherwise never be removed. Returns how many inboxes were dropped.
*/
pruneExcept(keep: Iterable<string>): number {
const alive = new Set(keep);
let dropped = 0;
for (const sessionId of [...this.inboxes.keys()]) {
if (alive.has(sessionId)) continue;
this.inboxes.delete(sessionId);
dropped++;
}
if (dropped > 0) this.onChange?.();
return dropped;
}
/** Whole-store snapshot for persistence, expired messages already swept. */
snapshot(): InboxSnapshot {
const inboxes: Record<string, StoredMessage[]> = {};
for (const sessionId of [...this.inboxes.keys()]) {
const queue = this.live(sessionId);
if (queue.length > 0) inboxes[sessionId] = queue.map((m) => ({ ...m }));
}
return { version: 1, inboxes };
}
/**
* Replace the store from a snapshot (server boot). Malformed input is skipped entry
* by entry rather than thrown: a corrupt file must not stop the server. Expired
* messages are not restored, and the caps hold for what is. Does not fire
* `onChange`: writing back what was just read would be a no-op at best.
*/
restore(snapshot: unknown): number {
const data = snapshot as Partial<InboxSnapshot> | null;
this.inboxes.clear();
if (!data || data.version !== 1 || typeof data.inboxes !== 'object' || data.inboxes === null) return 0;
const cutoff = this.now() - MESSAGE_TTL_MS;
let restored = 0;
for (const [sessionId, raw] of Object.entries(data.inboxes)) {
if (!Array.isArray(raw)) continue;
const queue: StoredMessage[] = [];
let bytes = 0;
for (const entry of raw) {
if (!isStoredMessage(entry) || entry.createdAt <= cutoff) continue;
const size = Buffer.byteLength(entry.text, 'utf8');
if (queue.length >= MAX_MESSAGES_PER_INBOX || bytes + size > MAX_INBOX_BYTES) break;
queue.push({ ...entry });
bytes += size;
}
if (queue.length === 0) continue;
this.inboxes.set(sessionId, queue);
restored += queue.length;
}
return restored;
}
/** Test hook: back to an empty store with no listener. */
resetForTests(): void {
this.inboxes.clear();
this.onChange = undefined;
}
/** The inbox with expired messages swept out (fires `onChange` when it swept any). */
private live(sessionId: string): StoredMessage[] {
const queue = this.inboxes.get(sessionId);
if (!queue) return [];
const cutoff = this.now() - MESSAGE_TTL_MS;
if (queue.length === 0 || queue[0].createdAt > cutoff) return queue;
// Oldest first, so the expired ones are a prefix.
const kept = queue.filter((m) => m.createdAt > cutoff);
this.store(sessionId, kept);
this.onChange?.();
return kept;
}
private store(sessionId: string, queue: StoredMessage[]): void {
if (queue.length === 0) this.inboxes.delete(sessionId);
else this.inboxes.set(sessionId, queue);
}
}
function view(m: StoredMessage): InboxMessage {
return { id: m.id, from: m.from, text: m.text, createdAt: m.createdAt, seen: m.seenAt !== null };
}
function countUnseen(queue: readonly StoredMessage[]): number {
let n = 0;
for (const m of queue) if (m.seenAt === null) n++;
return n;
}
function inboxBytes(queue: readonly StoredMessage[]): number {
let total = 0;
for (const m of queue) total += Buffer.byteLength(m.text, 'utf8');
return total;
}
function isStoredMessage(value: unknown): value is StoredMessage {
const m = value as Partial<StoredMessage> | null;
return (
!!m &&
typeof m.id === 'string' &&
(m.from === null || typeof m.from === 'string') &&
typeof m.text === 'string' &&
m.text.length > 0 &&
Buffer.byteLength(m.text, 'utf8') <= MAX_MESSAGE_BYTES &&
typeof m.createdAt === 'number' &&
Number.isFinite(m.createdAt) &&
(m.seenAt === null || typeof m.seenAt === 'number')
);
}
/** Process-wide store used by the routes and the server wiring. Tests construct their own. */
export const agentInbox = new AgentInbox();
+7
View File
@@ -2145,6 +2145,13 @@
<label class="set-chip" data-preview="header" data-preview-order="12"><input type="checkbox" id="appSettingsShowUltracodeAgents"><svg class="set-chip-ico" width="14" height="14" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="1.9" stroke-linecap="round" stroke-linejoin="round" aria-hidden="true"><circle cx="6" cy="6" r="2.5"/><circle cx="6" cy="18" r="2.5"/><circle cx="18" cy="12" r="2.5"/><path d="M8.2 7.2 15.6 11M8.2 16.8 15.6 13"/></svg><span>Ultracode Agents</span></label>
<label class="set-chip" data-preview="float" data-preview-order="1"><input type="checkbox" id="appSettingsUltracodeFloatingWindows"><svg class="set-chip-ico" width="14" height="14" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="1.9" stroke-linecap="round" stroke-linejoin="round" aria-hidden="true"><rect x="3" y="4" width="18" height="16" rx="2"/><path d="M3 9h18"/><circle cx="6" cy="6.5" r="0.6"/></svg><span>Ultracode Windows</span></label>
</div>
<div class="set-row" data-search="agent inbox messages persist restart disk codeman agent post">
<div class="set-row-text">
<span class="set-row-label">Keep agent inbox across restarts</span>
<span class="set-row-desc">Saves pending <code>codeman agent post</code> messages to <code>agent-inbox.json</code> in the data dir (readable by you only), so they survive a server restart. Off by default; turning it off deletes the file. Messages expire after 12 hours either way.</span>
</div>
<label class="switch switch-sm"><input type="checkbox" id="appSettingsAgentInboxPersist"><span class="slider"></span></label>
</div>
</div>
</div>
+3
View File
@@ -536,6 +536,8 @@ Object.assign(CodemanApp.prototype, {
document.getElementById('appSettingsShowUltracodeAgents').checked = settings.showUltracodeAgents ?? defaults.showUltracodeAgents ?? false;
// Approvals Inbox: synced, default OFF (opt-in; only an explicit true enables).
document.getElementById('appSettingsApprovalsInbox').checked = settings.approvalsInboxEnabled === true;
// Agent inbox snapshot on disk: synced, default OFF.
document.getElementById('appSettingsAgentInboxPersist').checked = settings.agentInboxPersistEnabled === true;
// Custom Model Endpoint Profiles: synced, default OFF. The toggle governs both
// the Run-menu picker's generated entries and this settings panel's visibility;
// the endpoint list itself is server state, loaded on demand below.
@@ -2921,6 +2923,7 @@ Object.assign(CodemanApp.prototype, {
configBackupEnabled: document.getElementById('appSettingsConfigBackup').checked,
configBackupDir: document.getElementById('appSettingsConfigBackupDir').value.trim(),
...this._configBackupRetentionPayload(_prev),
agentInboxPersistEnabled: document.getElementById('appSettingsAgentInboxPersist').checked,
readMyMindEnabled: document.getElementById('appSettingsReadMyMind').checked,
ultracodeFloatingWindows: document.getElementById('appSettingsUltracodeFloatingWindows').checked,
showMultiMonitorButton: document.getElementById('appSettingsShowMultiMonitorButton').checked,
+49 -1
View File
@@ -11,7 +11,8 @@ import fs from 'node:fs/promises';
import type { z } from 'zod';
import type { FastifyReply, FastifyRequest } from 'fastify';
import { Session } from '../session.js';
import { ApiErrorCode, createErrorResponse, type AuthUser, type SessionState } from '../types.js';
import { ApiErrorCode, createErrorResponse, type ApiResponse, type AuthUser, type SessionState } from '../types.js';
import type { WaitCapacityError } from './session-wait-registry.js';
import { MAX_CONCURRENT_SESSIONS } from '../config/map-limits.js';
import { parseRalphLoopConfig, extractCompletionPhrase } from '../ralph-config.js';
import { SseEvent } from './sse-events.js';
@@ -627,3 +628,50 @@ export function autoConfigureRalph(session: Session, workingDir: string, ctx: Ev
});
}
}
// ========== Long-poll helpers (shared by the wait routes and the inbox routes) ==========
/**
* Map a waiter-cap rejection to the code that tells the caller the truth.
*
* The two caps mean different things and warrant different recovery: `session` is
* genuinely about THIS session, while `owner` and `total` are process-wide budgets
* that say nothing about it. Reporting a global cap as `SESSION_BUSY` (409,
* documented as "Session is busy") sent an agent off to a different session to hit
* the identical error. `RATE_LIMITED` is the code whose whole meaning is "come back
* later", and clients and proxies already treat 429 that way.
*/
export function waitCapacityResponse(err: WaitCapacityError): ApiResponse<never> {
const code = err.scope === 'session' ? ApiErrorCode.SESSION_BUSY : ApiErrorCode.RATE_LIMITED;
// The registry's message already names the scope and the limit; passing it through
// verbatim keeps the wording in one place.
return createErrorResponse(code, err.message);
}
/**
* An `AbortController` that fires when the CLIENT goes away, and only then.
*
* Freeing an abandoned waiter matters because the documented pattern is a loop of
* short waits: `curl --max-time 30 ".../wait?timeout=600000"` abandons a live waiter
* every iteration until the cap is hit and an innocent session reports busy. Same for
* any proxy that cuts the connection.
*
* ⚠️ **It must listen on the RESPONSE, not the request.** `req.raw` emits `'close'`
* as soon as the request body has finished streaming, which on a POST is BEFORE the
* handler ever blocks — measured at +1ms with `aborted: false`, indistinguishable
* from a real hang-up at +0ms. Wiring the abort there cancels every send-and-wait
* instantly and silently kills the feature (it survives on GET only because a GET has
* no body to finish). `reply.raw` emits `'close'` both when the response completes
* and when the socket dies, and `writableFinished` is what tells those apart: true
* only if the response actually went out. The guard is load-bearing, not defensive.
*
* `app.inject()` never emits `'close'` at all, so this is only observable over real
* HTTP — which is why the regression test for it binds a port.
*/
export function abortOnClientHangUp(reply: FastifyReply): AbortController {
const controller = new AbortController();
reply.raw.on('close', () => {
if (!reply.raw.writableFinished) controller.abort();
});
return controller;
}
+181
View File
@@ -0,0 +1,181 @@
/**
* @fileoverview Agent inbox routes: post, read (optionally long-polling), ack, and an
* owner-scoped summary of counts. Store semantics live in `agent-inbox.ts`.
*
* Who is calling decides what a read or an ack does, so it is resolved here and never
* taken from the body. `resolveParentSessionId()` reads the `X-Codeman-Parent-Session`
* header `codeman agent` sends on every request and accepts it only for a live session
* the caller can see that has the same owner as the receiver (exact id, or a unique
* prefix of at least 8). Anything else is "no session": a post is from the user, and a
* read is a peek. What that buys: in multi-user mode a caller cannot name another
* owner's session. In single-user mode every local process holds the same
* credentials, so the header is a hint the client sets about itself.
*
* - `POST /api/sessions/:id/inbox` stores a message. `from` is the resolved caller or
* null. A full inbox is 409 CONFLICT; the oldest message is never dropped.
* - `GET /api/sessions/:id/inbox[?wait=ms]` returns every pending message. Only the
* receiver's own read marks them seen. With `wait`, it parks while nothing is
* UNSEEN (a seen-but-unacked message does not end the wait, or a receiver holding
* one would spin).
* - `POST /api/sessions/:id/inbox/ack` removes messages by id, idempotently, and only
* for the receiver itself (403 otherwise).
* - `GET /api/agent-inbox/summary` reports `pending`, `unseen`, `oldestUnseenAt` and
* `parked` per session the caller can see. Counts only, never text, and kept out of
* `toState()` so a post does not fan out a `session:updated` to every browser.
*
* Nothing here writes to a pane.
*
* @module web/routes/inbox-routes
*/
import { FastifyInstance } from 'fastify';
import { ApiErrorCode, createErrorResponse } from '../../types.js';
import { InboxAckSchema, InboxPostSchema, InboxReadQuerySchema } from '../schemas.js';
import {
abortOnClientHangUp,
canAccessOwned,
findSessionOrFail,
getAuthUser,
ownerFor,
parseBody,
resolveParentSessionId,
waitCapacityResponse,
} from '../route-helpers.js';
import { agentInbox, MAX_INBOX_BYTES, MAX_MESSAGE_BYTES, MAX_MESSAGES_PER_INBOX } from '../agent-inbox.js';
import { sessionWaits, WaitCapacityError, type InboxWaitResult } from '../session-wait-registry.js';
import { clampWaitMs } from '../../config/agent-wait.js';
import type { SessionPort } from '../ports/index.js';
/** Per-session counts the summary route reports. */
export interface InboxSessionSummary {
pending: number;
unseen: number;
oldestUnseenAt: number | null;
/** The receiver itself is parked on `GET …/inbox?wait=` right now. */
parked: boolean;
}
export function registerInboxRoutes(app: FastifyInstance, ctx: SessionPort): void {
/** The resolved caller is the receiving session itself. */
const callerIsReceiver = (req: Parameters<typeof resolveParentSessionId>[1], id: string, owner?: string) =>
resolveParentSessionId(ctx, req, undefined, owner) === id;
app.post('/api/sessions/:id/inbox', async (req) => {
const { id } = req.params as { id: string };
const session = findSessionOrFail(ctx, id, req);
const { text } = parseBody(InboxPostSchema, req.body, 'Invalid inbox message');
const from = resolveParentSessionId(ctx, req, undefined, session.owner) ?? null;
const result = agentInbox.post(id, from, text);
if (!result.ok) {
switch (result.reason) {
case 'empty':
return createErrorResponse(ApiErrorCode.INVALID_INPUT, 'text must not be empty');
case 'message-too-large':
return createErrorResponse(
ApiErrorCode.INVALID_INPUT,
`Message is larger than ${MAX_MESSAGE_BYTES} bytes (UTF-8). Write it to a file in the workspace and post the path.`
);
case 'inbox-full-count':
return createErrorResponse(
ApiErrorCode.CONFLICT,
`Inbox of ${id} is full (${MAX_MESSAGES_PER_INBOX} messages). Nothing was stored; the receiver has to ack before more can be posted.`
);
case 'inbox-full-bytes':
return createErrorResponse(
ApiErrorCode.CONFLICT,
`Inbox of ${id} is full (${MAX_INBOX_BYTES} bytes of pending text). Nothing was stored; the receiver has to ack before more can be posted.`
);
}
}
// After the store, so a parked reader that wakes finds the message.
sessionWaits.notifyInbox(id);
return {
success: true,
data: { sessionId: id, message: result.message, pending: result.pending, unseen: result.unseen },
};
});
app.get('/api/sessions/:id/inbox', async (req, reply) => {
const { id } = req.params as { id: string };
const session = findSessionOrFail(ctx, id, req);
const query = parseBody(InboxReadQuerySchema, req.query);
const receiver = callerIsReceiver(req, id, session.owner);
// A receiver loops on this URL with identical parameters; a cached 200 would turn
// the loop into a busy spin (same reason as GET …/wait).
reply.header('Cache-Control', 'no-store');
let wait: InboxWaitResult | undefined;
if (query.wait !== undefined) {
const timeoutMs = clampWaitMs(query.wait);
if (agentInbox.unseenCount(id) > 0) {
wait = {
posted: false,
timedOut: false,
immediate: true,
ended: false,
aborted: false,
waitedMs: 0,
timeoutMs,
};
} else {
const abort = abortOnClientHangUp(reply);
try {
wait = await sessionWaits.waitForInbox(id, {
timeoutMs,
owner: ownerFor(req),
abortSignal: abort.signal,
receiver,
});
} catch (err) {
if (err instanceof WaitCapacityError) return waitCapacityResponse(err);
throw err;
}
}
}
// Read AFTER the wait: this is what the caller gets, so it must be current. A wait
// that ended because the session was deleted reads an empty inbox.
const read = agentInbox.read(id, { receiver });
return { success: true, data: { sessionId: id, receiver, ...read, ...(wait ? { wait } : {}) } };
});
app.post('/api/sessions/:id/inbox/ack', async (req) => {
const { id } = req.params as { id: string };
const session = findSessionOrFail(ctx, id, req);
const { ids } = parseBody(InboxAckSchema, req.body);
if (!callerIsReceiver(req, id, session.owner)) {
// A no-op here would let a coordinator believe it cleared a worker's mail, and
// would hide a client that forgot the header. Refusing says both out loud.
return createErrorResponse(
ApiErrorCode.FORBIDDEN,
`Only ${id} itself can acknowledge its inbox (X-Codeman-Parent-Session must resolve to it). Reading as anyone else is a peek.`
);
}
const removed = agentInbox.ack(id, ids);
return {
success: true,
data: { sessionId: id, removed, pending: agentInbox.pendingCount(id), unseen: agentInbox.unseenCount(id) },
};
});
app.get('/api/agent-inbox/summary', async (req, reply) => {
reply.header('Cache-Control', 'no-store');
const user = getAuthUser(req);
const counts = agentInbox.summary();
const sessions: Record<string, InboxSessionSummary> = {};
for (const session of ctx.sessions.values()) {
if (!canAccessOwned(user, session.owner)) continue;
const c = counts.get(session.id);
const parked = sessionWaits.inboxWaiterCount(session.id, true) > 0;
if (!c && !parked) continue;
sessions[session.id] = {
pending: c?.pending ?? 0,
unseen: c?.unseen ?? 0,
oldestUnseenAt: c?.oldestUnseenAt ?? null,
parked,
};
}
return { success: true, data: { sessions } };
});
}
+1
View File
@@ -12,6 +12,7 @@ export { registerCronRoutes } from './cron-routes.js';
export { registerSystemRoutes } from './system-routes.js';
export { registerHookEventRoutes } from './hook-event-routes.js';
export { registerApprovalRoutes } from './approval-routes.js';
export { registerInboxRoutes } from './inbox-routes.js';
export { registerRebootRestoreRoutes } from './reboot-restore-routes.js';
export { registerReadMyMindRoutes } from './readmymind-routes.js';
export { registerGitStatusRoutes } from './git-status-routes.js';
+3 -46
View File
@@ -4,7 +4,7 @@
* auto-clear, auto-compact, image watcher, flicker filter, and logout.
*/
import { FastifyInstance, type FastifyReply, type FastifyRequest } from 'fastify';
import { FastifyInstance, type FastifyRequest } from 'fastify';
import { z } from 'zod';
import { join, dirname, extname, basename, resolve as resolvePath } from 'node:path';
import { homedir } from 'node:os';
@@ -115,6 +115,8 @@ import {
sessionCapacityMessage,
SETTINGS_PATH,
validatePathWithinBase,
waitCapacityResponse,
abortOnClientHangUp,
} from '../route-helpers.js';
import { buildAgentCaseMarker, writeAgentCaseMarker } from '../../agent-case-marker.js';
import { canUsernameRunPrivilegedCommands, resolveClaudeModeForUsername } from '../../user-store.js';
@@ -738,23 +740,6 @@ function parseWaitQuery<T>(schema: z.ZodType<T>, query: unknown, label: string):
});
}
/**
* Map a waiter-cap rejection to the code that tells the caller the truth.
*
* The two caps mean different things and warrant different recovery: `session` is
* genuinely about THIS session, while `owner` and `total` are process-wide budgets
* that say nothing about it. Reporting a global cap as `SESSION_BUSY` (409,
* documented as "Session is busy") sent an agent off to a different session to hit
* the identical error. `RATE_LIMITED` is the code whose whole meaning is "come back
* later", and clients and proxies already treat 429 that way.
*/
function waitCapacityResponse(err: WaitCapacityError): ApiResponse<never> {
const code = err.scope === 'session' ? ApiErrorCode.SESSION_BUSY : ApiErrorCode.RATE_LIMITED;
// The registry's message already names the scope and the limit; passing it through
// verbatim keeps the wording in one place.
return createErrorResponse(code, err.message);
}
/**
* The signal a session is ALREADY emitting, corrected for liveness.
*
@@ -931,34 +916,6 @@ export function _paneDeathWatcherCount(): number {
return paneDeathWatchers.size;
}
/**
* An `AbortController` that fires when the CLIENT goes away, and only then.
*
* Freeing an abandoned waiter matters because the documented pattern is a loop of
* short waits: `curl --max-time 30 ".../wait?timeout=600000"` abandons a live waiter
* every iteration until the cap is hit and an innocent session reports busy. Same for
* any proxy that cuts the connection.
*
* ⚠️ **It must listen on the RESPONSE, not the request.** `req.raw` emits `'close'`
* as soon as the request body has finished streaming, which on a POST is BEFORE the
* handler ever blocks — measured at +1ms with `aborted: false`, indistinguishable
* from a real hang-up at +0ms. Wiring the abort there cancels every send-and-wait
* instantly and silently kills the feature (it survives on GET only because a GET has
* no body to finish). `reply.raw` emits `'close'` both when the response completes
* and when the socket dies, and `writableFinished` is what tells those apart: true
* only if the response actually went out. The guard is load-bearing, not defensive.
*
* `app.inject()` never emits `'close'` at all, so this is only observable over real
* HTTP — which is why the regression test for it binds a port.
*/
function abortOnClientHangUp(reply: FastifyReply): AbortController {
const controller = new AbortController();
reply.raw.on('close', () => {
if (!reply.raw.writableFinished) controller.abort();
});
return controller;
}
/**
* Inject the agent skill into a case on create, surfacing only the REFUSALS.
*
+5
View File
@@ -64,6 +64,7 @@ import { AUTH_COOKIE_NAME } from '../middleware/auth.js';
import { QR_AUTH_FAILURE_MAX } from '../../config/tunnel-config.js';
import { AUTH_SESSION_TTL_MS } from '../../config/auth-config.js';
import { resolveTerminalHistoryConfig } from '../../config/terminal-history.js';
import { agentInboxPersistence } from '../agent-inbox-persistence.js';
/**
* Defaults for `POST /api/deepseek/install-profile`.
@@ -1129,6 +1130,10 @@ export function registerSystemRoutes(
'Workflow run watcher'
);
// Agent inbox snapshot (default OFF). From `merged` like the toggles above: a
// partial PUT that omits the key must not switch it off and delete the file.
agentInboxPersistence.setEnabled(merged.agentInboxPersistEnabled === true);
// Handle image watcher toggle dynamically
toggleService((merged.imageWatcherEnabled as boolean) ?? false, imageWatcher, 'Image watcher', () => {
// Re-watch all active sessions that have image watcher enabled
+32
View File
@@ -1537,6 +1537,12 @@ export const SettingsUpdateSchema = z
configBackupKeepCount: z.number().int().min(2).max(500).optional(),
/** Also drop snapshots older than this many days; 0 = no age limit. */
configBackupKeepDays: z.number().int().min(0).max(3650).optional(),
/**
* Agent inbox snapshot on disk (src/web/agent-inbox-persistence.ts). SYNCED, default OFF:
* off means no file is written, and switching it off deletes the snapshot (the in-memory
* inboxes stay). On, pending messages between agents survive a server restart.
*/
agentInboxPersistEnabled: z.boolean().optional(),
/**
* Read My Mind predictor model override. Empty/absent = the AI-checker
* default (opus: prediction quality is the product and it runs only on an
@@ -1836,6 +1842,32 @@ export const SessionWaitOutputQuerySchema = z.object({
timeout: z.coerce.number().int().positive().optional(),
});
// ========== Agent Inbox Routes ==========
/**
* Body for `POST /api/sessions/:id/inbox`. The byte cap (`MAX_MESSAGE_BYTES`) is the
* store's, checked there in UTF-8 bytes; this bound is only the outer fence so a
* multi-megabyte body never reaches it. Multi-line text is fine: it is stored and
* handed back as data, never typed.
*/
export const InboxPostSchema = z
.object({
text: z.string().min(1, 'text must not be empty').max(16_384),
})
.strict();
/** Query for `GET /api/sessions/:id/inbox`. `wait` follows the wait routes' `timeout`: clamped, never a 400 for size. */
export const InboxReadQuerySchema = z.object({
wait: z.coerce.number().int().positive().optional(),
});
/** Body for `POST /api/sessions/:id/inbox/ack`: the ids a read handed out. */
export const InboxAckSchema = z
.object({
ids: z.array(z.string().min(1).max(64)).min(1).max(200),
})
.strict();
// ========== Session Mutation Routes ==========
/** PUT /api/sessions/:id/name */
+2
View File
@@ -100,6 +100,8 @@ const SENSITIVE_PATTERNS: RegExp[] = [
/\/\.codeman[^/]*\/settings\.json$/,
/\/\.codeman[^/]*\/push-keys\.json$/,
/\/\.codeman[^/]*\/intents\.json$/,
// Pending agent-inbox messages (opt-in snapshot): same text as prompts, same 0600.
/\/\.codeman[^/]*\/agent-inbox\.json$/,
];
/**
+28
View File
@@ -116,6 +116,8 @@ import {
} from './session-listener-wiring.js';
import { sessionWaits } from './session-wait-registry.js';
import { intentStore } from '../intent-store.js';
import { agentInbox } from './agent-inbox.js';
import { agentInboxPersistence } from './agent-inbox-persistence.js';
import { AI_CHECK_MODEL } from '../config/ai-defaults.js';
import { approvalInbox } from './approval-inbox.js';
import { SCREEN_BLOCKED_SOURCE } from './screen-blocked.js';
@@ -201,6 +203,7 @@ import {
registerScheduledRoutes,
registerHookEventRoutes,
registerApprovalRoutes,
registerInboxRoutes,
registerRebootRestoreRoutes,
registerReadMyMindRoutes,
registerGitStatusRoutes,
@@ -1229,6 +1232,7 @@ export class WebServer extends EventEmitter {
registerScheduledRoutes(this.app, ctx);
registerHookEventRoutes(this.app, ctx);
registerApprovalRoutes(this.app, ctx);
registerInboxRoutes(this.app, ctx);
registerRebootRestoreRoutes(this.app, ctx);
registerReadMyMindRoutes(this.app, ctx);
registerGitStatusRoutes(this.app, ctx);
@@ -1686,6 +1690,10 @@ export class WebServer extends EventEmitter {
// The 'exit' here is NOT redundant with the PTY-exit listener: listeners are
// detached a few lines above, before `session.stop()`, so on a delete the
// session's own exit event never reaches the registry.
// The agent inbox goes with a DELETED session only. killMux=false is shutdown or
// detach, and a detached session comes back to the mail posted while it was away.
// Before cancelAll, so a reader released by it reads an empty inbox.
if (killMux) agentInbox.drop(sessionId);
sessionWaits.notifySignal(sessionId, 'exit');
sessionWaits.cancelAll(sessionId);
approvalInbox.resolveForSession(sessionId, 'session_ended');
@@ -2211,6 +2219,13 @@ export class WebServer extends EventEmitter {
return settings.agentSkillEnabled === true;
}
// Whether pending agent-inbox mail is snapshotted to disk (synced
// `agentInboxPersistEnabled` setting). Default OFF: an absent key writes nothing.
private async getAgentInboxPersistEnabled(): Promise<boolean> {
const settings = await this.readSettings();
return settings.agentInboxPersistEnabled === true;
}
// Whether a Claude session installs Codeman's hooks block into its workspace
// (synced `workspaceHooksEnabled` setting). Default ON — an absent key means a
// user who has never seen this setting, and OFF for them would mean no tab
@@ -3041,6 +3056,10 @@ export class WebServer extends EventEmitter {
// This prevents race conditions where clients connect before state is ready
// CRITICAL: Skip in test mode to prevent tests from picking up user sessions
if (!this.testMode) {
// Pending agent mail first (opt-in snapshot), so it is in place before any
// route can answer. Pruned against the restored sessions in finalizeRestoredState.
const restoredMail = agentInboxPersistence.load(await this.getAgentInboxPersistEnabled());
if (restoredMail > 0) console.log(`[AgentInbox] restored ${restoredMail} pending message(s)`);
const restored = await this.restoreMuxSessions();
await this.finalizeRestoredState(restored);
@@ -4013,6 +4032,13 @@ export class WebServer extends EventEmitter {
this.tabLayouts.markRestorationComplete();
await this.cleanupStaleSessions();
await this.tabLayouts.reconcileAfterRestoration();
// Same gate as the tab layouts: only a restore that succeeded may decide which
// inboxes are orphans. A failed one returns above and leaves every message alone.
// Kept: live sessions plus every record state.json still holds (a session that
// was not re-adopted this boot but is kept for recovery keeps its mail).
const keep = new Set([...this.sessions.keys(), ...Object.keys(this.store.getSessions())]);
const pruned = agentInbox.pruneExcept(keep);
if (pruned > 0) console.log(`[AgentInbox] dropped ${pruned} inbox(es) of sessions that did not come back`);
}
/**
@@ -4189,6 +4215,8 @@ export class WebServer extends EventEmitter {
sessionWaits.cancelEverything();
approvalInbox.stop();
this.keepWarmService.stop();
// A post within the debounce window before shutdown would otherwise not reach disk.
agentInboxPersistence.flush();
// Same reason as `cancelEverything` above: an in-flight wake is awaited by a request,
// and `app.close()` (the last line of this method) does not abort in-flight requests —
// so without this a restart during a wake waits out the readiness poll.
+163 -7
View File
@@ -7,11 +7,16 @@
* lets a request *block* until something happens instead, so an orchestrating
* agent can `curl` and wait. Plan: `docs/agent-control-plan.md`.
*
* Two waiter kinds:
* Three waiter kinds:
* - **signal waiters** resolve on the first of a requested set of lifecycle
* signals (`idle`, `working`, `stop`, `blocked`, `exit`).
* - **output waiters** resolve when a literal string appears in a session's
* ANSI-stripped output.
* - **inbox waiters** resolve when a message is posted to the session's agent
* inbox (`agent-inbox.ts`). The registry knows nothing about messages: the
* route decides whether to park (nothing unseen) and `notifyInbox()` is the
* post's "something arrived". Kept here so `GET .../inbox?wait=` is charged to
* the same per-session / per-owner / total caps as the other two.
*
* ## What "ANSI-stripped output" means here
*
@@ -390,6 +395,12 @@ export interface OutputWaitResult extends WaitResultBase {
snippet: string | null;
}
/** Outcome of an inbox wait. */
export interface InboxWaitResult extends WaitResultBase {
/** True when a post released the wait (the caller should read the inbox now). */
posted: boolean;
}
/**
* Thrown when a waiter would exceed a configured cap.
*
@@ -458,6 +469,15 @@ export interface OutputWaitOptions extends WaitOptionsBase {
initialText?: string;
}
export interface InboxWaitOptions extends WaitOptionsBase {
/**
* The caller IS the receiving session (as resolved by the route), not a coordinator
* peeking. Counted separately so the inbox summary's `parked` means "the receiver is
* listening", which is what a sender deciding whether to nudge needs to know.
*/
receiver?: boolean;
}
export interface SessionWaitRegistryOptions {
maxWaitersPerSession?: number;
maxWaitersPerOwner?: number;
@@ -483,6 +503,11 @@ interface SignalWaiter extends WaiterBase {
settle: (result: SignalWaitResult) => void;
}
interface InboxWaiter extends WaiterBase {
receiver: boolean;
settle: (result: InboxWaitResult) => void;
}
interface OutputWaiter extends WaiterBase {
/** Original-case needle, kept for reporting. */
needle: string;
@@ -505,6 +530,7 @@ interface OutputWaiter extends WaiterBase {
export class SessionWaitRegistry {
private readonly signalWaiters = new Map<string, Set<SignalWaiter>>();
private readonly outputWaiters = new Map<string, Set<OutputWaiter>>();
private readonly inboxWaiters = new Map<string, Set<InboxWaiter>>();
/**
* Live waiter count per owner. A derived count would mean walking every waiter in
* the process on each registration, so it is maintained incrementally instead:
@@ -544,20 +570,34 @@ export class SessionWaitRegistry {
return this.outputWaiters.get(sessionId)?.size ?? 0;
}
/** Pending waiters of both kinds for a session. */
waiterCount(sessionId: string): number {
return this.signalWaiterCount(sessionId) + this.outputWaiterCount(sessionId);
/**
* Pending inbox waiters for a session; with `receiverOnly`, only those whose caller
* is the receiving session itself (the inbox summary's `parked`).
*/
inboxWaiterCount(sessionId: string, receiverOnly = false): number {
const set = this.inboxWaiters.get(sessionId);
if (!set) return 0;
if (!receiverOnly) return set.size;
let count = 0;
for (const waiter of set) if (waiter.receiver) count++;
return count;
}
/** Pending waiters of both kinds across every session. */
/** Pending waiters of every kind for a session. */
waiterCount(sessionId: string): number {
return this.signalWaiterCount(sessionId) + this.outputWaiterCount(sessionId) + this.inboxWaiterCount(sessionId);
}
/** Pending waiters of every kind across every session. */
totalWaiterCount(): number {
let total = 0;
for (const set of this.signalWaiters.values()) total += set.size;
for (const set of this.outputWaiters.values()) total += set.size;
for (const set of this.inboxWaiters.values()) total += set.size;
return total;
}
/** Pending waiters of both kinds charged to one owner. */
/** Pending waiters of every kind charged to one owner. */
ownerWaiterCount(owner: string): number {
return this.ownerCounts.get(owner) ?? 0;
}
@@ -920,6 +960,105 @@ export class SessionWaitRegistry {
}
}
// ── Inbox waits ──
/**
* Block until a message is posted to this session's inbox, the timeout elapses, or
* the session ends.
*
* There is no "already satisfied" shortcut here: whether the inbox holds unseen mail
* is the store's business, so the route checks that first and only calls this when
* there is nothing to hand out. A post that lands between that check and this call
* is not lost either: both run in the same tick, and `notifyInbox` is called by the
* post route after the message is stored.
*
* @throws {WaitCapacityError} when a waiter cap is already reached.
*/
waitForInbox(sessionId: string, options: InboxWaitOptions): Promise<InboxWaitResult> {
const base = { posted: false, timedOut: false, immediate: false, waitedMs: 0, timeoutMs: options.timeoutMs };
if (options.abortSignal?.aborted) return Promise.resolve({ ...base, ended: true, aborted: true });
if (this.stopped) return Promise.resolve({ ...base, ended: true, aborted: false });
this.assertCapacity(sessionId, options.owner);
return new Promise<InboxWaitResult>((resolve) => {
let set = this.inboxWaiters.get(sessionId);
if (!set) {
set = new Set<InboxWaiter>();
this.inboxWaiters.set(sessionId, set);
}
const waiter: InboxWaiter = {
receiver: options.receiver === true,
startedAt: Date.now(),
timeoutMs: options.timeoutMs,
owner: options.owner,
timer: setTimeout(() => {
this.removeInboxWaiter(sessionId, waiter);
resolve({
posted: false,
timedOut: true,
immediate: false,
ended: false,
aborted: false,
waitedMs: Date.now() - waiter.startedAt,
timeoutMs: waiter.timeoutMs,
});
}, options.timeoutMs),
settle: resolve,
};
set.add(waiter);
this.chargeOwner(options.owner);
waiter.detachAbort = this.attachAbort(options.abortSignal, () => {
this.removeInboxWaiter(sessionId, waiter);
resolve({
posted: false,
timedOut: false,
immediate: false,
ended: true,
aborted: true,
waitedMs: Date.now() - waiter.startedAt,
timeoutMs: waiter.timeoutMs,
});
});
});
}
/**
* A message was posted to this session's inbox: release every inbox waiter on it.
*
* @returns how many waiters were resolved.
*/
notifyInbox(sessionId: string): number {
const set = this.inboxWaiters.get(sessionId);
if (!set || set.size === 0) return 0;
let resolved = 0;
for (const waiter of [...set]) {
this.removeInboxWaiter(sessionId, waiter);
waiter.settle({
posted: true,
timedOut: false,
immediate: false,
ended: false,
aborted: false,
waitedMs: Date.now() - waiter.startedAt,
timeoutMs: waiter.timeoutMs,
});
resolved++;
}
return resolved;
}
private removeInboxWaiter(sessionId: string, waiter: InboxWaiter): void {
clearTimeout(waiter.timer);
waiter.detachAbort?.();
const set = this.inboxWaiters.get(sessionId);
if (!set) return;
if (!set.delete(waiter)) return;
this.releaseOwner(waiter.owner);
if (set.size === 0) this.inboxWaiters.delete(sessionId);
}
// ── Teardown ──
/**
@@ -969,6 +1108,23 @@ export class SessionWaitRegistry {
}
}
const inbox = this.inboxWaiters.get(sessionId);
if (inbox) {
for (const waiter of [...inbox]) {
this.removeInboxWaiter(sessionId, waiter);
waiter.settle({
posted: false,
timedOut: false,
immediate: false,
ended: true,
aborted: false,
waitedMs: Date.now() - waiter.startedAt,
timeoutMs: waiter.timeoutMs,
});
resolved++;
}
}
this.pendingAnsi.delete(sessionId);
return resolved;
}
@@ -999,7 +1155,7 @@ export class SessionWaitRegistry {
cancelEverything(): number {
this.stopped = true;
let resolved = 0;
const ids = new Set([...this.signalWaiters.keys(), ...this.outputWaiters.keys()]);
const ids = new Set([...this.signalWaiters.keys(), ...this.outputWaiters.keys(), ...this.inboxWaiters.keys()]);
for (const id of ids) resolved += this.cancelAll(id);
return resolved;
}