perf: remove history scans from task and chat queries (#16278)

## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Task reads and external chat notices must remain responsive as run history grows. > - The chat notice sweep casts indexed UUID columns to text. > - Run redaction and task provenance filter large JSON contexts with expressions that lack matching indexes. > - These queries scan history and read large stored contexts on common requests. > - This pull request adapts two existing fixes to current master and adds the missing provenance index. > - The benefit is less database work with the same query results and access rules. ## Linked Issues or Issue Description Refs #15927. This change adapts that chat sweep fix to the extracted inbound notice module and retains the contributor credit. Refs #12499. This change adds that missing nested-context index through a current, concurrent migration. It retains the contributor credit. This combined change replaces those older layouts for the measured query paths. It also covers task creation provenance, which those pull requests do not address. **What happened?** The chat notice sweep scans wakeup history. Issue redaction scans run contexts because only one branch of its JSON condition has an index. The task creation-source filter also scans contexts because its native and legacy ID expression lacks an index. **Expected behavior** The queries use indexed lookups and return the same rows. Company access, task origin precedence, secret redaction, and chat notice ordering remain the same. **Steps to reproduce** Run `pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-slow-queries.ts --baseline-ref=f8af817f8d`. The benchmark creates disposable PostgreSQL data and compares the old query with the current query on the same fixture. ## What Changed - Compare chat UUID columns directly with safely parsed payload IDs. Reject malformed and noncanonical IDs without a cast failure. - Add the nested `paperclipIssue.id` run-context index and the exact native/legacy task-provenance expression index. - Build both indexes concurrently. Register the migration for top-level execution and invalid-index retry repair. - Extend chat payload and migration regression coverage. - Add a repeatable synthetic SQL benchmark and document its scope and command. - Keep the five newest schema snapshots. ## Verification - Focused local checks: 76 selected tests passed in five files. These cover FIFO notices, malformed IDs, conversation state, redaction, provenance, and company boundaries. - Disposable PostgreSQL 18.1 fixture: 335 tasks, 1,400 runs, 13,000 wakeups, 377 actions, and 192 KiB synthetic contexts. - Five warm samples per query, after two warmups. Every returned SQL column matched before and after, compared as positional tuples so duplicate unnamed projections are retained. | Query | Before, median | After, median | Improvement | | --- | ---: | ---: | ---: | | Chat notice candidates | 26.21 ms | 6.43 ms | 4.1× | | Task creation-source list | 159.69 ms | 6.51 ms | 24.5× | | Issue redaction contexts | 131.88 ms | 0.34 ms | 389.1× | The run-history context projection returned the same 200 rows and kept the same index plan and buffer counts. Its timings were noisy (37.30 ms before and 56.78 ms after in the last run; 34.25 ms and 34.14 ms in the previous run). No improvement is claimed for this control. Its repeated JSON projection remains a separate optimization opportunity. The populated plans use the wakeup primary key, the provenance expression index, and a BitmapOr over both redaction indexes. These are local SQL measurements. They do not prove an equivalent change in hosted API latency. - `pnpm -r typecheck` and `pnpm build` passed locally. - The full local `pnpm test:run` was started and then interrupted (exit 130) when the Mac reached its PostgreSQL shared-memory limit. It did not complete and is not reported as passed. The final SQL benchmark passed after its disposable database resources were released. The full general and serialized test suites passed in CI. - All CI checks for final head `193acca6abaa992acad12f4b631bd31dd3234641` passed, including full Vitest coverage, E2E checks, typecheck, build, migration order, and lifecycle boundaries. - Greptile reviewed that exact head at 5/5 with no actionable comments. ## Risks - Both indexes add storage and write cost. Initial builds read stored run contexts and can take time on large histories. - Concurrent index builds avoid holding a write-blocking table lock for the whole build. The existing migration retry path removes invalid partial indexes before retrying. - The chat parser accepts only canonical lowercase UUID text. This preserves the prior text comparison, including rejection of uppercase and malformed values. - Staging has not been migrated or deployed. Local PostgreSQL and synthetic data do not reproduce hosted load or network latency. ## Model Used - OpenAI Codex, based on GPT-6. The exact serving model ID and context window were not exposed in this session. Used reasoning, repository tools, browser inspection, and local code execution. The adapted fixes retain their original contributor credits. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <[email protected]>
2026-11-07 06:16:06 -10:00 · 2026-10-14 07:37:51 -05:00
1 parent 5498e8ba64
commit e9b6c4d44c
@@ -2,6 +2,32 @@

 Paperclip uses PostgreSQL via [Drizzle ORM](https://orm.drizzle.team/). There are three ways to run the database, from simplest to most production-ready.

+### Task query performance
+
+Run-context lookups use expression indexes for both `context_snapshot.issueId`
+and `context_snapshot.paperclipIssue.id`. Task creation provenance has its own
+company-scoped expression index that follows the query's native issue ID and
+legacy `issueId`/`taskId`/`taskKey` precedence. Migration 0345 builds the two new
+indexes concurrently and can resume an interrupted build without blocking run
+writes for the full build. The indexes add storage and run-write overhead; they
+do not change task visibility or secret redaction.
+
+The chat wake-notice sweep compares UUID columns directly against guarded,
+canonical lowercase UUID payload values. Malformed and noncanonical payloads
+still match nothing. The query can use existing primary keys without adding a
+wakeup-history index.
+
+To compare all three queries on the same disposable local PostgreSQL fixture:
+
+```sh
+pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-slow-queries.ts --baseline-ref=origin/master
+```
+
+Use a revision before these fixes for `--baseline-ref` after they merge. The
+benchmark loads that revision's chat query, uses synthetic data only, compares
+complete returned rows, and prints timings and `EXPLAIN (ANALYZE, BUFFERS)` plans.
+Local results measure query execution, not staging request latency.
+
 ## 1. Embedded PostgreSQL — zero config

 If you don't set `DATABASE_URL`, the server automatically starts an embedded PostgreSQL instance and manages a local data directory.
@@ -473,6 +473,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([
  "0313_private_task_access.sql",
  "0314_private_task_draft_assets.sql",
  "0332_little_night_nurse.sql",
+  "0345_luxuriant_carmella_unuscione.sql",
 ]);

 async function applyPendingMigrationsManually(
@@ -27,6 +27,8 @@ d("heartbeat context_snapshot expression index migration", () => {
    expect(names).toContain("heartbeat_runs_company_ctx_issue_created_idx");
    expect(names).toContain("heartbeat_runs_company_ctx_task_created_idx");
    expect(names).toContain("heartbeat_runs_company_ctx_taskkey_created_idx");
+    expect(names).toContain("heartbeat_runs_company_ctx_paperclip_issue_created_idx");
+    expect(names).toContain("heartbeat_runs_company_origin_issue_idx");
    expect(names).toContain("agent_wakeup_requests_company_payload_issue_idx");

    await sql.unsafe("SET enable_seqscan = off");
@@ -52,6 +54,20 @@ d("heartbeat context_snapshot expression index migration", () => {
    const taskKeyText = taskKeyPlan.map((r) => Object.values(r)[0]).join("\n");
    expect(taskKeyText).toContain("heartbeat_runs_company_ctx_taskkey_created_idx");

+    // An empty table can use any company-prefix index for the OR. Check the
+    // new branch directly here; the populated benchmark proves BitmapOr use.
+    const redactionPlan = await sql.unsafe(
+      "EXPLAIN SELECT id FROM heartbeat_runs WHERE company_id = '00000000-0000-0000-0000-000000000001' AND context_snapshot -> 'paperclipIssue' ->> 'id' = 'x'",
+    );
+    const redactionText = redactionPlan.map((r) => Object.values(r)[0]).join("\n");
+    expect(redactionText).toContain("heartbeat_runs_company_ctx_paperclip_issue_created_idx");
+
+    const originPlan = await sql.unsafe(
+      "EXPLAIN SELECT id FROM heartbeat_runs WHERE company_id = '00000000-0000-0000-0000-000000000001' AND coalesce(native_issue_id::text, nullif(context_snapshot->>'issueId', ''), nullif(context_snapshot->>'taskId', ''), nullif(context_snapshot->>'taskKey', '')) IN ('x', 'SRC-1')",
+    );
+    expect(originPlan.map((r) => Object.values(r)[0]).join("\n"))
+      .toContain("heartbeat_runs_company_origin_issue_idx");
+
    const wakePlan = await sql.unsafe(
      "EXPLAIN SELECT id FROM agent_wakeup_requests WHERE company_id = '00000000-0000-0000-0000-000000000001' AND status = 'deferred_issue_execution' AND payload ->> 'issueId' = 'x' LIMIT 1",
    );
@@ -63,6 +79,7 @@ d("heartbeat context_snapshot expression index migration", () => {
    for (const migration of [
      "./migrations/0209_heartbeat_context_snapshot_indexes.sql",
      "./migrations/0210_heartbeat_context_taskkey_index.sql",
+      "./migrations/0345_luxuriant_carmella_unuscione.sql",
    ]) {
      const migrationSql = await readFile(
        fileURLToPath(new URL(migration, import.meta.url)),
@@ -0,0 +1,9 @@
+-- Keep both JSON branches indexable without blocking run writes during the build.
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "heartbeat_runs_company_ctx_paperclip_issue_created_idx" ON "heartbeat_runs" USING btree ("company_id",("context_snapshot" -> 'paperclipIssue' ->> 'id'),"created_at" DESC NULLS LAST);--> statement-breakpoint
+-- Match the creation-source query's exact native/legacy precedence.
+CREATE INDEX CONCURRENTLY IF NOT EXISTS "heartbeat_runs_company_origin_issue_idx" ON "heartbeat_runs" USING btree ("company_id",coalesce(
+        "native_issue_id"::text,
+        nullif("context_snapshot"->>'issueId', ''),
+        nullif("context_snapshot"->>'taskId', ''),
+        nullif("context_snapshot"->>'taskKey', '')
+      ));
@@ -2402,6 +2402,13 @@
      "when": 1791930226792,
      "tag": "0344_wide_blue_marvel",
      "breakpoints": true
+    },
+    {
+      "idx": 345,
+      "version": "7",
+      "when": 1791978393174,
+      "tag": "0345_luxuriant_carmella_unuscione",
+      "breakpoints": true
    }
  ]
 }
@@ -189,5 +189,17 @@ export const heartbeatRuns = pgTable(
      sql`(${table.contextSnapshot} ->> 'taskKey')`,
      table.createdAt.desc(),
    ),
+    companyCtxPaperclipIssueCreatedIdx: index("heartbeat_runs_company_ctx_paperclip_issue_created_idx")
+      .on(table.companyId, sql`(${table.contextSnapshot} -> 'paperclipIssue' ->> 'id')`, table.createdAt.desc())
+      .concurrently(),
+    // Match creation provenance precedence, including empty legacy IDs.
+    companyOriginIssueIdx: index("heartbeat_runs_company_origin_issue_idx")
+      .on(table.companyId, sql`coalesce(
+        ${table.nativeIssueId}::text,
+        nullif(${table.contextSnapshot}->>'issueId', ''),
+        nullif(${table.contextSnapshot}->>'taskId', ''),
+        nullif(${table.contextSnapshot}->>'taskKey', '')
+      )`)
+      .concurrently(),
  }),
 );
@@ -0,0 +1,184 @@
+/** Disposable PostgreSQL repro. Run: pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-slow-queries.ts */
+import assert from "node:assert/strict";
+import { randomBytes, randomUUID } from "node:crypto";
+import { execFileSync } from "node:child_process";
+import { promises as fs } from "node:fs";
+import path from "node:path";
+import { fileURLToPath, pathToFileURL } from "node:url";
+import { createRequire } from "node:module";
+const { and, eq, or, sql } = createRequire(new URL("../server/package.json", import.meta.url))("drizzle-orm");
+import {
+  agents, agentWakeupRequests, chatActions, chatConversations, chatEndpoints,
+  companies, createDb, heartbeatRuns, issues, startEmbeddedPostgresTestDatabase,
+  toolApplications, toolConnections, type Db,
+} from "../packages/db/src/index.js";
+import { issueService } from "../server/src/services/issues.js";
+import { issueReadSqlCondition } from "../server/src/services/authorization.js";
+import { createInboundWakeNotices } from "../server/src/services/chat-channels/inbound/wakeup-notices.js";
+
+console.error("[benchmark] imports complete");
+
+const migration = await fs.readFile(new URL("../packages/db/src/migrations/0345_luxuriant_carmella_unuscione.sql", import.meta.url), "utf8");
+const indexes = migration.split("--> statement-breakpoint").map((statement) => statement.trim()).filter(Boolean);
+const baselineRef = process.argv.find((arg) => arg.startsWith("--baseline-ref="))?.slice("--baseline-ref=".length) ?? "origin/master";
+const baselineRevision = execFileSync("git", ["rev-parse", "--verify", `${baselineRef}^{commit}`], { encoding: "utf8" }).trim();
+// The database helper uses unreferenced retry/cleanup timers. A standalone
+// benchmark must stay alive until startup succeeds or reports its real error.
+const keepAlive = setInterval(() => {}, 1000);
+const database = await startEmbeddedPostgresTestDatabase("slow-query-benchmark-");
+console.error("[benchmark] database ready");
+const db = createDb(database.connectionString);
+let baselineModuleDir: string | undefined;
+type Query = { sql: string; params: unknown[] };
+
+// Capture the real service SQL before execution; never dispatch notices or runs.
+async function captureQuery(invoke: (captureDb: Db) => Promise<unknown>): Promise<Query> {
+  let captured: Query | undefined;
+  const stop = new Error("query captured");
+  const captureDb = new Proxy(db, {
+    get(target, key) {
+      if (key !== "select") return Reflect.get(target, key);
+      return (...fields: Parameters<Db["select"]>) => {
+        const builder = db.select(...fields);
+        const from = builder.from.bind(builder);
+        builder.from = ((...args: Parameters<typeof from>) => {
+          const query = from(...args);
+          query.execute = async () => { captured = query.toSQL(); throw stop; };
+          return query;
+        }) as typeof builder.from;
+        return builder;
+      };
+    },
+  });
+  try { await invoke(captureDb); } catch (error) { if (error !== stop) throw error; }
+  assert.ok(captured, "service must produce a SQL query");
+  return captured;
+}
+
+async function measure(query: Query) {
+  const samples: number[] = [];
+  let rows: unknown[] = [];
+  for (let sample = 0; sample < 7; sample++) {
+    const start = performance.now();
+    // Positional rows retain every column, including unnamed SQL projections
+    // whose duplicate names would overwrite one another in object results.
+    rows = await db.$client.unsafe(query.sql, query.params as never[]).values();
+    if (sample > 1) samples.push(performance.now() - start);
+  }
+  const [explain] = await db.$client.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${query.sql}`, query.params as never[]);
+  samples.sort((a, b) => a - b);
+  return { medianMs: samples[2], rows, explain: explain["QUERY PLAN"] };
+}
+
+try {
+  // Load only the old query module against current dependencies. Keep it in a
+  // temporary server folder so bare package imports resolve within the worktree.
+  const noticeModulePath = "server/src/services/chat-channels/inbound/wakeup-notices.ts";
+  const noticeModuleUrl = new URL(`../${noticeModulePath}`, import.meta.url);
+  const oldModule = execFileSync("git", ["show", `${baselineRevision}:${noticeModulePath}`], { encoding: "utf8" });
+  const rewritten = oldModule.replace(/from "(\.{1,2}\/[^\"]+)"/g,
+    (_match, specifier: string) => `from ${JSON.stringify(fileURLToPath(new URL(specifier, noticeModuleUrl)))}`);
+  baselineModuleDir = await fs.mkdtemp(path.join(fileURLToPath(new URL("../server/scripts/", import.meta.url)), ".slow-query-baseline-"));
+  const baselineModuleFile = path.join(baselineModuleDir, "wakeup-notices.ts");
+  await fs.writeFile(baselineModuleFile, rewritten);
+  const { createInboundWakeNotices: baselineNotices } = await import(pathToFileURL(baselineModuleFile).href) as typeof import("../server/src/services/chat-channels/inbound/wakeup-notices.js");
+  console.error("[benchmark] baseline module ready");
+  const companyId = randomUUID();
+  const agentId = randomUUID();
+  const sourceId = randomUUID();
+  const endpointId = randomUUID();
+  const conversationId = randomUUID();
+  const applicationId = randomUUID();
+  const connectionId = randomUUID();
+  await db.insert(companies).values({ id: companyId, name: "Query benchmark", issuePrefix: "BENCH" });
+  await db.insert(agents).values({ id: agentId, companyId, name: "Fixture", role: "engineer", adapterType: "codex_local" });
+  await db.insert(issues).values({ id: sourceId, companyId, title: "Source", identifier: "BENCH-1", assigneeAgentId: agentId });
+  await db.insert(toolApplications).values({ id: applicationId, companyId, name: "Fixture", type: "native" });
+  await db.insert(toolConnections).values({ id: connectionId, companyId, applicationId, name: "Fixture", uid: connectionId, transport: "chat_sdk", connectionPurpose: "channel" });
+  await db.insert(chatEndpoints).values({ id: endpointId, companyId, connectionId, provider: "slack", publicId: endpointId, assignedAgentId: agentId, status: "active" });
+  await db.insert(chatConversations).values({ id: conversationId, companyId, endpointId, issueId: sourceId, externalConversationId: "fixture", externalThreadId: "fixture", externalLabel: "Fixture", state: "waiting" });
+
+  // Similar row counts to the investigation. Every context has 192 KiB of
+  // incompressible synthetic prompt text, so JSON filtering must visit TOAST.
+  const prompt = randomBytes(144 * 1024).toString("base64");
+  await db.$client.unsafe(`INSERT INTO heartbeat_runs (company_id, agent_id, status, native_issue_id, created_at, context_snapshot)
+    SELECT $1::uuid, $2::uuid, 'succeeded', CASE WHEN n = 3 THEN $3::uuid END, now() + n * interval '1 millisecond',
+      jsonb_build_object('prompt', $4::text) || CASE
+        WHEN n = 1 THEN jsonb_build_object('issueId', $3::text, 'paperclipSecretRedactions', jsonb_build_array(jsonb_build_object('fixture', 'top-level')))
+        WHEN n = 2 THEN jsonb_build_object('paperclipIssue', jsonb_build_object('id', $3::text), 'paperclipSecretRedactions', jsonb_build_array(jsonb_build_object('fixture', 'nested')))
+        WHEN n = 3 THEN jsonb_build_object('issueId', gen_random_uuid()::text)
+        WHEN n = 4 THEN jsonb_build_object('taskId', $3::text)
+        WHEN n = 5 THEN jsonb_build_object('taskKey', 'BENCH-1')
+        ELSE jsonb_build_object('issueId', gen_random_uuid()::text) END
+    FROM generate_series(1, 1400) n`, [companyId, agentId, sourceId, prompt]);
+  const originatingRuns = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).limit(5);
+  await db.insert(issues).values(Array.from({ length: 334 }, (_, n) => ({
+    companyId, title: `Candidate ${n}`, createdByUserId: "fixture",
+    originRunId: originatingRuns[n % originatingRuns.length].id,
+  })));
+  const wakes = Array.from({ length: 13000 }, (_, n) => ({
+    id: randomUUID(), companyId, agentId, source: "chat", status: n < 50 ? "deferred_issue_execution" : "succeeded",
+  }));
+  for (let offset = 0; offset < wakes.length; offset += 500) await db.insert(agentWakeupRequests).values(wakes.slice(offset, offset + 500));
+  await db.insert(chatActions).values(wakes.slice(0, 377).map((wake, n) => ({
+    id: wake.id, companyId, endpointId, conversationId, kind: "inbound_wakeup", providerActionId: wake.id,
+    status: "processed", payload: n % 3 === 0 ? { coalesced: true } : {},
+  })));
+  // Cover coalesced ownership as well as ordinary receipts.
+  for (let n = 1; n < 50; n += 2) await db.update(agentWakeupRequests)
+    .set({ payload: { coalescedIntoWakeupRequestId: wakes[n - 1].id } }).where(eq(agentWakeupRequests.id, wakes[n].id));
+  for (const [n, ownerId] of [[51, "not-a-uuid"], [52, wakes[0].id.toUpperCase()], [53, ""], [54, null]] as const) {
+    await db.update(agentWakeupRequests).set({ payload: { coalescedIntoWakeupRequestId: ownerId } }).where(eq(agentWakeupRequests.id, wakes[n].id));
+  }
+  const noticeQuery = (factory: typeof createInboundWakeNotices) => captureQuery(async (captureDb) => factory(captureDb, {
+    isShuttingDown: () => false, enqueueFailedChatRetryPublications: async () => 0,
+  } as never).enqueueInboundWakeupPublications(200));
+  const notice = await noticeQuery(baselineNotices);
+  const fixedNotice = await noticeQuery(createInboundWakeNotices);
+  const list = await captureQuery(async (captureDb) => issueService(captureDb).list(companyId, {
+    createdFromIssueId: sourceId, sortField: "id", sortDir: "asc", limit: 200,
+    readCondition: await issueReadSqlCondition(db, { type: "board", userId: "fixture", source: "cloud_tenant" }),
+  }));
+  const redactions = db.select({ contextSnapshot: sql`jsonb_build_object('paperclipSecretRedactions', ${heartbeatRuns.contextSnapshot}->'paperclipSecretRedactions')` })
+    .from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, companyId), or(
+      sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${sourceId}`,
+      sql`${heartbeatRuns.contextSnapshot}->'paperclipIssue'->>'id' = ${sourceId}`,
+    ))).toSQL();
+
+  // The summary history query projects these eight context fields. Keep its
+  // projection as a control: these indexes do not remove per-row JSON work.
+  const runHistory = db.select({ id: heartbeatRuns.id,
+    contextIssueId: sql`coalesce(${heartbeatRuns.issueId}::text, ${heartbeatRuns.contextSnapshot}->>'issueId')`,
+    ...Object.fromEntries(["taskId", "taskKey", "commentId", "wakeCommentId", "wakeReason", "wakeSource", "wakeTriggerDetail"]
+      .map((key) => [key, sql`${heartbeatRuns.contextSnapshot}->>${key}`])),
+  }).from(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId)).orderBy(sql`${heartbeatRuns.createdAt} desc`).limit(200).toSQL();
+  const queries = { chatNotice: notice, createdFromIssueList: list, issueRedactions: redactions, runHistory };
+  // Remove only these candidate indexes from this disposable fixture. This
+  // also permits repeatable comparisons after the migration is installed.
+  for (const definition of indexes) await db.$client.unsafe(`DROP INDEX IF EXISTS ${definition.match(/"([^"]+)"/)![0]}`);
+  await db.$client.unsafe("ANALYZE");
+  const baseline: Record<string, Awaited<ReturnType<typeof measure>>> = {};
+  for (const [name, query] of Object.entries(queries)) baseline[name] = await measure(query);
+  for (const definition of indexes) await db.$client.unsafe(definition);
+  await db.$client.unsafe("ANALYZE");
+  const results = [];
+  for (const [name, query] of Object.entries(queries)) {
+    const treatment = await measure(name === "chatNotice" ? fixedNotice : query);
+    // Unordered SQL returns a set, not a guaranteed order. Compare full rows.
+    const normalize = (rows: unknown[]) => rows.map((row) => JSON.stringify(row)).sort();
+    assert.deepEqual(normalize(treatment.rows), normalize(baseline[name].rows));
+    results.push({ name, baselineMs: baseline[name].medianMs, treatmentMs: treatment.medianMs,
+      speedup: baseline[name].medianMs / treatment.medianMs, equalRows: treatment.rows.length,
+      baselinePlan: baseline[name].explain, treatmentPlan: treatment.explain });
+  }
+  console.log(JSON.stringify({ baselineRevision, postgres: (await db.$client`SELECT version()`)[0].version,
+    fixture: { issues: 335, runs: 1400, wakes: 13000, actions: 377, promptBytes: Buffer.byteLength(prompt) }, results }, null, 2));
+} finally {
+  try {
+    await db.$client.end({ timeout: 0 });
+    await database.cleanup();
+    if (baselineModuleDir) await fs.rm(baselineModuleDir, { recursive: true, force: true });
+  } finally {
+    clearInterval(keepAlive);
+  }
+}
@@ -69275,6 +69275,72 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => {
    }
  });

+  it.each(["not-a-uuid", "", "uppercase"])("skips a pre-run FIFO candidate with a noncanonical coalesced owner id (%s)", async (ownerId) => {
+    const context = await deferredChatQueueFixture();
+    try {
+      const [receipt] = await db
+        .select()
+        .from(agentWakeupRequests)
+        .where(eq(agentWakeupRequests.id, context.action.id));
+      await db
+        .update(agentWakeupRequests)
+        .set({
+          payload: {
+            ...receipt.payload,
+            coalescedIntoWakeupRequestId: ownerId === "uppercase" ? context.action.id.toUpperCase() : ownerId,
+          },
+        })
+        .where(eq(agentWakeupRequests.id, context.action.id));
+      await expect(
+        context.service.enqueueInboundWakeupPublications(),
+      ).resolves.toBe(0);
+      await expect(
+        db
+          .select()
+          .from(chatPublications)
+          .where(eq(chatPublications.endpointId, context.endpoint.id)),
+      ).resolves.toEqual([]);
+    } finally {
+      await context.service.shutdown();
+    }
+  });
+
+  it("keeps the pre-run FIFO sweep running when a valid owner's action has comment ids that are not uuids", async () => {
+    const context = await deferredChatQueueFixture();
+    try {
+      // Publish the queued notice first. The removed-comment and queued-notice
+      // checks then have real rows to compare the malformed ids against.
+      await context.service.processPendingPublications();
+      await expect(
+        db
+          .select({ state: chatPublications.state })
+          .from(chatPublications)
+          .where(eq(chatPublications.endpointId, context.endpoint.id)),
+      ).resolves.toEqual([{ state: "published" }]);
+      await db
+        .update(chatActions)
+        .set({
+          payload: {
+            ...context.action.payload,
+            issueId: "not-a-uuid",
+            commentId: "not-a-uuid",
+          },
+        })
+        .where(eq(chatActions.id, context.action.id));
+      await expect(
+        context.service.enqueueInboundWakeupPublications(),
+      ).resolves.toBe(0);
+      await expect(
+        db
+          .select({ state: chatPublications.state })
+          .from(chatPublications)
+          .where(eq(chatPublications.endpointId, context.endpoint.id)),
+      ).resolves.toEqual([{ state: "published" }]);
+    } finally {
+      await context.service.shutdown();
+    }
+  });
+
  it("never promises a queued follow-up before durable admission succeeds", async () => {
    const context = await deferredChatQueueFixture(true);
    try {
@@ -1,6 +1,6 @@
 import type { DbOrTransaction } from "../runtime-types.js";
 import { notifyChatPublicationWork } from "../../chat-work-notifications.js";
-import { and, asc, eq, inArray, isNull, or, sql } from "drizzle-orm";
+import { and, asc, eq, inArray, isNull, or, sql, type SQL } from "drizzle-orm";
 import { alias } from "drizzle-orm/pg-core";
 import type { Db } from "@paperclipai/db";
 import { agentWakeupRequests, chatActions, chatDeliveries, chatMessageLinks, chatPublications, issueComments } from "@paperclipai/db";
@@ -135,10 +135,17 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) {
    const retryInserted = await enqueueFailedChatRetryPublications(limit);
    const owner = alias(agentWakeupRequests, "chat_notice_owner");
    const cursor = inboundWakeNoticeCursor;
+    // Keep UUID columns indexable. Match only canonical lowercase UUID text,
+    // preserving the old comparison while malformed payloads match nothing.
+    const payloadUuid = (value: SQL) =>
+      sql`(case when ${value} ~ '^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$' then (${value})::uuid end)`;
+    const payloadIssueId = payloadUuid(sql`${chatActions.payload}->>'issueId'`);
+    const payloadCommentId = payloadUuid(sql`${chatActions.payload}->>'commentId'`);
+    const coalescedOwnerId = sql`${agentWakeupRequests.payload}->>'coalescedIntoWakeupRequestId'`;
    const sourceRemoved = sql`exists (select 1 from issue_comments removed_comment
      where removed_comment.company_id = ${chatActions.companyId}
-        and removed_comment.issue_id::text = ${chatActions.payload}->>'issueId'
-        and removed_comment.id::text = ${chatActions.payload}->>'commentId'
+        and removed_comment.issue_id = ${payloadIssueId}
+        and removed_comment.id = ${payloadCommentId}
        and removed_comment.deleted_at is not null)`;
    const hasVisibleQueue = sql`exists (select 1 from chat_publications queued_notice
      join chat_message_links visible_queue on visible_queue.publication_id = queued_notice.id
@@ -148,7 +155,7 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) {
        and visible_queue.provider_message_id = queued_notice.provider_message_id
        and visible_queue.direction = 'outbound' and visible_queue.comment_id is null
      where queued_notice.company_id = ${chatActions.companyId}
-        and queued_notice.comment_id::text = ${chatActions.payload}->>'commentId'
+        and queued_notice.comment_id = ${payloadCommentId}
        and queued_notice.state = 'published'
        and queued_notice.idempotency_key = 'wake:' || ${owner.id}::text || ':queued:' ||
          ${chatActions.endpointId}::text || ':' || ${chatActions.conversationId}::text)`;
@@ -163,7 +170,7 @@ export function createInboundWakeNotices(db: Db, options: NoticeOptions) {
      )
      .innerJoin(
        owner,
-        sql`${owner.id}::text = coalesce(${agentWakeupRequests.payload}->>'coalescedIntoWakeupRequestId', ${agentWakeupRequests.id}::text)`,
+        sql`${owner.id} = case when ${coalescedOwnerId} is null then ${agentWakeupRequests.id} else ${payloadUuid(coalescedOwnerId)} end`,
      )
      .where(
        and(