From 9f3a5456acda24cab7040daa67cefb321dd73f15 Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Sun, 25 Oct 2026 15:53:18 -0700 Subject: [PATCH] perf(server): bound run reads in authentication and conversation recovery (#17663) Read compact run identity during authentication and compact turn state during conversation recovery. Preserve fresh checks, durable steering and reply evidence, locking, and complete stored payloads. Validate 184 targeted PostgreSQL and lifecycle tests, workspace typecheck, server build, full CI, and Apex 5/5. Co-Authored-By: Paperclip --- doc/DATABASE.md | 12 ++ .../agent-auth-identity-transfer.test.ts | 102 +++++++++++++++++ .../conversation-settlement-transfer.test.ts | 105 ++++++++++++++++++ server/src/__tests__/run-identity.test.ts | 22 ++-- server/src/middleware/auth.ts | 3 +- server/src/services/agent-conversations.ts | 31 +++++- server/src/services/recovery/service.ts | 4 +- server/src/services/run-identity.ts | 4 +- 8 files changed, 265 insertions(+), 18 deletions(-) create mode 100644 server/src/__tests__/agent-auth-identity-transfer.test.ts create mode 100644 server/src/__tests__/conversation-settlement-transfer.test.ts diff --git a/doc/DATABASE.md b/doc/DATABASE.md index e35cf130381..d5f970ac23d 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -49,12 +49,24 @@ and leaves all existing cases passive. Source identities are not grants. ### Task query performance +Agent request authentication uses a compact run-identity read. It omits saved +execution payloads, results, and log excerpts from the query response. The read +still locks the task before the run and reconciles pending steering from durable +acknowledgements on every request. Default identity reads retain the full run. +Stored payloads and separate native tool authorization checks remain unchanged. + Native event acknowledgements read a count and maximum through the existing source-sequence index. A stream with gaps uses indexed lookups for its first missing successor. The acknowledgement keeps its contiguous-prefix rule without returning every historical sequence number. All events stay stored. Counting the index entries still grows with the stream length. +Conversation recovery reads only the saved run fields used to settle a turn. +Settlement locks and reads current conversation state without loading the task +description or comment bodies. It still checks the durable reply or pending +interaction, session generation, owning run, and newer user messages. Saved +execution payloads, task text, and comments remain stored in full. + The attention feed checks for newer runs with company-scoped existence probes. The probes match the failed run's agent and issue or task context. They preserve the existing millisecond comparison and empty-context rules. Each batch holds diff --git a/server/src/__tests__/agent-auth-identity-transfer.test.ts b/server/src/__tests__/agent-auth-identity-transfer.test.ts new file mode 100644 index 00000000000..819b23dc64a --- /dev/null +++ b/server/src/__tests__/agent-auth-identity-transfer.test.ts @@ -0,0 +1,102 @@ +import { randomUUID } from "node:crypto"; +import express from "express"; +import request from "supertest"; +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { agents, companies, createDb, heartbeatRunEvents, heartbeatRuns, issueComments, issues } from "@paperclipai/db"; +import { createDatabaseResponseCapture, installDatabaseResponseCapture } from "@paperclipai/db/response-capture"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { createLocalAgentJwt } from "../agent-auth-jwt.js"; +import { actorMiddleware } from "../middleware/auth.js"; +import { errorHandler } from "../middleware/error-handler.js"; +import { captureRunIdentity, initializeRunIdentity, reserveSteeredIdentity } from "../services/run-identity.js"; + +const support = await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("agent authentication identity transfer", () => { + let database: Awaited>; + let db: ReturnType; + beforeAll(async () => { + vi.stubEnv("PAPERCLIP_AGENT_JWT_SECRET", "identity-transfer-test-secret"); + vi.stubEnv("PAPERCLIP_INSTANCE_ID", "default"); + database = await startEmbeddedPostgresTestDatabase("paperclip-auth-identity-transfer-"); + db = createDb(database.connectionString); + }, 30_000); + afterAll(async () => { + vi.unstubAllEnvs(); + await db?.$client.end({ timeout: 5 }); + await database?.cleanup(); + }, 60_000); + + async function seed() { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Authentication transfer", issuePrefix: companyId.slice(0, 8) }); + await db.insert(agents).values({ id: agentId, companyId, name: "Reader", role: "engineer", adapterType: "codex_local", status: "idle" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Authentication task", responsibleUserId: "owner" }); + const padding = "retained input ".repeat(32_768); + const payloads = { + contextSnapshot: { issueId, launchInput: padding }, + runnerProfileJson: { nativeExecutionInput: { savedInput: padding } }, + resultJson: { savedResult: padding }, stdoutExcerpt: padding, stderrExcerpt: padding, + }; + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", ...payloads }); + const identity = await initializeRunIdentity(db, { companyId, runId, issueId, messageIds: [], responsibleUserId: "owner", cause: "instruction" }); + const app = express(); + app.use(actorMiddleware(db, { deploymentMode: "authenticated", resolveSession: async () => null })); + app.get("/actor", (req, res) => res.json(req.actor)); + app.patch("/actor", (req, res) => res.json(req.actor)); + app.use(errorHandler); + const token = createLocalAgentJwt(agentId, companyId, "codex_local", runId, "stale-user"); + expect(token).not.toBeNull(); + return { companyId, agentId, issueId, runId, app, token: token!, identity, payloads }; + } + + it("bounds actual middleware responses and preserves the complete stored run", async () => { + const fixture = await seed(); + const events: Array> = [], now = Date.now(); + const capture = createDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event)); + installDatabaseResponseCapture(db, { heartbeatRuns }, capture); + try { + const response = await request(fixture.app).get("/actor").set("Authorization", `Bearer ${fixture.token}`); + expect(response.status).toBe(200); + expect(response.body).toMatchObject({ type: "agent", runId: fixture.runId, + identityContextId: fixture.identity.id, onBehalfOfUserId: "owner" }); + } finally { capture.flush(); } + const reads = events.filter(event => event.event === "paperclip.db_response_capture.entry.v1" + && event.table === "heartbeat_runs" && (event.operation === "select" || event.operation === "execute")); + expect(reads.reduce((sum, event) => sum + Number(event.calls), 0)).toBeGreaterThanOrEqual(3); + expect(reads.reduce((sum, event) => sum + Number(event.responseValueBytes), 0)).toBeLessThan(16_384); + expect((await captureRunIdentity(db, fixture)).run).toMatchObject(fixture.payloads); + }); + + it("blocks uncertain steering, accepts its durable acknowledgement, and rechecks cancellation", async () => { + const fixture = await seed(), messageId = randomUUID(); + await db.insert(issueComments).values({ id: messageId, companyId: fixture.companyId, issueId: fixture.issueId, + authorUserId: "next-user", body: "Next instruction" }); + const pending = await reserveSteeredIdentity(db, { ...fixture, messageId }); + expect(pending).not.toBeNull(); + const blocked = await request(fixture.app).get("/actor").set("Authorization", `Bearer ${fixture.token}`); + expect(blocked.status).toBe(409); + await db.insert(heartbeatRunEvents).values({ companyId: fixture.companyId, agentId: fixture.agentId, runId: fixture.runId, + seq: 1, eventType: "item.completed", sourceEventId: randomUUID(), + payload: { prpEvent: { turnId: "auth-turn", itemId: `auth-turn:steer:${messageId}`, + payload: { kind: "steering_acknowledgement" } } } }); + const accepted = await request(fixture.app).get("/actor").set("Authorization", `Bearer ${fixture.token}`); + expect(accepted.status).toBe(200); + expect(accepted.body).toMatchObject({ identityContextId: pending!.id, onBehalfOfUserId: "next-user" }); + await db.update(heartbeatRuns).set({ status: "cancelled" }).where(eq(heartbeatRuns.id, fixture.runId)); + const revoked = await request(fixture.app).patch("/actor").set("Authorization", `Bearer ${fixture.token}`); + expect(revoked.status).toBe(403); + expect(revoked.body.code).toBe("agent_run_cancelled"); + const [stored] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + expect(stored).toMatchObject(fixture.payloads); + }); + + it.each(["company", "agent", "missing", "finished"] as const)("rejects %s run identity for the compact scope", async (fault) => { + const fixture = await seed(); + if (fault === "finished") await db.update(heartbeatRuns).set({ status: "succeeded" }).where(eq(heartbeatRuns.id, fixture.runId)); + const input = { companyId: fault === "company" ? randomUUID() : fixture.companyId, + agentId: fault === "agent" ? randomUUID() : fixture.agentId, + runId: fault === "missing" ? randomUUID() : fixture.runId }; + await expect(captureRunIdentity(db, input, { readScope: "request_authentication" })).rejects.toThrow(/active run/); + }); +}); diff --git a/server/src/__tests__/conversation-settlement-transfer.test.ts b/server/src/__tests__/conversation-settlement-transfer.test.ts new file mode 100644 index 00000000000..6a128b00e95 --- /dev/null +++ b/server/src/__tests__/conversation-settlement-transfer.test.ts @@ -0,0 +1,105 @@ +import { randomUUID } from "node:crypto"; +import { and, eq, sql } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { agents, companies, createDb, heartbeatRuns, issueComments, issueThreadInteractions, issues } from "@paperclipai/db"; +import { createDatabaseResponseCapture, installDatabaseResponseCapture } from "@paperclipai/db/response-capture"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { settleConversationTurnForRun } from "../services/agent-conversations.js"; + +const support = await getEmbeddedPostgresTestSupport(); +(support.supported ? describe : describe.skip)("conversation settlement transfer", () => { + let database: Awaited>; + let db: ReturnType; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("paperclip-conversation-settlement-"); + db = createDb(database.connectionString); + }, 30_000); + afterAll(async () => { await db?.$client.end({ timeout: 5 }); await database?.cleanup(); }, 60_000); + + async function seed(reply: "response" | "interaction" | "reset" | "none" = "response") { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(), wakeId = randomUUID(); + const padding = "saved conversation ".repeat(32_768); + await db.insert(companies).values({ id: companyId, name: "Settlement transfer", issuePrefix: companyId.slice(0, 8) }); + await db.insert(agents).values({ id: agentId, companyId, name: "Conversation agent", role: "engineer" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Conversation", description: padding, + status: "in_progress", conversationState: "active", conversationUserId: "conversation-owner", + conversationAgentId: agentId, assigneeAgentId: agentId, conversationSessionGeneration: 3 }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "succeeded", + contextSnapshot: { issueId, conversationSessionGeneration: 3, wakeCommentId: wakeId, + conversationReset: reply === "reset", savedInput: padding }, + runnerProfileJson: { savedProfile: padding }, resultJson: { savedResult: padding }, + stdoutExcerpt: padding, stderrExcerpt: padding }); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId)); + const wakeTime = new Date(Date.now() - 10_000); + await db.insert(issueComments).values({ id: wakeId, companyId, issueId, authorUserId: "conversation-owner", + body: padding, createdAt: wakeTime }); + if (reply === "response") await db.insert(issueComments).values({ companyId, issueId, + createdByRunId: runId, authorAgentId: agentId, origin: "comment", body: padding }); + if (reply === "interaction") await db.insert(issueThreadInteractions).values({ companyId, issueId, + sourceRunId: runId, createdByAgentId: agentId, status: "pending", kind: "request_confirmation", payload: { version: 1 } }); + return { companyId, agentId, issueId, runId, wakeId, wakeTime }; + } + + async function sourceDigests(input: Awaited>) { + return db.execute(sql`select + (select encode(sha256(convert_to(jsonb_build_array(context_snapshot,runner_profile_json,result_json,stdout_excerpt,stderr_excerpt)::text,'UTF8')),'hex') + from heartbeat_runs where id=${input.runId}::uuid) as run_digest, + (select encode(sha256(convert_to(description,'UTF8')),'hex') from issues where id=${input.issueId}::uuid) as description_digest, + (select encode(sha256(convert_to(jsonb_agg(jsonb_build_array(id,body) order by id)::text,'UTF8')),'hex') + from issue_comments where issue_id=${input.issueId}::uuid) as comments_digest`); + } + + it.each(["response", "interaction", "reset"] as const)("bounds %s settlement reads and retains all saved text and run payloads", async reply => { + const input = await seed(reply), before = await sourceDigests(input); + const events: Array> = [], now = Date.now(); + const capture = createDatabaseResponseCapture({ start: now - 100, end: now + 59_000 }, event => events.push(event)); + installDatabaseResponseCapture(db, { heartbeatRuns, issues, issueComments }, capture); + try { expect(await settleConversationTurnForRun(db, input.companyId, input.runId)).toBe(true); } + finally { capture.flush(); } + const reads = events.filter(event => event.event === "paperclip.db_response_capture.entry.v1" + && ["heartbeat_runs", "issues", "issue_comments"].includes(String(event.table)) + && (event.operation === "select" || event.operation === "execute")); + expect(reads.reduce((sum, event) => sum + Number(event.calls), 0)).toBeGreaterThanOrEqual(4); + expect(reads.reduce((sum, event) => sum + Number(event.responseValueBytes), 0)).toBeLessThan(16_384); + const [issue] = await db.select({ status: issues.status, state: issues.conversationState }).from(issues).where(eq(issues.id, input.issueId)); + expect(issue).toEqual({ status: "in_review", state: "waiting" }); + expect(await sourceDigests(input)).toEqual(before); + }); + + it.each(["conversationReplyBoundaryCommentId", "wakeCommentId", "commentId"] as const)("keeps later user messages active through %s", async key => { + const input = await seed(); + await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: input.issueId, + conversationSessionGeneration: 3, [key]: input.wakeId } }).where(eq(heartbeatRuns.id, input.runId)); + await db.insert(issueComments).values({ companyId: input.companyId, issueId: input.issueId, + authorUserId: "conversation-owner", body: "A new instruction", createdAt: new Date(input.wakeTime.getTime() + 1000) }); + expect(await settleConversationTurnForRun(db, input.companyId, input.runId)).toBe(true); + const [issue] = await db.select({ status: issues.status, state: issues.conversationState }).from(issues).where(eq(issues.id, input.issueId)); + expect(issue).toEqual({ status: "in_progress", state: "active" }); + }); + + it.each(["missing", "company", "issue-company", "generation", "owner", "failed", "no-reply"] as const)("refuses to settle a %s mismatch", async fault => { + const input = await seed(fault === "no-reply" ? "none" : "response"); + if (fault === "generation") await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: input.issueId, + conversationSessionGeneration: 2, wakeCommentId: input.wakeId } }).where(eq(heartbeatRuns.id, input.runId)); + if (fault === "owner") { + const otherRun = randomUUID(); + await db.insert(heartbeatRuns).values({ id: otherRun, companyId: input.companyId, agentId: input.agentId }); + await db.update(issues).set({ executionRunId: otherRun }).where(eq(issues.id, input.issueId)); + } + if (fault === "failed") await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, input.runId)); + if (fault === "issue-company") { + const otherCompany = randomUUID(), otherIssue = randomUUID(), otherAgent = randomUUID(); + await db.insert(companies).values({ id: otherCompany, name: "Other company", issuePrefix: otherCompany.slice(0, 8) }); + await db.insert(agents).values({ id: otherAgent, companyId: otherCompany, name: "Other agent" }); + await db.insert(issues).values({ id: otherIssue, companyId: otherCompany, title: "Other conversation", + conversationUserId: "other-owner", conversationAgentId: otherAgent, assigneeAgentId: otherAgent, + conversationState: "active", conversationSessionGeneration: 3 }); + await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: otherIssue, conversationSessionGeneration: 3, + conversationReset: true } }).where(eq(heartbeatRuns.id, input.runId)); + } + expect(await settleConversationTurnForRun(db, fault === "company" ? randomUUID() : input.companyId, + fault === "missing" ? randomUUID() : input.runId)).toBe(false); + const [issue] = await db.select({ status: issues.status }).from(issues).where(and(eq(issues.id, input.issueId), eq(issues.companyId, input.companyId))); + expect(issue.status).toBe("in_progress"); + }); +}); diff --git a/server/src/__tests__/run-identity.test.ts b/server/src/__tests__/run-identity.test.ts index b371c59f043..069139a8123 100644 --- a/server/src/__tests__/run-identity.test.ts +++ b/server/src/__tests__/run-identity.test.ts @@ -29,14 +29,14 @@ const support = await getEmbeddedPostgresTestSupport(); } return { companyId, agentId, issueId, runId, messageIds }; } - it("retains typed GitHub trust bindings and keeps other identity reads complete", async () => { + it.each(["github_credentials", "request_authentication"] as const)("retains typed trust bindings for %s and keeps other identity reads complete", async (readScope) => { const input = await seed(); await initializeRunIdentity(db, { ...input, responsibleUserId: "owner", cause: "queued" }); for (const value of [null, true, 42, " padded ", [], {}, { future: { opaque: true } }]) { const contextSnapshot = { issueId: input.issueId, projectId: value, executionPolicy: value, savedInput: { retained: true } }; const runnerProfileJson = { nativeExecutionInput: { opaque: value }, savedInput: "retained" }; await db.update(heartbeatRuns).set({ contextSnapshot, runnerProfileJson }).where(eq(heartbeatRuns.id, input.runId)); - const narrow = await captureRunIdentity(db, input, { readScope: "github_credentials" }); + const narrow = await captureRunIdentity(db, input, { readScope }); expect(narrow.run.contextSnapshot).toEqual({ issueId: input.issueId, taskId: null, projectId: value, executionPolicy: value }); expect(narrow.run.runnerProfileJson).toBeNull(); const full = await captureRunIdentity(db, input); @@ -123,24 +123,25 @@ const support = await getEmbeddedPostgresTestSupport(); }, ); - it("holds acquisition during uncertain steering, preserves snapshots, and never rewinds on replay", async () => { + it.each([undefined, "github_credentials", "request_authentication"] 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" }); - const startedUnderA = await captureRunIdentity(db, input); + const startedUnderA = await captureRunIdentity(db, input, options); const b = await reserveSteeredIdentity(db, { ...input, messageId: input.messageIds[1] }); expect(b).not.toBeNull(); - await expect(captureRunIdentity(db, input)).rejects.toThrow(/reconciled/); + await expect(captureRunIdentity(db, input, options)).rejects.toThrow(/reconciled/); await expect(reserveSteeredIdentity(db, { ...input, messageId: input.messageIds[2] })).rejects.toThrow(/reconciled/); await db.transaction(async tx => { await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.runId)).for("update"); await acceptSteeredIdentity(tx, b!); }); - expect((await captureRunIdentity(db, input)).context?.responsibleUserId).toBe("B"); + expect((await captureRunIdentity(db, input, options)).context?.responsibleUserId).toBe("B"); expect(startedUnderA.context?.responsibleUserId).toBe("A"); const a = await reserveSteeredIdentity(db, { ...input, messageId: input.messageIds[2] }); await acceptSteeredIdentity(db, a!); await acceptSteeredIdentity(db, b!); - expect((await captureRunIdentity(db, input)).context?.responsibleUserId).toBe("A"); + expect((await captureRunIdentity(db, input, options)).context?.responsibleUserId).toBe("A"); }); it("keeps rejected steering unchanged and preserves the originating identity on retry", async () => { const input = await seed(); @@ -175,16 +176,17 @@ const support = await getEmbeddedPostgresTestSupport(); expect(captured.run.responsibleUserId).toBeNull(); expect(captured.context?.parentContextId).toBe(origin.id); }); - it("recovers an uncertain acknowledgement from the authenticated native event journal", async () => { + it.each([undefined, "github_credentials", "request_authentication"] 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" }); const pending = await reserveSteeredIdentity(db, { ...input, messageId: input.messageIds[1] }); - await expect(captureRunIdentity(db, input)).rejects.toThrow(/reconciled/); + await expect(captureRunIdentity(db, input, options)).rejects.toThrow(/reconciled/); await db.insert(heartbeatRunEvents).values({ companyId: input.companyId, runId: input.runId, agentId: input.agentId, seq: 1, eventType: "item.completed", sourceEventId: randomUUID(), payload: { prpEvent: { turnId: "turn-1", itemId: `turn-1:steer:${pending!.messageId}`, payload: {kind: "steering_acknowledgement"} } }, }); - expect((await captureRunIdentity(db, input)).context?.responsibleUserId).toBe("B"); + expect((await captureRunIdentity(db, input, options)).context?.responsibleUserId).toBe("B"); await db.update(heartbeatRuns).set({status: "succeeded"}).where(eq(heartbeatRuns.id, input.runId)); expect((await reserveSteeredIdentity(db, { ...input, messageId: input.messageIds[1] }))?.status).toBe("accepted"); }); diff --git a/server/src/middleware/auth.ts b/server/src/middleware/auth.ts index 8504fc3c853..86347bb40d5 100644 --- a/server/src/middleware/auth.ts +++ b/server/src/middleware/auth.ts @@ -429,7 +429,8 @@ export function actorMiddleware(db: Db, opts: ActorMiddlewareOptions): RequestHa return; } if (identityRun?.activeIdentityContextId && identityRun.status === "running") { - const captured = await captureRunIdentity(db, { companyId: claims.company_id, agentId: claims.sub, runId: claims.run_id }); + const captured = await captureRunIdentity(db, { companyId: claims.company_id, agentId: claims.sub, runId: claims.run_id }, + { readScope: "request_authentication" }); identityRun.activeIdentityContextId = captured.context?.id ?? null; identityRun.responsibleUserId = captured.context?.responsibleUserId ?? null; } diff --git a/server/src/services/agent-conversations.ts b/server/src/services/agent-conversations.ts index a2ec8f74b01..b6cfd1a7337 100644 --- a/server/src/services/agent-conversations.ts +++ b/server/src/services/agent-conversations.ts @@ -282,7 +282,7 @@ export async function prepareConversationTurn( /** Finalizers only park a turn with a durable response, interaction, or processed /new. */ export async function settleConversationTurn( db: Db, - run: typeof heartbeatRuns.$inferSelect, + run: Pick, ) { if (run.status !== "succeeded") return false; const context = run.contextSnapshot ?? {}; @@ -291,7 +291,12 @@ export async function settleConversationTurn( let publication: ActivityPublication | null = null; const settled = await db.transaction(async (tx) => { const [issue] = await tx - .select() + .select({ + id: issues.id, companyId: issues.companyId, status: issues.status, + conversationAgentId: issues.conversationAgentId, conversationUserId: issues.conversationUserId, + conversationState: issues.conversationState, conversationSessionGeneration: issues.conversationSessionGeneration, + executionRunId: issues.executionRunId, + }) .from(issues) .where(and(eq(issues.id, issueId), eq(issues.companyId, run.companyId))) .for("update"); @@ -344,7 +349,7 @@ export async function settleConversationTurn( const [wake] = typeof wakeId === "string" ? await tx - .select() + .select({ id: issueComments.id }) .from(issueComments) .where(eq(issueComments.id, wakeId)) : []; @@ -407,6 +412,26 @@ export async function settleConversationTurn( return settled; } +export async function settleConversationTurnForRun( + db: Db, + companyId: string, + runId: string, +) { + const [run] = await db.select({ + id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, status: heartbeatRuns.status, + contextSnapshot: sql | null>`case when jsonb_typeof(${heartbeatRuns.contextSnapshot}) = 'object' + then jsonb_build_object( + 'issueId', ${heartbeatRuns.contextSnapshot}->'issueId', + 'conversationReset', ${heartbeatRuns.contextSnapshot}->'conversationReset', + 'conversationSessionGeneration', ${heartbeatRuns.contextSnapshot}->'conversationSessionGeneration', + 'conversationReplyBoundaryCommentId', ${heartbeatRuns.contextSnapshot}->'conversationReplyBoundaryCommentId', + 'wakeCommentId', ${heartbeatRuns.contextSnapshot}->'wakeCommentId', + 'commentId', ${heartbeatRuns.contextSnapshot}->'commentId' + ) else null end`, + }).from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, companyId))); + return run ? settleConversationTurn(db, run) : false; +} + /** Fresh provider context uses only messages in this session, up to this turn. */ export async function conversationReplay( db: Db, diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index 488ef813306..3bbf40ab0d2 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -21,7 +21,7 @@ import { import { CONTROLLER_LEASE_LOST_FAILURE_CODE, hasLiveLegacyController } from "../legacy-controller-lease.js"; import { isCompletedOnboardingHandoffWake } from "../chat-completion-delivery.js"; import { instanceSettingsService } from "../instance-settings.js"; -import { isWaitingConversation, settleConversationTurn, deliverConversationComments } from "../agent-conversations.js"; +import { isWaitingConversation, settleConversationTurnForRun, deliverConversationComments } from "../agent-conversations.js"; import { and, asc, @@ -4582,7 +4582,7 @@ export function recoveryService( if (issue.conversationAgentId) { const lastRun = await getLatestIssueRun(issue.companyId, issue.id); if (lastRun?.status === "succeeded") { - if (await settleConversationTurn(db, (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, lastRun.id)))[0]!)) { + if (await settleConversationTurnForRun(db, issue.companyId, lastRun.id)) { const [current] = await db.select().from(issues).where(eq(issues.id, issue.id)); if (current) Object.assign(issue, current); } diff --git a/server/src/services/run-identity.ts b/server/src/services/run-identity.ts index b1e27d1ae72..2f1e9e97d65 100644 --- a/server/src/services/run-identity.ts +++ b/server/src/services/run-identity.ts @@ -434,13 +434,13 @@ export async function acceptSteeredIdentity( export async function captureRunIdentity( db: Db, input: { companyId: string; runId: string; agentId: string }, - options?: { readScope: "github_credentials" }, + options?: { readScope: "github_credentials" | "request_authentication" }, ) { // 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" ? { + .select(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',