mirror of
https://github.com/Ark0N/Codeman.git
synced 2026-11-14 02:29:33 +01:00
fix(agent-inbox): review round 1
- codeman agent post: accept text that starts with "-" (markdown from a file), --json and -h stay options. - DELETE /api/sessions/:id on a persisted-only row drops the inbox a detach kept. - inbox --wait: wait.ended is exit 3 only when the session list no longer has the session; a restart or detach ending the wait is exit 2 (wait again). - A read whose client hung up during ?wait= marks nothing seen. - InboxPostSchema bound is 4x the byte cap, so every oversize message gets the store's "post a file path" answer. - Store: restore sorts oldest first; the TTL sweep checks every message, not only the first. - Docs: admins can act as any session of the receiver's owner (now tested for admin and non-admin); a peek with ?wait= does not block; exit codes. - Tests: partial PUT /api/settings keeps persistence on. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
1 parent
c6dfaa2cac
commit
a1a33cf050
15 files changed
+209
-31
No files matched your search
+26
-4
@@ -890,6 +890,17 @@ export interface InboxReadOptions {
|
||||
waitMs?: number;
|
||||
}
|
||||
|
||||
/** True only when the session list was read and does not contain `id`. */
|
||||
async function sessionIsGone(deps: AgentDeps, id: string): Promise<boolean> {
|
||||
try {
|
||||
const res = await deps.request(deps.ctx, { method: 'GET', path: '/api/v1/sessions' });
|
||||
if (!res.json?.success) return false;
|
||||
return !((res.json.data as SessionRow[] | undefined) ?? []).some((s) => s.id === id);
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/** Relative age for humans: `12s`, `4m`, `3h`. */
|
||||
function age(createdAt: number, now: number): string {
|
||||
const s = Math.max(0, Math.round((now - createdAt) / 1000));
|
||||
@@ -901,8 +912,8 @@ function age(createdAt: number, now: number): string {
|
||||
/**
|
||||
* `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.
|
||||
* seen. Exit 0 with mail, 2 when a `--wait` ran out with nothing unseen or the server
|
||||
* cut it short (restart, detach), 3 when the session is gone from the list.
|
||||
*/
|
||||
export async function agentInboxRead(deps: AgentDeps, options: InboxReadOptions): Promise<number> {
|
||||
const target = await resolveSessionId(deps, options.id ?? deps.ctx.selfId);
|
||||
@@ -924,7 +935,12 @@ export async function agentInboxRead(deps: AgentDeps, options: InboxReadOptions)
|
||||
}
|
||||
| undefined;
|
||||
const messages = data?.messages ?? [];
|
||||
const code = data?.wait?.ended ? EXIT.dead : data?.wait?.timedOut ? EXIT.timeout : EXIT.ok;
|
||||
// `ended` is also what a server restart or a detach answers. Exit 3 tells a receiver
|
||||
// loop to stop, so it needs the session list to confirm the session is gone; while
|
||||
// it is still listed (or the list cannot be had, as during a restart) the wait was
|
||||
// only cut short, which is exit 2: loop again.
|
||||
const gone = data?.wait?.ended ? await sessionIsGone(deps, target.id) : false;
|
||||
const code = gone ? EXIT.dead : data?.wait?.timedOut || data?.wait?.ended ? EXIT.timeout : EXIT.ok;
|
||||
if (deps.json) {
|
||||
emitJson(deps, data ?? {});
|
||||
return code;
|
||||
@@ -936,8 +952,10 @@ export async function agentInboxRead(deps: AgentDeps, options: InboxReadOptions)
|
||||
deps.io.out(palette.muted(`── ${m.id} from ${sender}, ${age(m.createdAt, now)} ago, `) + tag);
|
||||
deps.io.out(m.text);
|
||||
}
|
||||
if (data?.wait?.ended) {
|
||||
if (gone) {
|
||||
deps.io.err(palette.err(`${GLYPH.fail} the session went away during the wait`));
|
||||
} else if (data?.wait?.ended) {
|
||||
deps.io.err(palette.muted('(the server ended the wait early, e.g. a restart; exit 2, wait again)'));
|
||||
} 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) {
|
||||
@@ -1165,6 +1183,10 @@ export function registerAgentCommands(program: Command): Command {
|
||||
"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')
|
||||
// A message is often a file (`"$(cat task.md)"`) and markdown starts with `- ` or
|
||||
// `---` often enough; without this commander takes it for an unknown option.
|
||||
// `--json` and `-h` are still options: text that is exactly one goes after `--`.
|
||||
.allowUnknownOption()
|
||||
.action((id: string, words: string[], options: { json?: boolean }) =>
|
||||
run(Boolean(options.json), (deps) => {
|
||||
const message = postTextFromArgs(words);
|
||||
|
||||
@@ -228,8 +228,12 @@ export class AgentInbox {
|
||||
if (!Array.isArray(raw)) continue;
|
||||
const queue: StoredMessage[] = [];
|
||||
let bytes = 0;
|
||||
for (const entry of raw) {
|
||||
if (!isStoredMessage(entry) || entry.createdAt <= cutoff) continue;
|
||||
// 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 });
|
||||
@@ -253,8 +257,9 @@ export class AgentInbox {
|
||||
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.
|
||||
// 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?.();
|
||||
|
||||
@@ -8,8 +8,10 @@
|
||||
* 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.
|
||||
* owner's session (an admin, who can see every session, can act as any session of
|
||||
* the receiver's owner, as admins can everywhere else). 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.
|
||||
@@ -135,8 +137,9 @@ export function registerInboxRoutes(app: FastifyInstance, ctx: SessionPort): voi
|
||||
}
|
||||
|
||||
// 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 });
|
||||
// that ended because the session was deleted reads an empty inbox. A client that
|
||||
// hung up never sees this answer, so its read must not mark anything seen.
|
||||
const read = agentInbox.read(id, { receiver: receiver && !wait?.aborted });
|
||||
return { success: true, data: { sessionId: id, receiver, ...read, ...(wait ? { wait } : {}) } };
|
||||
});
|
||||
|
||||
|
||||
@@ -119,6 +119,7 @@ import {
|
||||
abortOnClientHangUp,
|
||||
} from '../route-helpers.js';
|
||||
import { buildAgentCaseMarker, writeAgentCaseMarker } from '../../agent-case-marker.js';
|
||||
import { agentInbox } from '../agent-inbox.js';
|
||||
import { canUsernameRunPrivilegedCommands, resolveClaudeModeForUsername } from '../../user-store.js';
|
||||
import { clampEnvOverridesForOwner } from '../../session-env-clamp.js';
|
||||
import { enabledClis, getCli } from '../../config/cli-registry/registry.js';
|
||||
@@ -1680,6 +1681,9 @@ export function registerSessionRoutes(
|
||||
// below re-looks-up the record by id, so the returned SessionState is unused here.
|
||||
findPersistedSessionOrFail(ctx.store, id, req);
|
||||
ctx.store.demoteOrRemoveSession(id);
|
||||
// A detached session kept its agent inbox (server.ts _doCleanupSession drops it
|
||||
// only on killMux); retiring the row is the delete, so the mail goes now.
|
||||
agentInbox.drop(id);
|
||||
// Mirrors the broadcast at the tail of the live-session cleanup path
|
||||
// (_doCleanupSession in server.ts) — without it, other open tabs keep
|
||||
// showing the retired row until their next unrelated fetch.
|
||||
|
||||
+7
-2
@@ -1847,12 +1847,17 @@ export const SessionWaitOutputQuerySchema = z.object({
|
||||
/**
|
||||
* 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
|
||||
* multi-megabyte body never reaches it. It is 4× the byte cap in characters (a UTF-8
|
||||
* character is at most 4 bytes), so every over-size message gets the store's answer,
|
||||
* which tells the sender what to do instead. 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),
|
||||
text: z
|
||||
.string()
|
||||
.min(1, 'text must not be empty')
|
||||
.max(4 * 16_384),
|
||||
})
|
||||
.strict();
|
||||
|
||||
|
||||
+3
-2
@@ -1691,8 +1691,9 @@ export class WebServer extends EventEmitter {
|
||||
// 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.
|
||||
// detach: mail posted before that is still there when the session is re-attached
|
||||
// (while detached, the routes answer 404, so nothing new arrives). Before
|
||||
// cancelAll, so a reader released by it reads an empty inbox.
|
||||
if (killMux) agentInbox.drop(sessionId);
|
||||
sessionWaits.notifySignal(sessionId, 'exit');
|
||||
sessionWaits.cancelAll(sessionId);
|
||||
|
||||
Reference in new issue
Block a user