perf(server): scope run identity reads to operation needs (#17931)

## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Tool and connection operations capture the current accepted requester. > - Capture locks the task and run, then reconciles pending steering. > - Most callers use identity or task fields from the returned run. > - The default query also transfers saved prompts, runner configuration, results, and logs. > - This pull request gives each production caller an explicit read scope. > - Narrow scopes keep their authority checks and return only selected columns. > - The Muse project path still reads its complete saved execution input. ## Linked Issues or Issue Description **What existing behavior does this improve?** Operation-time run identity capture for native tools, gateway sessions, project and workspace access, Slack, voice, connection management, and connection token admission. **Current behavior** These callers use the default full run read. Large saved execution documents cross the database connection even when the caller only needs the current accepted requester or task binding. **Proposed behavior** Use typed identity, task, broker, and project-execution scopes. Keep task-before-run locks, company and agent predicates, active status validation, and pending-steering reconciliation. Return the complete saved native input only where project admission uses it. **Reason and benefit** A synthetic PostgreSQL 17.11 fixture matches a wide run with a 142 KB context and a 77 KB runner profile. Its locked run DataRow falls from 225,101 bytes to 190–605 bytes for authority reads. The project scope returns 66,004 bytes because it must retain a complete 64 KB execution input. These are local DataRow measurements, not a forecast of production billing or latency. **Breaking changes** None to public APIs, saved data, or authority policy. Internal narrow return types omit columns that were not selected. The default full compatibility read and the existing GitHub-credential and authentication scopes keep their current contracts. **Additional context** The public issue and PR duplicate searches found no matching open change. This is one independent query optimization. It does not require other performance PRs. ## What Changed - Add typed scopes for accepted identity metadata, task authorization, token attribution, and saved project execution input. - Assign explicit scopes at all production `captureRunIdentity` call sites. - Keep the full native execution input for Muse project admission while omitting unrelated profile, context, result, and log data. - Narrow the broker's initial read and the existing Slack and connection-management historical reads to the same required fields. - Extend steering and acknowledgement regressions to all new scopes. Add real PostgreSQL projection, malformed JSON, tenant and agent, session-generation, and saved-input coverage. - Add a service-generated SQL benchmark with exact raw DataRow accounting, retained-field equality, plans, and alternating transaction timings. - Document the scope contracts and reproducible benchmark. ## Verification - Baseline: `623f1dc289a14a699ebbe3f09b94179d38fc2ae2`. - `vitest run server/src/__tests__/run-identity-projections.test.ts server/src/__tests__/run-identity.test.ts`: 54 tests passed on an owned disposable PostgreSQL 17.11 database. - The tests retain task-before-run and audit foreign-key lock behavior. Every scope rejects uncertain steering, accepts its durable acknowledgement, and preserves requester replay ordering. - A full-row mutation makes the narrow identity regression fail. The restored treatment passes. - `pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-run-identity-projections.ts --baseline-ref=623f1dc289a14a699ebbe3f09b94179d38fc2ae2` captures the real locked SQL for both versions. - Every retained column and selected JSON field matches the baseline. Identity contexts match. The complete stored run remains unchanged. Capture keeps the same five queries. | Scope | Before DataRow bytes | After DataRow bytes | Reduction | | --- | ---: | ---: | ---: | | Identity | 225,101 | 190 | 99.92% | | Task authorization | 225,101 | 410 | 99.82% | | Token broker | 225,101 | 605 | 99.73% | | Project execution input | 225,101 | 66,004 | 70.68% | - DataRow bytes include row framing and raw text-format column values. They exclude response metadata, TLS, and TCP. - Two warmups and seven alternating samples measure the complete capture transaction. Several timing ranges overlap. The primary claim is byte reduction; production latency is not verified. - Initial fixture runs exposed an empty external schema and an incomplete synthetic conversation binding. The test setup now applies migrations and supplies the required user/state fields. The final 54 tests passed. - `pnpm -r typecheck` and `pnpm build` passed locally. - Four existing skill-setup tests passed. Four assembled connection-management/native-tool consumer cases passed with real JWT/API-key authentication. These include per-request identity freshness and task-scoped activity authority. - A CI-only stale mock assertion expected two capture arguments; it now verifies the explicit identity scope. A repository-wide capture mock audit found no other stale argument assertions. - Review follow-up: the projection suite uses the existing embedded-support guard while always enabling an explicitly supplied external database. The audit-lock regression merges URL query parameters and retains existing PostgreSQL options. - `vitest run server/src/__tests__/run-identity-projections.test.ts server/src/__tests__/run-identity.test.ts server/src/__tests__/company-skills-routes.test.ts server/src/__tests__/project-skill-setup-identity.test.ts`: 123 tests passed. The external PostgreSQL URL included both `application_name` and `options`; lock tests verified `lock_timeout=1s` and retained `statement_timeout=5s`. - Full local tests are not run because the host's embedded PostgreSQL setup is unavailable. Greptile reviewed `06e1dd53a3fb24df74d5fbfb2ad397f291f1e30f` at 5/5 with zero unresolved threads. The full final-head CI matrix passes. All 58 checks are terminal: 55 success, one neutral, and two optional Storybook skips. - No migration, lockfile, workflow, new polling, or cache change. ## Risks A narrow read must retain each caller's actual fields. The caller audit includes Slack's wake-request fallback, conversation generation, legacy broker attribution, and the Muse parser's complete native input. JSON selection retains value types, including malformed legacy values. The saved-input scope preserves missing-key behavior. The default full read still returns the complete stored row for compatibility. Locks and live authority checks are unchanged. ## Model Used OpenAI GPT-6, accessed through Codex. The exact runtime model ID and context-window size were not exposed in this session. Used repository inspection, code execution, and PostgreSQL validation. ## 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 <noreply@paperclip.ing>
2026-10-28 12:04:41 +01:00 · 2026-10-27 16:31:21 -05:00
1 parent 56bc470ab2
commit c3a9a8e92a
@@ -0,0 +1,35 @@
+# Run identity reads
+
+`captureRunIdentity` locks the task before the run. It validates company, agent,
+and active status. Pending steering still requires its durable acknowledgement.
+The accepted requester is read for each operation. No scope adds a cache or skips
+these checks.
+
+Callers choose a typed read scope:
+
+| Scope | Retained run data | Consumers |
+| --- | --- | --- |
+| `identity` | Run and accepted requester identifiers, status | Gateway, native connection tools, connection intents, project skill setup |
+| `task_authorization` | Identity, task bindings, wake request, conversation generation, skill-test marker | Projects, workspace repositories, Slack, voice, connection management |
+| `token_broker` | Identity, invocation source, task/project/routine and requester attribution | Connection token broker |
+| `project_execution` | Task authorization plus runtime mode and the complete saved native execution input | Muse project admission |
+
+Narrow return types omit unselected columns. JSON projections preserve the types
+of retained values. The default full read remains available for compatibility and
+explicit inspection. Existing GitHub-credential and request-authentication scopes
+keep their current contracts. Execution preparation and recovery retain their
+separate complete-input reads.
+
+The synthetic real-PostgreSQL benchmark captures the actual locked query and
+compares every retained column, context and identity result. It keeps the same
+five-query reconciliation transaction, checks the unchanged stored run, records
+plans and alternating timings, and measures raw PostgreSQL DataRow bytes:
+
+```sh
+pnpm --filter @paperclipai/server exec tsx ../scripts/benchmark-run-identity-projections.ts --baseline-ref=623f1dc289a14a699ebbe3f09b94179d38fc2ae2
+```
+
+The default uses disposable embedded PostgreSQL. `PAPERCLIP_TEST_DATABASE_URL`
+can select a disposable local database whose name ends in `_test`. DataRow bytes
+exclude response metadata, TLS and TCP. Byte reduction does not establish a
+production latency or billing change.
@@ -0,0 +1,120 @@
+import assert from "node:assert/strict";
+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 { applyPendingMigrations, createDb, startEmbeddedPostgresTestDatabase } from "../packages/db/src/index.js";
+import { captureRunIdentity } from "../server/src/services/run-identity.js";
+import { captureIdentityQueries, seedIdentityProjection, type IdentityQuery } from "../server/src/__tests__/helpers/run-identity-projection-fixture.js";
+
+const root = fileURLToPath(new URL("../", import.meta.url));
+const git = (args: string[]) => execFileSync("git", args, { cwd: root, encoding: "utf8" });
+const baselineRef = process.argv.find(arg => arg.startsWith("--baseline-ref="))?.slice(15);
+assert.ok(baselineRef, "Pass an explicit pre-fix --baseline-ref=<commit>");
+const baselineRevision = git(["rev-parse", "--verify", `${baselineRef}^{commit}`]).trim();
+const modulePath = "server/src/services/run-identity.ts";
+const unchangedHelpers = ["run-task-binding.ts", "task-retry.ts", "issue-queued-comment-queue.ts"];
+for (const name of unchangedHelpers) {
+  const helper = `server/src/services/${name}`;
+  assert.equal(git(["show", `${baselineRevision}:${helper}`]), await fs.readFile(path.join(root, helper), "utf8"),
+    `Baseline helper ${name} changed; isolate it before benchmarking`);
+}
+const keepAlive = setInterval(() => {}, 1_000);
+let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
+let db: ReturnType<typeof createDb> | undefined;
+let baselineDir: string | undefined;
+try {
+  const oldSource = git(["show", `${baselineRevision}:${modulePath}`]);
+  const moduleUrl = new URL(`../${modulePath}`, import.meta.url);
+  baselineDir = await fs.mkdtemp(path.join(root, "server/scripts/.identity-projection-baseline-"));
+  const baselineFile = path.join(baselineDir, "run-identity.ts");
+  await fs.writeFile(baselineFile, oldSource.replace(/from "(\.{1,2}\/[^\"]+)"/g,
+    (_match, specifier: string) => `from ${JSON.stringify(fileURLToPath(new URL(specifier, moduleUrl)))}`));
+  const baseline = await import(pathToFileURL(baselineFile).href) as typeof import("../server/src/services/run-identity.js");
+  const externalUrl = process.env.PAPERCLIP_TEST_DATABASE_URL;
+  if (externalUrl) {
+    const parsed = new URL(externalUrl);
+    assert.ok(["127.0.0.1", "localhost"].includes(parsed.hostname) && /_test$/.test(parsed.pathname),
+      "External benchmark database must be a disposable local *_test database");
+    await applyPendingMigrations(externalUrl);
+    db = createDb(externalUrl);
+  } else {
+    temporary = await startEmbeddedPostgresTestDatabase("identity-projection-benchmark-");
+    db = createDb(temporary.connectionString);
+  }
+  const f = await seedIdentityProjection(db);
+  await db.$client.unsafe("ANALYZE");
+  const beforeQueries: IdentityQuery[] = [];
+  const before = await baseline.captureRunIdentity(captureIdentityQueries(db, beforeQueries), f);
+  const findRunQuery = (queries: IdentityQuery[]) => {
+    const query = queries.find(query => query.sql.includes('from "heartbeat_runs"') && query.sql.includes("for no key update"));
+    assert.ok(query, "Capture the actual locked identity query");
+    return query;
+  };
+  const beforeQuery = findRunQuery(beforeQueries);
+  const dataRowBytes = async (query: IdentityQuery) => {
+    const rows = await db!.$client.unsafe(query.sql, query.params as never[]).raw();
+    return rows.reduce((total, row) => total + 7 + 4 * row.length + row.reduce((bytes, value) => bytes + (value?.byteLength ?? 0), 0), 0);
+  };
+  const baselineBytes = await dataRowBytes(beforeQuery);
+  async function plan(query: IdentityQuery) {
+    const [result] = await db!.$client.unsafe(`EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) ${query.sql}`, query.params as never[]);
+    return result!["QUERY PLAN"];
+  }
+  const results = [];
+  for (const readScope of ["identity", "task_authorization", "token_broker", "project_execution"] as const) {
+    const afterQueries: IdentityQuery[] = [];
+    const after = await captureRunIdentity(captureIdentityQueries(db, afterQueries), f, { readScope });
+    assert.deepEqual(after.context, before.context);
+    for (const [key, value] of Object.entries(after.run)) {
+      if (key === "runnerProfileJson") assert.deepEqual(value, { nativeExecutionInput: before.run.runnerProfileJson?.nativeExecutionInput });
+      else if (key !== "contextSnapshot") assert.deepEqual(value, before.run[key as keyof typeof before.run], `Retained ${key}`);
+      else {
+        const snapshot = value as Record<string, unknown>;
+        for (const [name, selected] of Object.entries(snapshot)) {
+          if (name !== "paperclipIssue") assert.deepEqual(selected, before.run.contextSnapshot?.[name] ?? null, `Retained context ${name}`);
+          else for (const [nested, item] of Object.entries(selected as Record<string, unknown>))
+            assert.deepEqual(item, (before.run.contextSnapshot?.paperclipIssue as Record<string, unknown>)?.[nested] ?? null, `Retained nested ${nested}`);
+        }
+      }
+    }
+    const afterQuery = findRunQuery(afterQueries), afterBytes = await dataRowBytes(afterQuery);
+    assert.ok(afterBytes < baselineBytes * (readScope === "project_execution" ? 0.5 : 0.05),
+      "Fixture acceptance: remove at least 95% of authority bytes or 50% while preserving complete execution input");
+    assert.equal(afterQueries.length, beforeQueries.length, "No queries added or removed from identity reconciliation");
+    const samples = { baseline: [] as number[], treatment: [] as number[] };
+    for (let n = 0; n < 9; n++) {
+      for (const version of n % 2 ? ["treatment", "baseline"] as const : ["baseline", "treatment"] as const) {
+        const start = performance.now();
+        if (version === "baseline") await baseline.captureRunIdentity(db, f);
+        else await captureRunIdentity(db, f, { readScope });
+        if (n >= 2) samples[version].push(performance.now() - start);
+      }
+    }
+    const stats = (values: number[]) => {
+      const sorted = [...values].sort((a, b) => a - b);
+      return { samplesMs: values, medianMs: sorted[Math.floor(sorted.length / 2)], minMs: sorted[0], maxMs: sorted.at(-1) };
+    };
+    results.push({ readScope, baselineBytes, treatmentBytes: afterBytes, reductionPercent: 100 * (1 - afterBytes / baselineBytes),
+      retainedColumnsAndContextEqual: true, identityContextEqual: true, queryCount: afterQueries.length,
+      generatedSql: { baseline: beforeQuery, treatment: afterQuery },
+      timings: { scope: "complete captureRunIdentity service transaction", warmups: 2, samples: 7, order: "alternating",
+        baseline: stats(samples.baseline), treatment: stats(samples.treatment) },
+      plans: { baseline: await plan(beforeQuery), treatment: await plan(afterQuery) } });
+  }
+  assert.deepEqual((await captureRunIdentity(db, f)).run, before.run, "Complete stored run unchanged");
+  console.log(JSON.stringify({ baselineRevision, treatmentRevision: git(["rev-parse", "HEAD"]).trim(),
+    treatmentDiff: git(["diff", "HEAD", "--", "server/src"]), unchangedHelpers,
+    postgresVersion: (await db.$client.unsafe("SELECT version()"))[0]!.version,
+    settings: await db.$client.unsafe("SELECT name, setting FROM pg_settings WHERE name IN ('block_size','shared_buffers','work_mem','jit') ORDER BY name"),
+    fixture: { runs: 1, synthetic: true, contextBytes: Buffer.byteLength(JSON.stringify(f.payloads.contextSnapshot)),
+      profileBytes: Buffer.byteLength(JSON.stringify(f.payloads.runnerProfileJson)) },
+    measurement: "PostgreSQL DataRow messages from raw text-format buffers; excludes metadata, TLS and TCP", results,
+  }, null, 2));
+} finally {
+  try {
+    await db?.$client.end({ timeout: 0 });
+    await temporary?.cleanup();
+    if (baselineDir) await fs.rm(baselineDir, { recursive: true, force: true });
+  } finally { clearInterval(keepAlive); }
+}
@@ -1264,7 +1264,7 @@ describe("company skill mutation permissions", () => {
    const response = await request(createApp(agent)).post("/api/companies/company-1/skills/import")
      .send({ source: "https://github.com/acme/public-skills" });
    expect(response.status, JSON.stringify(response.body)).toBe(201);
-    expect(mockCaptureRunIdentity).toHaveBeenCalledWith(expect.anything(), { companyId: "company-1", agentId: agent.agentId, runId: "run-1" });
+    expect(mockCaptureRunIdentity).toHaveBeenCalledWith(expect.anything(), { companyId: "company-1", agentId: agent.agentId, runId: "run-1" }, { readScope: "identity" });
    expect(mockSkillSourceService.importFromUrl).toHaveBeenCalledWith("company-1", expect.anything(),
      expect.objectContaining({ setup: { userId: "validated-owner", agentId: agent.agentId, localTrusted: false } }));
  });
@@ -0,0 +1,41 @@
+import { randomBytes, randomUUID } from "node:crypto";
+import { agents, companies, heartbeatRuns, issueComments, issues, type Db } from "@paperclipai/db";
+import { initializeRunIdentity } from "../../services/run-identity.js";
+
+export async function seedIdentityProjection(db: Db) {
+  const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(), messageId = randomUUID();
+  await db.insert(companies).values({ id: companyId, name: "Synthetic identity", issuePrefix: companyId });
+  await db.insert(agents).values({ id: agentId, companyId, name: "Identity reader", adapterType: "process" });
+  await db.insert(issues).values({ id: issueId, companyId, title: "Identity task", assigneeAgentId: agentId,
+    conversationAgentId: agentId, conversationUserId: "operator", conversationState: "active", conversationSessionGeneration: 3, workMode: "standard" });
+  const payloads = { contextSnapshot: { issueId, conversationSessionGeneration: 3, paperclipSkillTest: false,
+    projectId: randomUUID(), routineId: randomUUID(), paperclipIssue: { id: issueId, projectId: randomUUID(),
+      responsibleUserId: "nested-user", document: randomBytes(8_192).toString("base64") },
+    savedInput: randomBytes(96 * 1_024).toString("base64") },
+    runnerProfileJson: { nativeExecutionInput: { savedInput: randomBytes(48 * 1_024).toString("base64") },
+      unrelatedProfileDocument: randomBytes(8_192).toString("base64") },
+    resultJson: { retained: randomBytes(4_096).toString("base64") }, stdoutExcerpt: "retained stdout", stderrExcerpt: "retained stderr" };
+  await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", nativeIssueId: issueId,
+    runtimeMode: "native", invocationSource: "automation", ...payloads });
+  await db.insert(issueComments).values({ id: messageId, companyId, issueId, authorUserId: "next-user", body: "New request" });
+  const context = await initializeRunIdentity(db, { companyId, runId, issueId, responsibleUserId: "operator", cause: "instruction" });
+  return { companyId, agentId, issueId, runId, messageId, context, payloads };
+}
+
+export type IdentityQuery = { sql: string; params: unknown[] };
+export function captureIdentityQueries(db: Db, queries: IdentityQuery[]): Db {
+  return new Proxy(db, { get(target, key) {
+    if (key === "transaction") return (action: (tx: Db) => Promise<unknown>) =>
+      target.transaction(tx => action(captureIdentityQueries(tx as unknown as Db, queries)));
+    if (key !== "select") return Reflect.get(target, key);
+    return (...fields: Parameters<Db["select"]>) => {
+      const builder = target.select(...fields), from = builder.from.bind(builder);
+      builder.from = ((...args: Parameters<typeof from>) => {
+        const selection = from(...args), execute = selection.execute.bind(selection);
+        selection.execute = async () => { queries.push(selection.toSQL()); return execute(); };
+        return selection;
+      }) as typeof builder.from;
+      return builder;
+    };
+  }});
+}
@@ -13,7 +13,7 @@ it('keeps responsibility without searching personal GitHub grants when the run h
 it('pins only the connection and grant already authorized for the active run', async () => {
  capture.mockResolvedValue({ run: { responsibleUserId: 'person' }, context: { github: { status: 'available', connectionId: 'selected', grantId: 'grant' } } });
  expect(await projectSkillSetupIdentity(db, 'company', actor)).toMatchObject({ refreshUserId: 'person', connectionId: 'selected', refreshGrantId: 'grant' });
-  expect(capture).toHaveBeenCalledWith(db, { companyId: 'company', agentId: 'agent', runId: 'run' });
+  expect(capture).toHaveBeenCalledWith(db, { companyId: 'company', agentId: 'agent', runId: 'run' }, { readScope: 'identity' });
 });
 it('does not reuse an unavailable run credential', async () => {
  capture.mockResolvedValue({ run: { responsibleUserId: 'person' }, context: { github: { status: 'unavailable', connectionId: 'old', grantId: 'old' } } });
@@ -0,0 +1,116 @@
+import { randomUUID } from "node:crypto";
+import { eq } from "drizzle-orm";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+import { applyPendingMigrations, createDb, heartbeatRuns, issues, type Db } from "@paperclipai/db";
+import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
+import { captureIdentityQueries, seedIdentityProjection, type IdentityQuery } from "./helpers/run-identity-projection-fixture.js";
+import { captureRunIdentity } from "../services/run-identity.js";
+import { projectToolContext } from "../services/project-tool-context.js";
+import { boundRunIssueId } from "../services/run-task-binding.js";
+
+const externalDatabaseUrl = process.env.PAPERCLIP_TEST_DATABASE_URL;
+const support = externalDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
+const scopes = ["identity", "task_authorization", "token_broker"] as const;
+(support.supported ? describe : describe.skip)("operation identity read projections", () => {
+  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
+  let db: Db;
+  beforeAll(async () => {
+    if (externalDatabaseUrl) {
+      await applyPendingMigrations(externalDatabaseUrl);
+      db = createDb(externalDatabaseUrl);
+    } else {
+      temporary = await startEmbeddedPostgresTestDatabase("identity-projection-");
+      db = createDb(temporary.connectionString);
+    }
+  }, 90_000);
+  afterAll(async () => { await db?.$client.end({ timeout: 0 }); await temporary?.cleanup(); });
+
+  it.each(scopes)("bounds %s SQL results and preserves the stored execution", async readScope => {
+    const f = await seedIdentityProjection(db), queries: IdentityQuery[] = [];
+    const full = await captureRunIdentity(db, f);
+    const narrow = await captureRunIdentity(captureIdentityQueries(db, queries), f, { readScope });
+    expect(narrow.context).toEqual(full.context);
+    expect(narrow.run).toMatchObject({ id: f.runId, companyId: f.companyId, agentId: f.agentId,
+      status: "running", responsibleUserId: "operator", activeIdentityContextId: full.run.activeIdentityContextId });
+    expect(narrow.run).not.toHaveProperty("runnerProfileJson");
+    expect(narrow.run).not.toHaveProperty("resultJson");
+    const query = queries.find(query => query.sql.includes('from "heartbeat_runs"') && query.sql.includes("for no key update"))!;
+    const rows = await db.$client.unsafe(query.sql, query.params as never[]).raw();
+    expect(rows).toHaveLength(1);
+    expect(rows[0]!.reduce((bytes, value) => bytes + (value?.byteLength ?? 0), 0)).toBeLessThan(2_048);
+    expect((await captureRunIdentity(db, f)).run).toEqual(full.run);
+  });
+
+  it("preserves task binding types and legacy non-object contexts", async () => {
+    const f = await seedIdentityProjection(db);
+    for (const snapshot of [null, false, 42, "text", [], ["value"], {}, ...[null, false, 42, " 3 ", [], {}, { nested: true }]
+      .map(value => ({ issueId: f.issueId, conversationSessionGeneration: value, paperclipSkillTest: value }))]) {
+      await db.$client.unsafe("UPDATE heartbeat_runs SET context_snapshot = $1::jsonb WHERE id = $2", [JSON.stringify(snapshot), f.runId]);
+      const full = await captureRunIdentity(db, f), narrow = await captureRunIdentity(db, f, { readScope: "task_authorization" });
+      expect(boundRunIssueId(narrow.run)).toBe(boundRunIssueId(full.run));
+      expect(Number(narrow.run.contextSnapshot?.conversationSessionGeneration ?? 0)).toBe(Number(full.run.contextSnapshot?.conversationSessionGeneration ?? 0));
+      expect(Boolean(narrow.run.contextSnapshot?.paperclipSkillTest)).toBe(Boolean(full.run.contextSnapshot?.paperclipSkillTest));
+    }
+    for (const issueId of [true, 123, [], {}, "malformed"]) {
+      await db.$client.unsafe("UPDATE heartbeat_runs SET context_snapshot = $1::jsonb WHERE id = $2", [JSON.stringify({ issueId }), f.runId]);
+      await expect(captureRunIdentity(db, f)).rejects.toThrow("task identity is invalid");
+      await expect(captureRunIdentity(db, f, { readScope: "task_authorization" })).rejects.toThrow("task identity is invalid");
+    }
+  });
+
+  it("retains broker fields and nested attribution without document payloads", async () => {
+    const f = await seedIdentityProjection(db);
+    for (const value of [null, true, 42, " padded ", [], {}, { retained: true }]) {
+      const snapshot = { issueId: f.issueId, projectId: value, routineId: value, responsibleUserId: value,
+        responsible_user_id: value, paperclipIssue: { id: value, projectId: value, responsibleUserId: value,
+          responsible_user_id: value, savedDocument: f.payloads.contextSnapshot.savedInput } };
+      await db.update(heartbeatRuns).set({ contextSnapshot: snapshot }).where(eq(heartbeatRuns.id, f.runId));
+      const narrow = await captureRunIdentity(db, f, { readScope: "token_broker" });
+      expect(narrow.run.invocationSource).toBe("automation");
+      expect(narrow.run.contextSnapshot).toEqual({ ...snapshot, paperclipIssue: { id: value,
+        projectId: value, responsibleUserId: value, responsible_user_id: value } });
+    }
+    for (const paperclipIssue of [null, true, 42, "legacy", []]) {
+      await db.update(heartbeatRuns).set({ contextSnapshot: { paperclipIssue } }).where(eq(heartbeatRuns.id, f.runId));
+      expect((await captureRunIdentity(db, f, { readScope: "token_broker" })).run.contextSnapshot?.paperclipIssue).toBeNull();
+    }
+  });
+
+  it.each(scopes.flatMap(readScope => (["company", "agent", "run", "finished"] as const).map(fault => ({ readScope, fault }))))(
+    "denies $fault mismatches for $readScope", async ({ readScope, fault }) => {
+      const f = await seedIdentityProjection(db);
+      if (fault === "finished") await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, f.runId));
+      await expect(captureRunIdentity(db, { ...f, ...(fault === "company" ? { companyId: randomUUID() } :
+        fault === "agent" ? { agentId: randomUUID() } : fault === "run" ? { runId: randomUUID() } : {}) }, { readScope })).rejects.toThrow("active run");
+    },
+  );
+
+  it("preserves every saved native input shape without unrelated profile fields", async () => {
+    const f = await seedIdentityProjection(db);
+    for (const nativeExecutionInput of [null, true, 42, "legacy", [], {}, { opaque: { retained: true } }]) {
+      await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeExecutionInput, unrelated: "drop" } }).where(eq(heartbeatRuns.id, f.runId));
+      expect((await captureRunIdentity(db, f, { readScope: "project_execution" })).run.runnerProfileJson).toEqual({ nativeExecutionInput });
+    }
+    await db.update(heartbeatRuns).set({ runnerProfileJson: { unrelated: "drop" } }).where(eq(heartbeatRuns.id, f.runId));
+    expect((await captureRunIdentity(db, f, { readScope: "project_execution" })).run.runnerProfileJson).toEqual({});
+    for (const profile of [null, true, 42, "legacy", []]) {
+      await db.$client.unsafe("UPDATE heartbeat_runs SET runner_profile_json = $1::jsonb WHERE id = $2", [JSON.stringify(profile), f.runId]);
+      expect((await captureRunIdentity(db, f, { readScope: "project_execution" })).run.runnerProfileJson).toBeNull();
+    }
+  });
+
+  it("keeps the complete execution input for the project admission consumer", async () => {
+    const f = await seedIdentityProjection(db);
+    const actor = { type: "agent" as const, source: "agent_jwt" as const, companyId: f.companyId, agentId: f.agentId, runId: f.runId };
+    const narrow = await projectToolContext(db, actor);
+    expect(narrow.run).not.toHaveProperty("runnerProfileJson");
+    const full = await projectToolContext(db, actor, true, "Project", { includeExecutionInput: true });
+    expect(full.run.runnerProfileJson).toEqual({ nativeExecutionInput: f.payloads.runnerProfileJson.nativeExecutionInput });
+    expect(full.run.contextSnapshot).toEqual(narrow.run.contextSnapshot);
+    expect(full.run).not.toHaveProperty("resultJson");
+    expect(full.run.contextSnapshot).not.toHaveProperty("savedInput");
+    await db.update(issues).set({ conversationSessionGeneration: 4 }).where(eq(issues.id, f.issueId));
+    await expect(projectToolContext(db, actor)).rejects.toThrow("session has changed");
+    await expect(projectToolContext(db, actor, true, "Project", { includeExecutionInput: true })).rejects.toThrow("session has changed");
+  });
+});
@@ -1,19 +1,23 @@
 import { randomUUID } from "node:crypto";
 import { eq, sql } from "drizzle-orm";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
-import { agentWakeupRequests, agents, companies, createDb, heartbeatRuns, heartbeatRunEvents, issueComments, issueThreadInteractions, issues, secretAccessEvents } from "@paperclipai/db";
+import { agentWakeupRequests, agents, companies, applyPendingMigrations, createDb, closeRegisteredClients, heartbeatRuns, heartbeatRunEvents, issueComments, issueThreadInteractions, issues, secretAccessEvents } from "@paperclipai/db";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
 import { acceptSteeredIdentity, captureRunIdentity, initializeRunIdentity, listRunIdentityContexts, rejectSteeredIdentity, reserveSteeredIdentity } from "../services/run-identity.js";

-const support = await getEmbeddedPostgresTestSupport();
+const externalDatabaseUrl = process.env.PAPERCLIP_TEST_DATABASE_URL;
+const support = externalDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
 (support.supported ? describe : describe.skip)("durable execution identity", () => {
  let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
  let db: ReturnType<typeof createDb>;
  beforeAll(async () => {
-    database = await startEmbeddedPostgresTestDatabase("paperclip-run-identity-");
-    db = createDb(database.connectionString);
+    if (externalDatabaseUrl) { await applyPendingMigrations(externalDatabaseUrl); db = createDb(externalDatabaseUrl); }
+    else {
+      database = await startEmbeddedPostgresTestDatabase("paperclip-run-identity-");
+      db = createDb(database.connectionString);
+    }
  }, 30_000);
-  afterAll(async () => { await database?.cleanup(); }, 60_000);
+  afterAll(async () => { if (externalDatabaseUrl) await closeRegisteredClients(externalDatabaseUrl); await database?.cleanup(); }, 60_000);

  async function seed() {
    const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID();
@@ -123,7 +127,7 @@ const support = await getEmbeddedPostgresTestSupport();
    },
  );

-  it.each([undefined, "github_credentials", "request_authentication"] as const)("holds acquisition during uncertain steering and never rewinds on replay with %s", async (readScope) => {
+  it.each([undefined, "github_credentials", "request_authentication", "identity", "task_authorization", "token_broker", "project_execution"] as const)("holds acquisition during uncertain steering and never rewinds on replay with %s", async (readScope) => {
    const options = readScope ? { readScope } : undefined;
    const input = await seed();
    await initializeRunIdentity(db, { ...input, messageIds: [input.messageIds[0]], responsibleUserId: "A", cause: "instruction" });
@@ -176,7 +180,7 @@ const support = await getEmbeddedPostgresTestSupport();
    expect(captured.run.responsibleUserId).toBeNull();
    expect(captured.context?.parentContextId).toBe(origin.id);
  });
-  it.each([undefined, "github_credentials", "request_authentication"] as const)("recovers an uncertain native acknowledgement with %s", async (readScope) => {
+  it.each([undefined, "github_credentials", "request_authentication", "identity", "task_authorization", "token_broker", "project_execution"] as const)("recovers an uncertain native acknowledgement with %s", async (readScope) => {
    const options = readScope ? { readScope } : undefined;
    const input = await seed();
    await initializeRunIdentity(db, { ...input, messageIds: [], responsibleUserId: "A", cause: "instruction" });
@@ -260,9 +264,9 @@ const support = await getEmbeddedPostgresTestSupport();
    await expect(initialization).resolves.toMatchObject({ responsibleUserId: "A" });
  });

-  it.each(["initialize", "capture"] as const)(
-    "can %s identity while an audit append holds foreign-key locks",
-    async (operation) => {
+  it.each((["initialize", "capture"] as const).flatMap(operation => [false, true].map(withQuery => ({ operation, withQuery }))))(
+    "can $operation identity while an audit append holds foreign-key locks (query: $withQuery)",
+    async ({ operation, withQuery }) => {
      const input = await seed();
      if (operation === "capture") {
        await initializeRunIdentity(db, { ...input, messageIds: [], responsibleUserId: "A", cause: "instruction" });
@@ -270,11 +274,22 @@ const support = await getEmbeddedPostgresTestSupport();
      // A real append holds KEY SHARE on both parent rows until it commits.
      // Bound the other connection's wait so a conflicting lock fails this
      // regression instead of leaving both transactions waiting for each other.
-      const identityDb = createDb(`${database.connectionString}?options=-c%20lock_timeout%3D1000`, {
+      const identityUrl = new URL(externalDatabaseUrl ?? database.connectionString);
+      if (withQuery) {
+        identityUrl.searchParams.set("application_name", "identity-query-test");
+        identityUrl.searchParams.set("options", [identityUrl.searchParams.get("options"), "-c statement_timeout=5000"].filter(Boolean).join(" "));
+      }
+      identityUrl.searchParams.set("options", [identityUrl.searchParams.get("options"), "-c lock_timeout=1000"].filter(Boolean).join(" "));
+      const identityDb = createDb(identityUrl.toString(), {
        maxConnections: 1,
      });
      const [settings] = await identityDb.execute(sql`show lock_timeout`);
      expect(settings?.lock_timeout).toBe("1s");
+      if (withQuery) {
+        expect(identityUrl.searchParams.get("application_name")).toBe("identity-query-test");
+        const [timeouts] = await identityDb.execute(sql`show statement_timeout`);
+        expect(timeouts?.statement_timeout).toBe("5s");
+      }
      await db.transaction(async (audit) => {
        await audit.insert(secretAccessEvents).values({
          companyId: input.companyId,
@@ -423,7 +423,7 @@ export function projectRoutes(db: Db) {
        // The receipt lock can wait too. Keep current authority behind every
        // admission wait and ahead of both receipt replay and project creation.
        await withExternalExecutionAdmission(tx, actorCompanyId, agentId, runId, async () => {
-          const current = await projectToolContext(tx as unknown as Db, museActor, true);
+          const current = await projectToolContext(tx as unknown as Db, museActor, true, "Project", { includeExecutionInput: true });
          try {
            const execution = parseNativeExecutionInput(current.run.runnerProfileJson?.nativeExecutionInput);
            const broker = museExternalAgents(db);
@@ -214,7 +214,7 @@ export function connectionIntentService(db: Db) {
      throw forbidden("Runtime tool token is no longer active");
    }
    if (run.activeIdentityContextId && !failedAuthRun) {
-      const current = await captureRunIdentity(db, { companyId: run.companyId, agentId: run.agentId, runId: run.id });
+      const current = await captureRunIdentity(db, { companyId: run.companyId, agentId: run.agentId, runId: run.id }, { readScope: "identity" });
      run = { ...run, responsibleUserId: current.run.responsibleUserId };
    }
    if (!run.responsibleUserId) throw forbidden("This task needs a responsible user to connect a service");
@@ -3,7 +3,7 @@ import { isUuidLike } from "@paperclipai/shared";
 import { agents, agentApiKeys, heartbeatRuns, runIdentityContexts, issues, projects, toolApplications, toolCatalogEntries, toolConnections, toolConnectionInstalls, toolProfiles, toolProfileEntries, toolProfileBindings, toolMcpGateways, aiConnectionPools, connectionGrants, routines, type Db } from "@paperclipai/db";
 import { forbidden } from "../errors.js";
 import { authorizationService, canActorReadIssuePrivacy, canActorReadProjectPrivacy, type AuthorizationActor, type AuthorizationDecision } from "./authorization.js";
-import { captureRunIdentity } from "./run-identity.js";
+import { captureRunIdentity, runIdentityTaskFields, type RunIdentityCapture } from "./run-identity.js";
 import type { LogActivityInput } from "./activity-log.js";
 import type { ConnectionManagementAction, ConnectionManagementResource } from "./connection-management-policy.js";

@@ -17,7 +17,7 @@ export type ConnectionManagementContext = {
  reauthorize: (continuation?: { afterHumanAction: true }) => Promise<void>;
 };

-type CapturedConnectionRun = Omit<Awaited<ReturnType<typeof captureRunIdentity>>, "context"> & { context: Awaited<ReturnType<typeof captureRunIdentity>>["context"] | null };
+type CapturedConnectionRun = RunIdentityCapture<"task_authorization">;
 const connectionReadRuns = new WeakMap<ConnectionManagementContext, CapturedConnectionRun>();

 export function responsibleConnectionUser(actor: AuthorizationActor): string | null {
@@ -51,12 +51,12 @@ export function connectionManagementService(db: Db) {
        const reusable = options.capabilityProbe && !options.afterHumanAction && options.readContext?.actor === actor && options.readContext.resource.companyId === resource.companyId
          ? connectionReadRuns.get(options.readContext) : undefined;
        const captured = reusable ?? (options.afterHumanAction ? await (async () => {
-          const [run] = await db.select().from(heartbeatRuns).where(and(eq(heartbeatRuns.id, actor.runId!), eq(heartbeatRuns.agentId, actor.agentId!), eq(heartbeatRuns.companyId, resource.companyId)));
+          const [run] = await db.select(runIdentityTaskFields).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, actor.runId!), eq(heartbeatRuns.agentId, actor.agentId!), eq(heartbeatRuns.companyId, resource.companyId)));
          if (!run || !["running", "succeeded"].includes(run.status)) throw forbidden("The originating run cannot continue this authentication");
          const [pending] = await db.select().from(runIdentityContexts).where(and(eq(runIdentityContexts.runId, run.id), eq(runIdentityContexts.status, "pending")));
          if (pending) throw forbidden("Run identity is being reconciled");
          return { run, context: null };
-        })() : await captureRunIdentity(db, { companyId: resource.companyId, agentId: actor.agentId, runId: actor.runId }));
+        })() : await captureRunIdentity(db, { companyId: resource.companyId, agentId: actor.agentId, runId: actor.runId }, { readScope: "task_authorization" }));
        capturedRun = captured;
        const { run, context } = captured;
        if ((context?.responsibleUserId ?? run.responsibleUserId) !== responsibleUserId) throw forbidden("Run responsible-user authority has changed");
@@ -8,7 +8,7 @@ import {
  type chatConversations,
 } from "@paperclipai/db";
 import { and, eq, inArray } from "drizzle-orm";
-import { captureRunIdentity } from "../run-identity.js";
+import { captureRunIdentity, runIdentityTaskFields } from "../run-identity.js";
 import { forbidden } from "../../errors.js";
 import { boundRunIssueId } from "../run-task-binding.js";
 import { chatRunSourceMessageId } from "../chat-channels/run-origin.js";
@@ -110,7 +110,7 @@ export async function slackRunOrigin(db: Db, binding: SlackTaskBinding) {
    if (!approved || !binding.identityContextId)
      throw forbidden("Slack approval is not executing for this task");
    const [run] = await db
-      .select()
+      .select(runIdentityTaskFields)
      .from(heartbeatRuns)
      .where(
        and(
@@ -141,7 +141,7 @@ export async function slackRunOrigin(db: Db, binding: SlackTaskBinding) {
      );
    captured = { run, context };
  } else {
-    captured = await captureRunIdentity(db, binding);
+    captured = await captureRunIdentity(db, binding, { readScope: "task_authorization" });
    if (
      binding.identityContextId &&
      binding.identityContextId !== captured.context?.id
@@ -134,7 +134,7 @@ export function executionWorkspaceRepositoryService(db: Db) {
    let localTrusted = input.actor.source === "local_implicit";
    if (input.actor.type === "agent") {
      if (input.actor.source !== "agent_jwt" || !input.actor.agentId || !input.actor.runId || input.actor.companyId !== input.companyId) throw forbidden("Repository preparation requires an authenticated task run");
-      const identity = await captureRunIdentity(db, { companyId: input.companyId, agentId: input.actor.agentId, runId: input.actor.runId });
+      const identity = await captureRunIdentity(db, { companyId: input.companyId, agentId: input.actor.agentId, runId: input.actor.runId }, { readScope: "task_authorization" });
      const runIssueId = identity.run.nativeIssueId ?? identity.run.contextSnapshot?.issueId;
      if (runIssueId !== input.issueId) throw forbidden("Repository preparation is scoped to the current task");
      userId = identity.run.responsibleUserId;
@@ -458,7 +458,7 @@ export class PaperclipRunnerToolAuthority {
    }
    if (RUNTIME_CONNECTION_TOOL_DEFINITIONS.some((tool) => tool.name === call.tool)) {
      await this.#boundContext();
-      const { run } = await captureRunIdentity(this.db, this.binding);
+      const { run } = await captureRunIdentity(this.db, this.binding, { readScope: "identity" });
      if (!run.responsibleUserId) throw forbidden("This task needs a responsible user before requesting a connection");
      const claims = {
        sub: this.binding.agentId, company_id: this.binding.companyId,
@@ -794,9 +794,9 @@ export class PaperclipRunnerToolAuthority {
      }
      case "report_progress": return this.#reportProgress(input);
      case "request_human_input": return this.#requestHumanInput(input,
-        (await captureRunIdentity(this.db, this.binding)).context?.id ?? null);
+        (await captureRunIdentity(this.db, this.binding, { readScope: "identity" })).context?.id ?? null);
      case "create_task": return this.#createTask(input,
-        (await captureRunIdentity(this.db, this.binding)).context?.id ?? null);
+        (await captureRunIdentity(this.db, this.binding, { readScope: "identity" })).context?.id ?? null);
      case "reassign_task": return this.#reassignTask(input);
      case "set_task_title": return this.#setTaskTitle(input);
      case "set_task_monitor": return this.#setTaskMonitor(input);
@@ -12,7 +12,7 @@ export async function projectSkillSetupIdentity(db: Db, companyId: string, actor
  let connectionId: string | null = null, refreshGrantId: string | null = null;
  if (actor?.type === 'agent' && actor.source === 'agent_jwt') {
    if (!actor.runId || !actor.agentId || actor.companyId !== companyId) throw forbidden('Project skill setup requires an authenticated agent run.');
-    const identity = await captureRunIdentity(db, { companyId, agentId: actor.agentId, runId: actor.runId });
+    const identity = await captureRunIdentity(db, { companyId, agentId: actor.agentId, runId: actor.runId }, { readScope: 'identity' });
    userId = identity.run.responsibleUserId;
    // Only reuse the connection already authorized by managed GitHub delivery
    // for this active run. Responsibility alone cannot discover another grant.
@@ -2,16 +2,21 @@ import type { Request } from "express";
 import { and, eq } from "drizzle-orm";
 import { issues, type Db } from "@paperclipai/db";
 import { forbidden } from "../errors.js";
-import { captureRunIdentity } from "./run-identity.js";
+import { captureRunIdentity, type RunIdentityCapture } from "./run-identity.js";

+type ProjectToolResult<Run> = { run: Run; issue: typeof issues.$inferSelect; userId: string | null; localTrusted: boolean };
 /** Resolve authority from the authenticated run, never caller-supplied user/task IDs. */
-export async function projectToolContext(db: Db, actor: Request["actor"], write = false, resource = "Project") {
+export function projectToolContext(db: Db, actor: Request["actor"], write?: boolean, resource?: string): Promise<ProjectToolResult<RunIdentityCapture<"task_authorization">["run"]>>;
+export function projectToolContext(db: Db, actor: Request["actor"], write: boolean, resource: string, options: { includeExecutionInput: true }): Promise<ProjectToolResult<RunIdentityCapture<"project_execution">["run"]>>;
+export async function projectToolContext(db: Db, actor: Request["actor"], write = false, resource = "Project", options?: { includeExecutionInput: true }) {
  if (actor.type !== "agent" || actor.source !== "agent_jwt" || !actor.runId || !actor.agentId || !actor.companyId) {
    throw forbidden(`${resource} tools require an authenticated agent run`);
  }
  // Acquire task/run locks before checking mode and session generation. A reset,
  // cancellation, or steering update cannot race a committing project mutation.
-  const identity = await captureRunIdentity(db, { companyId: actor.companyId, agentId: actor.agentId, runId: actor.runId });
+  const binding = { companyId: actor.companyId, agentId: actor.agentId, runId: actor.runId };
+  const identity = options?.includeExecutionInput ? await captureRunIdentity(db, binding, { readScope: "project_execution" }) :
+    await captureRunIdentity(db, binding, { readScope: "task_authorization" });
  const run = identity.run;
  const snapshot = run.contextSnapshot ?? {};
  const issueId = run.nativeIssueId ?? (typeof snapshot.issueId === "string" ? snapshot.issueId : null);
@@ -465,16 +465,71 @@ export async function acceptSteeredIdentity(
  if (accepted) await activate(executor, accepted);
 }

-export async function captureRunIdentity(
+const runIdentityFields = {
+  id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
+  status: heartbeatRuns.status, responsibleUserId: heartbeatRuns.responsibleUserId,
+  activeIdentityContextId: heartbeatRuns.activeIdentityContextId,
+};
+
+export const runIdentityTaskFields = {
+  ...runIdentityFields, nativeIssueId: heartbeatRuns.nativeIssueId, issueId: heartbeatRuns.issueId,
+  wakeupRequestId: heartbeatRuns.wakeupRequestId,
+  contextSnapshot: sql<Record<string, unknown> | null>`case when jsonb_typeof(${heartbeatRuns.contextSnapshot}) = 'object'
+    then jsonb_build_object('issueId', ${heartbeatRuns.contextSnapshot}->'issueId',
+      'taskId', ${heartbeatRuns.contextSnapshot}->'taskId',
+      'conversationSessionGeneration', ${heartbeatRuns.contextSnapshot}->'conversationSessionGeneration',
+      'paperclipSkillTest', ${heartbeatRuns.contextSnapshot}->'paperclipSkillTest') else null end`,
+};
+
+export const runIdentityBrokerFields = {
+  ...runIdentityFields, invocationSource: heartbeatRuns.invocationSource,
+  contextSnapshot: sql<Record<string, unknown> | null>`case when jsonb_typeof(${heartbeatRuns.contextSnapshot}) = 'object'
+    then jsonb_build_object('issueId', ${heartbeatRuns.contextSnapshot}->'issueId',
+      'projectId', ${heartbeatRuns.contextSnapshot}->'projectId', 'routineId', ${heartbeatRuns.contextSnapshot}->'routineId',
+      'responsibleUserId', ${heartbeatRuns.contextSnapshot}->'responsibleUserId',
+      'responsible_user_id', ${heartbeatRuns.contextSnapshot}->'responsible_user_id',
+      'paperclipIssue', case when jsonb_typeof(${heartbeatRuns.contextSnapshot}->'paperclipIssue') = 'object'
+        then jsonb_build_object('id', ${heartbeatRuns.contextSnapshot}->'paperclipIssue'->'id',
+          'projectId', ${heartbeatRuns.contextSnapshot}->'paperclipIssue'->'projectId',
+          'responsibleUserId', ${heartbeatRuns.contextSnapshot}->'paperclipIssue'->'responsibleUserId',
+          'responsible_user_id', ${heartbeatRuns.contextSnapshot}->'paperclipIssue'->'responsible_user_id') else null end)
+    else null end`,
+};
+
+const runIdentityProjectFields = {
+  ...runIdentityTaskFields, runtimeMode: heartbeatRuns.runtimeMode,
+  runnerProfileJson: sql<Record<string, unknown> | null>`case when jsonb_typeof(${heartbeatRuns.runnerProfileJson}) = 'object'
+    then case when ${heartbeatRuns.runnerProfileJson} ? 'nativeExecutionInput'
+      then jsonb_build_object('nativeExecutionInput', ${heartbeatRuns.runnerProfileJson}->'nativeExecutionInput')
+      else '{}'::jsonb end else null end`,
+};
+
+type FullIdentityRun = typeof heartbeatRuns.$inferSelect;
+type IdentityRun = Pick<FullIdentityRun, keyof typeof runIdentityFields>;
+type TaskIdentityRun = Pick<FullIdentityRun, keyof typeof runIdentityTaskFields>;
+type BrokerIdentityRun = Pick<FullIdentityRun, keyof typeof runIdentityBrokerFields>;
+type ProjectIdentityRun = Pick<FullIdentityRun, keyof typeof runIdentityProjectFields>;
+export type RunIdentityReadScope = "github_credentials" | "request_authentication" | "identity" | "task_authorization" | "token_broker" | "project_execution";
+export type RunIdentityCapture<Scope extends RunIdentityReadScope | undefined = undefined> = {
+  run: Scope extends "identity" ? IdentityRun : Scope extends "task_authorization" ? TaskIdentityRun :
+    Scope extends "token_broker" ? BrokerIdentityRun : Scope extends "project_execution" ? ProjectIdentityRun : FullIdentityRun;
+  context: RunIdentityContext | null;
+};
+
+export async function captureRunIdentity<Scope extends RunIdentityReadScope | undefined = undefined>(
  db: Db,
  input: { companyId: string; runId: string; agentId: string },
-  options?: { readScope: "github_credentials" | "request_authentication" },
-) {
+  options?: { readScope: Scope },
+): Promise<RunIdentityCapture<Scope>> {
  // Lock acquisition serializes with steering delivery and its durable acknowledgement.
  return db.transaction(async (tx) => {
    await lockIdentityTask(tx, input.companyId, input.runId);
    const [run] = await tx
-      .select(options?.readScope === "github_credentials" || options?.readScope === "request_authentication" ? {
+      .select(options?.readScope === "identity" ? runIdentityFields :
+        options?.readScope === "task_authorization" ? runIdentityTaskFields :
+        options?.readScope === "token_broker" ? runIdentityBrokerFields :
+        options?.readScope === "project_execution" ? runIdentityProjectFields :
+        options?.readScope === "github_credentials" || options?.readScope === "request_authentication" ? {
        ...getTableColumns(heartbeatRuns),
        contextSnapshot: sql<Record<string, unknown> | null>`case when jsonb_typeof(${heartbeatRuns.contextSnapshot}) = 'object'
          then jsonb_build_object('issueId', ${heartbeatRuns.contextSnapshot}->'issueId',
@@ -528,7 +583,7 @@ export async function captureRunIdentity(
            ),
          )
      : [];
-    return { run, context: context ?? null };
+    return { run, context: context ?? null } as RunIdentityCapture<Scope>;
  });
 }

@@ -1,5 +1,5 @@
 import type { DbTransaction } from "../types.js";
-import { captureRunIdentity } from "../../run-identity.js";
+import { captureRunIdentity, runIdentityBrokerFields } from "../../run-identity.js";
 import { and, eq, gte, inArray, sql } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
 import { connectionTokenIssuances, companyMemberships, heartbeatRuns, toolConnections, toolProfileBindings, toolProfileEntries, toolProfiles } from "@paperclipai/db";
@@ -75,11 +75,11 @@ export function createTokenBrokerAdmission(db: Db, now: () => Date) {
    runId: string;
  }) {
    const [initialRun] = await db
-      .select()
+      .select(runIdentityBrokerFields)
      .from(heartbeatRuns)
      .where(eq(heartbeatRuns.id, input.runId));
    const run = initialRun?.activeIdentityContextId
-      ? (await captureRunIdentity(db, input)).run
+      ? (await captureRunIdentity(db, input, { readScope: "token_broker" })).run
      : initialRun;
    if (
      !run ||
@@ -2199,7 +2199,7 @@ export function createToolGatewayService(
      companyId: session.companyId,
      agentId: session.agentId,
      runId: session.runId,
-    });
+    }, { readScope: "identity" });
    return {
      ...session,
      identityContextId: captured.context?.id,
@@ -19,7 +19,7 @@ export function registerSpekoVoiceRuntime(db: Db, start: (binding: Binding, invo
 async function resolveAuthority(db: Db, session: SpekoTaskSession, endpointId: string) {
  if (!(await instanceSettingsService(db).getExperimental()).enableChatConnectors) throw forbidden("Voice connections are disabled");
  if (!runtimes.has(db) || !session.agentId || !session.runId || !session.issueId) throw forbidden("Speko tools require an active task run");
-  const { run, context } = await captureRunIdentity(db, { companyId: session.companyId, agentId: session.agentId, runId: session.runId });
+  const { run, context } = await captureRunIdentity(db, { companyId: session.companyId, agentId: session.agentId, runId: session.runId }, { readScope: "task_authorization" });
  if (boundRunIssueId(run) !== session.issueId || !run.responsibleUserId || session.identityContextId && session.identityContextId !== context?.id) throw forbidden("The accepted requester or task changed");
  const [endpoint] = await db.select().from(chatEndpoints).where(and(eq(chatEndpoints.companyId, session.companyId), eq(chatEndpoints.id, endpointId), eq(chatEndpoints.provider, "speko"), eq(chatEndpoints.assignedAgentId, session.agentId), eq(chatEndpoints.status, "active")));
  if (!endpoint) throw forbidden("This agent has no active Speko connection");