diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 67b0b47f82c..172130221c4 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2268,7 +2268,9 @@ posts its final answer to the conversation even when the task stays held: a pending or revision-requested approval, a pending execution stage, or a dependency blocker. A held task does not present the turn. Only a pending thread interaction (a question or confirmation card) presents the turn in place -of its prose. +of its prose. Background presentation recovery applies the same rule, so a +restart between the status commit and the live presentation posts the held +turn's answer once, or records that its card presented the turn. A delegated-task completion callback can finish native execution before its accepted final reply is saved. The completion outbox must retain that callback's diff --git a/server/src/services/heartbeat/run-completion.ts b/server/src/services/heartbeat/run-completion.ts index 425e681f025..fccf30ccc9e 100644 --- a/server/src/services/heartbeat/run-completion.ts +++ b/server/src/services/heartbeat/run-completion.ts @@ -45,7 +45,7 @@ import { isUnresolvedWorkspaceBaseRefError, readUnresolvedWorkspaceBaseRefDiagno import { runnerGoalService } from "../runner-goals.js"; import { resolveChatRunPresentationAuthorizationReason } from "../chat-run-publications.js"; import { hasChatRunOwnedProviderInteraction } from "../chat-interaction-arbitration.js"; -import { conversationTurnAnswered } from "../native-runtime/native-governance.js"; +import { conversationTurnPresentsAnswer } from "../native-runtime/native-governance.js"; import { readAcceptedTaskContinuationSummary } from "../heartbeat-run-summary.js"; import { SANDBOX_PROVIDER_PLUGIN_NOT_READY_REASON } from "../recovery/stranded-notice.js"; import { MAX_TURN_CONTINUATION_RETRY_REASON, isResolvedInteractionContinuationWakeContext } from "../../modules/run-dispatch/index.js"; @@ -847,14 +847,11 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC interactionOwnsResponse: Boolean(issueId && readAcceptedTaskContinuationSummary(persistedResultJson) && await hasChatRunOwnedProviderInteraction(db, { companyId: livenessRun.companyId, issueId, runId: livenessRun.id })), conversationTurnFinished: Boolean(issueId) && isConversation(issueContext) && livenessRun.status === "succeeded" - && (persistedResultJson?.finalizationReasonCode === "conversation_turn_finished" - // A turn left waiting on an approval, an execution stage or a - // blocker still answered the person. Only its own question card - // presents it instead. - || (persistedResultJson?.finalizationPhase === "committed" - && await measureRunPhase("response.resolve", () => conversationTurnAnswered({ - db, companyId: livenessRun.companyId, issueId: issueId!, runId: livenessRun.id, - })))), + && await measureRunPhase("response.resolve", () => conversationTurnPresentsAnswer({ + db, companyId: livenessRun.companyId, issueId: issueId!, runId: livenessRun.id, + finalizationPhase: persistedResultJson?.finalizationPhase, + finalizationReasonCode: persistedResultJson?.finalizationReasonCode, + })), existingComment: existingRunComment, finalAgentMessage, preferFinalResponseOverExistingComment: 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 index b0d4ce5481a..4d048122992 100644 --- a/server/src/services/native-runtime/agent-chat-response-recovery.test.ts +++ b/server/src/services/native-runtime/agent-chat-response-recovery.test.ts @@ -1,7 +1,7 @@ import { randomUUID } from "node:crypto"; import { and, eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; -import { agents, authUsers, companies, completionContracts, createDb, heartbeatRunEvents, heartbeatRuns, issueComments, issues, +import { agents, approvals, issueApprovals, statusDecisions, authUsers, companies, completionContracts, createDb, heartbeatRunEvents, heartbeatRuns, issueComments, issues, nativeRunFinalizations, nativeRunResults, issueThreadInteractions, chatTaskHandoffs, chatCompletionDeliveries, agentWakeupRequests, type Db } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js"; @@ -26,7 +26,8 @@ describe.skipIf(!support.supported)("committed Agent Chat reply recovery", () => }, 60_000); afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); - async function fixture(conversation = true, options: { succeed?: boolean; disposition?: "done" | "needs_review" } = {}) { + async function fixture(conversation = true, options: { succeed?: boolean; disposition?: "done" | "needs_review" | "yielded"; + beforeFinalize?: (ids: { companyId: string; issueId: string; runId: string; agentId: string }) => Promise } = {}) { 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`, @@ -54,8 +55,11 @@ describe.skipIf(!support.supported)("committed Agent Chat reply recovery", () => payload: { kind: "agentMessage", channel: "final", providerPhase: "final", text: reply } }; await db.insert(heartbeatRunEvents).values({ companyId, runId, agentId, seq: 170, eventType: "item.completed", payload: { prpEvent } }); const disposition = options.disposition ?? "done"; + await options.beforeFinalize?.({ companyId, issueId, runId, agentId }); await port.completeRun({ result: { schema: "paperclip.run_result.v1", reportedWorkDisposition: disposition, - summary: "A completed hire, with its final message preserved separately.", completionClaim: { contractRevision: "test", + summary: "A completed hire, with its final message preserved separately.", + ...(disposition === "yielded" ? { continuation: { kind: "response_wake", summary: "Wait for the hire decision.", idempotencyKey: "hire-wait" } } : {}), + completionClaim: { contractRevision: "test", objectiveSatisfied: true, criteria: [], remainingWork: [] }, evidence: [], verification: [], attentionRequests: disposition === "needs_review" ? [{ kind: "review", ownerClass: "human", summary: "Review the completed work." }] : [], artifacts: [] }, terminal: { schema: "paperclip.prp.terminal.v1", runTerminalState: "succeeded", turnTerminalState: "completed", reportedWorkDisposition: disposition }, turnId: "hire-turn" }); @@ -706,4 +710,87 @@ describe.skipIf(!support.supported)("committed Agent Chat reply recovery", () => await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(true); expect(await comments(f.runId)).toHaveLength(1); }); + + // A chat turn can end while its task stays held: a resubmitted approval or + // an execution stage keeps the status, but the turn still answered the + // person. A restart between the status commit and the live presentation + // must not lose that answer. The run is saved exactly as the previous + // release's finalizer saves it, with no presentation recorded yet. + type Hold = "pending approval" | "approval sent back for revision" | "execution stage" | "own question card"; + async function heldTurn(hold: Hold, disposition: "yielded" | "done" = "yielded") { + return fixture(true, { disposition, beforeFinalize: async ({ companyId, issueId, runId, agentId }) => { + if (hold.includes("approval")) { + const [approval] = await db.insert(approvals).values({ companyId, type: "hire_agent", + status: hold === "pending approval" ? "pending" : "revision_requested", payload: { name: "Researcher" } }).returning(); + await db.insert(issueApprovals).values({ companyId, issueId, approvalId: approval.id }); + } + if (hold === "execution stage") await db.update(issues).set({ executionState: { status: "pending" } }).where(eq(issues.id, issueId)); + if (hold === "own question card") { + await db.insert(issueThreadInteractions).values({ companyId, issueId, kind: "ask_user_questions", status: "pending", + sourceRunId: runId, createdByAgentId: agentId, payload: { version: 1, questions: [{ id: "model", prompt: "Which model?", + selectionMode: "single", required: true, options: [{ id: "flash", label: "Flash" }] }] } }); + } + } }); + } + it.each([ + ["pending approval", "yielded"], ["approval sent back for revision", "yielded"], ["execution stage", "yielded"], + ["pending approval", "done"], + ] as const)( + "recovers a held conversation turn's answer exactly once after a restart: %s (%s)", async (hold, disposition) => { + const f = await heldTurn(hold, disposition); + const [decision] = await db.select().from(statusDecisions).where(eq(statusDecisions.runId, f.runId)); + expect(decision.reasonCode).not.toBe("conversation_turn_finished"); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].resultJson) + .not.toHaveProperty("presentationDecision"); + expect(await comments(f.runId)).toEqual([]); + 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 db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].resultJson).toMatchObject({ + presentationDecision: { chosenSource: "final_agent_message", commentAction: "create", commentId: expect.any(String) } }); + expect(await work.load(f.input)).toEqual([]); + await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(false); + expect(await comments(f.runId)).toHaveLength(1); + }, + ); + it("does not repost a held conversation turn's answer that the live path already presented", async () => { + const f = await heldTurn("pending approval"); + const [saved] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, authorAgentId: f.agentId, + createdByRunId: f.runId, body: f.reply }).returning(); + await db.update(heartbeatRuns).set({ resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({ + presentationDecision: { chosenSource: "final_agent_message", commentAction: "create", commentId: saved.id, reasonCodes: [] } })}::jsonb` }) + .where(eq(heartbeatRuns.id, f.runId)); + expect(await nativeResponsePresentationWork(db).load(f.input)).toEqual([]); + await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(false); + expect(await comments(f.runId)).toEqual([saved]); + }); + it("keeps a held turn's presentation that the previous release's live path already recorded", async () => { + // Before the live answered rule, a held yielded turn was recorded as a + // control-plane wait. That decision stands; recovery never re-presents it. + const f = await heldTurn("pending approval"); + await db.update(heartbeatRuns).set({ resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({ + presentationDecision: { chosenSource: "none", commentAction: "none", reasonCodes: ["yielded_control_plane_wait"] } })}::jsonb` }) + .where(eq(heartbeatRuns.id, f.runId)); + expect(await nativeResponsePresentationWork(db).load(f.input)).toEqual([]); + await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(false); + expect(await comments(f.runId)).toEqual([]); + }); + it("settles a held conversation turn presented by its own question card without posting its prose", async () => { + const f = await heldTurn("own question card"); + const cards = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.issueId, f.issueId)); + expect(cards).toEqual([expect.objectContaining({ kind: "ask_user_questions", status: "pending", sourceRunId: f.runId })]); + 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((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0].resultJson).toMatchObject({ + presentationDecision: { chosenSource: "none", commentAction: "none" } }); + // Settled like the live path: the card is the turn's response. + expect(await issue(f.issueId)).toMatchObject({ conversationState: "waiting" }); + expect(await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.issueId, f.issueId))).toEqual(cards); + expect(await work.load(f.input)).toEqual([]); + await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(false); + expect(await comments(f.runId)).toEqual([]); + }); }); diff --git a/server/src/services/native-runtime/native-governance.ts b/server/src/services/native-runtime/native-governance.ts index 84b7c5c5b80..9be92563bf6 100644 --- a/server/src/services/native-runtime/native-governance.ts +++ b/server/src/services/native-runtime/native-governance.ts @@ -103,3 +103,23 @@ export async function conversationTurnAnswered(input: { // status gate, but it does not present the turn. return !(await pendingNativeInteraction(input)); } + +/** + * The one rule for whether a successful conversation turn's prose is its + * answer. Live run completion and background presentation recovery both use + * it, so a restart between the status commit and the live presentation + * presents the same reply the live path would have. + */ +export async function conversationTurnPresentsAnswer(input: { + db: Db; + companyId: string; + issueId: string; + runId: string; + finalizationPhase: unknown; + finalizationReasonCode: unknown; +}): Promise { + if (input.finalizationReasonCode === "conversation_turn_finished") return true; + // A turn left waiting on an approval, an execution stage or a blocker still + // answered the person. Only its own question card presents it instead. + return input.finalizationPhase === "committed" && await conversationTurnAnswered(input); +} 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 87da12c4ff9..5ea7b2a3e2c 100644 --- a/server/src/services/native-runtime/native-response-presentation-query.ts +++ b/server/src/services/native-runtime/native-response-presentation-query.ts @@ -1,5 +1,5 @@ import { QueryBuilder } from "drizzle-orm/pg-core"; -import { and, asc, eq, exists, gt, inArray, isNotNull, isNull, notExists, or, sql } from "drizzle-orm"; +import { and, asc, eq, exists, gt, inArray, isNotNull, isNull, ne, notExists, or, sql } from "drizzle-orm"; import { chatActions, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, statusDecisions, statusDecisionEffects } from "@paperclipai/db"; import type { NativeResponsePresentationScope } from "../native-response-presentation-notifications.js"; @@ -42,9 +42,11 @@ export function nativeResponsePresentationCandidates(scope?: NativeResponsePrese or( and(eq(heartbeatRuns.status, "succeeded"), or(sql`(coalesce(${issues.conversationAgentId}::text, '') = '' or coalesce(${issues.conversationUserId}, '') = '')`, - and(eq(statusDecisions.reasonCode, "conversation_turn_finished"), + // An unsettled conversation turn. A held turn keeps the status its + // gate committed; recovery applies the live path's answered rule. + and(or(ne(statusDecisions.reasonCode, "conversation_turn_finished"), eq(issues.status, "in_progress")), eq(issues.conversationAgentId, heartbeatRuns.agentId), eq(issues.conversationUserId, heartbeatRuns.responsibleUserId), - eq(issues.status, "in_progress"), eq(issues.conversationState, "active"), + 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'`, @@ -65,6 +67,9 @@ export function nativeResponsePresentationCandidates(scope?: NativeResponsePrese and(sql`(coalesce(${issues.conversationAgentId}::text, '') = '' or coalesce(${issues.conversationUserId}, '') = '')`, sql`${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' = 'needs_review'`, sql`${nativeRunResults.resultJson} #>> '{terminal,reportedWorkDisposition}' = 'needs_review'`), + and(sql`(coalesce(${issues.conversationAgentId}::text, '') <> '' and coalesce(${issues.conversationUserId}, '') <> '')`, + sql`${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' = 'yielded'`, + sql`${nativeRunResults.resultJson} #>> '{terminal,reportedWorkDisposition}' = 'yielded'`), )), and(sql`${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' = 'yielded'`, sql`${nativeRunResults.resultJson} #>> '{terminal,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 1fae0f2768f..069a415d072 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -15,7 +15,7 @@ import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-pre import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { readRequiredQualityReview } from "../required-quality-review.js"; import { NativeDeliverableEvidenceError, validateNativeDeliverableEvidence } from "./native-deliverable-feedback.js"; -import { pendingNativeGovernance } from "./native-governance.js"; +import { conversationTurnPresentsAnswer, pendingNativeGovernance } from "./native-governance.js"; import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; import { onboardingFirstTaskCompletionAllowed } from "../onboarding-first-task-admission.js"; import { notifyExecutionRunWork } from "../execution-control-notifications.js"; @@ -830,6 +830,7 @@ async function recoverCommittedNativeChatResponseInScope( let agentId: string | null = null; let materialized = false; let conversationReply = false; + let settledWithoutReply = false; const entered = beginRunPhase("response.transaction.entry"); try { materialized = await measureRunPhase("response.transaction", () => db.transaction(async (transaction) => { @@ -996,9 +997,13 @@ async function recoverCommittedNativeChatResponseInScope( return true; } const conversation = isConversation(issue); - const currentConversation = conversation && decision.reasonCode === "conversation_turn_finished" + // The turn is still the conversation's unsettled current turn. A held + // turn keeps whatever status its gate committed; a finished turn + // preserves in_progress. + const currentConversation = conversation && issue.conversationAgentId === run.agentId && issue.conversationUserId === run.responsibleUserId - && issue.status === "in_progress" && issue.conversationState === "active" + && (decision.reasonCode !== "conversation_turn_finished" || issue.status === "in_progress") + && issue.conversationState === "active" && (!issue.executionRunId || issue.executionRunId === run.id) && run.contextSnapshot?.issueId === issue.id && typeof run.contextSnapshot.conversationSessionGeneration === "number" @@ -1013,7 +1018,10 @@ async function recoverCommittedNativeChatResponseInScope( terminal.runTerminalState === "succeeded" && terminal.turnTerminalState === "completed" && ((result.reportedWorkDisposition === "done" && terminal.reportedWorkDisposition === "done") || - (!conversation && result.reportedWorkDisposition === "needs_review" && terminal.reportedWorkDisposition === "needs_review")) && + (!conversation && result.reportedWorkDisposition === "needs_review" && terminal.reportedWorkDisposition === "needs_review") || + // A conversation turn can yield while an approval, an execution + // stage or a blocker holds its task; it still answered the person. + (conversation && result.reportedWorkDisposition === "yielded" && terminal.reportedWorkDisposition === "yielded")) && (await resolveChatRunPresentationAuthorizationReason(tx, input)) === "internal_agent_write" ) { const rows = await tx.select({ seq: heartbeatRunEvents.seq, payload: heartbeatRunEvents.payload }) @@ -1049,6 +1057,10 @@ async function recoverCommittedNativeChatResponseInScope( )).orderBy(desc(issueComments.createdAt), desc(issueComments.id)); const resolved = resolveHeartbeatRunResponse({ resultJson: { ...record(run.resultJson), nativeResult: result }, + // The same answered rule as live run completion. + conversationTurnFinished: conversation && await conversationTurnPresentsAnswer({ + db: tx, ...input, finalizationPhase: coordinator.phase, finalizationReasonCode: decision.reasonCode, + }), existingComment: findHeartbeatRunCompletionComment(comments, run.resultJson), finalAgentMessage, }); @@ -1063,6 +1075,19 @@ async function recoverCommittedNativeChatResponseInScope( }).where(eq(heartbeatRuns.id, run.id)); return false; } + if (conversation && resolved.decision.commentAction === "none") { + // The turn's own question or confirmation card presents it, as on + // the live path. Record that decision once so the turn settles and + // leaves the pending presentation set. + await tx.update(heartbeatRuns).set({ + resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({ + presentationDecision: { ...resolved.decision, + reasonCodes: [...resolved.decision.reasonCodes, "committed_conversation_interaction_presented"] }, + })}::jsonb`, updatedAt: new Date(), + }).where(eq(heartbeatRuns.id, run.id)); + settledWithoutReply = true; + return false; + } if (!resolved.text || resolved.decision.commentAction !== "create") return false; await notifyChatRunMilestones(tx, input); const comment = await issueService(db).addComment(input.issueId, resolved.text, @@ -1152,6 +1177,7 @@ async function recoverCommittedNativeChatResponseInScope( } finally { entered(); } + if (settledWithoutReply) await settleConversationTurnForRun(db, input.companyId, input.runId); if (materialized && agentId) { if (conversationReply) await settleConversationTurnForRun(db, input.companyId, input.runId); publishChatPublicationCommitSignal({