From ad2304dc881db2bb46bc7002b2e88612cdec40c8 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Mon, 9 Nov 2026 01:08:10 -0600 Subject: [PATCH] fix(chat): recover a held conversation turn's answer after a restart (#20204) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - People talk to agents in Agent Chat. The agent's final message becomes the chat reply. > - A chat turn can end while its task stays held: a pending approval, an execution stage or a dependency blocker keeps the status. > - #20179 made live run completion post that held turn's answer. It added `conversationTurnAnswered()`. > - Background presentation recovery still required a `conversation_turn_finished` decision and a `done` result. > - If the server restarts between the status commit and the live presentation, recovery skips the held turn. The person never gets an answer. > - Restarts are product failures. A committed answer must not be lost. > - This pull request makes recovery use the same answered rule as the live path, through one shared helper. > - The benefit is that a held turn's answer reaches the chat exactly once, with or without a restart. ## Linked Issues or Issue Description Refs #20179 (live fix; this PR closes the "Known gap" in its Risks section). **What happened** In an Agent Chat, the agent resubmitted a hire approval and ended its turn with a yielded result. The finalizer committed `governed_response_waiting` and moved the task to `in_review`. The server restarted before run completion posted the reply. Background presentation recovery found the run as a candidate, but declined it, because it accepted only `conversation_turn_finished` with a `done` result. No reply was posted. The run stayed a pending presentation candidate. **Expected behavior** After a restart, recovery posts the held turn's final answer exactly once. It never posts a second copy when the live path already presented the answer. When the turn's own question or confirmation card presents it, recovery posts nothing and records that decision, as the live path does. **Steps to reproduce** 1. Open an Agent Chat. Link a pending hire approval to the chat task. 2. Let the agent end the turn with a yielded result and a `response_wake` continuation. The status decision commits as `governed_response_waiting`. 3. Stop the server after the status commit and before run completion presents the reply. 4. Start the server. Presentation recovery runs. 5. See that the chat shows no reply. ## What Changed - `server/src/services/native-runtime/native-governance.ts`: add `conversationTurnPresentsAnswer()`. It is the one rule for whether a successful conversation turn's prose is its answer: `conversation_turn_finished`, or a committed decision with no pending card of the turn (`conversationTurnAnswered()`). - `server/src/services/heartbeat/run-completion.ts`: the live path calls the shared helper. No change in behavior. - `server/src/services/native-runtime/native-run-finalizer.ts` (`recoverCommittedNativeChatResponse`): - A current conversation turn no longer needs `conversation_turn_finished`. The turn must still be the unsettled current turn: same agent, user, session generation, `conversationState: active`, and no other execution run. A `conversation_turn_finished` turn still needs `in_progress`. - A conversation turn with a `yielded` result is accepted next to `done`. - The resolver gets `conversationTurnFinished` from the shared helper. - When the turn's own card presents it, recovery records `presentationDecision: none` (reason `committed_conversation_interaction_presented`) and settles the conversation turn, like the live path. The run then leaves the pending presentation set. Before, such a run stayed a candidate. - `server/src/services/native-runtime/native-response-presentation-query.ts`: the candidate prefilter matches the same turns: any committed reason for an active conversation turn, and `yielded` conversation results. - `server/src/services/native-runtime/agent-chat-response-recovery.test.ts`: restart regressions with the real finalizer and a real database. - `doc/execution-semantics.md`: state that recovery applies the same rule. ## Verification - Failing first. With only the new tests applied, 5 of 7 new cases failed: pending approval, revision-requested approval, execution stage (all yielded), pending approval with `done`, and the own-card case. The run was a candidate, but recovery posted nothing and did not settle it. - After the fix: `cd server && npx vitest run src/services/native-runtime/agent-chat-response-recovery.test.ts src/services/heartbeat/run-completion.test.ts src/__tests__/idle-native-finalization.test.ts src/services/presentation-ownership-read.test.ts src/services/chat-monitor-continuation.test.ts src/services/native-runtime/native-preserved-monitor-wait.test.ts src/services/native-runtime/native-response-publication-retry.test.ts src/services/heartbeat/run-lifecycle.test.ts src/__tests__/scoped-deadline-work.test.ts src/services/native-runtime/paperclip-control-plane-port.test.ts src/__tests__/heartbeat-run-summary.test.ts src/__tests__/agent-conversations.test.ts`: 12 files, 627 tests passed. - The new cases check: - Held turns: the run is a candidate, the check posts the exact final answer once, the presentation decision is recorded, the run leaves the candidate set, and a second recovery returns false with one comment. - Live already presented: no candidate, no second comment. - Own question card: no comment, `presentationDecision: none`, the card stays pending, the conversation settles to `waiting`, and the run leaves the candidate set. - Previous-release live decision `none` (`yielded_control_plane_wait`): kept, not re-presented. - `tsc --noEmit -p server`: no errors in changed files. Four unrelated errors remain in this local worktree (`esbuild` not linked in `native-harness-prestage.test.ts`, and a linked-package type mismatch in `run-dispatch/adapters/postgres.ts`). - Turn start: no change. The helper runs only in run completion (after the provider turn ends) and in background presentation recovery. Nothing new runs before `turn.start`, and no files move. The live path keeps the same queries: it short-circuits on `conversation_turn_finished`, and otherwise runs the same card read as before. ## Risks - Low to medium. Recovery can now post the answer of a conversation turn whose task is held. This matches the live path since #20179. - A card-presented turn found by recovery is now settled with `presentationDecision: none`. Before, it stayed a pending candidate and kept the presentation worker (and idle-sleep checks) busy. - A `conversation_turn_finished` turn keeps the `in_progress` check. Held turns accept any status their gate committed. The unsettled-turn checks (active conversation state, session generation, execution run) still block stale or superseded turns. - Previous-release state: no persisted shape changes. A run committed by the previous release with no saved presentation decision (the restart window) is recovered by the new code. The test fixture uses the unchanged finalizer, so it saves exactly the previous release's shape. A run whose live path already recorded a decision keeps it. A held turn that the previous release recorded as `yielded_control_plane_wait` is not re-presented, so no late duplicate or stale reply appears. - Not changed: generic stranded-work recovery in `recovery/service.ts` still skips only `conversation_turn_finished` runs during the publication window. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Anthropic Claude Opus 5.5 (`claude-opus-5-5`), with tool use (shell, file edits, test runs) in Claude Code. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Opus 5.5 --- doc/execution-semantics.md | 4 +- .../src/services/heartbeat/run-completion.ts | 15 ++- .../agent-chat-response-recovery.test.ts | 93 ++++++++++++++++++- .../native-runtime/native-governance.ts | 20 ++++ .../native-response-presentation-query.ts | 11 ++- .../native-runtime/native-run-finalizer.ts | 34 ++++++- 6 files changed, 157 insertions(+), 20 deletions(-) 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({