test(queue): remove dependent indexes in legacy Slack fixture

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

@@ -12,6 +12,7 @@ import { networkReceiptsSchema } from "./migrations/161_network_receipts.js";
 import { sessionAddressReadIndexesSchema } from "./migrations/162_session_address_read_indexes.js";
 import { networkInstallationsSchema } from "./migrations/163_network_installations.js";
 import { queueItemsReplyToIndexSchema } from "./migrations/164_queue_items_reply_to_index.js";
+import { queueListReadIndexesSchema } from "./migrations/166_queue_list_read_indexes.js";
 import { activityGrantsSchema } from "./migrations/143_activity_grants.js";
 import { environmentBindingsSchema } from "./migrations/144_environment_bindings.js";
 import { nodePermissionsToSchema } from "./migrations/145_node_permissions_to.js";
@@ -331,4 +332,5 @@ export const ALL_MIGRATIONS: Migration[] = [
   sessionAddressReadIndexesSchema,
   networkInstallationsSchema,
   queueItemsReplyToIndexSchema,
+  queueListReadIndexesSchema,
 ];

packages/daemon/src/db/migrations/166_queue_list_read_indexes.ts

@@ -0,0 +1,18 @@
+import type { Migration } from "../migrate.js";
+
+/** Bound queue selection by its existing order, and notification reads by their eligible history. */
+export const queueListReadIndexesSchema: Migration = {
+  name: "166_queue_list_read_indexes.sql",
+  sql: `
+    CREATE INDEX IF NOT EXISTS idx_queue_items_created
+      ON queue_items(ts_created DESC);
+    CREATE INDEX IF NOT EXISTS idx_queue_items_active_created
+      ON queue_items(CASE WHEN state IN ('pending', 'in-progress', 'blocked') THEN 0 ELSE 1 END, ts_created DESC);
+    CREATE INDEX IF NOT EXISTS idx_queue_items_active_updated
+      ON queue_items(CASE WHEN state IN ('pending', 'in-progress', 'blocked') THEN 0 ELSE 1 END, ts_updated DESC);
+    CREATE INDEX IF NOT EXISTS idx_queue_transitions_owner_notification_qitem
+      ON queue_transitions(qitem_id, transition_id DESC) WHERE owner_notification_level IS NOT NULL;
+    CREATE INDEX IF NOT EXISTS idx_queue_transitions_archive_owner_notification_qitem
+      ON queue_transitions_archive(qitem_id, transition_id DESC) WHERE owner_notification_level IS NOT NULL;
+  `,
+};

packages/daemon/src/domain/queue-recovery.ts

@@ -42,7 +42,9 @@ export function freshRecoveryId(db: Database.Database, qitemId: string, overdueD
  * idx_queue_items_recovery_tags' WHERE clause (migration 099) byte for byte, which is what lets the
  * planner read that small index instead of scanning and parsing every row. */
 const RECOVERY_TAG_ROWS = `(tags LIKE '%"recovery-for:%' OR tags LIKE '%\\%')`;
-const RECOVERY_ORDER = "ORDER BY CASE WHEN state IN ('pending','in-progress','blocked') THEN 0 ELSE 1 END, ts_updated DESC, qitem_id DESC LIMIT 1";
+// Unary + preserves the numeric CASE order but prevents a prefix match on the full-history listing
+// index. Both standalone lookups and the batch need the small recovery-tag index instead.
+const RECOVERY_ORDER = "ORDER BY +CASE WHEN state IN ('pending','in-progress','blocked') THEN 0 ELSE 1 END, ts_updated DESC, qitem_id DESC LIMIT 1";
 /** A listing's recovery index, built by SQLite's own json_each (the same values, and the same truthy excludeTag
  * test, as the per-row query): for each string tag value, the rows that carry it in RECOVERY_ORDER, and each row's
  * string values. Built lazily, on the first lookup inside `withRecoveryIndex`, so a listing that never asks (a

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

@@ -188,6 +188,7 @@ export const migrationsForFullTestDb = [coreSchema, bindingsSessionsSchema, even
  * its exclusion); the guard's redundant-exclusion check then keeps the two in sync.
  */
 export const migrationsForFullTestDbExclusions: Record<string, string> = {
+  "166_queue_list_read_indexes.sql": "read indexes depend on owner-notification columns/archive omitted here (054/076); startup and queue-list-order-indexes.test.ts exercise ALL_MIGRATIONS.",
   "156_queue_erasure_link_indexes.sql": "indexes outbox_entries, inbox_entries and the stream tables (all excluded from the core edge) for queue erasure's lookups; erasure tests migrate the canonical full schema.",
   "082_archive_identity_provenance.sql": "archive extension; 054 is excluded from this core fixture. Archive identity regressions use ALL_MIGRATIONS.",
   "026_inbox_entries.sql": "inbox subsystem table — not on the shared core edge (inbox suites migrate it inline).",

packages/daemon/test/queue-list-order-indexes.test.ts

@@ -0,0 +1,164 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import Database from "better-sqlite3";
+import { migrate } from "../src/db/migrate.js";
+import { ALL_MIGRATIONS } from "../src/db/all-migrations.js";
+import { EventBus } from "../src/domain/event-bus.js";
+import { QueueRepository, type QueueListOptions } from "../src/domain/queue-repository.js";
+import { RECOVERY_INDEX_SQL, findQueueRecovery } from "../src/domain/queue-recovery.js";
+import { QueueTransitionLog } from "../src/domain/queue-transition-log.js";
+import { runStuckSweep } from "../src/domain/queue-stuck-sweep.js";
+import { OutboxHandler } from "../src/domain/outbox-handler.js";
+import { RigRepository } from "../src/domain/rig-repository.js";
+
+describe("bounded queue lists use their ordering indexes", () => {
+  let db: Database.Database;
+  let repo: QueueRepository;
+
+  beforeEach(() => {
+    db = new Database(":memory:");
+    migrate(db, ALL_MIGRATIONS);
+    repo = new QueueRepository(db, new EventBus(db), { loadHumanRegistry: () => ({ ok: true, entities: [] }) });
+    const insert = db.prepare(`INSERT INTO queue_items
+      (qitem_id, ts_created, ts_updated, source_session, destination_session, state, body, tags)
+      VALUES (?, ?, ?, 'sender@r', 'worker@r', ?, ?, ?)`);
+    db.transaction(() => {
+      for (let i = 0; i < 240; i++) {
+        // Equal timestamps across states exercise the existing scan's tie order too.
+        const ts = new Date(Date.UTC(2026, 10, 12, 0, 0, Math.floor(i / 3))).toISOString();
+        const tags = i % 7 === 0 ? ["first", "second"] : i % 2 ? ["first"] : ["second"];
+        insert.run(`q-${i}`, ts, ts, ["pending", "done", "blocked"][i % 3], "body".repeat(256), JSON.stringify(tags));
+      }
+    })();
+  });
+  afterEach(() => { vi.restoreAllMocks(); db.close(); });
+
+  function selection(options: QueueListOptions) {
+    const prepare = db.prepare.bind(db);
+    let selected: { sql: string; args: unknown[] } | undefined;
+    const spy = vi.spyOn(db, "prepare").mockImplementation((sql: string) => {
+      const statement = prepare(sql);
+      if (/^SELECT (?:\*|qitem_id,).* FROM queue_items\s+(?:WHERE\b|ORDER BY\b)/s.test(sql)) {
+        const all = statement.all.bind(statement);
+        statement.all = (...args: unknown[]) => { selected = { sql, args }; return all(...args); };
+      }
+      return statement;
+    });
+    let ids: string[];
+    try { ids = repo.list(options).map((item) => item.qitemId); }
+    finally { spy.mockRestore(); }
+    expect(selected).toBeDefined();
+    const { sql, args } = selected!;
+    const legacy = prepare(sql.replace("FROM queue_items", "FROM queue_items NOT INDEXED"))
+      .all(...args) as Array<{ qitem_id: string }>;
+    expect(ids).toEqual(legacy.map((row) => row.qitem_id));
+    const plan = prepare(`EXPLAIN QUERY PLAN ${sql}`).all(...args) as Array<{ detail: string }>;
+    return { ids, details: plan.map((row) => row.detail).join("\n") };
+  }
+
+  it.each([false, true])("does not scan and sort all rows for a bounded list (compact=%s)", (compact) => {
+    const { ids, details } = selection({ compact, limit: 6 });
+    expect(ids).toHaveLength(6);
+    expect(details).toContain("USING INDEX");
+    expect(details).not.toContain("TEMP B-TREE");
+  });
+
+  it.each([false, true])("keeps active rows first without sorting all rig history (compact=%s)", (compact) => {
+    const { ids, details } = selection({ rig: "r", compact, limit: 6 });
+    expect(ids).toHaveLength(6);
+    expect(details).toContain("USING INDEX");
+    expect(details).not.toContain("TEMP B-TREE");
+  });
+
+  it.each([false, true])("keeps scoped update-time ordering indexed (compact=%s)", (compact) => {
+    const { ids, details } = selection({ rig: "r", sortByUpdated: true, compact, limit: 6 });
+    expect(ids).toHaveLength(6);
+    expect(details).toContain("USING INDEX");
+    expect(details).not.toContain("TEMP B-TREE");
+  });
+
+  it("looks for owner notifications without reading ordinary live or archived transitions", () => {
+    const log = new QueueTransitionLog(db);
+    const prepare = db.prepare.bind(db);
+    let query = "";
+    const spy = vi.spyOn(db, "prepare").mockImplementation((sql: string) => {
+      if (sql.includes("ROW_NUMBER()") && sql.includes("owner_notification_level IS NOT NULL")) query = sql;
+      return prepare(sql);
+    });
+    try { expect(log.ownerNotificationsForQitems(["q-0"])).toEqual(new Map()); }
+    finally { spy.mockRestore(); }
+    expect(query).not.toBe("");
+    const plan = prepare(`EXPLAIN QUERY PLAN ${query}`).all(JSON.stringify(["q-0"])) as Array<{ detail: string }>;
+    const details = plan.map((row) => row.detail).join("\n");
+    expect(details).toContain("idx_queue_transitions_owner_notification_qitem");
+    expect(details).toContain("idx_queue_transitions_archive_owner_notification_qitem");
+  });
+
+  it("still applies both exact tags before the bound, including zero and unbounded limits", () => {
+    for (const limit of [0, 1, 6, -1]) {
+      const { ids } = selection({ rig: "r", tag: ["first", "second"], limit });
+      expect(ids.every((id) => Number(id.slice(2)) % 7 === 0)).toBe(true);
+      expect(ids).toHaveLength(limit < 0 ? 35 : Math.min(limit, 35));
+    }
+  });
+
+  it("keeps recovery lookup on its small partial index when listing indexes exist", () => {
+    const plan = db.prepare(`EXPLAIN QUERY PLAN ${RECOVERY_INDEX_SQL}`).all() as Array<{ detail: string }>;
+    expect(plan.some((row) => row.detail.includes("idx_queue_items_recovery_tags"))).toBe(true);
+  });
+
+  async function pointPlan(prefix: string, run: () => unknown | Promise<unknown>) {
+    const prepare = db.prepare.bind(db);
+    let selected: { sql: string; args: unknown[] } | undefined;
+    const spy = vi.spyOn(db, "prepare").mockImplementation((sql: string) => {
+      const statement = prepare(sql);
+      if (sql.startsWith(prefix)) {
+        const get = statement.get.bind(statement);
+        statement.get = (...args: unknown[]) => { selected = { sql, args }; return get(...args); };
+      }
+      return statement;
+    });
+    try { await run(); } finally { spy.mockRestore(); }
+    expect(selected).toBeDefined();
+    const plan = prepare(`EXPLAIN QUERY PLAN ${selected!.sql}`).all(...selected!.args) as Array<{ detail: string }>;
+    expect(plan.some((row) => row.detail.includes("idx_queue_items_recovery_tags"))).toBe(true);
+  }
+
+  it("keeps standalone recovery hits and misses on the recovery partial index", async () => {
+    db.prepare("UPDATE queue_items SET tags = ? WHERE qitem_id = 'q-0'").run(JSON.stringify(["recovery-for:subject"]));
+    for (const subject of ["subject", "no-match"]) {
+      for (const exclude of [undefined, "excluded"]) {
+        await pointPlan("SELECT qitem_id, state, ts_updated FROM queue_items", () => {
+          expect(findQueueRecovery(db, subject, exclude)?.qitemId ?? null).toBe(subject === "subject" ? "q-0" : null);
+        });
+      }
+    }
+  });
+
+  async function sweepForPlan(state: "pending" | "in-progress") {
+    db.exec("DELETE FROM queue_items");
+    repo.attachOutbox(new OutboxHandler(db));
+    if (state === "pending") {
+      const rigs = new RigRepository(db);
+      const rig = rigs.createRig("r");
+      const node = rigs.addNode(rig.id, "worker", { runtime: "claude-code" });
+      db.prepare("INSERT INTO sessions (id,node_id,session_name,status) VALUES ('plan-session',?,'worker@r','running')").run(node.id);
+    }
+    const past = new Date(Date.now() - 120 * 60_000).toISOString();
+    const insert = db.prepare(`INSERT INTO queue_items
+      (qitem_id, ts_created, ts_updated, source_session, destination_session, state, body, claimed_at, closure_required_at)
+      VALUES (?, ?, ?, 'sender@r', 'worker@r', ?, 'work', ?, ?)`);
+    for (let i = 0; i < (state === "pending" ? 3 : 1); i++) {
+      insert.run(`candidate-${i}`, past, past, state, state === "in-progress" ? past : null, state === "in-progress" ? past : null);
+    }
+    return runStuckSweep({ db, queueRepo: repo, resolveOrchestrator: () => null, resolveRecoveryAgentSeat: () => null,
+      resolveOperatorSeat: () => null, isRegisteredHost: () => false, unclaimedAgeMinutes: 60, log: () => {} });
+  }
+
+  it("keeps the per-row finding lookup on the recovery partial index", async () => {
+    await pointPlan("SELECT qitem_id, source_session, state, ts_updated FROM queue_items", () => sweepForPlan("in-progress"));
+  });
+
+  it("keeps the latest backlog finding lookup on the recovery partial index", async () => {
+    await pointPlan("SELECT qitem_id, state, ts_updated, summary, body, destination_session, tags FROM queue_items", () => sweepForPlan("pending"));
+  });
+});

packages/daemon/test/slack-sweep-selection.test.ts

@@ -96,6 +96,8 @@ describe("Slack sweep selection", () => {
   it("keeps a legacy schema without owner classification quiet", async () => {
     row("legacy");
     for (const table of ["queue_transitions", "queue_transitions_archive"]) {
+      // The simulated pre-classification schema also predates its partial index.
+      db.exec(`DROP INDEX idx_${table}_owner_notification_qitem`);
       db.exec(`ALTER TABLE ${table} DROP COLUMN owner_notification_level; ALTER TABLE ${table} DROP COLUMN owner_notification_kind`);
     }
     repo = new QueueRepository(db, new EventBus(db), { loadHumanRegistry: () => registry });