diff --git a/doc/run-identity-reads.md b/doc/run-identity-reads.md new file mode 100644 index 00000000000..a41b252b411 --- /dev/null +++ b/doc/run-identity-reads.md @@ -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. diff --git a/scripts/benchmark-run-identity-projections.ts b/scripts/benchmark-run-identity-projections.ts new file mode 100644 index 00000000000..4ddc589ef3b --- /dev/null +++ b/scripts/benchmark-run-identity-projections.ts @@ -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="); +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> | undefined; +let db: ReturnType | 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; + 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)) + assert.deepEqual(item, (before.run.contextSnapshot?.paperclipIssue as Record)?.[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); } +} diff --git a/server/src/__tests__/company-skills-routes.test.ts b/server/src/__tests__/company-skills-routes.test.ts index f40bfa0f8ce..11fe51213a7 100644 --- a/server/src/__tests__/company-skills-routes.test.ts +++ b/server/src/__tests__/company-skills-routes.test.ts @@ -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 } })); }); diff --git a/server/src/__tests__/helpers/run-identity-projection-fixture.ts b/server/src/__tests__/helpers/run-identity-projection-fixture.ts new file mode 100644 index 00000000000..5072e37060f --- /dev/null +++ b/server/src/__tests__/helpers/run-identity-projection-fixture.ts @@ -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) => + target.transaction(tx => action(captureIdentityQueries(tx as unknown as Db, queries))); + if (key !== "select") return Reflect.get(target, key); + return (...fields: Parameters) => { + const builder = target.select(...fields), from = builder.from.bind(builder); + builder.from = ((...args: Parameters) => { + 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; + }; + }}); +} diff --git a/server/src/__tests__/project-skill-setup-identity.test.ts b/server/src/__tests__/project-skill-setup-identity.test.ts index 844c4f7df71..d08cd9ec867 100644 --- a/server/src/__tests__/project-skill-setup-identity.test.ts +++ b/server/src/__tests__/project-skill-setup-identity.test.ts @@ -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' } } }); diff --git a/server/src/__tests__/run-identity-projections.test.ts b/server/src/__tests__/run-identity-projections.test.ts new file mode 100644 index 00000000000..1e7132029a2 --- /dev/null +++ b/server/src/__tests__/run-identity-projections.test.ts @@ -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> | 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"); + }); +}); diff --git a/server/src/__tests__/run-identity.test.ts b/server/src/__tests__/run-identity.test.ts index 069139a8123..1cc8ed20f7a 100644 --- a/server/src/__tests__/run-identity.test.ts +++ b/server/src/__tests__/run-identity.test.ts @@ -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>; let db: ReturnType; 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, diff --git a/server/src/routes/projects.ts b/server/src/routes/projects.ts index c6479982335..272051bf521 100644 --- a/server/src/routes/projects.ts +++ b/server/src/routes/projects.ts @@ -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); diff --git a/server/src/services/connection-intents.ts b/server/src/services/connection-intents.ts index 1788c0ebd60..f0d143861e5 100644 --- a/server/src/services/connection-intents.ts +++ b/server/src/services/connection-intents.ts @@ -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"); diff --git a/server/src/services/connection-management.ts b/server/src/services/connection-management.ts index fcc532476be..0c1740f04ba 100644 --- a/server/src/services/connection-management.ts +++ b/server/src/services/connection-management.ts @@ -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; }; -type CapturedConnectionRun = Omit>, "context"> & { context: Awaited>["context"] | null }; +type CapturedConnectionRun = RunIdentityCapture<"task_authorization">; const connectionReadRuns = new WeakMap(); 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"); diff --git a/server/src/services/connectors/slack-authority.ts b/server/src/services/connectors/slack-authority.ts index 06da6455d35..ee19440f45e 100644 --- a/server/src/services/connectors/slack-authority.ts +++ b/server/src/services/connectors/slack-authority.ts @@ -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 diff --git a/server/src/services/execution-workspace-repositories.ts b/server/src/services/execution-workspace-repositories.ts index fae172296ff..4d4cfe16b86 100644 --- a/server/src/services/execution-workspace-repositories.ts +++ b/server/src/services/execution-workspace-repositories.ts @@ -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; diff --git a/server/src/services/native-runtime/paperclip-runner-tool-authority.ts b/server/src/services/native-runtime/paperclip-runner-tool-authority.ts index aa6863205fb..71f9e3ea02b 100644 --- a/server/src/services/native-runtime/paperclip-runner-tool-authority.ts +++ b/server/src/services/native-runtime/paperclip-runner-tool-authority.ts @@ -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); diff --git a/server/src/services/project-skill-source-bindings.ts b/server/src/services/project-skill-source-bindings.ts index 29ff8bb1875..259e0dda705 100644 --- a/server/src/services/project-skill-source-bindings.ts +++ b/server/src/services/project-skill-source-bindings.ts @@ -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. diff --git a/server/src/services/project-tool-context.ts b/server/src/services/project-tool-context.ts index a0cc9468351..89be328b052 100644 --- a/server/src/services/project-tool-context.ts +++ b/server/src/services/project-tool-context.ts @@ -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; 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["run"]>>; +export function projectToolContext(db: Db, actor: Request["actor"], write: boolean, resource: string, options: { includeExecutionInput: true }): Promise["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); diff --git a/server/src/services/run-identity.ts b/server/src/services/run-identity.ts index 166b1ae8cac..f72cc974ca7 100644 --- a/server/src/services/run-identity.ts +++ b/server/src/services/run-identity.ts @@ -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 | 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 | 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 | 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; +type TaskIdentityRun = Pick; +type BrokerIdentityRun = Pick; +type ProjectIdentityRun = Pick; +export type RunIdentityReadScope = "github_credentials" | "request_authentication" | "identity" | "task_authorization" | "token_broker" | "project_execution"; +export type RunIdentityCapture = { + 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( db: Db, input: { companyId: string; runId: string; agentId: string }, - options?: { readScope: "github_credentials" | "request_authentication" }, -) { + options?: { readScope: Scope }, +): Promise> { // 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 | 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; }); } diff --git a/server/src/services/tool-access/token-broker/admission.ts b/server/src/services/tool-access/token-broker/admission.ts index 5b30a3fd915..b4a98042001 100644 --- a/server/src/services/tool-access/token-broker/admission.ts +++ b/server/src/services/tool-access/token-broker/admission.ts @@ -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 || diff --git a/server/src/services/tool-gateway.ts b/server/src/services/tool-gateway.ts index ff5e1028ffc..57763f8b8d5 100644 --- a/server/src/services/tool-gateway.ts +++ b/server/src/services/tool-gateway.ts @@ -2199,7 +2199,7 @@ export function createToolGatewayService( companyId: session.companyId, agentId: session.agentId, runId: session.runId, - }); + }, { readScope: "identity" }); return { ...session, identityContextId: captured.context?.id, diff --git a/server/src/services/voice/speko-agent-tools.ts b/server/src/services/voice/speko-agent-tools.ts index 6332329fe35..7f0ce74b78f 100644 --- a/server/src/services/voice/speko-agent-tools.ts +++ b/server/src/services/voice/speko-agent-tools.ts @@ -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");