/** * @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; } // ─── Store ─────────────────────────────────────────────────────────────────── export class AgentInbox { private readonly inboxes = new Map(); /** 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 { const out = new Map(); 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): 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 = {}; 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 | 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; // Oldest first whatever order the file has, so the caps keep the oldest mail, // which is the rule for posts too. const entries = raw .filter((entry): entry is StoredMessage => isStoredMessage(entry) && entry.createdAt > cutoff) .sort((a, b) => a.createdAt - b.createdAt); for (const entry of entries) { 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; // Every message, not only queue[0]: a clock stepping back can put an older // createdAt behind a newer one. At most 200 entries, so the scan is cheap. if (!queue.some((m) => m.createdAt <= cutoff)) return queue; 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 | 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();