perf(native): narrow backup eligibility to selected runs (#20599)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Server replacement checks whether retained native sessions still
depend on local harness data.
> - This check joins selected task sessions, runs, backup stamps, and
durable backup records.
> - PostgreSQL can start from backup leases and scan an issue's full run
history before applying the selected-run condition.
> - A measured instance spent 9.56 seconds in that selector and exceeded
an 8-second statement limit.
> - This pull request selects the exact current run identities first,
then applies the existing lease and backup checks.
> - The same receipt validation and eligibility decisions remain in
place.

## Linked Issues or Issue Description

**What happened?**

Startup eligibility timed out while selecting native harness backup
obligations. The plan scanned historical runs for each lease before
checking the task session's selected run.

**Expected behavior**

The selector should use the selected run identities without scanning
unrelated run history. It must return the same session, scope, company,
and receipt flag.

**Steps to reproduce**

Use a database with retained native task sessions, many historical runs
on the same issues, and version-two backup stamps. Run startup
eligibility with an 8-second statement timeout. The lease-first plan can
exceed that limit.

**Paperclip version or commit**

The local baseline is `d0c30e69a9cea912db7c5290c5d585fdd63ec228`.

**Deployment mode**

Self-hosted server with PostgreSQL.

## What Changed

- Keep `AS MATERIALIZED` as an intentional join-order fence: build
selected task-session and last-run identities before joining backup
leases.
- Keep only narrow identity columns in that materialized set. Read the
JSON receipt flag through the selected run's primary key after the lease
and current-backup filters, so profiles for unrelated identities do not
widen the intermediate rows.
- Preserve company and agent bindings, exact normalized-session text
comparison, current-backup exclusion, scope deduplication, ordering, and
the 100-row page.
- Add real PostgreSQL regressions for full ordered-row equivalence,
nullable JSON receipt flags, malformed and uppercase session bindings,
cross-company rows, duplicate stamps, page tails, and invalid receipts.
- Existing saved state carries forward under the same validation rules.
No schema or persisted identity changes are required.

## Verification

- The two focused PostgreSQL regressions passed locally. Command: `pnpm
exec vitest run
server/src/services/native-runtime/native-harness-backup-obligations.integration.test.ts
--silent`.
- Worktree preflight passed. The focused regressions (2/2) and `node
server/scripts/typecheck-layers.mjs` passed on final head `c739353b22`
after rebase.
- One bounded read-only pair on PostgreSQL 17.11 returned the same 54
complete ordered rows: baseline 9,564 ms, candidate 298 ms (32.1 times
faster). The plan changed from lease-to-issue-history scans to
selected-run primary-key lookups and a hash join.
- A synthetic PostgreSQL 18.1 fixture already chose an efficient
baseline plan. Across five alternating warm pairs, its median was 104.6
ms before and 109.4 ms after. This neutral control used 22,800 runs,
14,100 leases, 3,173 task sessions, and 5,460 backups; all 54 complete
rows matched. The empty control also matched.
- These measurements cover the selector. A later read-only
assembled-call probe on a newer serving release substituted only this
selector: the selector took 544 ms for 54 rows, while the full
eligibility call took 56.1 seconds and still reported a controller
migration blocker. Other checks remain outside this PR; this PR does not
claim to bring the whole request within its deadline.
- Required CI and fresh review results will be recorded here before
handoff.

## Risks

- Materialization adds a selected-identity pass. Databases that already
choose an efficient plan may see little benefit, as the local neutral
control shows.
- Pagination and receipt validation must stay exact. The regression
crosses the 100-row boundary and checks that an invalid final scope
still blocks eligibility.

## Model Used

OpenAI Codex, GPT-6 family. The exact model ID and context window are
unavailable in this session. The agent used reasoning, code editing,
shell execution, PostgreSQL diagnostics, and GitHub tools.

## 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
- [ ] All Paperclip CI gates are green
- [ ] 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 <noreply@paperclip.ing>

server/src/services/native-runtime/native-harness-backup-migration.ts

@@ -15,15 +15,19 @@ import { hasValidNativeHarnessFilesystemIndependence } from "./native-harness-fi
 export async function hasLocalNativeHarnessObligations(db: Db): Promise<boolean> {
   let afterSession = "", afterScope = "";
   for (;;) {
-    const rows = await db.execute<{ company_id: string; selected_session_id: string; scope_digest: string; has_receipt: boolean }>(sql`select distinct
+    const rows = await db.execute<{ company_id: string; selected_session_id: string; scope_digest: string; has_receipt: boolean }>(sql`with selected_sessions as materialized (
+      select s.company_id, s.id, s.last_run_id, r.native_issue_id, r.native_session_id
+      from agent_task_sessions s
+      join heartbeat_runs r on r.id = s.last_run_id and r.company_id = s.company_id and r.agent_id = s.agent_id
+      where s.adapter_type = 'paperclip_runner'
+    ) select distinct
     s.company_id, s.id::text as selected_session_id, coalesce(e.metadata->'nativeHarnessBackup'->>'sessionScopeSha256', '') as scope_digest,
-    r.runner_profile_json ? 'nativeHarnessFilesystemIndependence' as has_receipt from agent_task_sessions s
-    join heartbeat_runs r on r.id = s.last_run_id and r.company_id = s.company_id and r.agent_id = s.agent_id
-    join environment_leases e on e.company_id = r.company_id and e.issue_id = r.native_issue_id
-      and e.metadata->'nativeHarnessBackup'->>'normalizedSessionId' = r.native_session_id::text
-    where s.adapter_type = 'paperclip_runner'
-      and e.metadata->'nativeHarnessBackup'->>'schema' = 'paperclip.native-harness-backup-stamp.v2'
-      and not exists (select 1 from native_harness_backups b where b.company_id = r.company_id
+    r.runner_profile_json ? 'nativeHarnessFilesystemIndependence' as has_receipt from selected_sessions s
+    join heartbeat_runs r on r.id = s.last_run_id and r.company_id = s.company_id
+    join environment_leases e on e.company_id = s.company_id and e.issue_id = s.native_issue_id
+      and e.metadata->'nativeHarnessBackup'->>'normalizedSessionId' = s.native_session_id::text
+    where e.metadata->'nativeHarnessBackup'->>'schema' = 'paperclip.native-harness-backup-stamp.v2'
+      and not exists (select 1 from native_harness_backups b where b.company_id = s.company_id
         and b.scope_digest = e.metadata->'nativeHarnessBackup'->>'sessionScopeSha256' and b.status = 'current')
       and (s.id::text, coalesce(e.metadata->'nativeHarnessBackup'->>'sessionScopeSha256', '')) > (${afterSession}, ${afterScope})
     order by selected_session_id, scope_digest limit 100`);

server/src/services/native-runtime/native-harness-backup-obligations.integration.test.ts

@@ -0,0 +1,128 @@
+import { randomUUID } from "node:crypto";
+import { sql } from "drizzle-orm";
+import { PgDialect } from "drizzle-orm/pg-core";
+import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
+import { agentTaskSessions, agents, companies, createDb, environmentLeases, heartbeatRuns, issues, nativeHarnessBackups, type Db } from "@paperclipai/db";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js";
+import { hasLocalNativeHarnessObligations } from "./native-harness-backup-migration.js";
+import { hasValidNativeHarnessFilesystemIndependence } from "./native-harness-filesystem-independence.js";
+
+vi.mock("./native-harness-filesystem-independence.js", () => ({ hasValidNativeHarnessFilesystemIndependence: vi.fn() }));
+const support = await getEmbeddedPostgresTestSupport(), dialect = new PgDialect();
+const baseline = dialect.sqlToQuery(sql`select distinct
+  s.company_id, s.id::text as selected_session_id, coalesce(e.metadata->'nativeHarnessBackup'->>'sessionScopeSha256', '') as scope_digest,
+  r.runner_profile_json ? 'nativeHarnessFilesystemIndependence' as has_receipt from agent_task_sessions s
+  join heartbeat_runs r on r.id = s.last_run_id and r.company_id = s.company_id and r.agent_id = s.agent_id
+  join environment_leases e on e.company_id = r.company_id and e.issue_id = r.native_issue_id
+    and e.metadata->'nativeHarnessBackup'->>'normalizedSessionId' = r.native_session_id::text
+  where s.adapter_type = 'paperclip_runner'
+    and e.metadata->'nativeHarnessBackup'->>'schema' = 'paperclip.native-harness-backup-stamp.v2'
+    and not exists (select 1 from native_harness_backups b where b.company_id = r.company_id
+      and b.scope_digest = e.metadata->'nativeHarnessBackup'->>'sessionScopeSha256' and b.status = 'current')
+    and (s.id::text, coalesce(e.metadata->'nativeHarnessBackup'->>'sessionScopeSha256', '')) > (${""}, ${""})
+  order by selected_session_id, scope_digest limit 100`);
+
+describe.skipIf(!support.supported)("native harness backup obligation selection", () => {
+  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>, db: ReturnType<typeof createDb>;
+  beforeAll(async () => {
+    database = await startEmbeddedPostgresTestDatabase("native-obligation-query-");
+    db = createDb(database.connectionString, { maxConnections: 1 });
+  }, 90_000);
+  beforeEach(() => { vi.mocked(hasValidNativeHarnessFilesystemIndependence).mockReset().mockResolvedValue(true); });
+  afterEach(async () => { await db.execute(sql`truncate native_harness_backups, agent_task_sessions, environment_leases, heartbeat_runs, issues, agents, companies cascade`); });
+  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); });
+
+  async function fixture() {
+    const companyId = randomUUID(), foreignCompanyId = randomUUID(), agentId = randomUUID(), otherAgentId = randomUUID();
+    const foreignAgentId = randomUUID(), issueId = randomUUID(), runId = randomUUID();
+    const sessionId = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa";
+    await db.insert(companies).values([{ id: companyId, name: "Selected", issuePrefix: "SEL" },
+      { id: foreignCompanyId, name: "Foreign", issuePrefix: "FOR" }]);
+    await db.insert(agents).values([{ id: agentId, companyId, name: "Selected" }, { id: otherAgentId, companyId, name: "Other" },
+      { id: foreignAgentId, companyId: foreignCompanyId, name: "Foreign" }]);
+    await db.insert(issues).values({ id: issueId, companyId, title: "Retained" });
+    await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issueId, nativeSessionId: sessionId,
+      runnerProfileJson: { nativeHarnessFilesystemIndependence: {} } });
+    const [selected] = await db.insert(agentTaskSessions).values({ companyId, agentId, adapterType: "paperclip_runner",
+      taskKey: randomUUID(), lastRunId: runId }).returning();
+    return { companyId, foreignCompanyId, agentId, otherAgentId, foreignAgentId, issueId, runId, sessionId, selected };
+  }
+  async function stamp(f: Awaited<ReturnType<typeof fixture>>, scope: unknown, extra: Record<string, unknown> = {}, companyId = f.companyId) {
+    await db.insert(environmentLeases).values({ companyId, issueId: f.issueId, status: "expired", metadata: {
+      nativeHarnessBackup: { schema: "paperclip.native-harness-backup-stamp.v2", normalizedSessionId: f.sessionId,
+        sessionScopeSha256: scope, ...extra },
+    } });
+  }
+  async function query() {
+    let captured!: ReturnType<PgDialect["sqlToQuery"]>;
+    await hasLocalNativeHarnessObligations({ execute: async (statement: Parameters<Db["execute"]>[0]) => {
+      captured = dialect.sqlToQuery(statement as Parameters<PgDialect["sqlToQuery"]>[0]); return [];
+    } } as unknown as Db);
+    return captured;
+  }
+  async function pages(statement: ReturnType<PgDialect["sqlToQuery"]>) {
+    const all: unknown[][] = []; let cursor = ["", ""];
+    for (;;) {
+      const rows = await db.$client.unsafe(statement.sql, cursor).values();
+      all.push(...rows);
+      if (rows.length < 100) return all;
+      cursor = [String(rows.at(-1)![1]), String(rows.at(-1)![2])];
+    }
+  }
+
+  it("preserves complete rows, tenant and agent bindings, JSON semantics, deduplication and pagination", async () => {
+    const f = await fixture();
+    await db.insert(agentTaskSessions).values(Array.from({ length: 105 }, () => ({ companyId: f.companyId, agentId: f.agentId,
+      adapterType: "paperclip_runner", taskKey: randomUUID(), lastRunId: f.runId })));
+    await db.insert(agentTaskSessions).values([
+      { companyId: f.companyId, agentId: f.otherAgentId, adapterType: "paperclip_runner", taskKey: randomUUID(), lastRunId: f.runId },
+      { companyId: f.foreignCompanyId, agentId: f.foreignAgentId, adapterType: "paperclip_runner", taskKey: randomUUID(), lastRunId: f.runId },
+      { companyId: f.companyId, agentId: f.agentId, adapterType: "codex_local", taskKey: randomUUID(), lastRunId: f.runId },
+      { companyId: f.companyId, agentId: f.agentId, adapterType: "paperclip_runner", taskKey: randomUUID() },
+    ]);
+    const flags: Array<[Record<string, unknown> | null, boolean | null]> = [[{}, false], [null, null]];
+    const flagSessions: Array<{ id: string; flag: boolean | null }> = [];
+    for (const [profile, flag] of flags) {
+      const [run] = await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, nativeIssueId: f.issueId,
+        nativeSessionId: f.sessionId, runnerProfileJson: profile }).returning();
+      const [selected] = await db.insert(agentTaskSessions).values({ companyId: f.companyId, agentId: f.agentId,
+        adapterType: "paperclip_runner", taskKey: randomUUID(), lastRunId: run.id }).returning();
+      flagSessions.push({ id: selected.id, flag });
+    }
+    for (const scope of ["covered", "foreign-only", "pending-only", null, 7, { retained: true }]) await stamp(f, scope);
+    await stamp(f, "foreign-only");
+    for (const extra of [{ normalizedSessionId: f.sessionId.toUpperCase() }, { normalizedSessionId: "malformed" },
+      { normalizedSessionId: null }, { normalizedSessionId: 7 }, { schema: "old" }]) await stamp(f, "excluded", extra);
+    await stamp(f, "foreign-lease", {}, f.foreignCompanyId);
+    await db.insert(nativeHarnessBackups).values([
+      { companyId: f.companyId, issueId: f.issueId, runId: f.runId, scopeDigest: "covered", status: "current", manifestSha256: "m", reference: {} },
+      { companyId: f.foreignCompanyId, issueId: f.issueId, runId: f.runId, scopeDigest: "foreign-only", status: "current", manifestSha256: "m", reference: {} },
+      { companyId: f.companyId, issueId: f.issueId, runId: f.runId, scopeDigest: "pending-only", status: "pending", manifestSha256: "m", reference: {} },
+      { companyId: f.companyId, issueId: f.issueId, runId: f.runId, scopeDigest: "", status: "current", manifestSha256: "m", reference: {} },
+    ]);
+    const candidate = await query(), before = await pages(baseline), after = await pages(candidate);
+    expect(after).toEqual(before);
+    expect(after).toHaveLength(108 * 5);
+    expect(new Set(after.map(row => row[2]))).toEqual(new Set(["", "7", "{\"retained\": true}", "foreign-only", "pending-only"]));
+    expect(after.every(row => row[0] === f.companyId)).toBe(true);
+    expect(after.filter(row => row[1] === f.selected.id).every(row => row[3] === true)).toBe(true);
+    for (const { id, flag } of flagSessions) expect(after.filter(row => row[1] === id).map(row => row[3])).toEqual(Array(5).fill(flag));
+  });
+
+  it("validates each uncovered scope across pages and still refuses invalid or absent receipts", async () => {
+    const f = await fixture(), scopes = Array.from({ length: 105 }, (_, index) => index.toString().padStart(64, "0"));
+    await db.insert(environmentLeases).values(scopes.map(scope => ({ companyId: f.companyId, issueId: f.issueId,
+      metadata: { nativeHarnessBackup: { schema: "paperclip.native-harness-backup-stamp.v2", normalizedSessionId: f.sessionId, sessionScopeSha256: scope } } })));
+    await expect(hasLocalNativeHarnessObligations(db)).resolves.toBe(false);
+    expect(vi.mocked(hasValidNativeHarnessFilesystemIndependence).mock.calls.map(([, input]) => input.scopeDigest)).toEqual(scopes);
+    expect(vi.mocked(hasValidNativeHarnessFilesystemIndependence).mock.calls.every(([, input]) =>
+      input.companyId === f.companyId && input.selectedSessionId === f.selected.id)).toBe(true);
+    vi.mocked(hasValidNativeHarnessFilesystemIndependence).mockClear().mockImplementation(async (_db, input) => input.scopeDigest !== scopes[104]);
+    await expect(hasLocalNativeHarnessObligations(db)).resolves.toBe(true);
+    expect(hasValidNativeHarnessFilesystemIndependence).toHaveBeenCalledTimes(105);
+    await db.update(heartbeatRuns).set({ runnerProfileJson: {} });
+    vi.mocked(hasValidNativeHarnessFilesystemIndependence).mockClear();
+    await expect(hasLocalNativeHarnessObligations(db)).resolves.toBe(true);
+    expect(hasValidNativeHarnessFilesystemIndependence).not.toHaveBeenCalled();
+  });
+});