mirror of
https://github.com/Ark0N/Codeman.git
synced 2026-11-15 10:59:32 +01:00
Conflict in src/web/routes/session-routes.ts (the import block) resolved by keeping both sides: #680's claudeProjectDirName import stays, and the fastify import takes #690's form without FastifyReply, since #690 moved abortOnClientHangUp (its only user here) into route-helpers.ts. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Codeman maintainer committed
commit
e739b2e9d4
39 files changed
+2664
-86
No files matched your search
@@ -0,0 +1,104 @@
|
||||
/**
|
||||
* @fileoverview Opt-in agent inbox snapshot (src/web/agent-inbox-persistence.ts).
|
||||
*
|
||||
* Off writes nothing, switching off deletes the file and keeps memory, on writes 0600
|
||||
* through tmp + rename, and a boot with the setting on restores what was pending.
|
||||
*/
|
||||
import { describe, it, expect, beforeEach, afterEach } from 'vitest';
|
||||
import { existsSync, mkdtempSync, readFileSync, rmSync, statSync, writeFileSync } from 'node:fs';
|
||||
import { join } from 'node:path';
|
||||
import { tmpdir } from 'node:os';
|
||||
import { AgentInbox } from '../src/web/agent-inbox.js';
|
||||
import { AgentInboxPersistence } from '../src/web/agent-inbox-persistence.js';
|
||||
|
||||
describe('AgentInboxPersistence', () => {
|
||||
let dir: string;
|
||||
let file: string;
|
||||
let inbox: AgentInbox;
|
||||
let persistence: AgentInboxPersistence;
|
||||
|
||||
beforeEach(() => {
|
||||
dir = mkdtempSync(join(tmpdir(), 'codeman-inbox-'));
|
||||
file = join(dir, 'agent-inbox.json');
|
||||
inbox = new AgentInbox();
|
||||
// Debounce 0 so a test can wait one macrotask instead of sleeping.
|
||||
persistence = new AgentInboxPersistence(inbox, () => file, 0);
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
persistence.dispose();
|
||||
rmSync(dir, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
const tick = () => new Promise((r) => setTimeout(r, 5));
|
||||
|
||||
it('writes nothing while off (the default)', async () => {
|
||||
persistence.load(false);
|
||||
inbox.post('s', null, 'hello');
|
||||
await tick();
|
||||
expect(existsSync(file)).toBe(false);
|
||||
});
|
||||
|
||||
it('writes a 0600 snapshot when on, and no tmp file is left behind', async () => {
|
||||
persistence.load(true);
|
||||
inbox.post('s', null, 'hello');
|
||||
await tick();
|
||||
expect(JSON.parse(readFileSync(file, 'utf-8')).inboxes.s[0].text).toBe('hello');
|
||||
expect(statSync(file).mode & 0o777).toBe(0o600);
|
||||
expect(existsSync(`${file}.tmp`)).toBe(false);
|
||||
});
|
||||
|
||||
it('switching on writes the current inboxes at once', () => {
|
||||
persistence.load(false);
|
||||
inbox.post('s', null, 'posted while off');
|
||||
persistence.setEnabled(true);
|
||||
expect(JSON.parse(readFileSync(file, 'utf-8')).inboxes.s).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('switching off deletes the snapshot and keeps the in-memory inboxes', async () => {
|
||||
persistence.load(true);
|
||||
inbox.post('s', null, 'hello');
|
||||
await tick();
|
||||
expect(existsSync(file)).toBe(true);
|
||||
persistence.setEnabled(false);
|
||||
expect(existsSync(file)).toBe(false);
|
||||
expect(inbox.pendingCount('s')).toBe(1);
|
||||
inbox.post('s', null, 'after off');
|
||||
await tick();
|
||||
expect(existsSync(file)).toBe(false);
|
||||
});
|
||||
|
||||
it('a boot with the setting on restores pending mail', async () => {
|
||||
persistence.load(true);
|
||||
inbox.post('s', 'sender', 'survives');
|
||||
await tick();
|
||||
persistence.dispose();
|
||||
|
||||
const next = new AgentInbox();
|
||||
const nextPersistence = new AgentInboxPersistence(next, () => file, 0);
|
||||
expect(nextPersistence.load(true)).toBe(1);
|
||||
expect(next.read('s', { receiver: false }).messages[0]).toMatchObject({ from: 'sender', text: 'survives' });
|
||||
nextPersistence.dispose();
|
||||
});
|
||||
|
||||
it('a boot with the setting off removes a file left from earlier and restores nothing', () => {
|
||||
writeFileSync(file, JSON.stringify({ version: 1, inboxes: { s: [] } }));
|
||||
expect(persistence.load(false)).toBe(0);
|
||||
expect(existsSync(file)).toBe(false);
|
||||
});
|
||||
|
||||
it('an unreadable file is an empty inbox, not a failed boot', () => {
|
||||
writeFileSync(file, '{not json');
|
||||
expect(persistence.load(true)).toBe(0);
|
||||
});
|
||||
|
||||
it('flush writes a pending change before shutdown', () => {
|
||||
const slow = new AgentInboxPersistence(inbox, () => file, 60_000);
|
||||
slow.load(true);
|
||||
inbox.post('s', null, 'just before shutdown');
|
||||
expect(existsSync(file)).toBe(false);
|
||||
slow.flush();
|
||||
expect(JSON.parse(readFileSync(file, 'utf-8')).inboxes.s).toHaveLength(1);
|
||||
slow.dispose();
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,112 @@
|
||||
/**
|
||||
* @fileoverview How server.ts treats the agent inbox across a session's lifecycle.
|
||||
*
|
||||
* - Delete (`killMux=true`) drops the inbox; detach and shutdown (`killMux=false`)
|
||||
* keep it, because a detached session comes back.
|
||||
* - Boot pruning of orphaned inboxes runs only after a successful restore, and keeps
|
||||
* every session state.json still holds plus every session on the reboot-restore
|
||||
* offer (its record is already pruned, and a rebuild reuses the id). A failed
|
||||
* restore leaves all mail alone.
|
||||
*
|
||||
* Uses a bare `WebServer.prototype` object with just the fields these paths touch,
|
||||
* like tab-layout-restore-gate.test.ts.
|
||||
*/
|
||||
import { describe, it, expect, beforeEach, afterEach } from 'vitest';
|
||||
import { WebServer } from '../src/web/server.js';
|
||||
import { agentInbox } from '../src/web/agent-inbox.js';
|
||||
import { rebootRestoreRegistry } from '../src/web/reboot-restore-registry.js';
|
||||
import type { RebootRestoreEntry } from '../src/reboot-restore.js';
|
||||
import type { SessionState } from '../src/types.js';
|
||||
|
||||
type CleanupServer = {
|
||||
_doCleanupSession(sessionId: string, killMux: boolean, reason?: string): Promise<void>;
|
||||
finalizeRestoredState(restored: boolean): Promise<void>;
|
||||
[key: string]: unknown;
|
||||
};
|
||||
|
||||
function bareServer(live: string[] = [], persisted: string[] = []): CleanupServer {
|
||||
const server = Object.create(WebServer.prototype) as CleanupServer;
|
||||
Object.assign(server, {
|
||||
sessions: new Map(live.map((id) => [id, {}])),
|
||||
keepWarmService: { remove: () => {} },
|
||||
store: {
|
||||
getSession: () => null,
|
||||
getSessions: () => Object.fromEntries(persisted.map((id) => [id, {}])),
|
||||
removeRalphState: () => {},
|
||||
},
|
||||
tabLayouts: {
|
||||
markRestorationComplete: () => {},
|
||||
markRestorationFailed: () => {},
|
||||
reconcileAfterRestoration: async () => {},
|
||||
},
|
||||
cleanupStaleSessions: async () => 0,
|
||||
_customModelDisplacedNotified: new Set(),
|
||||
respawnControllers: new Map(),
|
||||
respawnTimers: new Map(),
|
||||
pendingRespawnStarts: new Map(),
|
||||
runSummaryTrackers: new Map(),
|
||||
stopTranscriptWatcher: () => {},
|
||||
persistDeb: { cancelKey: () => {} },
|
||||
sse: { cleanupSessionBatches: () => {} },
|
||||
broadcast: () => {},
|
||||
});
|
||||
return server;
|
||||
}
|
||||
|
||||
/** A reboot-restore offer entry; the prune reads only its `sessionId`. */
|
||||
function offerEntry(sessionId: string): RebootRestoreEntry {
|
||||
return {
|
||||
sessionId,
|
||||
workingDir: `/tmp/codeman-agent-inbox-offer/${sessionId}`,
|
||||
mode: 'claude',
|
||||
resumeConversationId: `conv-${sessionId}`,
|
||||
state: { id: sessionId, workingDir: `/tmp/codeman-agent-inbox-offer/${sessionId}` } as SessionState,
|
||||
};
|
||||
}
|
||||
|
||||
describe('agent inbox server wiring', () => {
|
||||
beforeEach(() => agentInbox.resetForTests());
|
||||
afterEach(() => {
|
||||
agentInbox.resetForTests();
|
||||
rebootRestoreRegistry.reset();
|
||||
});
|
||||
|
||||
it('a delete drops the inbox; a detach (killMux=false) keeps it', async () => {
|
||||
const server = bareServer();
|
||||
agentInbox.post('detached', null, 'still here after detach');
|
||||
agentInbox.post('deleted', null, 'gone with the session');
|
||||
await server._doCleanupSession('detached', false, 'test');
|
||||
await server._doCleanupSession('deleted', true, 'test');
|
||||
expect(agentInbox.pendingCount('detached')).toBe(1);
|
||||
expect(agentInbox.pendingCount('deleted')).toBe(0);
|
||||
});
|
||||
|
||||
it('a failed restore prunes nothing', async () => {
|
||||
const server = bareServer();
|
||||
agentInbox.post('not-restored', null, 'm');
|
||||
await server.finalizeRestoredState(false);
|
||||
expect(agentInbox.pendingCount('not-restored')).toBe(1);
|
||||
});
|
||||
|
||||
it('a successful restore prunes orphans and keeps live and persisted sessions', async () => {
|
||||
const server = bareServer(['live'], ['kept-for-recovery']);
|
||||
for (const id of ['live', 'kept-for-recovery', 'orphan']) agentInbox.post(id, null, 'm');
|
||||
await server.finalizeRestoredState(true);
|
||||
expect(agentInbox.pendingCount('live')).toBe(1);
|
||||
expect(agentInbox.pendingCount('kept-for-recovery')).toBe(1);
|
||||
expect(agentInbox.pendingCount('orphan')).toBe(0);
|
||||
});
|
||||
|
||||
it('a successful restore keeps the inbox of a session on the reboot-restore offer', async () => {
|
||||
// After a host reboot the dead session is in neither `this.sessions` nor state.json
|
||||
// (cleanupStaleSessions already pruned its record), only in the offer, and the
|
||||
// rebuild reuses its id, so its mail must still be there when it comes back.
|
||||
const server = bareServer();
|
||||
rebootRestoreRegistry.set([offerEntry('on-offer')]);
|
||||
for (const id of ['on-offer', 'orphan']) agentInbox.post(id, null, 'm');
|
||||
await server.finalizeRestoredState(true);
|
||||
expect(rebootRestoreRegistry.list(() => true).map((e) => e.sessionId)).toEqual(['on-offer']);
|
||||
expect(agentInbox.pendingCount('on-offer')).toBe(1);
|
||||
expect(agentInbox.pendingCount('orphan')).toBe(0);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,279 @@
|
||||
/**
|
||||
* @fileoverview The agent inbox store (src/web/agent-inbox.ts), pure: no routes, no IO.
|
||||
*
|
||||
* Each block pins one rule from discussion #558: read until ack, seen only for the
|
||||
* receiver, byte and count caps that refuse instead of dropping, a TTL counted from
|
||||
* each message's own timestamp, and a snapshot that survives malformed input.
|
||||
*/
|
||||
import { describe, it, expect, beforeEach } from 'vitest';
|
||||
import {
|
||||
AgentInbox,
|
||||
MAX_INBOX_BYTES,
|
||||
MAX_MESSAGE_BYTES,
|
||||
MAX_MESSAGES_PER_INBOX,
|
||||
MESSAGE_TTL_MS,
|
||||
} from '../src/web/agent-inbox.js';
|
||||
|
||||
const RX = 'receiver-session';
|
||||
|
||||
describe('AgentInbox', () => {
|
||||
let now: number;
|
||||
let inbox: AgentInbox;
|
||||
|
||||
beforeEach(() => {
|
||||
now = 1_000_000;
|
||||
inbox = new AgentInbox(() => now);
|
||||
});
|
||||
|
||||
const post = (text: string, from: string | null = 'sender') => {
|
||||
const r = inbox.post(RX, from, text);
|
||||
if (!r.ok) throw new Error(`post refused: ${r.reason}`);
|
||||
return r.message;
|
||||
};
|
||||
|
||||
describe('read until ack', () => {
|
||||
it('a read removes nothing; the same messages come back until acked', () => {
|
||||
const a = post('first');
|
||||
const b = post('second');
|
||||
expect(inbox.read(RX, { receiver: true }).messages.map((m) => m.id)).toEqual([a.id, b.id]);
|
||||
expect(inbox.read(RX, { receiver: true }).messages.map((m) => m.id)).toEqual([a.id, b.id]);
|
||||
expect(inbox.pendingCount(RX)).toBe(2);
|
||||
expect(inbox.ack(RX, [a.id])).toBe(1);
|
||||
expect(inbox.read(RX, { receiver: true }).messages.map((m) => m.id)).toEqual([b.id]);
|
||||
});
|
||||
|
||||
it('ack is idempotent: an id that is already gone or never existed is not an error', () => {
|
||||
const a = post('x');
|
||||
expect(inbox.ack(RX, [a.id])).toBe(1);
|
||||
expect(inbox.ack(RX, [a.id])).toBe(0);
|
||||
expect(inbox.ack(RX, ['no-such-id'])).toBe(0);
|
||||
expect(inbox.ack('no-such-session', [a.id])).toBe(0);
|
||||
});
|
||||
|
||||
it('ack removes only the given ids, never a message posted after the read', () => {
|
||||
const a = post('read me');
|
||||
inbox.read(RX, { receiver: true });
|
||||
const late = post('arrived after the read');
|
||||
inbox.ack(RX, [a.id]);
|
||||
expect(inbox.read(RX, { receiver: false }).messages.map((m) => m.id)).toEqual([late.id]);
|
||||
});
|
||||
|
||||
it('keeps multi-line text byte for byte', () => {
|
||||
const text = 'line one\nline two\r\n\ttabbed\n';
|
||||
post(text);
|
||||
expect(inbox.read(RX, { receiver: true }).messages[0].text).toBe(text);
|
||||
});
|
||||
});
|
||||
|
||||
describe('seen belongs to the receiver', () => {
|
||||
it('a receiver read marks seen; the flag on each message is the state BEFORE this read', () => {
|
||||
post('m');
|
||||
const first = inbox.read(RX, { receiver: true });
|
||||
expect(first.messages[0].seen).toBe(false);
|
||||
expect(first.unseen).toBe(0);
|
||||
const second = inbox.read(RX, { receiver: true });
|
||||
expect(second.messages[0].seen).toBe(true);
|
||||
});
|
||||
|
||||
it('a peek changes nothing: unseen stays unseen', () => {
|
||||
post('m');
|
||||
const peek = inbox.read(RX, { receiver: false });
|
||||
expect(peek.messages[0].seen).toBe(false);
|
||||
expect(peek.unseen).toBe(1);
|
||||
expect(inbox.unseenCount(RX)).toBe(1);
|
||||
expect(inbox.summary().get(RX)).toMatchObject({ pending: 1, unseen: 1 });
|
||||
});
|
||||
|
||||
it('a message posted after a receiver read is unseen again', () => {
|
||||
post('a');
|
||||
inbox.read(RX, { receiver: true });
|
||||
expect(inbox.unseenCount(RX)).toBe(0);
|
||||
post('b');
|
||||
expect(inbox.unseenCount(RX)).toBe(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe('caps refuse, never drop the oldest', () => {
|
||||
it(`refuses message ${MAX_MESSAGES_PER_INBOX + 1} and keeps the first one`, () => {
|
||||
const first = post('0');
|
||||
for (let i = 1; i < MAX_MESSAGES_PER_INBOX; i++) post(String(i));
|
||||
expect(inbox.post(RX, null, 'one too many')).toEqual({ ok: false, reason: 'inbox-full-count' });
|
||||
const ids = inbox.read(RX, { receiver: false }).messages.map((m) => m.id);
|
||||
expect(ids).toHaveLength(MAX_MESSAGES_PER_INBOX);
|
||||
expect(ids[0]).toBe(first.id);
|
||||
});
|
||||
|
||||
it('counts the per-message cap in UTF-8 bytes, not characters', () => {
|
||||
// 'é' is 2 bytes: half the byte budget in characters already fills it.
|
||||
expect(inbox.post(RX, null, 'é'.repeat(MAX_MESSAGE_BYTES / 2)).ok).toBe(true);
|
||||
expect(inbox.post(RX, null, 'é'.repeat(MAX_MESSAGE_BYTES / 2 + 1))).toEqual({
|
||||
ok: false,
|
||||
reason: 'message-too-large',
|
||||
});
|
||||
});
|
||||
|
||||
it('refuses a post that would cross the per-inbox byte cap', () => {
|
||||
const chunk = 'x'.repeat(MAX_MESSAGE_BYTES);
|
||||
const fit = Math.floor(MAX_INBOX_BYTES / MAX_MESSAGE_BYTES);
|
||||
for (let i = 0; i < fit; i++) post(chunk);
|
||||
expect(inbox.post(RX, null, 'x'.repeat(MAX_INBOX_BYTES - fit * MAX_MESSAGE_BYTES + 1))).toEqual({
|
||||
ok: false,
|
||||
reason: 'inbox-full-bytes',
|
||||
});
|
||||
expect(inbox.pendingCount(RX)).toBe(fit);
|
||||
});
|
||||
|
||||
it('an ack makes room again', () => {
|
||||
const ids: string[] = [];
|
||||
for (let i = 0; i < MAX_MESSAGES_PER_INBOX; i++) ids.push(post(String(i)).id);
|
||||
inbox.ack(RX, [ids[0]]);
|
||||
expect(inbox.post(RX, null, 'now it fits').ok).toBe(true);
|
||||
});
|
||||
|
||||
it('refuses empty text', () => {
|
||||
expect(inbox.post(RX, null, '')).toEqual({ ok: false, reason: 'empty' });
|
||||
});
|
||||
});
|
||||
|
||||
describe('TTL from the message timestamp', () => {
|
||||
it('a message older than the TTL is gone, read or not', () => {
|
||||
post('old');
|
||||
now += MESSAGE_TTL_MS / 2;
|
||||
const fresh = post('fresh');
|
||||
now += MESSAGE_TTL_MS / 2 + 1;
|
||||
expect(inbox.read(RX, { receiver: false }).messages.map((m) => m.id)).toEqual([fresh.id]);
|
||||
});
|
||||
|
||||
it('a restore keeps createdAt, so a restart does not restart the clock', () => {
|
||||
post('m');
|
||||
const snap = JSON.parse(JSON.stringify(inbox.snapshot()));
|
||||
now += MESSAGE_TTL_MS - 1000;
|
||||
const restarted = new AgentInbox(() => now);
|
||||
expect(restarted.restore(snap)).toBe(1);
|
||||
now += 1001;
|
||||
expect(restarted.pendingCount(RX)).toBe(0);
|
||||
});
|
||||
|
||||
it('does not restore a message that expired while the server was down', () => {
|
||||
post('m');
|
||||
const snap = inbox.snapshot();
|
||||
now += MESSAGE_TTL_MS + 1;
|
||||
expect(new AgentInbox(() => now).restore(snap)).toBe(0);
|
||||
});
|
||||
|
||||
it('sweeps an expired message that is not first (clock stepped back)', () => {
|
||||
const late = post('posted first, stamped later');
|
||||
now -= 1000; // the clock steps back
|
||||
const early = post('posted second, stamped earlier');
|
||||
now += MESSAGE_TTL_MS + 1000 - (late.createdAt - early.createdAt) / 2;
|
||||
// `early` is past the TTL, `late` is not, and `early` sits behind `late`.
|
||||
expect(inbox.read(RX, { receiver: false }).messages.map((m) => m.id)).toEqual([late.id]);
|
||||
});
|
||||
|
||||
it('expiry fires onChange so the snapshot catches up', () => {
|
||||
post('m');
|
||||
let changes = 0;
|
||||
inbox.onChange = () => changes++;
|
||||
now += MESSAGE_TTL_MS + 1;
|
||||
inbox.summary();
|
||||
expect(changes).toBe(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe('summary', () => {
|
||||
it('reports counts and the oldest unseen time, never text', () => {
|
||||
const a = post('secret one');
|
||||
now += 10;
|
||||
post('secret two');
|
||||
inbox.read(RX, { receiver: true });
|
||||
now += 10;
|
||||
const c = post('secret three');
|
||||
const row = inbox.summary().get(RX);
|
||||
expect(row).toEqual({ pending: 3, unseen: 1, oldestUnseenAt: c.createdAt });
|
||||
expect(JSON.stringify(row)).not.toContain('secret');
|
||||
expect(a.createdAt).toBeLessThan(c.createdAt);
|
||||
});
|
||||
|
||||
it('leaves out sessions with an empty inbox', () => {
|
||||
const a = post('m');
|
||||
inbox.ack(RX, [a.id]);
|
||||
expect(inbox.summary().has(RX)).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
describe('lifecycle', () => {
|
||||
it('drop discards the inbox and fires onChange', () => {
|
||||
post('m');
|
||||
let changes = 0;
|
||||
inbox.onChange = () => changes++;
|
||||
inbox.drop(RX);
|
||||
expect(inbox.pendingCount(RX)).toBe(0);
|
||||
expect(changes).toBe(1);
|
||||
});
|
||||
|
||||
it('pruneExcept drops only inboxes of sessions not kept', () => {
|
||||
inbox.post('keep-me', null, 'a');
|
||||
inbox.post('gone', null, 'b');
|
||||
expect(inbox.pruneExcept(['keep-me'])).toBe(1);
|
||||
expect(inbox.pendingCount('keep-me')).toBe(1);
|
||||
expect(inbox.pendingCount('gone')).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe('snapshot / restore', () => {
|
||||
it('round-trips messages, senders and the seen state', () => {
|
||||
const a = post('from a session', 'sender-1');
|
||||
post('from the user', null);
|
||||
inbox.read(RX, { receiver: true });
|
||||
const c = post('unseen');
|
||||
const restored = new AgentInbox(() => now);
|
||||
expect(restored.restore(JSON.parse(JSON.stringify(inbox.snapshot())))).toBe(3);
|
||||
const read = restored.read(RX, { receiver: false });
|
||||
expect(read.messages.map((m) => [m.id, m.from, m.seen])).toEqual([
|
||||
[a.id, 'sender-1', true],
|
||||
[expect.any(String), null, true],
|
||||
[c.id, 'sender', false],
|
||||
]);
|
||||
});
|
||||
|
||||
it('skips malformed entries and files instead of throwing', () => {
|
||||
const restored = new AgentInbox(() => now);
|
||||
expect(restored.restore(null)).toBe(0);
|
||||
expect(restored.restore({ version: 2, inboxes: {} })).toBe(0);
|
||||
expect(restored.restore({ version: 1, inboxes: { s: 'nope' } })).toBe(0);
|
||||
const good = { id: 'g', from: null, text: 'ok', createdAt: now, seenAt: null };
|
||||
expect(
|
||||
restored.restore({
|
||||
version: 1,
|
||||
inboxes: {
|
||||
s: [
|
||||
good,
|
||||
{ id: 1 },
|
||||
{ ...good, id: 'big', text: 'x'.repeat(MAX_MESSAGE_BYTES + 1) },
|
||||
{ ...good, text: '' },
|
||||
],
|
||||
},
|
||||
})
|
||||
).toBe(1);
|
||||
});
|
||||
|
||||
it('restores oldest first whatever the file order, so the caps keep the oldest', () => {
|
||||
const msg = (id: string, createdAt: number) => ({ id, from: null, text: 'x', createdAt, seenAt: null });
|
||||
const restored = new AgentInbox(() => now);
|
||||
restored.restore({ version: 1, inboxes: { s: [msg('new', now - 10), msg('old', now - 20)] } });
|
||||
expect(restored.read('s', { receiver: false }).messages.map((m) => m.id)).toEqual(['old', 'new']);
|
||||
});
|
||||
|
||||
it('holds the count cap on restore', () => {
|
||||
const many = Array.from({ length: MAX_MESSAGES_PER_INBOX + 5 }, (_, i) => ({
|
||||
id: `m${i}`,
|
||||
from: null,
|
||||
text: 'x',
|
||||
createdAt: now,
|
||||
seenAt: null,
|
||||
}));
|
||||
const restored = new AgentInbox(() => now);
|
||||
expect(restored.restore({ version: 1, inboxes: { s: many } })).toBe(MAX_MESSAGES_PER_INBOX);
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -13,8 +13,11 @@ import { afterAll, beforeAll, describe, expect, it } from 'vitest';
|
||||
import {
|
||||
AgentGuardError,
|
||||
EXIT,
|
||||
agentAck,
|
||||
agentInboxRead,
|
||||
agentInterrupt,
|
||||
agentLs,
|
||||
agentPost,
|
||||
agentRead,
|
||||
agentRm,
|
||||
agentSend,
|
||||
@@ -28,13 +31,16 @@ import {
|
||||
deleteRefusal,
|
||||
describeFailure,
|
||||
httpRequest,
|
||||
inboxCell,
|
||||
inputRefusal,
|
||||
isSelfSession,
|
||||
MIN_ID_PREFIX_LENGTH,
|
||||
parsePositiveInt,
|
||||
postTextFromArgs,
|
||||
registerAgentCommands,
|
||||
resolveAgentContext,
|
||||
stripAnsi,
|
||||
terminalSafeText,
|
||||
waitExitCode,
|
||||
type AgentContext,
|
||||
type AgentDeps,
|
||||
@@ -607,6 +613,196 @@ describe('session id prefixes', () => {
|
||||
});
|
||||
});
|
||||
|
||||
describe('agent inbox verbs', () => {
|
||||
const MSG = {
|
||||
id: 'm-1',
|
||||
from: OTHER,
|
||||
text: 'line one\nline two',
|
||||
createdAt: 1_700_000_000_000 - 90_000,
|
||||
seen: false,
|
||||
};
|
||||
|
||||
it('ls merges the inbox summary into an INBOX column', async () => {
|
||||
const deps = fakeDeps((o) =>
|
||||
o.path === '/api/v1/agent-inbox/summary'
|
||||
? ok({
|
||||
sessions: {
|
||||
[SELF]: { pending: 2, unseen: 1, oldestUnseenAt: 1, parked: false },
|
||||
[OTHER]: { pending: 0, unseen: 0, oldestUnseenAt: null, parked: true },
|
||||
},
|
||||
})
|
||||
: ok([
|
||||
{ id: SELF, mode: 'claude', status: 'busy', name: 'me' },
|
||||
{ id: OTHER, mode: 'codex', status: 'idle', name: 'worker' },
|
||||
])
|
||||
);
|
||||
expect(await agentLs(deps)).toBe(EXIT.ok);
|
||||
const text = deps.out.join('\n');
|
||||
expect(text).toMatch(/ID\s+MODE\s+STATUS\s+INBOX\s+NAME/);
|
||||
expect(text).toMatch(/058ee7b5\s+claude\s+busy\s+2 \(1 new\)\s+me/);
|
||||
expect(text).toMatch(/94990c6d\s+codex\s+idle\s+parked\s+worker/);
|
||||
});
|
||||
|
||||
it('ls --json carries each session’s inbox row (null when it has none)', async () => {
|
||||
const deps = fakeDeps(
|
||||
(o) =>
|
||||
o.path === '/api/v1/agent-inbox/summary'
|
||||
? ok({ sessions: { [OTHER]: { pending: 1, unseen: 1, oldestUnseenAt: 5, parked: false } } })
|
||||
: ok([{ id: SELF }, { id: OTHER }]),
|
||||
true
|
||||
);
|
||||
await agentLs(deps);
|
||||
const parsed = JSON.parse(deps.out.join('')) as Array<{ id: string; inbox: unknown }>;
|
||||
expect(parsed.map((s) => s.inbox)).toEqual([null, { pending: 1, unseen: 1, oldestUnseenAt: 5, parked: false }]);
|
||||
});
|
||||
|
||||
it('ls still lists sessions when the server has no summary route', async () => {
|
||||
const deps = fakeDeps((o) =>
|
||||
o.path === '/api/v1/agent-inbox/summary' ? apiError(404, 'NOT_FOUND', 'Route not found') : ok([{ id: OTHER }])
|
||||
);
|
||||
expect(await agentLs(deps)).toBe(EXIT.ok);
|
||||
expect(deps.out.join('\n')).not.toContain('INBOX');
|
||||
});
|
||||
|
||||
it('inboxCell', () => {
|
||||
expect(inboxCell(undefined)).toBe('');
|
||||
expect(inboxCell({ pending: 3, unseen: 0, oldestUnseenAt: null, parked: false })).toBe('3');
|
||||
expect(inboxCell({ pending: 3, unseen: 2, oldestUnseenAt: 1, parked: true })).toBe('3 (2 new), parked');
|
||||
});
|
||||
|
||||
it('post sends the text as data to the resolved session and never to /input', async () => {
|
||||
const deps = fakeDeps([ok({ message: MSG, pending: 1, unseen: 1 })]);
|
||||
expect(await agentPost(deps, { id: OTHER, text: 'line one\nline two' })).toBe(EXIT.ok);
|
||||
expect(deps.calls).toEqual([
|
||||
expect.objectContaining({
|
||||
method: 'POST',
|
||||
path: `/api/v1/sessions/${OTHER}/inbox`,
|
||||
body: { text: 'line one\nline two' },
|
||||
}),
|
||||
]);
|
||||
expect(deps.out.join('')).toMatch(/posted to 94990c6d \(1 pending, 1 unseen\)/);
|
||||
});
|
||||
|
||||
it('post passes a full inbox (409) through as an error', async () => {
|
||||
const deps = fakeDeps([apiError(409, 'CONFLICT', 'Inbox of x is full (200 messages).')]);
|
||||
expect(await agentPost(deps, { id: OTHER, text: 'x' })).toBe(EXIT.error);
|
||||
expect(deps.err.join('')).toMatch(/CONFLICT: Inbox of x is full/);
|
||||
});
|
||||
|
||||
it('post takes ONE argument and refuses an empty one before any request', async () => {
|
||||
expect(postTextFromArgs(['a\nb'])).toEqual({ text: 'a\nb' });
|
||||
expect(postTextFromArgs(['a', 'b'])).toEqual({ error: expect.stringMatching(/ONE argument, got 2/) });
|
||||
const deps = fakeDeps([]);
|
||||
expect(await agentPost(deps, { id: OTHER, text: '' })).toBe(EXIT.refused);
|
||||
expect(deps.calls).toEqual([]);
|
||||
});
|
||||
|
||||
it('inbox reads this session’s own inbox by default and prints full ids plus the ack hint', async () => {
|
||||
const deps = fakeDeps([ok({ receiver: true, messages: [MSG], pending: 1, unseen: 0 })]);
|
||||
expect(await agentInboxRead(deps, {})).toBe(EXIT.ok);
|
||||
expect(deps.calls[0]).toMatchObject({ method: 'GET', path: `/api/v1/sessions/${SELF}/inbox` });
|
||||
expect(deps.calls[0].query).toEqual({ wait: undefined });
|
||||
const text = deps.out.join('\n');
|
||||
expect(text).toMatch(/── m-1 from 94990c6d, 2m ago, new/);
|
||||
expect(text).toContain('line one\nline two');
|
||||
expect(deps.err.join('')).toMatch(/codeman agent ack <id>/);
|
||||
});
|
||||
|
||||
it('inbox prints message text with escape sequences and control characters neutralized', async () => {
|
||||
const hostile = {
|
||||
...MSG,
|
||||
text: 'title \x1b]0;pwned\x07ok\r\nred \x1b[31mtext\x1b[0m\rover\bwrite\ttab\x9b2J\nend',
|
||||
};
|
||||
const deps = fakeDeps([ok({ receiver: true, messages: [hostile], pending: 1, unseen: 0 })]);
|
||||
expect(await agentInboxRead(deps, {})).toBe(EXIT.ok);
|
||||
// out[0] is the `── <id> from …` header (styled by the palette), out[1] the text.
|
||||
expect(deps.out).toHaveLength(2);
|
||||
expect(deps.out[1]).not.toMatch(/[\x00-\x08\x0b-\x1f\x7f-\x9f]/);
|
||||
expect(deps.out[1]).toBe('title ok\nred text\\x0dover\\x08write\ttab\\x9b2J\nend');
|
||||
});
|
||||
|
||||
it('terminalSafeText keeps newline and tab and escapes every other control', () => {
|
||||
expect(terminalSafeText('a\nb\tc')).toBe('a\nb\tc');
|
||||
expect(terminalSafeText('a\r\nb')).toBe('a\nb');
|
||||
expect(terminalSafeText('\x1b[2J\x1bc\x00\x7f')).toBe('\\x1bc\\x00\\x7f');
|
||||
expect(terminalSafeText('ünïcödé ✓')).toBe('ünïcödé ✓');
|
||||
});
|
||||
|
||||
it('inbox --json hands the text out as stored', async () => {
|
||||
const raw = { ...MSG, text: 'red \x1b[31mtext' };
|
||||
const deps = fakeDeps([ok({ receiver: true, messages: [raw], pending: 1, unseen: 0 })], true);
|
||||
await agentInboxRead(deps, {});
|
||||
expect((JSON.parse(deps.out.join('')) as { messages: Array<{ text: string }> }).messages[0].text).toBe(
|
||||
'red \x1b[31mtext'
|
||||
);
|
||||
});
|
||||
|
||||
it('inbox <id> is a peek and says so', async () => {
|
||||
const deps = fakeDeps([ok([{ id: OTHER }]), ok({ receiver: false, messages: [MSG], pending: 1, unseen: 1 })]);
|
||||
expect(await agentInboxRead(deps, { id: OTHER.slice(0, 8) })).toBe(EXIT.ok);
|
||||
expect(deps.calls[1].path).toBe(`/api/v1/sessions/${OTHER}/inbox`);
|
||||
expect(deps.err.join('')).toMatch(/peek: nothing was marked seen/);
|
||||
});
|
||||
|
||||
it('inbox --wait passes the budget and maps timeout to exit 2', async () => {
|
||||
const timedOut = fakeDeps([ok({ receiver: true, messages: [], pending: 0, unseen: 0, wait: { timedOut: true } })]);
|
||||
expect(await agentInboxRead(timedOut, { waitMs: 30_000 })).toBe(EXIT.timeout);
|
||||
expect(timedOut.calls[0]).toMatchObject({ query: { wait: 30_000 }, timeoutMs: 40_000 });
|
||||
});
|
||||
|
||||
describe('a wait the server ended', () => {
|
||||
const ended = ok({ receiver: true, messages: [], pending: 0, unseen: 0, wait: { ended: true } });
|
||||
|
||||
it('is exit 3 only when the session list confirms the session is gone', async () => {
|
||||
const deps = fakeDeps([ended, ok([{ id: OTHER }])]);
|
||||
expect(await agentInboxRead(deps, { waitMs: 30_000 })).toBe(EXIT.dead);
|
||||
expect(deps.calls[1]).toMatchObject({ method: 'GET', path: '/api/v1/sessions' });
|
||||
expect(deps.err.join('')).toMatch(/went away/);
|
||||
});
|
||||
|
||||
it('is exit 2 (wait again) while the session is still listed: a restart or detach', async () => {
|
||||
const deps = fakeDeps([ended, ok([{ id: SELF }])]);
|
||||
expect(await agentInboxRead(deps, { waitMs: 30_000 })).toBe(EXIT.timeout);
|
||||
expect(deps.err.join('')).toMatch(/ended the wait early/);
|
||||
});
|
||||
|
||||
it('is exit 2 when the list cannot be had (the server is still coming back)', async () => {
|
||||
const deps = fakeDeps([ended, apiError(503, 'INTERNAL_ERROR', 'restarting')]);
|
||||
expect(await agentInboxRead(deps, { waitMs: 30_000 })).toBe(EXIT.timeout);
|
||||
});
|
||||
});
|
||||
|
||||
it('ack posts the ids to this session’s own inbox', async () => {
|
||||
const deps = fakeDeps([ok({ removed: 2, pending: 0 })]);
|
||||
expect(await agentAck(deps, { ids: ['m-1', ' m-2 '] })).toBe(EXIT.ok);
|
||||
expect(deps.calls[0]).toMatchObject({
|
||||
method: 'POST',
|
||||
path: `/api/v1/sessions/${SELF}/inbox/ack`,
|
||||
body: { ids: ['m-1', 'm-2'] },
|
||||
});
|
||||
expect(deps.out.join('')).toMatch(/acked 2 \(0 still pending\)/);
|
||||
});
|
||||
|
||||
it('ack refuses without ids and passes a 403 through', async () => {
|
||||
const none = fakeDeps([]);
|
||||
expect(await agentAck(none, { ids: [' '] })).toBe(EXIT.refused);
|
||||
expect(none.calls).toEqual([]);
|
||||
const forbidden = fakeDeps([apiError(403, 'FORBIDDEN', 'Only x itself can acknowledge its inbox')]);
|
||||
expect(await agentAck(forbidden, { ids: ['m-1'] })).toBe(EXIT.error);
|
||||
});
|
||||
|
||||
it('none of the inbox verbs writes to a pane', async () => {
|
||||
const deps = fakeDeps((o) =>
|
||||
o.path === '/api/v1/sessions' ? ok([{ id: OTHER }]) : ok({ messages: [], pending: 0, unseen: 0 })
|
||||
);
|
||||
await agentPost(deps, { id: OTHER, text: 'x' });
|
||||
await agentInboxRead(deps, {});
|
||||
await agentInboxRead(deps, { id: OTHER, waitMs: 1000 });
|
||||
await agentAck(deps, { ids: ['m'] });
|
||||
for (const call of deps.calls) expect(call.path).not.toMatch(/\/input$/);
|
||||
});
|
||||
});
|
||||
|
||||
describe('help texts carry the traps the skill documents', () => {
|
||||
const agent = registerAgentCommands(new Command());
|
||||
const sub = (name: string) => agent.commands.find((c) => c.name() === name)!;
|
||||
@@ -621,6 +817,26 @@ describe('help texts carry the traps the skill documents', () => {
|
||||
expect(sub('send').description()).toContain('send <id> -- "- fix the bug"');
|
||||
});
|
||||
|
||||
it('post takes text that starts with "-" (markdown from a file) without --', async () => {
|
||||
const program = new Command();
|
||||
program.exitOverride();
|
||||
const post = registerAgentCommands(program).commands.find((c) => c.name() === 'post')!;
|
||||
const got: Array<{ words: string[]; json: boolean }> = [];
|
||||
post.action((_id: string, words: string[], o: { json?: boolean }) => {
|
||||
got.push({ words, json: Boolean(o.json) });
|
||||
});
|
||||
for (const text of ['- [ ] item\nnext', '---\ntitle: a', '-v is great']) {
|
||||
await program.parseAsync(['node', 'codeman', 'agent', 'post', 'abcdefgh', text]);
|
||||
}
|
||||
await program.parseAsync(['node', 'codeman', 'agent', 'post', 'abcdefgh', 'hi', '--json']);
|
||||
expect(got).toEqual([
|
||||
{ words: ['- [ ] item\nnext'], json: false },
|
||||
{ words: ['---\ntitle: a'], json: false },
|
||||
{ words: ['-v is great'], json: false },
|
||||
{ words: ['hi'], json: true },
|
||||
]);
|
||||
});
|
||||
|
||||
it('rm does not claim a lineage check it does not make', () => {
|
||||
expect(sub('rm').description()).toMatch(/^Delete any session except this one/);
|
||||
});
|
||||
|
||||
@@ -0,0 +1,281 @@
|
||||
/**
|
||||
* @fileoverview Agent inbox routes (src/web/routes/inbox-routes.ts) via app.inject().
|
||||
*
|
||||
* The caller is whoever `X-Codeman-Parent-Session` resolves to (live session, same
|
||||
* owner as the receiver). These tests pin what that decides: who a post is from,
|
||||
* whether a read marks seen or is a peek, and who may ack. Plus the wait rule (parks
|
||||
* while nothing is UNSEEN), the full-inbox 409 and the owner-scoped summary.
|
||||
*
|
||||
* The routes use the process-wide `agentInbox` and `sessionWaits`; every test resets
|
||||
* the inbox and cancels its waiters.
|
||||
*/
|
||||
import { describe, it, expect, beforeEach, afterEach } from 'vitest';
|
||||
import Fastify, { type FastifyInstance } from 'fastify';
|
||||
import fastifyCookie from '@fastify/cookie';
|
||||
import { registerInboxRoutes } from '../../src/web/routes/inbox-routes.js';
|
||||
import { agentInbox, MAX_MESSAGES_PER_INBOX } from '../../src/web/agent-inbox.js';
|
||||
import { sessionWaits } from '../../src/web/session-wait-registry.js';
|
||||
import { installRouteErrorHandler } from '../../src/web/route-error-handler.js';
|
||||
import { httpStatusForErrorCode, type ApiErrorCode, type AuthUser } from '../../src/types.js';
|
||||
import { createMockRouteContext, createMockSession, type MockRouteContext } from '../mocks/index.js';
|
||||
|
||||
const RX = 'aaaaaaaa-1111-4111-8111-111111111111';
|
||||
const TX = 'bbbbbbbb-2222-4222-8222-222222222222';
|
||||
const OTHER = 'cccccccc-3333-4333-8333-333333333333';
|
||||
|
||||
interface Harness {
|
||||
app: FastifyInstance;
|
||||
ctx: MockRouteContext;
|
||||
}
|
||||
|
||||
let currentUser: AuthUser | undefined;
|
||||
|
||||
async function createHarness(): Promise<Harness> {
|
||||
const app = Fastify({ logger: false });
|
||||
await app.register(fastifyCookie);
|
||||
const ctx = createMockRouteContext({ sessionId: RX });
|
||||
for (const id of [TX, OTHER]) ctx.sessions.set(id, createMockSession(id) as never);
|
||||
app.decorateRequest('authUser', undefined);
|
||||
app.addHook('onRequest', async (req) => {
|
||||
req.authUser = currentUser;
|
||||
});
|
||||
registerInboxRoutes(app, ctx as never);
|
||||
// server.ts's envelope hook: an error envelope carries its 4xx.
|
||||
app.addHook('preSerialization', (req, reply, payload: unknown, done) => {
|
||||
const p = payload as { success?: unknown; errorCode?: unknown } | null;
|
||||
if (p && p.success === false && reply.statusCode === 200 && typeof p.errorCode === 'string') {
|
||||
reply.code(httpStatusForErrorCode(p.errorCode as ApiErrorCode));
|
||||
}
|
||||
done(null, payload);
|
||||
});
|
||||
installRouteErrorHandler(app);
|
||||
await app.ready();
|
||||
return { app, ctx };
|
||||
}
|
||||
|
||||
const as = (caller?: string) => (caller ? { 'x-codeman-parent-session': caller } : {});
|
||||
|
||||
describe('agent inbox routes', () => {
|
||||
let h: Harness;
|
||||
|
||||
const post = (to: string, text: string, caller?: string) =>
|
||||
h.app.inject({ method: 'POST', url: `/api/sessions/${to}/inbox`, payload: { text }, headers: as(caller) });
|
||||
const read = (of: string, caller?: string, query = '') =>
|
||||
h.app.inject({ method: 'GET', url: `/api/sessions/${of}/inbox${query}`, headers: as(caller) });
|
||||
const ack = (of: string, ids: string[], caller?: string) =>
|
||||
h.app.inject({ method: 'POST', url: `/api/sessions/${of}/inbox/ack`, payload: { ids }, headers: as(caller) });
|
||||
const data = (res: { body: string }) => JSON.parse(res.body).data;
|
||||
|
||||
beforeEach(async () => {
|
||||
agentInbox.resetForTests();
|
||||
currentUser = undefined;
|
||||
delete process.env.CODEMAN_MULTIUSER;
|
||||
h = await createHarness();
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
for (const id of [RX, TX, OTHER]) sessionWaits.cancelAll(id);
|
||||
await h.app.close();
|
||||
agentInbox.resetForTests();
|
||||
delete process.env.CODEMAN_MULTIUSER;
|
||||
});
|
||||
|
||||
describe('from', () => {
|
||||
it('is the resolved caller session (exact id or 8-char prefix)', async () => {
|
||||
expect(data(await post(RX, 'hi', TX)).message.from).toBe(TX);
|
||||
expect(data(await post(RX, 'hi', TX.slice(0, 8))).message.from).toBe(TX);
|
||||
});
|
||||
|
||||
it('is null (the user) without a header, for an unknown id, or for a short prefix', async () => {
|
||||
expect(data(await post(RX, 'a')).message.from).toBeNull();
|
||||
expect(data(await post(RX, 'b', 'dddddddd-0000-4000-8000-000000000000')).message.from).toBeNull();
|
||||
expect(data(await post(RX, 'c', TX.slice(0, 7))).message.from).toBeNull();
|
||||
});
|
||||
|
||||
it('a body field cannot set it: the schema rejects unknown keys', async () => {
|
||||
const res = await h.app.inject({
|
||||
method: 'POST',
|
||||
url: `/api/sessions/${RX}/inbox`,
|
||||
payload: { text: 'x', from: TX },
|
||||
});
|
||||
expect(res.statusCode).toBe(400);
|
||||
});
|
||||
|
||||
it('multi-user, non-admin: a session the caller cannot see never resolves as the sender', async () => {
|
||||
process.env.CODEMAN_MULTIUSER = '1';
|
||||
currentUser = { username: 'alice', role: 'user' };
|
||||
(h.ctx.sessions.get(RX) as unknown as { owner: string }).owner = 'alice';
|
||||
(h.ctx.sessions.get(TX) as unknown as { owner: string }).owner = 'bob';
|
||||
(h.ctx.sessions.get(OTHER) as unknown as { owner: string }).owner = 'alice';
|
||||
expect(data(await post(RX, 'x', TX)).message.from).toBeNull();
|
||||
expect(data(await post(RX, 'y', OTHER)).message.from).toBe(OTHER);
|
||||
// Nor can alice read as bob's session or ack for it: bob's inbox is a 404 to her.
|
||||
expect((await read(TX, TX)).statusCode).toBe(404);
|
||||
expect((await ack(TX, ['m'], TX)).statusCode).toBe(404);
|
||||
});
|
||||
|
||||
it('multi-user, admin: may act as any session of the receiver’s owner (admins see everything)', async () => {
|
||||
process.env.CODEMAN_MULTIUSER = '1';
|
||||
currentUser = { username: 'root', role: 'admin' };
|
||||
(h.ctx.sessions.get(RX) as unknown as { owner: string }).owner = 'bob';
|
||||
(h.ctx.sessions.get(TX) as unknown as { owner: string }).owner = 'bob';
|
||||
expect(data(await post(RX, 'x', TX)).message.from).toBe(TX);
|
||||
expect(data(await read(RX, RX)).receiver).toBe(true);
|
||||
});
|
||||
|
||||
it('multi-user: a session of another owner never resolves as the sender', async () => {
|
||||
process.env.CODEMAN_MULTIUSER = '1';
|
||||
currentUser = { username: 'alice', role: 'admin' };
|
||||
(h.ctx.sessions.get(RX) as unknown as { owner: string }).owner = 'alice';
|
||||
(h.ctx.sessions.get(TX) as unknown as { owner: string }).owner = 'bob';
|
||||
(h.ctx.sessions.get(OTHER) as unknown as { owner: string }).owner = 'alice';
|
||||
expect(data(await post(RX, 'x', TX)).message.from).toBeNull();
|
||||
expect(data(await post(RX, 'y', OTHER)).message.from).toBe(OTHER);
|
||||
});
|
||||
});
|
||||
|
||||
describe('post', () => {
|
||||
it('keeps multi-line text as data', async () => {
|
||||
await post(RX, 'one\ntwo\n', TX);
|
||||
expect(data(await read(RX, RX)).messages[0].text).toBe('one\ntwo\n');
|
||||
});
|
||||
|
||||
it('a full inbox is 409 CONFLICT with a clear message, and the oldest stays', async () => {
|
||||
const first = data(await post(RX, 'first')).message.id;
|
||||
for (let i = 1; i < MAX_MESSAGES_PER_INBOX; i++) agentInbox.post(RX, null, String(i));
|
||||
const res = await post(RX, 'one too many');
|
||||
expect(res.statusCode).toBe(409);
|
||||
expect(JSON.parse(res.body)).toMatchObject({ errorCode: 'CONFLICT' });
|
||||
expect(JSON.parse(res.body).error).toMatch(/full .*Nothing was stored/);
|
||||
expect(data(await read(RX)).messages[0].id).toBe(first);
|
||||
});
|
||||
|
||||
it('a message over the byte cap is 400 with the store’s hint, multi-byte or not', async () => {
|
||||
for (const text of ['é'.repeat(8193), 'x'.repeat(16_385)]) {
|
||||
const res = await post(RX, text);
|
||||
expect(res.statusCode).toBe(400);
|
||||
expect(JSON.parse(res.body).error).toMatch(/Write it to a file in the workspace and post the path/);
|
||||
}
|
||||
});
|
||||
|
||||
it('404 for an unknown session', async () => {
|
||||
expect((await post('nope', 'x')).statusCode).toBe(404);
|
||||
});
|
||||
});
|
||||
|
||||
describe('read', () => {
|
||||
it('only the receiver marks seen; a coordinator read is a peek', async () => {
|
||||
await post(RX, 'task', TX);
|
||||
const peek = data(await read(RX, TX));
|
||||
expect(peek).toMatchObject({ receiver: false, pending: 1, unseen: 1 });
|
||||
expect(data(await read(RX)).unseen).toBe(1);
|
||||
const own = data(await read(RX, RX));
|
||||
expect(own).toMatchObject({ receiver: true, pending: 1, unseen: 0 });
|
||||
expect(own.messages[0].seen).toBe(false);
|
||||
expect(data(await read(RX, RX)).messages[0].seen).toBe(true);
|
||||
});
|
||||
|
||||
it('?wait= parks while nothing is unseen, even with a seen-but-unacked message', async () => {
|
||||
await post(RX, 'already seen', TX);
|
||||
await read(RX, RX);
|
||||
const parked = read(RX, RX, '?wait=5000');
|
||||
await new Promise((r) => setTimeout(r, 30));
|
||||
expect(sessionWaits.inboxWaiterCount(RX, true)).toBe(1);
|
||||
await post(RX, 'new', TX);
|
||||
const res = data(await parked);
|
||||
expect(res.wait).toMatchObject({ posted: true, timedOut: false });
|
||||
expect(res.messages.map((m: { text: string; seen: boolean }) => [m.text, m.seen])).toEqual([
|
||||
['already seen', true],
|
||||
['new', false],
|
||||
]);
|
||||
});
|
||||
|
||||
it('?wait= answers at once when something is unseen', async () => {
|
||||
await post(RX, 'm', TX);
|
||||
const res = data(await read(RX, RX, '?wait=60000'));
|
||||
expect(res.wait).toMatchObject({ immediate: true, waitedMs: 0, timeoutMs: 60000 });
|
||||
});
|
||||
|
||||
it('?wait= runs out as a 200 with timedOut, clamped to the wait floor', async () => {
|
||||
const res = await read(RX, RX, '?wait=1');
|
||||
expect(res.statusCode).toBe(200);
|
||||
expect(data(res).wait).toMatchObject({ timedOut: true, timeoutMs: 1000 });
|
||||
expect(res.headers['cache-control']).toBe('no-store');
|
||||
});
|
||||
|
||||
it('?wait= rejects junk', async () => {
|
||||
expect((await read(RX, RX, '?wait=-5')).statusCode).toBe(400);
|
||||
expect((await read(RX, RX, '?wait=abc')).statusCode).toBe(400);
|
||||
});
|
||||
|
||||
it('a parked read on a session that goes away answers ended, with an empty inbox', async () => {
|
||||
await post(RX, 'x');
|
||||
await read(RX, RX);
|
||||
const parked = read(RX, RX, '?wait=5000');
|
||||
await new Promise((r) => setTimeout(r, 30));
|
||||
// What server.ts does on delete: drop the inbox, then cancel the waiters.
|
||||
agentInbox.drop(RX);
|
||||
sessionWaits.cancelAll(RX);
|
||||
const res = data(await parked);
|
||||
expect(res.wait).toMatchObject({ ended: true });
|
||||
expect(res.messages).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
describe('ack', () => {
|
||||
it('removes by id for the receiver, idempotently', async () => {
|
||||
const id = data(await post(RX, 'm', TX)).message.id;
|
||||
expect(data(await ack(RX, [id], RX))).toMatchObject({ removed: 1, pending: 0 });
|
||||
const again = await ack(RX, [id, 'never-existed'], RX);
|
||||
expect(again.statusCode).toBe(200);
|
||||
expect(data(again)).toMatchObject({ removed: 0, pending: 0 });
|
||||
});
|
||||
|
||||
it('is 403 for anyone but the receiver, and removes nothing', async () => {
|
||||
const id = data(await post(RX, 'm', TX)).message.id;
|
||||
for (const caller of [TX, undefined]) {
|
||||
const res = await ack(RX, [id], caller);
|
||||
expect(res.statusCode).toBe(403);
|
||||
expect(JSON.parse(res.body).errorCode).toBe('FORBIDDEN');
|
||||
}
|
||||
expect(agentInbox.pendingCount(RX)).toBe(1);
|
||||
});
|
||||
|
||||
it('rejects an empty id list', async () => {
|
||||
expect((await ack(RX, [], RX)).statusCode).toBe(400);
|
||||
});
|
||||
});
|
||||
|
||||
describe('summary', () => {
|
||||
it('reports counts and parked per session, never text', async () => {
|
||||
await post(RX, 'secret', TX);
|
||||
await post(TX, 'other secret', RX);
|
||||
await read(TX, TX);
|
||||
const parked = read(OTHER, OTHER, '?wait=5000');
|
||||
// A coordinator peeking with ?wait= is not the receiver: not "parked".
|
||||
const peeking = read(RX, TX, '?wait=5000');
|
||||
await new Promise((r) => setTimeout(r, 30));
|
||||
const res = await h.app.inject({ method: 'GET', url: '/api/agent-inbox/summary' });
|
||||
expect(res.body).not.toContain('secret');
|
||||
expect(data(res).sessions).toEqual({
|
||||
[RX]: { pending: 1, unseen: 1, oldestUnseenAt: expect.any(Number), parked: false },
|
||||
[TX]: { pending: 1, unseen: 0, oldestUnseenAt: null, parked: false },
|
||||
[OTHER]: { pending: 0, unseen: 0, oldestUnseenAt: null, parked: true },
|
||||
});
|
||||
sessionWaits.cancelAll(OTHER);
|
||||
sessionWaits.cancelAll(RX);
|
||||
await Promise.all([parked, peeking]);
|
||||
});
|
||||
|
||||
it('multi-user: lists only sessions the caller can see', async () => {
|
||||
process.env.CODEMAN_MULTIUSER = '1';
|
||||
currentUser = { username: 'alice', role: 'user' };
|
||||
(h.ctx.sessions.get(RX) as unknown as { owner: string }).owner = 'alice';
|
||||
(h.ctx.sessions.get(TX) as unknown as { owner: string }).owner = 'bob';
|
||||
agentInbox.post(RX, null, 'mine');
|
||||
agentInbox.post(TX, null, 'bob’s');
|
||||
const res = await h.app.inject({ method: 'GET', url: '/api/agent-inbox/summary' });
|
||||
expect(Object.keys(data(res).sessions)).toEqual([RX]);
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -26,6 +26,7 @@ import { createMockRouteContext, type MockRouteContext } from '../mocks/index.js
|
||||
import { installRouteErrorHandler } from '../../src/web/route-error-handler.js';
|
||||
import { ApiErrorCode, httpStatusForErrorCode } from '../../src/types.js';
|
||||
import { Session } from '../../src/session.js';
|
||||
import { agentInbox } from '../../src/web/agent-inbox.js';
|
||||
|
||||
// Mock execFile so the send-key route's `tmux` invocation is observable (not run for real).
|
||||
const { execFile } = vi.hoisted(() => ({ execFile: vi.fn() }));
|
||||
@@ -985,6 +986,17 @@ describe('session-routes', () => {
|
||||
expect(harness.ctx.broadcast).toHaveBeenCalledWith('session:deleted', { id: 'ghost-session' });
|
||||
});
|
||||
|
||||
it('a persisted-only delete drops the agent inbox the detach kept', async () => {
|
||||
// Detach (killMux=false) keeps the inbox; retiring the row afterwards is the delete.
|
||||
agentInbox.resetForTests();
|
||||
agentInbox.post('ghost-session', null, 'mail from before the detach');
|
||||
vi.mocked(harness.ctx.store.getSession).mockReturnValueOnce({ id: 'ghost-session', owner: undefined } as never);
|
||||
const res = await harness.app.inject({ method: 'DELETE', url: '/api/sessions/ghost-session' });
|
||||
expect(res.statusCode).toBe(200);
|
||||
expect(agentInbox.pendingCount('ghost-session')).toBe(0);
|
||||
agentInbox.resetForTests();
|
||||
});
|
||||
|
||||
it('404s a persisted-only session id the state store does not recognize either', async () => {
|
||||
vi.mocked(harness.ctx.store.getSession).mockReturnValueOnce(null);
|
||||
const res = await harness.app.inject({
|
||||
|
||||
@@ -47,13 +47,15 @@ const { EXISTING_SETTINGS, subagentWatcher, imageWatcher, workflowRunWatcher } =
|
||||
};
|
||||
};
|
||||
return {
|
||||
// Persisted settings.json for these tests: two watchers ON, subagent tracking OFF.
|
||||
// Persisted settings.json for these tests: two watchers ON, subagent tracking OFF,
|
||||
// agent inbox snapshot ON.
|
||||
EXISTING_SETTINGS: {
|
||||
subagentTrackingEnabled: false,
|
||||
imageWatcherEnabled: true,
|
||||
showUltracodeAgents: true,
|
||||
promptCacheKeepWarm: false,
|
||||
promptCacheKeepWarmMaxHours: 2,
|
||||
agentInboxPersistEnabled: true,
|
||||
},
|
||||
subagentWatcher: makeWatcher(false),
|
||||
imageWatcher: makeWatcher(true),
|
||||
@@ -61,6 +63,9 @@ const { EXISTING_SETTINGS, subagentWatcher, imageWatcher, workflowRunWatcher } =
|
||||
};
|
||||
});
|
||||
|
||||
// setEnabled(false) deletes the snapshot file, so a wrong default here loses mail.
|
||||
const { agentInboxPersistence } = vi.hoisted(() => ({ agentInboxPersistence: { setEnabled: vi.fn() } }));
|
||||
|
||||
vi.mock('node:fs/promises', () => ({
|
||||
default: {
|
||||
readFile: vi.fn(async () => JSON.stringify(EXISTING_SETTINGS)),
|
||||
@@ -76,6 +81,7 @@ vi.mock('node:fs', async (importOriginal) => {
|
||||
vi.mock('../../src/subagent-watcher.js', () => ({ subagentWatcher }));
|
||||
vi.mock('../../src/image-watcher.js', () => ({ imageWatcher }));
|
||||
vi.mock('../../src/workflow-run-watcher.js', () => ({ workflowRunWatcher }));
|
||||
vi.mock('../../src/web/agent-inbox-persistence.js', () => ({ agentInboxPersistence }));
|
||||
|
||||
describe('PUT /api/settings — partial body must not reset service toggles', () => {
|
||||
let harness: RouteTestHarness;
|
||||
@@ -87,6 +93,7 @@ describe('PUT /api/settings — partial body must not reset service toggles', ()
|
||||
w.stop.mockClear();
|
||||
w.__resetRunning(); // running state, not just call records — see makeWatcher
|
||||
}
|
||||
agentInboxPersistence.setEnabled.mockClear();
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
@@ -167,4 +174,25 @@ describe('PUT /api/settings — partial body must not reset service toggles', ()
|
||||
expect((await put({ promptCacheKeepWarm: true })).statusCode).toBe(200);
|
||||
expect(harness.ctx.keepWarm.rescheduleAll).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it('keeps the agent inbox snapshot on when the body omits its key', async () => {
|
||||
const res = await harness.app.inject({
|
||||
method: 'PUT',
|
||||
url: '/api/settings',
|
||||
payload: { showPlanUsageLimits: true },
|
||||
});
|
||||
expect(res.statusCode).toBe(200);
|
||||
expect(agentInboxPersistence.setEnabled).toHaveBeenCalledWith(true);
|
||||
expect(agentInboxPersistence.setEnabled).not.toHaveBeenCalledWith(false);
|
||||
});
|
||||
|
||||
it('switches the agent inbox snapshot off when the body says so', async () => {
|
||||
const res = await harness.app.inject({
|
||||
method: 'PUT',
|
||||
url: '/api/settings',
|
||||
payload: { agentInboxPersistEnabled: false },
|
||||
});
|
||||
expect(res.statusCode).toBe(200);
|
||||
expect(agentInboxPersistence.setEnabled).toHaveBeenCalledWith(false);
|
||||
});
|
||||
});
|
||||
@@ -11,6 +11,7 @@
|
||||
import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest';
|
||||
import { createRouteTestHarness, type RouteTestHarness } from './_route-test-utils.js';
|
||||
import { registerSystemRoutes } from '../../src/web/routes/system-routes.js';
|
||||
import { agentInbox } from '../../src/web/agent-inbox.js';
|
||||
|
||||
const hostSensors = vi.hoisted(() => ({ readHostSensors: vi.fn(async (_request: unknown) => ({})) }));
|
||||
vi.mock('../../src/host-sensors.js', () => hostSensors);
|
||||
@@ -360,6 +361,60 @@ describe('system-routes', () => {
|
||||
expect(harness.ctx.store.cleanupSessionsByIds).toHaveBeenCalledWith(new Set());
|
||||
expect(runStaleSessionCleanup).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it('drops the agent inbox of every record it removes, and only those', async () => {
|
||||
agentInbox.resetForTests();
|
||||
agentInbox.post('stale-detached', null, 'mail for a detached session');
|
||||
agentInbox.post('still-kept', null, 'mail for a session that stays');
|
||||
harness.ctx.tabLayouts.runStaleSessionCleanup = vi.fn(
|
||||
async (_activeIds: Set<string>, action: (ids: ReadonlySet<string>) => unknown) =>
|
||||
action(new Set(['stale-detached']))
|
||||
);
|
||||
vi.mocked(harness.ctx.store.cleanupSessionsByIds).mockReturnValueOnce({
|
||||
count: 1,
|
||||
cleaned: [{ id: 'stale-detached', name: 'w9-gone' }],
|
||||
});
|
||||
try {
|
||||
const res = await harness.app.inject({ method: 'POST', url: '/api/cleanup-state' });
|
||||
expect(res.statusCode).toBe(200);
|
||||
expect(JSON.parse(res.body).cleanedSessions).toBe(1);
|
||||
expect(agentInbox.pendingCount('stale-detached')).toBe(0);
|
||||
expect(agentInbox.pendingCount('still-kept')).toBe(1);
|
||||
} finally {
|
||||
agentInbox.resetForTests();
|
||||
}
|
||||
});
|
||||
|
||||
it('keeps the agent inbox of a removed record whose tmux session the multiplexer still lists', async () => {
|
||||
// Detach keeps the tmux session and its mux entry; the next boot re-adopts it
|
||||
// under the same id even without a state.json record, so its mail must survive.
|
||||
agentInbox.resetForTests();
|
||||
agentInbox.post('stale-detached', null, 'mail for a session nothing can bring back');
|
||||
agentInbox.post('pane-still-alive', null, 'mail for a detached pane that boot re-adopts');
|
||||
harness.ctx.tabLayouts.runStaleSessionCleanup = vi.fn(
|
||||
async (_activeIds: Set<string>, action: (ids: ReadonlySet<string>) => unknown) =>
|
||||
action(new Set(['stale-detached', 'pane-still-alive']))
|
||||
);
|
||||
vi.mocked(harness.ctx.store.cleanupSessionsByIds).mockReturnValueOnce({
|
||||
count: 2,
|
||||
cleaned: [
|
||||
{ id: 'stale-detached', name: 'w9-gone' },
|
||||
{ id: 'pane-still-alive', name: 'w8-detached' },
|
||||
],
|
||||
});
|
||||
// The harness is rebuilt per test, so this mock does not leak into the next one.
|
||||
vi.mocked(harness.ctx.mux.getSession).mockImplementation(((id: string) =>
|
||||
id === 'pane-still-alive' ? { sessionId: id } : null) as unknown as typeof harness.ctx.mux.getSession);
|
||||
try {
|
||||
const res = await harness.app.inject({ method: 'POST', url: '/api/cleanup-state' });
|
||||
expect(res.statusCode).toBe(200);
|
||||
expect(JSON.parse(res.body).cleanedSessions).toBe(2);
|
||||
expect(agentInbox.pendingCount('stale-detached')).toBe(0);
|
||||
expect(agentInbox.pendingCount('pane-still-alive')).toBe(1);
|
||||
} finally {
|
||||
agentInbox.resetForTests();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
// ========== GET /api/subagents ==========
|
||||
|
||||
@@ -88,6 +88,7 @@ describe('isSensitivePath', () => {
|
||||
['codeman push keys (VAPID private)', `${HOME}/.codeman/push-keys.json`],
|
||||
['codeman intent profiles', `${HOME}/.codeman/intents.json`],
|
||||
['codeman intents on a named instance', `${HOME}/.codeman-beta/intents.json`],
|
||||
['codeman agent inbox snapshot', `${HOME}/.codeman/agent-inbox.json`],
|
||||
];
|
||||
|
||||
it.each(blocked)('blocks the %s', (_label, path) => {
|
||||
|
||||
@@ -1088,3 +1088,66 @@ describe('SessionWaitRegistry: teardown and leak prevention', () => {
|
||||
expect(reg.totalWaiterCount()).toBe(0);
|
||||
});
|
||||
});
|
||||
|
||||
describe('inbox waiters', () => {
|
||||
it('resolve with posted:true on notifyInbox, and only for that session', async () => {
|
||||
const reg = new SessionWaitRegistry();
|
||||
const a = reg.waitForInbox('a', { timeoutMs: 5_000 });
|
||||
const b = reg.waitForInbox('b', { timeoutMs: 50 });
|
||||
expect(reg.notifyInbox('a')).toBe(1);
|
||||
await expect(a).resolves.toMatchObject({ posted: true, timedOut: false, ended: false });
|
||||
await expect(b).resolves.toMatchObject({ posted: false, timedOut: true });
|
||||
expect(reg.totalWaiterCount()).toBe(0);
|
||||
});
|
||||
|
||||
it('count toward the per-session, per-owner and total caps like the other kinds', async () => {
|
||||
const reg = new SessionWaitRegistry({ maxWaitersPerSession: 2, maxWaitersPerOwner: 3, maxWaitersTotal: 4 });
|
||||
const pending = [
|
||||
reg.waitForSignal('s', { until: ['stop'], timeoutMs: 5_000, owner: 'alice' }),
|
||||
reg.waitForInbox('s', { timeoutMs: 5_000, owner: 'alice' }),
|
||||
];
|
||||
expect(() => reg.assertCapacity('s', 'alice')).toThrow(WaitCapacityError);
|
||||
expect(() => reg.waitForInbox('s', { timeoutMs: 5_000 })).toThrow(/scope: session/);
|
||||
pending.push(reg.waitForInbox('t', { timeoutMs: 5_000, owner: 'alice' }));
|
||||
expect(() => reg.waitForInbox('u', { timeoutMs: 5_000, owner: 'alice' })).toThrow(/scope: owner/);
|
||||
pending.push(reg.waitForInbox('u', { timeoutMs: 5_000, owner: 'bob' }));
|
||||
expect(() => reg.waitForInbox('v', { timeoutMs: 5_000 })).toThrow(/scope: total/);
|
||||
expect(reg.waiterCount('s')).toBe(2);
|
||||
reg.stop();
|
||||
await Promise.all(pending);
|
||||
expect(reg.ownerWaiterCount('alice')).toBe(0);
|
||||
});
|
||||
|
||||
it('counts receiver waiters separately (the summary’s parked)', () => {
|
||||
const reg = new SessionWaitRegistry();
|
||||
void reg.waitForInbox('s', { timeoutMs: 5_000, receiver: false });
|
||||
expect(reg.inboxWaiterCount('s')).toBe(1);
|
||||
expect(reg.inboxWaiterCount('s', true)).toBe(0);
|
||||
void reg.waitForInbox('s', { timeoutMs: 5_000, receiver: true });
|
||||
expect(reg.inboxWaiterCount('s', true)).toBe(1);
|
||||
reg.stop();
|
||||
});
|
||||
|
||||
it('cancelAll ends them with ended:true, and an abort frees the slot at once', async () => {
|
||||
const reg = new SessionWaitRegistry();
|
||||
const ended = reg.waitForInbox('s', { timeoutMs: 5_000 });
|
||||
expect(reg.cancelAll('s')).toBe(1);
|
||||
await expect(ended).resolves.toMatchObject({ ended: true, aborted: false, posted: false });
|
||||
|
||||
const controller = new AbortController();
|
||||
const aborted = reg.waitForInbox('s', { timeoutMs: 5_000, abortSignal: controller.signal });
|
||||
expect(reg.inboxWaiterCount('s')).toBe(1);
|
||||
controller.abort();
|
||||
await expect(aborted).resolves.toMatchObject({ ended: true, aborted: true });
|
||||
expect(reg.inboxWaiterCount('s')).toBe(0);
|
||||
});
|
||||
|
||||
it('after stop() a new inbox wait resolves at once with ended:true', async () => {
|
||||
const reg = new SessionWaitRegistry();
|
||||
const parked = reg.waitForInbox('s', { timeoutMs: 5_000 });
|
||||
reg.stop();
|
||||
await expect(parked).resolves.toMatchObject({ ended: true });
|
||||
await expect(reg.waitForInbox('s', { timeoutMs: 5_000 })).resolves.toMatchObject({ ended: true, posted: false });
|
||||
expect(reg.totalWaiterCount()).toBe(0);
|
||||
});
|
||||
});
|
||||
@@ -45,7 +45,12 @@ describe('tab layout restore gate', () => {
|
||||
};
|
||||
cleanupStaleSessions(): Promise<number>;
|
||||
finalizeRestoredState(restored: boolean): Promise<void>;
|
||||
sessions: Map<string, unknown>;
|
||||
store: { getSessions(): Record<string, unknown> };
|
||||
};
|
||||
// Read by the agent-inbox prune at the end of a successful restore.
|
||||
server.sessions = new Map();
|
||||
server.store = { getSessions: () => ({}) };
|
||||
server.tabLayouts = {
|
||||
markRestorationComplete: () => order.push('complete'),
|
||||
markRestorationFailed: () => order.push('failed'),
|
||||
|
||||
Reference in new issue
Block a user