feat: read a rig's outbox in one request (#2496)

A daemon client can now read a rig's messages in one request: `GET /api/queue/outbox/list?rig=<rig>` returns the messages sent from or to that rig's seats, newest first, with the same `limit`. `?senderSession=` still returns one sender's messages, unchanged, and when both are given `senderSession` is used. Migration 148 adds an index on the outbox's dispatch time with both session columns, so the rig read walks newest first, tests the match inside the index and stops at its limit instead of scanning and sorting the whole table. On a copy of a 109,622-row outbox, a busy rig's read went from 378 ms to 0.3 ms and an empty rig's from 30 ms to 19 ms; building the index took 0.37 s. The client reference now says queue SSE events are hints to read `queue/list?rig=` again, since some events name only the destination, and that the rig queue read scans its table, so read it occasionally. Limits: a rig matches by its `@<rig>` suffix with SQLite's ASCII case folding, as `queue/list?rig=` does, and a quiet rig's read still walks the index. Reported by @ziadeh in #2487 (item 1).

docs/reference/daemon-http-clients.md

@@ -86,6 +86,18 @@ All three are server-sent events (`text/event-stream`):
 
 ## Route details
 
+- **A rig's queue history:** `GET /api/queue/list?rig=<rig>&updatedSince=<ISO time>&sort=updated` returns the
+  rig's queue items (any whose source or destination seat is in the rig) updated at or after that time: open items
+  (`pending`, `in-progress`, `blocked`) first, then the rest by latest update. `limit` defaults to 100 and has no
+  upper cap. The list is cut after `limit` rows in that order, so when it comes back full, raise `limit` to read
+  them all. Each item's own history is `GET /api/queue/<qitemId>/transitions`. This read scans the queue table, so
+  read it occasionally rather than polling it. For live changes, treat `GET /api/queue/sse` events as hints to read
+  `queue/list?rig=` again. Not every event names both seats (`queue.claimed` and `queue.unclaimed` carry only the
+  destination), and not every state change is on that stream, so the events alone don't track a rig's items.
+- **A rig's messages:** `GET /api/queue/outbox/list?rig=<rig>` returns the messages sent from or to the rig's
+  seats, newest first; `?senderSession=<session>` still returns one sender's. One of the two is required; with
+  both, `senderSession` is used and `rig` is ignored. `limit` defaults to 100. No event is sent when a message is delivered, so read this list to see delivery
+  states.
 - **`GET /api/queue/recent-transitions`** returns at most 20 rows. A positive `limit` above 20 is capped at 20;
   a missing, zero, negative or unparseable `limit` means 20.
 - **`POST /api/transport/send`** takes `urgent: true`, the same as `rig send --urgent`. The daemon records who

packages/daemon/src/db/all-migrations.ts

@@ -3,6 +3,7 @@ import { environmentBindingsSchema } from "./migrations/144_environment_bindings
 import { nodePermissionsToSchema } from "./migrations/145_node_permissions_to.js";
 import { nodeStartupGuidanceFileSchema } from "./migrations/146_node_startup_guidance_file.js";
 import { managedBlockWritesSchema } from "./migrations/147_managed_block_writes.js";
+import { outboxDispatchedIndexSchema } from "./migrations/148_outbox_dispatched_index.js";
 import { nodeAccountHomeSchema } from "./migrations/141_node_account_home.js";
 import { escalationTargetRefsSchema } from "./migrations/126_escalation_target_refs.js";
 import { claudeTokenReadingsSchema } from "./migrations/134_claude_token_readings.js";
@@ -297,4 +298,5 @@ export const ALL_MIGRATIONS: Migration[] = [
   nodePermissionsToSchema,
   nodeStartupGuidanceFileSchema,
   managedBlockWritesSchema,
+  outboxDispatchedIndexSchema,
 ];

packages/daemon/src/db/migrations/148_outbox_dispatched_index.ts

@@ -0,0 +1,13 @@
+import type { Migration } from "../migrate.js";
+
+/**
+ * GET /api/queue/outbox/list?rig= matches session names by suffix, which no session index can serve. This index lets
+ * that read walk the outbox newest-first and stop at its limit, instead of scanning and sorting the whole table. The
+ * session columns ride along so the match is tested in the index, and only matching rows are read.
+ */
+export const outboxDispatchedIndexSchema: Migration = {
+  name: "148_outbox_dispatched_index.sql",
+  sql: `
+    CREATE INDEX IF NOT EXISTS idx_outbox_entries_ts_dispatched ON outbox_entries(ts_dispatched, sender_session, destination_session);
+  `,
+};

packages/daemon/src/domain/outbox-handler.ts

@@ -382,6 +382,18 @@ export class OutboxHandler {
     return rows.map((r) => this.rowToEntry(r));
   }
 
+  /** Messages a rig's seats sent or were sent, newest first: sender or destination `<member>@<rig>`. */
+  listForRig(rig: string, limit = 100): OutboxEntry[] {
+    const pattern = `%@${rig.replace(/[\\%_]/g, (ch) => `\\${ch}`)}`;
+    const rows = this.db
+      .prepare(
+        `SELECT * FROM outbox_entries WHERE sender_session LIKE ? ESCAPE '\\' OR destination_session LIKE ? ESCAPE '\\'
+         ORDER BY ts_dispatched DESC, rowid DESC LIMIT ?`
+      )
+      .all(pattern, pattern, limit) as OutboxEntryRow[];
+    return rows.map((r) => this.rowToEntry(r));
+  }
+
   /**
    * W1 (transactional closure): list still-`pending` rows whose outbox_id begins
    * with `idPrefix`, oldest first. The drain uses this to recover intents a crash

packages/daemon/src/routes/queue.ts

@@ -1476,9 +1476,12 @@ export function queueRoutes(): Hono {
 
   app.get("/outbox/list", (c) => {
     const senderSession = c.req.query("senderSession");
-    if (!senderSession) return c.json({ error: "senderSession is required" }, 400);
+    // A rig's messages, sent from or to its seats: the same scope as GET /list?rig=.
+    const rig = c.req.query("rig")?.trim();
+    if (!senderSession && !rig) return c.json({ error: "senderSession or rig is required" }, 400);
     const parsed = queryLimit(c);
     if (!parsed.ok) return parsed.response;
+    if (!senderSession) return c.json(getOutbox(c).listForRig(rig!, parsed.limit));
     const sessions = c.get("sessionRegistry" as never) as SessionRegistry | undefined;
     return c.json(getOutbox(c).listForSender(senderSession, parsed.limit, sessions?.sessionAddresses(senderSession)));
   });

packages/daemon/test/helpers/test-app.ts

@@ -183,6 +183,7 @@ export const migrationsForFullTestDbExclusions: Record<string, string> = {
   "086_classification_fields_and_attempts.sql": "extends 028 (excluded here) and adds the S02 attempt ledger — classifier suites migrate it inline.",
   "087_seat_delivery_guard.sql": "extends outbox_entries (027 is excluded); delivery-guard suites use ALL_MIGRATIONS. Revisit if a core-fixture consumer uses retained delivery.",
   "100_typing_guard_modes.sql": "extends outbox_entries and its typing guard (027 and 087 are excluded); guard-mode and retry suites use ALL_MIGRATIONS.",
+  "148_outbox_dispatched_index.sql": "indexes outbox_entries (027 is excluded); outbox suites migrate it inline.",
   "089_classification_identity_provenance.sql": "extends project_classifications (028 is excluded); classifier fixtures migrate it inline.",
   "129_stream_dispositions.sql": "companion of project_classifications (028 is excluded); stream-disposition suites migrate it inline.",
   "130_stream_subscriptions.sql": "stream reader subscriptions and retention marks; stream suites migrate it inline.",

packages/daemon/test/queue-routes.test.ts

@@ -1573,6 +1573,23 @@ describe("queue routes", () => {
     expect(data[0]!.body).toBe("fyi");
   });
 
+  it("outbox list for a rig returns what its seats sent and what was sent to them, in one read", async () => {
+    for (const [from, to, body] of [["a@r", "b@r", "inside"], ["c@other", "b@r", "into r"], ["c@other", "d@other", "elsewhere"]]) {
+      const record = await app.request("/api/queue/outbox/record", {
+        method: "POST", headers: { "Content-Type": "application/json", "X-OpenRig-Session": from! },
+        body: JSON.stringify({ senderSession: from, destinationSession: to, body }),
+      });
+      expect(record.status).toBe(201);
+    }
+    const list = await app.request("/api/queue/outbox/list?rig=r");
+    expect(list.status).toBe(200);
+    expect(((await list.json()) as Array<{ body: string }>).map((e) => e.body).sort()).toEqual(["inside", "into r"]);
+    // Control: the per-seat read is unchanged, one sender's messages wherever they went.
+    const seat = await app.request("/api/queue/outbox/list?senderSession=c@other");
+    expect(((await seat.json()) as Array<{ body: string }>).map((e) => e.body).sort()).toEqual(["elsewhere", "into r"]);
+    expect((await app.request("/api/queue/outbox/list")).status).toBe(400);
+  });
+
   it("queue, inbox and outbox list refuse only a non-numeric or unbindable limit (#586)", async () => {
     for (const body of ["one", "two", "three"]) {
       const res = await app.request("/api/queue/create", {