diff --git a/doc/DATABASE.md b/doc/DATABASE.md
index c3cf5eddf7c..6ee2badb769 100644
--- a/doc/DATABASE.md
+++ b/doc/DATABASE.md
@@ -1929,6 +1929,22 @@ delivery queries, including complete positional results and database plans.
 All data is synthetic. Query timings exclude process liveness probes, HTTP and
 provider work.
 
+The reconciler's lost-wake check for user comments reads in three stages.
+The first stage keeps recent user comments that no wake request carries. The
+second stage removes comments with a live or later run, and it requires an
+earlier run. Each stage probes native and legacy run bindings separately with
+existing indexes. Only the remaining comments reach the steer check, which
+reads run event payloads. An OR across both bindings let PostgreSQL hash the
+runs by company only and test each comment against every run of its company.
+No new index or migration is needed.
+
+```sh
+pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-comment-wake-check.ts --baseline-ref=02bfe71a6b
+```
+
+This disposable fixture compares the complete check result, the complete
+positional SQL rows, plans, and timings with and without nested-loop joins.
+
 ### Agent runtime synchronization journal
 
 Migration 0347 adds the opt-in `agent_runtime` resource stream. Database triggers
diff --git a/scripts/benchmark-comment-wake-check.ts b/scripts/benchmark-comment-wake-check.ts
new file mode 100644
index 00000000000..b8a475902a3
--- /dev/null
+++ b/scripts/benchmark-comment-wake-check.ts
@@ -0,0 +1,266 @@
+/**
+ * Local red/green proof for the reconciler's user-comment wake check
+ * (`checkUserCommentsWithoutWake` in the ownership invariant report).
+ *
+ * Loads the check from an explicit pre-change revision and from the working
+ * tree, runs both against the same synthetic PostgreSQL fixture, and asserts
+ * that the complete service result and the complete positional SQL rows are
+ * equal before timing anything.
+ *
+ *   pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-comment-wake-check.ts --baseline-ref=<commit>
+ *
+ * Options: --fixture=<name> (repeatable), --samples=<n>.
+ */
+import assert from "node:assert/strict";
+import { execFileSync } from "node:child_process";
+import { createHash, randomUUID } from "node:crypto";
+import { promises as fs } from "node:fs";
+import path from "node:path";
+import { fileURLToPath, pathToFileURL } from "node:url";
+import type { SQL } from "drizzle-orm";
+import { createDb, startEmbeddedPostgresTestDatabase, type Db } from "../packages/db/src/index.js";
+
+type Check = (db: Db, now: Date, limit: number, processStartedAt: Date) => Promise<unknown>;
+type Query = { sql: string; params: unknown[] };
+
+const baselineRef = process.argv.find(value => value.startsWith("--baseline-ref="))?.slice(15);
+assert.ok(baselineRef, "Pass an explicit pre-change --baseline-ref=<commit>");
+// Run every git command from the repository root: the documented command runs from server/.
+const root = fileURLToPath(new URL("../", import.meta.url));
+const baselineRevision = execFileSync("git", ["rev-parse", "--verify", `${baselineRef}^{commit}`], { encoding: "utf8", cwd: root }).trim();
+const samplesPerArm = Number(process.argv.find(value => value.startsWith("--samples="))?.slice(10) ?? 5);
+const sourcePath = "server/src/services/recovery/ownership-invariant-report.ts";
+// Every helper the check imports must be unchanged, or the baseline is contaminated.
+const changedRuntime = execFileSync("git", ["diff", "--name-only", baselineRevision, "--",
+  "server/src/services", "packages/db/src", "packages/shared/src", "pnpm-lock.yaml"], { encoding: "utf8", cwd: root })
+  .trim().split("\n").filter(Boolean);
+assert.deepEqual(changedRuntime.filter(file => file !== sourcePath), [], "The baseline must not import changed runtime helpers");
+
+const baselineSource = execFileSync("git", ["show", `${baselineRevision}:${sourcePath}`], { encoding: "utf8", cwd: root });
+const baselineDir = await fs.mkdtemp(path.join(root, "server/scripts/.comment-wake-baseline-"));
+const originalUrl = new URL(`../${sourcePath}`, import.meta.url);
+const rewritten = baselineSource.replace(/from "(\.{1,2}\/[^"]+)"/g,
+  (_match, specifier: string) => `from ${JSON.stringify(fileURLToPath(new URL(specifier, originalUrl)))}`)
+  + "\nexport { checkUserCommentsWithoutWake as __benchmarkCheck };\n";
+await fs.writeFile(path.join(baselineDir, "ownership-invariant-report.ts"), rewritten);
+const baseline = (await import(pathToFileURL(path.join(baselineDir, "ownership-invariant-report.ts")).href) as { __benchmarkCheck: Check }).__benchmarkCheck;
+const treatment = (await import("../server/src/services/recovery/ownership-invariant-report.js") as unknown as { checkUserCommentsWithoutWake: Check }).checkUserCommentsWithoutWake;
+
+const fixtures = [
+  // `steerWakeLostEvery`: a comment sent to a live run normally has a deferred
+  // wake that carries it. Steered comments on issues with n % steerWakeLostEvery = 2
+  // lost that wake (a restart), so only the steer check can rule them out.
+  // Few comments, shallow history: the check must stay cheap.
+  { name: "small", issues: 300, runs: 2_000, eventsPerRun: 6, hotIssues: 20, commentsPerHotIssue: 2, steerEvents: 40, payloadBytes: 1_024, contextBytes: 1_024, steerWakeLostEvery: 6 },
+  // A busy instance: deep run history and wide contexts, a few hundred recent comments.
+  { name: "representative", issues: 4_000, runs: 60_000, eventsPerRun: 6, hotIssues: 200, commentsPerHotIssue: 3, steerEvents: 400, payloadBytes: 2_048, contextBytes: 8_192, steerWakeLostEvery: 60 },
+  // The same instance after a restart lost the wakes of steered comments: more
+  // comments reach the run checks.
+  { name: "lost-steer-wakes", issues: 4_000, runs: 30_000, eventsPerRun: 10, hotIssues: 200, commentsPerHotIssue: 3, steerEvents: 400, payloadBytes: 2_048, contextBytes: 4_096, steerWakeLostEvery: 6 },
+  // Deeper history: twice the runs per company with the same recent traffic.
+  { name: "deep-history", issues: 6_000, runs: 120_000, eventsPerRun: 2, hotIssues: 300, commentsPerHotIssue: 3, steerEvents: 200, payloadBytes: 2_048, contextBytes: 8_192, steerWakeLostEvery: 60 },
+  // A busy hour: thousands of recent user comments over a deep run history.
+  { name: "busy-hour", issues: 6_000, runs: 100_000, eventsPerRun: 2, hotIssues: 1_500, commentsPerHotIssue: 3, steerEvents: 100, payloadBytes: 2_048, contextBytes: 2_048, steerWakeLostEvery: 60 },
+  // Stress: every steered comment lost its wake, and each steered run has thousands
+  // of large item events after the comment.
+  { name: "long-steered-runs", issues: 2_000, runs: 15_000, eventsPerRun: 8, hotIssues: 120, commentsPerHotIssue: 4, steerEvents: 2_000, payloadBytes: 4_096, contextBytes: 4_096, steerWakeLostEvery: 6 },
+];
+
+const keepAlive = setInterval(() => {}, 1000);
+const started = performance.now();
+const median = (values: number[]) => [...values].sort((a, b) => a - b)[Math.floor(values.length / 2)]!;
+const multiset = (rows: unknown[]) => rows.map(row => JSON.stringify(row)).sort();
+let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
+let db: Db | undefined;
+
+/** Synthetic, incompressible text of about `bytes` characters (md5 hex chain). */
+const filler = (bytes: number, seed: string) =>
+  `(select string_agg(md5(${seed}::text || ':' || g::text), '') from generate_series(1, ${Math.max(1, Math.ceil(bytes / 32))}) g)`;
+
+async function seed(db: Db, fixture: typeof fixtures[number], now: Date) {
+  const run = (text: string, params: unknown[] = []) => db.$client.unsafe(text, params as never[]);
+  for (const [company, scale] of [[randomUUID(), 1], [randomUUID(), 0.25]] as const) {
+    const prefix = `B${company.slice(0, 6).toUpperCase()}`;
+    const issueCount = Math.ceil(fixture.issues * scale), runCount = Math.ceil(fixture.runs * scale);
+    const hot = Math.ceil(fixture.hotIssues * scale);
+    await run(`insert into companies (id, name, issue_prefix) values ($1::uuid, 'Synthetic', $2)`, [company, prefix]);
+    await run(`insert into agents (company_id, name, adapter_type) select $1::uuid, 'agent-' || n, 'codex_local' from generate_series(1, 20) n`, [company]);
+    await run(`create temp table if not exists bench_agents (company_id uuid, n int, id uuid)`);
+    await run(`insert into bench_agents select company_id, row_number() over (order by id), id from agents where company_id = $1::uuid`, [company]);
+    // Issue n: status by n % 10; the first `hot` issues receive recent user comments.
+    await run(`insert into issues (company_id, title, status, assignee_agent_id, identifier, issue_number)
+      select $1::uuid, 'I-' || s.num, (array['done','done','todo','in_progress','done','cancelled','todo','in_progress','done','todo'])[1 + s.num % 10],
+        (select a.id from bench_agents a where a.company_id = $1::uuid and a.n = 1 + s.num % 20), $2 || '-' || s.num, s.num
+      from generate_series(1, $3::int) s(num)`, [company, prefix, issueCount]);
+    await run(`create temp table if not exists bench_issues (company_id uuid, n int, id uuid, agent_id uuid)`);
+    await run(`insert into bench_issues select company_id, issue_number, id, assignee_agent_id from issues where company_id = $1::uuid`, [company]);
+    // Run history over 14 days. Legacy runs bind by context, native runs by column, some by both.
+    await run(`insert into heartbeat_runs (company_id, agent_id, status, invocation_source, native_issue_id, context_snapshot,
+        created_at, started_at, finished_at)
+      select $1::uuid, i.agent_id, 'succeeded', 'assignment', case when r % 3 <> 0 then i.id end,
+        case when r % 3 = 1 then '{}'::jsonb else jsonb_build_object('issueId', i.id::text, 'prompt', ${filler(fixture.contextBytes, "r")}) end,
+        $2::timestamptz - interval '2 hours' - (r * interval '14 days' / $3::int), $2::timestamptz - interval '2 hours' - (r * interval '14 days' / $3::int),
+        $2::timestamptz - interval '2 hours' - (r * interval '14 days' / $3::int) + interval '20 minutes'
+      from generate_series(1, $3::int) r join bench_issues i on i.company_id = $1::uuid and i.n = 1 + (r * 7919) % $4::int`,
+      [company, now.toISOString(), runCount, issueCount]);
+    await run(`insert into heartbeat_run_events (company_id, run_id, agent_id, seq, event_type, created_at, payload)
+      select r.company_id, r.id, r.agent_id, e, (array['item.started','item.completed','item.started','log','item.completed'])[1 + e % 5],
+        r.started_at + e * interval '1 second', jsonb_build_object('prpEvent', jsonb_build_object('itemId', 'item-' || e, 'text', ${filler(fixture.payloadBytes, "r.id::text || e")}))
+      from heartbeat_runs r cross join generate_series(1, $2::int) e where r.company_id = $1::uuid`, [company, fixture.eventsPerRun]);
+    // Hot issues: one recent run each, then user comments in the last hour. Shape by issue n % 6:
+    // 0,1 normal (a wake carries each comment, a later run reads it); 2 steered (the run that
+    // was live at comment time acknowledged it in an item event); 3 lost (no wake, no later run:
+    // the reported case); 4 live run (queued); 5 carried by a goal wake context list.
+    await run(`insert into heartbeat_runs (company_id, agent_id, status, invocation_source, native_issue_id, context_snapshot, created_at, started_at, finished_at)
+      select i.company_id, i.agent_id, case when i.n % 6 = 4 then 'queued' else 'succeeded' end, 'assignment',
+        case when i.n % 2 = 0 then i.id end, jsonb_build_object('issueId', i.id::text, 'prompt', ${filler(fixture.contextBytes, "i.n")}),
+        $2::timestamptz - interval '58 minutes', $2::timestamptz - interval '58 minutes',
+        case when i.n % 6 = 2 then $2::timestamptz - interval '1 minute' when i.n % 6 = 4 then null else $2::timestamptz - interval '57 minutes' end
+      from bench_issues i where i.company_id = $1::uuid and i.n <= $3::int`, [company, now.toISOString(), hot]);
+    await run(`insert into issue_comments (company_id, issue_id, author_user_id, body, created_at)
+      select i.company_id, i.id, 'user-1', 'comment ' || k,
+        -- Distinct times: the check orders by creation time only.
+        $2::timestamptz - interval '55 minutes' + ((k - 1) * $3::int + i.n) * (interval '50 minutes' / ($3::int * $4::int))
+          + $5::int * interval '1 millisecond'
+      from bench_issues i cross join generate_series(1, $4::int) k where i.company_id = $1::uuid and i.n <= $3::int`,
+      [company, now.toISOString(), hot, fixture.commentsPerHotIssue, scale === 1 ? 0 : 500]);
+    // Older comment history outside the lookback window.
+    await run(`insert into issue_comments (company_id, issue_id, author_user_id, body, created_at)
+      select i.company_id, i.id, 'user-1', 'old', $2::timestamptz - interval '2 hours' - (k * interval '3 hours')
+      from bench_issues i cross join generate_series(1, 8) k where i.company_id = $1::uuid`, [company, now.toISOString()]);
+    await run(`insert into agent_wakeup_requests (company_id, agent_id, source, status, payload, created_at, updated_at)
+      select c.company_id, i.agent_id, 'automation', case when i.n % 6 = 2 then 'coalesced' else 'completed' end,
+        case when i.n % 6 = 5 then jsonb_build_object('issueId', c.issue_id::text, '_paperclipWakeContext', jsonb_build_object('wakeCommentIds', jsonb_build_array(c.id::text)))
+          else jsonb_build_object('issueId', c.issue_id::text, 'commentId', c.id::text) end, c.created_at, c.created_at
+      from issue_comments c join bench_issues i on i.id = c.issue_id
+      where c.company_id = $1::uuid and (i.n % 6 in (0, 1, 5) or (i.n % 6 = 2 and i.n % $3::int <> 2)
+        or c.created_at < $2::timestamptz - interval '2 hours')`, [company, now.toISOString(), fixture.steerWakeLostEvery]);
+    // Normal comments get a run after them.
+    await run(`insert into heartbeat_runs (company_id, agent_id, status, invocation_source, context_snapshot, created_at, started_at, finished_at)
+      select c.company_id, i.agent_id, 'succeeded', 'automation', jsonb_build_object('issueId', c.issue_id::text, 'commentId', c.id::text),
+        c.created_at + interval '2 seconds', c.created_at + interval '2 seconds', c.created_at + interval '30 seconds'
+      from issue_comments c join bench_issues i on i.id = c.issue_id
+      where c.company_id = $1::uuid and i.n % 6 in (0, 1) and c.created_at >= $2::timestamptz - interval '1 hour'`, [company, now.toISOString()]);
+    // Steered runs: many large item events after each comment; one acknowledges the comment.
+    await run(`insert into heartbeat_run_events (company_id, run_id, agent_id, seq, event_type, created_at, payload)
+      select r.company_id, r.id, r.agent_id, 1000 + e, (array['item.started','item.completed'])[1 + e % 2],
+        r.started_at + interval '2 minutes' + e * (interval '55 minutes' / $2::int),
+        jsonb_build_object('prpEvent', jsonb_build_object('itemId', 'item-' || e, 'text', ${filler(fixture.payloadBytes, "r.id::text || e")}))
+      from heartbeat_runs r join bench_issues i on i.company_id = r.company_id
+        and (r.native_issue_id = i.id or r.context_snapshot->>'issueId' = i.id::text)
+      cross join generate_series(1, $2::int) e
+      where r.company_id = $1::uuid and i.n <= $3::int and i.n % 6 = 2 and r.started_at >= $4::timestamptz - interval '1 hour'`,
+      [company, fixture.steerEvents, hot, now.toISOString()]);
+    await run(`insert into heartbeat_run_events (company_id, run_id, agent_id, seq, event_type, created_at, payload)
+      select c.company_id, r.id, r.agent_id, 100000 + row_number() over (partition by r.id order by c.created_at), 'item.completed', c.created_at + interval '3 seconds',
+        jsonb_build_object('prpEvent', jsonb_build_object('itemId', 'provider:steer:' || c.id::text, 'payload', jsonb_build_object('kind', 'steering_acknowledgement')))
+      from issue_comments c join bench_issues i on i.id = c.issue_id
+      join heartbeat_runs r on r.company_id = c.company_id and r.native_issue_id = c.issue_id and r.started_at < c.created_at and r.finished_at >= c.created_at
+      where c.company_id = $1::uuid and i.n % 6 = 2 and c.created_at >= $2::timestamptz - interval '1 hour'`, [company, now.toISOString()]);
+    // Closures of done tasks: agent closes even n, a person closes odd n.
+    await run(`insert into activity_log (company_id, actor_type, actor_id, action, entity_type, entity_id, details, created_at)
+      select i.company_id, case when i.n % 2 = 0 then 'agent' else 'user' end, 'actor', 'issue.updated', 'issue', i.id::text,
+        jsonb_build_object('toStatus', 'done'), $2::timestamptz - interval '3 hours'
+      from bench_issues i join issues s on s.id = i.id where i.company_id = $1::uuid and s.status = 'done'`, [company, now.toISOString()]);
+  }
+  await run("analyze");
+}
+
+function captureQuery(db: Db, check: Check, now: Date, limit: number, processStartedAt: Date): Promise<{ query: Query; result: unknown }> {
+  let query: Query | undefined;
+  const dialect = (db as unknown as { dialect: { sqlToQuery(sql: SQL): Query } }).dialect;
+  const captureDb = new Proxy(db, { get(target, key) {
+    if (key !== "execute") return Reflect.get(target, key);
+    return (statement: SQL) => { query = dialect.sqlToQuery(statement); return target.execute(statement); };
+  } });
+  return check(captureDb, now, limit, processStartedAt).then(result => { assert.ok(query, "query captured"); return { query: query!, result }; });
+}
+
+try {
+  database = await startEmbeddedPostgresTestDatabase("comment-wake-benchmark-");
+  db = createDb(database.connectionString, { maxConnections: 1, idleTimeoutSeconds: 0 });
+  await db.$client.unsafe("set client_min_messages = warning");
+  const version = await db.$client.unsafe("select version()");
+  const settings = await db.$client.unsafe("select name, setting from pg_settings where name in ('shared_buffers', 'work_mem', 'jit', 'random_page_cost') order by name");
+  const selectors = process.argv.filter(value => value.startsWith("--fixture=")).map(value => value.slice(10));
+  const results = [];
+  for (const fixture of fixtures.filter(value => !selectors.length || selectors.includes(value.name))) {
+    await db.$client.unsafe("TRUNCATE companies CASCADE");
+    await db.$client.unsafe("drop table if exists bench_agents, bench_issues");
+    const now = new Date(), processStartedAt = new Date(now.getTime() - 2 * 60 * 60_000), limit = 200;
+    const seedStart = performance.now();
+    await seed(db, fixture, now);
+    const seedMs = performance.now() - seedStart;
+    const counts = await db.$client.unsafe(`select (select count(*) from heartbeat_runs) runs, (select count(*) from heartbeat_run_events) events,
+      (select count(*) from issue_comments where created_at >= now() - interval '1 hour') recent_comments,
+      (select pg_size_pretty(pg_total_relation_size('heartbeat_run_events'))) events_size,
+      (select pg_size_pretty(pg_total_relation_size('heartbeat_runs'))) runs_size`);
+    const before = await captureQuery(db, baseline, now, limit, processStartedAt);
+    const after = await captureQuery(db, treatment, now, limit, processStartedAt);
+    assert.deepEqual(after.result, before.result, `${fixture.name}: complete service result differs`);
+    const rows = async (query: Query) => db!.$client.unsafe(query.sql, query.params as never[]).values();
+    const expected = await rows(before.query), actual = await rows(after.query);
+    assert.ok(expected.length > 0, `${fixture.name}: the fixture must report owed comments`);
+    // Order is part of the contract: compare positionally, then as multisets.
+    assert.deepEqual(actual, expected, `${fixture.name}: complete positional SQL output differs`);
+    assert.deepEqual(multiset(actual), multiset(expected));
+    const samples = { before: [] as number[], after: [] as number[] };
+    for (let iteration = 0; iteration < samplesPerArm + 2; iteration++) {
+      for (const arm of iteration % 2 ? ["after", "before"] as const : ["before", "after"] as const) {
+        const start = performance.now();
+        const result = await (arm === "before" ? baseline : treatment)(db, now, limit, processStartedAt);
+        const ms = performance.now() - start;
+        assert.deepEqual(result, before.result);
+        if (iteration >= 2) samples[arm].push(ms);
+      }
+    }
+    const plans: Record<string, unknown> = {};
+    const buffers: Record<string, number> = {};
+    for (const [arm, captured] of [["before", before], ["after", after]] as const) {
+      const [row] = await db.$client.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${captured.query.sql}`, captured.query.params as never[]);
+      plans[arm] = row["QUERY PLAN"];
+      const plan = (row["QUERY PLAN"] as { Plan: Record<string, number> }[])[0]!.Plan;
+      buffers[arm] = Number(plan["Shared Hit Blocks"]) + Number(plan["Shared Read Blocks"]);
+    }
+    const medians = { before: median(samples.before), after: median(samples.after) };
+    // Hash-join plan: without nested loops, PostgreSQL can join the baseline's
+    // OR-bound run lookups only by company, testing each comment against every
+    // run of its company. The treatment's per-comment probes have no such plan.
+    // One sample each, outside the timed interval, with the same result check.
+    const hashPlan: Record<string, number | string> = {};
+    if (!process.argv.includes("--skip-hash-plan")) {
+      await db.$client.unsafe("set enable_nestloop = off");
+      await db.$client.unsafe("set statement_timeout = '300s'");
+      try {
+        for (const [arm, captured] of [["before", before], ["after", after]] as const) {
+          const start = performance.now();
+          try {
+            assert.deepEqual(await rows(captured.query), expected, `${fixture.name}: hash-plan output differs`);
+            hashPlan[arm] = performance.now() - start;
+          } catch (error) {
+            if ((error as { code?: string }).code !== "57014") throw error;
+            hashPlan[arm] = "timeout after 300 s";
+          }
+        }
+      } finally {
+        await db.$client.unsafe("reset enable_nestloop");
+        await db.$client.unsafe("reset statement_timeout");
+      }
+    }
+    results.push({ fixture, counts: counts[0], seedMs, reportedRows: expected.length, queries: { before: before.query.sql, after: after.query.sql },
+      samples, medians, buffers, hashPlan, plans });
+    process.stderr.write(`${fixture.name}: ${JSON.stringify(counts[0])} owed=${expected.length}; `
+      + `median ${medians.before.toFixed(1)} -> ${medians.after.toFixed(1)} ms (${(medians.before / medians.after).toFixed(1)}x); `
+      + `buffers ${buffers.before} -> ${buffers.after}; hash plan ${JSON.stringify(hashPlan)}; samples before=${samples.before.map(v => v.toFixed(0)).join(",")} after=${samples.after.map(v => v.toFixed(0)).join(",")}\n`);
+  }
+  const treatmentSource = await fs.readFile(path.join(root, sourcePath));
+  const output = JSON.stringify({ baselineRevision, treatmentRevision: execFileSync("git", ["rev-parse", "HEAD"], { encoding: "utf8", cwd: root }).trim(),
+    sourcePath, baselineSha256: createHash("sha256").update(baselineSource).digest("hex"),
+    treatmentSha256: createHash("sha256").update(treatmentSource).digest("hex"), version, settings,
+    warmups: 2, timedSamplesPerArm: samplesPerArm, setupAndProofMs: performance.now() - started, results }, null, 2);
+  const out = process.argv.find(value => value.startsWith("--out="))?.slice(6);
+  if (out) await fs.writeFile(out, output); else console.log(output);
+} finally {
+  try { await db?.$client.end({ timeout: 0 }); await database?.cleanup(); await fs.rm(baselineDir, { recursive: true, force: true }); }
+  finally { clearInterval(keepAlive); }
+}
diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts
index e9e92261278..40e50b67da0 100644
--- a/server/src/__tests__/ownership-reconciliation-report.test.ts
+++ b/server/src/__tests__/ownership-reconciliation-report.test.ts
@@ -723,6 +723,64 @@ describe("report-only ownership reconciliation", () => {
       .toEqual([lost.commentId]);
   });
 
+  // The check reads a task's runs through two separate bindings (legacy
+  // context or native column) and reads event payloads only for comments that
+  // the indexed checks keep. Each binding must count alone, and only in the
+  // comment's company.
+  it("binds each task run by its legacy context or its native column in the user-comment wake check", async () => {
+    const otherCompanyId = randomUUID(), otherAgentId = randomUUID();
+    await db.insert(companies).values({ id: otherCompanyId, name: "Other company", issuePrefix: "OTH" });
+    await db.insert(agents).values({ id: otherAgentId, companyId: otherCompanyId, name: "Other", role: "engineer", adapterType: "codex_local" });
+    const task = async (title: string, runs: Array<{ binding: "context" | "native" | "both"; status?: string;
+      startedAt: Date; finishedAt?: Date | null; companyId?: string }>, commentAt: Date) => {
+      const issueId = randomUUID();
+      await db.insert(issues).values({ id: issueId, companyId, title, status: "todo", assigneeAgentId: agentId });
+      const runIds: string[] = [];
+      for (const run of runs) {
+        const runCompanyId = run.companyId ?? companyId;
+        const [row] = await db.insert(heartbeatRuns).values({ companyId: runCompanyId, agentId: runCompanyId === companyId ? agentId : otherAgentId,
+          status: run.status ?? "succeeded", invocationSource: "manual",
+          contextSnapshot: run.binding === "native" ? {} : { issueId },
+          nativeIssueId: run.binding === "context" ? null : issueId,
+          startedAt: run.startedAt, createdAt: run.startedAt, finishedAt: run.finishedAt === undefined ? minutesAgo(3) : run.finishedAt }).returning();
+        runIds.push(row!.id);
+      }
+      const [comment] = await db.insert(issueComments).values({ companyId, issueId, authorUserId: "user-1", body: title,
+        createdAt: commentAt }).returning();
+      return { issueId, runIds, commentId: comment!.id };
+    };
+    const steerEvent = (runId: string, commentId: string, eventType: string, createdAt: Date) => ({ companyId, runId, agentId,
+      seq: 1, eventType, stream: "system", level: "info", sourceEventId: randomUUID(), createdAt,
+      payload: { prpEvent: { itemId: `provider:steer:${commentId}`, payload: { kind: "steering_acknowledgement" } } } });
+    const lostNative = await task("lost, native prior run", [{ binding: "native", startedAt: minutesAgo(10), finishedAt: minutesAgo(9) }], minutesAgo(8));
+    const lostContext = await task("lost, legacy prior run", [{ binding: "context", startedAt: minutesAgo(10), finishedAt: minutesAgo(9) }], minutesAgo(7));
+    await task("no run at all", [], minutesAgo(8));
+    await task("only another company's run", [{ binding: "both", startedAt: minutesAgo(10), finishedAt: minutesAgo(9), companyId: otherCompanyId }], minutesAgo(8));
+    await task("native later run", [{ binding: "context", startedAt: minutesAgo(10), finishedAt: minutesAgo(9) },
+      { binding: "native", startedAt: minutesAgo(5) }], minutesAgo(8));
+    await task("legacy later run", [{ binding: "native", startedAt: minutesAgo(10), finishedAt: minutesAgo(9) },
+      { binding: "context", startedAt: minutesAgo(5) }], minutesAgo(8));
+    await task("native live run", [{ binding: "native", status: "running", startedAt: minutesAgo(10), finishedAt: null }], minutesAgo(8));
+    const nativeSteered = await task("native steered run", [{ binding: "native", startedAt: minutesAgo(10) }], minutesAgo(8));
+    const contextSteered = await task("legacy steered run", [{ binding: "context", startedAt: minutesAgo(10) }], minutesAgo(8));
+    const earlyAck = await task("acknowledged before the comment", [{ binding: "native", startedAt: minutesAgo(10) }], minutesAgo(6));
+    const logOnly = await task("named only in a log event", [{ binding: "context", startedAt: minutesAgo(10) }], minutesAgo(5));
+    const otherRunAck = await task("acknowledged by a run that started after it", [{ binding: "native", startedAt: minutesAgo(10), finishedAt: minutesAgo(9) }], minutesAgo(4));
+    await db.insert(heartbeatRunEvents).values([
+      steerEvent(nativeSteered.runIds[0]!, nativeSteered.commentId, "item.completed", minutesAgo(7)),
+      steerEvent(contextSteered.runIds[0]!, contextSteered.commentId, "item.started", minutesAgo(7)),
+      steerEvent(earlyAck.runIds[0]!, earlyAck.commentId, "item.completed", minutesAgo(7)),
+      steerEvent(logOnly.runIds[0]!, logOnly.commentId, "log", minutesAgo(4)),
+      // The run that names this comment ended before it: not a steer of this comment.
+      steerEvent(otherRunAck.runIds[0]!, otherRunAck.commentId, "item.completed", minutesAgo(3)),
+    ]);
+    const report = await runOwnershipReconciliationPass(db, { state: createOwnershipReportState(), record: false,
+      processStartedAt: minutesAgo(30) });
+    // Oldest first, as the check orders them.
+    expect(report.violations.filter(row => row.reason === "user_comment_without_wake").map(row => row.recordId))
+      .toEqual([lostNative.commentId, lostContext.commentId, earlyAck.commentId, logOnly.commentId, otherRunAck.commentId]);
+  });
+
   it("keeps a backstop deadline for a lifted pause still inside its grace window", async () => {
     const updatedAt = minutesAgo(1);
     await db.insert(agentWakeupRequests).values({ companyId, agentId, source: "automation", status: "deferred_issue_execution",
diff --git a/server/src/services/recovery/ownership-invariant-report.ts b/server/src/services/recovery/ownership-invariant-report.ts
index c00cb3b4517..8b2bae2353c 100644
--- a/server/src/services/recovery/ownership-invariant-report.ts
+++ b/server/src/services/recovery/ownership-invariant-report.ts
@@ -1,4 +1,4 @@
-import { and, asc, eq, inArray, isNotNull, isNull, lt, notExists, or, sql } from "drizzle-orm";
+import { and, asc, eq, inArray, isNotNull, isNull, lt, notExists, or, sql, type SQL } from "drizzle-orm";
 import {
   agentWakeupRequests,
   heartbeatRuns,
@@ -467,36 +467,53 @@ async function checkStaleDroppedComments(db: Db, now: Date, limit: number): Prom
  * finds a user comment on an assigned, non-conversation task with no live run
  * or lock, newer than every run of the task, that no wake request carries.
  * The reconciliation worker saves it as a deferred receipt and promotes it. */
-async function checkUserCommentsWithoutWake(db: Db, now: Date, limit: number, processStartedAt: Date): Promise<{ violations: InvariantViolation[]; nextDueAt: Date | null }> {
+export async function checkUserCommentsWithoutWake(db: Db, now: Date, limit: number, processStartedAt: Date): Promise<{ violations: InvariantViolation[]; nextDueAt: Date | null }> {
   const since = new Date(now.getTime() - INTENT_LOOKBACK_MS).toISOString();
+  // A run belongs to the task by its legacy context or its native column. Each
+  // binding is a separate indexed probe per comment: one OR across both lets
+  // PostgreSQL hash every run of the company and read each run's context.
+  // `offset 0` keeps each probe, and the steer check's event lookup, a
+  // per-row index lookup; without it PostgreSQL can also build a hashed
+  // subplan or a hash join over all runs or all events of the company.
+  const taskRuns = (alias: string, condition: SQL) => {
+    const run = sql.raw(alias);
+    return sql`(exists (select 1 from heartbeat_runs ${run} where ${run}.company_id = c.company_id
+        and ${run}.context_snapshot->>'issueId' = c.issue_id::text and ${condition} offset 0)
+      or exists (select 1 from heartbeat_runs ${run} where ${run}.company_id = c.company_id
+        and ${run}.native_issue_id = c.issue_id and ${condition} offset 0))`;
+  };
+  // Three stages, cheapest first. Most user comments have a wake (stage one)
+  // or a later run (stage two). Only the few comments left reach the steer
+  // check, which reads run event payloads.
   const rows = await db.execute<{ id: string; company_id: string; issue_id: string; created_at: string }>(sql`
-    select c.id, c.company_id, c.issue_id, c.created_at::text as created_at from issue_comments c
-    join issues i on i.id = c.issue_id and i.company_id = c.company_id
-    where c.author_user_id is not null and c.deleted_at is null
-      and c.author_agent_id is null and c.created_by_run_id is null
-      and c.created_at >= ${since}::timestamptz
-      and i.assignee_agent_id is not null and i.conversation_agent_id is null
-      and (i.status in ('todo', 'in_progress')
-        or (i.status = 'done' and ${taskClosedByAgentSql(sql`c.company_id`, sql`c.issue_id`)}))
-      and i.execution_run_id is null
-      and not exists (select 1 from heartbeat_runs steered join heartbeat_run_events steer_event
-          on steer_event.company_id = steered.company_id and steer_event.run_id = steered.id
-        where steered.company_id = c.company_id
-          and (steered.context_snapshot->>'issueId' = c.issue_id::text or steered.native_issue_id = c.issue_id)
-          and coalesce(steered.started_at, steered.created_at) <= c.created_at
-          and (steered.finished_at is null or steered.finished_at >= c.created_at)
+    with unwoken as materialized (
+      select c.id, c.company_id, c.issue_id, c.created_at from issue_comments c
+      join issues i on i.id = c.issue_id and i.company_id = c.company_id
+      where c.author_user_id is not null and c.deleted_at is null
+        and c.author_agent_id is null and c.created_by_run_id is null
+        and c.created_at >= ${since}::timestamptz
+        and i.assignee_agent_id is not null and i.conversation_agent_id is null
+        and (i.status in ('todo', 'in_progress')
+          or (i.status = 'done' and ${taskClosedByAgentSql(sql`c.company_id`, sql`c.issue_id`)}))
+        and i.execution_run_id is null
+        and not exists (select 1 from agent_wakeup_requests w where w.company_id = c.company_id
+          and w.payload->>'issueId' = c.issue_id::text
+          and (w.payload->>'commentId' = c.id::text
+            or coalesce(w.payload#>'{_paperclipWakeContext,wakeCommentIds}', '[]'::jsonb) ? c.id::text))
+    ), candidate as materialized (
+      select c.id, c.company_id, c.issue_id, c.created_at from unwoken c
+      where not ${taskRuns("r", sql`(r.status in ('queued', 'running', 'scheduled_retry')
+          or coalesce(r.started_at, r.created_at) >= c.created_at)`)}
+        and ${taskRuns("prior", sql`true`)}
+    )
+    select c.id, c.company_id, c.issue_id, c.created_at::text as created_at from candidate c
+    where not ${taskRuns("steered", sql`coalesce(steered.started_at, steered.created_at) <= c.created_at
+      and (steered.finished_at is null or steered.finished_at >= c.created_at)
+      and exists (select 1 from heartbeat_run_events steer_event
+        where steer_event.company_id = steered.company_id and steer_event.run_id = steered.id
           and steer_event.event_type in ('item.started', 'item.completed')
           and steer_event.created_at >= c.created_at
-          and strpos(steer_event.payload::text, c.id::text) > 0)
-      and exists (select 1 from heartbeat_runs prior where prior.company_id = c.company_id
-        and (prior.context_snapshot->>'issueId' = c.issue_id::text or prior.native_issue_id = c.issue_id))
-      and not exists (select 1 from heartbeat_runs r where r.company_id = c.company_id
-        and (r.context_snapshot->>'issueId' = c.issue_id::text or r.native_issue_id = c.issue_id)
-        and (r.status in ('queued', 'running', 'scheduled_retry') or coalesce(r.started_at, r.created_at) >= c.created_at))
-      and not exists (select 1 from agent_wakeup_requests w where w.company_id = c.company_id
-        and w.payload->>'issueId' = c.issue_id::text
-        and (w.payload->>'commentId' = c.id::text
-          or coalesce(w.payload#>'{_paperclipWakeContext,wakeCommentIds}', '[]'::jsonb) ? c.id::text))
+          and strpos(steer_event.payload::text, c.id::text) > 0 offset 0)`)}
     order by c.created_at limit ${Math.min(limit, MAX_RECORDS_PER_KIND)}`);
   // A comment inside the grace may still get its normal wake: check it again
   // when the grace ends (a saved deadline), and report only older ones. This