diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 13f3868869a..e8c98c6f6e5 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -2660,6 +2660,13 @@ An upgrade readiness report contains preliminary native evidence, not permission to remove a deployment volume. Keep the checkpoint pool bounded and measure all pools when qualifying database cost; the current size of two is a default. +Committed native Agent Chat turns recover their unanswered final reply through +the existing response presentation worker. Recovery verifies the accepted result +and final turn, agent, responsible user, native session and conversation generation. +It stores the exact final message before the normal conversation settlement moves +the chat to waiting. Later user input remains active. Recovery does not run the +agent again, replace a selected reply or restore a deleted reply. + For retained Codex event tails, keep source drain/accounting and target admission separate. `native-staged-attachment-recovery.ts` validates private source/target header evidence; `native-recovery-usage-delivery.ts` binds the source's accounting diff --git a/server/src/services/native-runtime/agent-chat-response-recovery.test.ts b/server/src/services/native-runtime/agent-chat-response-recovery.test.ts new file mode 100644 index 00000000000..1c2cc4399a3 --- /dev/null +++ b/server/src/services/native-runtime/agent-chat-response-recovery.test.ts @@ -0,0 +1,149 @@ +import { randomUUID } from "node:crypto"; +import { and, eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { agents, authUsers, companies, completionContracts, createDb, heartbeatRunEvents, heartbeatRuns, issueComments, issues, + nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; +import { NativeRunCoordinatorStore } from "./native-run-coordinator-store.js"; +import { finalizeNativeRun, recoverCommittedNativeChatResponse } from "./native-run-finalizer.js"; +import { nativeResponsePresentationWork } from "./native-response-presentation-work.js"; + +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_AGENT_CHAT_RESPONSE_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("committed Agent Chat reply recovery", () => { + let database: Awaited> | undefined; + let db: Db; + beforeAll(async () => { + if (!external) database = await startEmbeddedPostgresTestDatabase("agent-chat-response-"); + db = createDb(external ?? database!.connectionString); + }, 60_000); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + + async function fixture(conversation = true) { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); + const sessionId = randomUUID(), runnerId = randomUUID(), userId = `fixture-${randomUUID()}`, wakeId = randomUUID(); + await db.insert(authUsers).values({ id: userId, name: "Conversation owner", email: `${userId}@example.test`, + createdAt: new Date("2020-01-01T00:00:00Z"), updatedAt: new Date("2020-01-01T00:00:00Z") }); + await db.insert(companies).values({ id: companyId, name: "Reply recovery fixture", issuePrefix: companyId.slice(0, 8) }); + await db.insert(agents).values({ id: agentId, companyId, name: "Conversation agent", adapterType: "paperclip_runner" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Conversation reply", status: "in_progress", assigneeAgentId: agentId, + ...(conversation ? { conversationAgentId: agentId, conversationUserId: userId, conversationState: "active", conversationSessionGeneration: 0 } : {}) }); + await db.insert(issueComments).values({ id: wakeId, companyId, issueId, authorUserId: userId, body: "Hire one teammate and tell me when ready.", + createdAt: new Date("2020-01-01T00:00:00Z") }); + const [contract] = await db.insert(completionContracts).values({ companyId, issueId, revision: 1, + schemaVersion: "paperclip.completion-contract.v1", policyVersion: "test", risk: "standard", completionAuthority: "server_arbiter", + incompleteCriteriaPolicy: "preserve_non_terminal", contractJson: { revision: "test", objective: "Answer the user", criteria: [] }, + canonicalSha256: "fixture-conversation-contract", createdByActorType: "system", createdByActorId: "test" }).returning(); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: issueId, nativeSessionId: sessionId, + runnerInstanceId: runnerId, completionContractId: contract.id, completionContractSha256: contract.canonicalSha256, + status: "running", runtimeMode: "native", responsibleUserId: userId, + contextSnapshot: { issueId, source: "issue.comment", wakeCommentId: wakeId, ...(conversation ? { conversationSessionGeneration: 0 } : {}) } }); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId)); + const port = new NativeRunCoordinatorStore(db, { companyId, issueId, runId, agentId, normalizedSessionId: sessionId, + runnerSourceInstanceId: runnerId, completionContractId: contract.id, completionContractSha256: contract.canonicalSha256, + completionContractRevision: "test", completionContractCriterionIds: [] }); + const reply = "The teammate is hired and ready for your next instruction."; + const prpEvent = { runId, normalizedSessionId: sessionId, turnId: "hire-turn", sourceEventId: "fixture-final-answer", + payload: { kind: "agentMessage", channel: "final", providerPhase: "final", text: reply } }; + await db.insert(heartbeatRunEvents).values({ companyId, runId, agentId, seq: 170, eventType: "item.completed", payload: { prpEvent } }); + await port.completeRun({ result: { schema: "paperclip.run_result.v1", reportedWorkDisposition: "done", + summary: "A completed hire, with its final message preserved separately.", completionClaim: { contractRevision: "test", + objectiveSatisfied: true, criteria: [], remainingWork: [] }, evidence: [], verification: [], attentionRequests: [], artifacts: [] }, + terminal: { schema: "paperclip.prp.terminal.v1", runTerminalState: "succeeded", turnTerminalState: "completed", reportedWorkDisposition: "done" }, turnId: "hire-turn" }); + await finalizeNativeRun({ db, runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: false }); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date("2020-01-01T00:02:00Z") }).where(eq(heartbeatRuns.id, runId)); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)))[0].phase).toBe("committed"); + const input = { companyId, issueId, runId }; + return { ...input, input, agentId, userId, wakeId, prpEvent, reply }; + } + const comments = (runId: string) => db.select().from(issueComments).where(eq(issueComments.createdByRunId, runId)); + const issue = async (issueId: string) => (await db.select().from(issues).where(eq(issues.id, issueId)))[0]; + + it("finds a committed final Agent Chat turn, persists the exact reply once and only then settles waiting", async () => { + const f = await fixture(); + expect(await comments(f.runId)).toEqual([]); + expect(await issue(f.issueId)).toMatchObject({ status: "in_progress", conversationState: "active" }); + const work = nativeResponsePresentationWork(db), entries = await work.load(f.input); + expect(entries).toEqual([expect.objectContaining({ id: f.runId })]); + await expect(work.check(entries[0])).resolves.toBeNull(); + expect(await comments(f.runId)).toEqual([expect.objectContaining({ body: f.reply, authorAgentId: f.agentId, createdByRunId: f.runId })]); + expect(await issue(f.issueId)).toMatchObject({ status: "in_review", conversationState: "waiting" }); + expect(await work.load(f.input)).toEqual([]); + await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(false); + expect(await comments(f.runId)).toHaveLength(1); + }); + it("keeps a later user message actionable after recovering the earlier exact reply", async () => { + const f = await fixture(); + await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, authorUserId: f.userId, body: "Another instruction", + createdAt: new Date("2020-01-01T00:01:00Z") }); + await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(true); + expect(await comments(f.runId)).toHaveLength(1); + expect(await issue(f.issueId)).toMatchObject({ status: "in_progress", conversationState: "active" }); + }); + it("does not duplicate a reply under concurrent recovery", async () => { + const f = await fixture(); + expect((await Promise.all([recoverCommittedNativeChatResponse(db, f.input), recoverCommittedNativeChatResponse(db, f.input)])).sort()) + .toEqual([false, true]); + expect(await comments(f.runId)).toHaveLength(1); + }); + it.each(["skip", "generation", "agent", "user", "execution", "waiting", "turn", "event-run", "session", "channel", "kind", "digest", "partial"])( + "keeps %s mismatches unchanged instead of publishing an unauthorized or incomplete reply", async fault => { + const f = await fixture(); + if (["skip", "generation"].includes(fault)) await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId, + source: "issue.comment", wakeCommentId: f.wakeId, conversationSessionGeneration: fault === "generation" ? 1 : 0, + ...(fault === "skip" ? { skipIssueComment: true } : {}) } }).where(eq(heartbeatRuns.id, f.runId)); + if (fault === "user") await db.update(heartbeatRuns).set({ responsibleUserId: "another-user" }).where(eq(heartbeatRuns.id, f.runId)); + if (fault === "agent") { + const other = randomUUID(); await db.insert(agents).values({ id: other, companyId: f.companyId, name: "Other agent" }); + await db.update(issues).set({ conversationAgentId: other, assigneeAgentId: other }).where(eq(issues.id, f.issueId)); + } + if (fault === "execution") { + const other = randomUUID(); await db.insert(heartbeatRuns).values({ id: other, companyId: f.companyId, agentId: f.agentId }); + await db.update(issues).set({ executionRunId: other }).where(eq(issues.id, f.issueId)); + } + if (fault === "waiting") await db.update(issues).set({ executionRunId: null, status: "in_review", conversationState: "waiting" }).where(eq(issues.id, f.issueId)); + if (["turn", "event-run", "session", "channel", "kind"].includes(fault)) { + const event = structuredClone(f.prpEvent); + if (fault === "turn") event.turnId = "foreign-turn"; + if (fault === "event-run") event.runId = randomUUID(); + if (fault === "session") event.normalizedSessionId = randomUUID(); + if (fault === "channel") event.payload.channel = "analysis"; + if (fault === "kind") event.payload.kind = "reasoning"; + await db.update(heartbeatRunEvents).set({ payload: { prpEvent: event } }).where(and(eq(heartbeatRunEvents.runId, f.runId), eq(heartbeatRunEvents.seq, 170))); + } + if (fault === "digest") await db.update(nativeRunResults).set({ canonicalSha256: "invalid-digest" }).where(eq(nativeRunResults.runId, f.runId)); + if (fault === "partial") await db.update(nativeRunFinalizations).set({ phase: "copying_back" }).where(eq(nativeRunFinalizations.runId, f.runId)); + const before = await issue(f.issueId); + await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(false); + expect(await comments(f.runId)).toEqual([]); expect(await issue(f.issueId)).toEqual(before); + }, + ); + it.each([false, true])("preserves a selected reply without replacing or resurrecting it (deleted=%s)", async deleted => { + const f = await fixture(); + const [saved] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, authorAgentId: f.agentId, + createdByRunId: f.runId, body: "Already selected", ...(deleted ? { deletedAt: new Date("2020-01-01T00:03:00Z") } : {}) }).returning(); + await db.update(heartbeatRuns).set({ resultJson: { presentationDecision: { commentId: saved.id, commentAction: "create" } } }).where(eq(heartbeatRuns.id, f.runId)); + await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(false); + expect(await comments(f.runId)).toEqual([saved]); + }); + it("never crosses company scope and keeps ordinary internal task response recovery intact", async () => { + const f = await fixture(false); + await expect(recoverCommittedNativeChatResponse(db, { ...f.input, companyId: randomUUID() })).resolves.toBe(false); + const work = nativeResponsePresentationWork(db), entries = await work.load(f.input); + expect(entries).toHaveLength(1); await expect(work.check(entries[0])).resolves.toBeNull(); + expect(await comments(f.runId)).toEqual([expect.objectContaining({ body: f.reply })]); + expect((await issue(f.issueId)).conversationAgentId).toBeNull(); + }); + it("rolls back a partial reply write without changing the waiting state or consuming recovery", async () => { + const f = await fixture(), abort = new Error("fixture-commit-aborted"); + const failing = new Proxy(db, { get(target, key) { return key === "transaction" + ? (callback: Parameters[0]) => db.transaction(async tx => { await callback(tx); throw abort; }) + : Reflect.get(target, key); } }); + await expect(recoverCommittedNativeChatResponse(failing, f.input)).rejects.toBe(abort); + expect(await comments(f.runId)).toEqual([]); + expect(await issue(f.issueId)).toMatchObject({ status: "in_progress", conversationState: "active" }); + await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(true); + expect(await comments(f.runId)).toHaveLength(1); + }); +}); diff --git a/server/src/services/native-runtime/native-response-presentation-query.ts b/server/src/services/native-runtime/native-response-presentation-query.ts index d4daaba61bc..f3a476a287a 100644 --- a/server/src/services/native-runtime/native-response-presentation-query.ts +++ b/server/src/services/native-runtime/native-response-presentation-query.ts @@ -1,6 +1,6 @@ import { QueryBuilder } from "drizzle-orm/pg-core"; -import { and, asc, eq, gt, inArray, isNotNull, isNull, notExists, or, sql } from "drizzle-orm"; -import { heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, statusDecisionEffects } from "@paperclipai/db"; +import { and, asc, eq, exists, gt, inArray, isNotNull, isNull, notExists, or, sql } from "drizzle-orm"; +import { heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, statusDecisionEffects } from "@paperclipai/db"; import type { NativeResponsePresentationScope } from "../native-response-presentation-notifications.js"; export function nativeResponsePresentationCandidates(scope?: NativeResponsePresentationScope, after?: string, runId?: string) { @@ -30,7 +30,24 @@ export function nativeResponsePresentationCandidates(scope?: NativeResponsePrese sql`${nativeRunResults.resultJson} #>> '{terminal,turnTerminalState}' = 'completed'`, or( and(eq(heartbeatRuns.status, "succeeded"), - sql`(coalesce(${issues.conversationAgentId}::text, '') = '' or coalesce(${issues.conversationUserId}, '') = '')`, + or(sql`(coalesce(${issues.conversationAgentId}::text, '') = '' or coalesce(${issues.conversationUserId}, '') = '')`, + and(eq(statusDecisions.reasonCode, "conversation_turn_finished"), + eq(issues.conversationAgentId, heartbeatRuns.agentId), eq(issues.conversationUserId, heartbeatRuns.responsibleUserId), + eq(issues.status, "in_progress"), eq(issues.conversationState, "active"), + or(isNull(issues.executionRunId), eq(issues.executionRunId, heartbeatRuns.id)), + sql`${heartbeatRuns.contextSnapshot}->>'issueId' = ${issues.id}::text`, + sql`jsonb_typeof(${heartbeatRuns.contextSnapshot}->'conversationSessionGeneration') = 'number'`, + sql`${heartbeatRuns.contextSnapshot}->>'conversationSessionGeneration' = ${issues.conversationSessionGeneration}::text`, + exists(db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, heartbeatRuns.companyId), eq(heartbeatRunEvents.runId, heartbeatRuns.id), + eq(heartbeatRunEvents.agentId, heartbeatRuns.agentId), eq(heartbeatRunEvents.eventType, "item.completed"), + sql`${heartbeatRunEvents.payload} #>> '{prpEvent,runId}' = ${heartbeatRuns.id}::text`, + sql`${heartbeatRunEvents.payload} #>> '{prpEvent,normalizedSessionId}' = ${heartbeatRuns.nativeSessionId}::text`, + sql`${heartbeatRunEvents.payload} #>> '{prpEvent,turnId}' = ${nativeRunResults.turnId}`, + sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,kind}' = 'agentMessage'`, + sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,channel}' = 'final'`, + sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,text}' ~ '[^[:space:]]'`, + ))))), sql`${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' = 'done'`, sql`${nativeRunResults.resultJson} #>> '{terminal,reportedWorkDisposition}' = 'done'`), and(sql`${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' = 'yielded'`, diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index 7112a6afcff..d39a3b682bb 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -29,7 +29,7 @@ import { readPersistedNativeFailureEvidence } from "./native-provider-failure-ev import { dismissAutomaticCompletionReviews } from "./automatic-completion-reviews.js"; import { findSupersededExternalReviewRequests } from "./superseded-external-review-requests.js"; import { getNativeReviewAssignment, readNativeReviewAssignmentContext } from "./native-review-participant.js"; -import { conversationNativeDecision, isConversation } from "../agent-conversations.js"; +import { conversationNativeDecision, isConversation, settleConversationTurnForRun } from "../agent-conversations.js"; import { issueTreeControlService } from "../issue-tree-control.js"; import { randomUUID } from "node:crypto"; import { preserveNativeWorkspaceExportLease } from "./native-workspace-export-resume.js"; @@ -747,6 +747,7 @@ async function recoverCommittedNativeChatResponseInScope( ): Promise { let agentId: string | null = null; let materialized = false; + let conversationReply = false; const entered = beginRunPhase("response.transaction.entry"); try { materialized = await measureRunPhase("response.transaction", () => db.transaction(async (transaction) => { @@ -789,6 +790,7 @@ async function recoverCommittedNativeChatResponseInScope( id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + responsibleUserId: heartbeatRuns.responsibleUserId, nativeIssueId: heartbeatRuns.nativeIssueId, nativeSessionId: heartbeatRuns.nativeSessionId, completionContractId: heartbeatRuns.completionContractId, @@ -803,6 +805,8 @@ async function recoverCommittedNativeChatResponseInScope( contextSnapshot: sql | null>`case when ${heartbeatRuns.contextSnapshot} is null then null else jsonb_build_object( 'skipIssueComment', ${heartbeatRuns.contextSnapshot}->'skipIssueComment', + 'issueId', ${heartbeatRuns.contextSnapshot}->'issueId', + 'conversationSessionGeneration', ${heartbeatRuns.contextSnapshot}->'conversationSessionGeneration', 'source', ${heartbeatRuns.contextSnapshot}->'source', 'externalChatContinuation', ${heartbeatRuns.contextSnapshot}->'externalChatContinuation', 'paperclipWake', jsonb_build_object('externalInteractionContinuation', @@ -909,12 +913,16 @@ async function recoverCommittedNativeChatResponseInScope( agentId = run.agentId; return true; } - // Copyback can be owned by reconciliation before the live heartbeat - // reaches its presentation step. Recover a completed internal task's - // exact final reply from the same accepted turn, without rerunning work - // or granting external-chat publication authority. + const conversation = isConversation(issue); + const currentConversation = conversation && decision.reasonCode === "conversation_turn_finished" + && issue.conversationAgentId === run.agentId && issue.conversationUserId === run.responsibleUserId + && issue.status === "in_progress" && issue.conversationState === "active" + && (!issue.executionRunId || issue.executionRunId === run.id) + && run.contextSnapshot?.issueId === issue.id + && typeof run.contextSnapshot.conversationSessionGeneration === "number" + && run.contextSnapshot.conversationSessionGeneration === issue.conversationSessionGeneration; if ( - !isConversation(issue) && + (!conversation || currentConversation) && run.status === "succeeded" && record(run.contextSnapshot).skipIssueComment !== true && !isExternalChatPresentationContext(run.contextSnapshot) && @@ -929,6 +937,9 @@ async function recoverCommittedNativeChatResponseInScope( eq(heartbeatRunEvents.companyId, input.companyId), eq(heartbeatRunEvents.runId, run.id), eq(heartbeatRunEvents.eventType, "item.completed"), + conversation ? and(eq(heartbeatRunEvents.agentId, run.agentId), + sql`${heartbeatRunEvents.payload}->'prpEvent'->>'runId' = ${run.id}`, + sql`${heartbeatRunEvents.payload}->'prpEvent'->>'normalizedSessionId' = ${run.nativeSessionId}`) : undefined, sql`${heartbeatRunEvents.payload}->'prpEvent'->>'turnId' = ${accepted.turnId}`, sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,kind}' = 'agentMessage'`, sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,channel}' = 'final'`, @@ -963,6 +974,7 @@ async function recoverCommittedNativeChatResponseInScope( })}::jsonb`, updatedAt: new Date(), }).where(eq(heartbeatRuns.id, run.id)); agentId = run.agentId; + conversationReply = conversation; return true; } if ( @@ -1039,12 +1051,14 @@ async function recoverCommittedNativeChatResponseInScope( } finally { entered(); } - if (materialized && agentId) + if (materialized && agentId) { + if (conversationReply) await settleConversationTurnForRun(db, input.companyId, input.runId); publishChatPublicationCommitSignal({ ...input, agentId, eventType: "run.presentation.resolved", }); + } return materialized; }