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',