diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 2ddd05e6a19..61dc59a1f84 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1342,8 +1342,11 @@ apply for ordinary tasks. In native standing conversations, a completed final response or accepted non-yielded semantic response takes precedence over earlier tool comments. Those comments stay in history. Identical text reuses its existing comment. With no qualified final, the explicit comment remains the response. -Governed waits and interaction cards keep ownership. The presentation decision -records the conversation-specific choice. +Governed waits and interaction cards keep ownership. A server-proven finished +conversation may still have a yielded semantic result. Its completed final reply +remains visible; a completed unknown-channel terminal reply is used only under +that finished-turn authority and after the explicit final and non-yielded summary. +The presentation decision records the conversation-specific choice. Stream closure without a turn terminal is not proof of success. Event replay uses the existing source receipts and never repeats provider work merely to recover recorded output. diff --git a/server/src/__tests__/heartbeat-run-summary.test.ts b/server/src/__tests__/heartbeat-run-summary.test.ts index fa19e01a1b5..52718b535b8 100644 --- a/server/src/__tests__/heartbeat-run-summary.test.ts +++ b/server/src/__tests__/heartbeat-run-summary.test.ts @@ -572,6 +572,22 @@ describe("resolveHeartbeatRunResponse", () => { }); }); + it.each([ + { conversationTurnFinished: true, interactionOwnsResponse: false, channel: "final" as const, visible: true }, + { conversationTurnFinished: true, interactionOwnsResponse: false, channel: "unknown" as const, visible: true }, + { conversationTurnFinished: false, interactionOwnsResponse: false, channel: "unknown" as const, visible: false }, + { conversationTurnFinished: true, interactionOwnsResponse: true, channel: "unknown" as const, visible: false }, + ])("preserves server-finished conversation authority for a yielded result: %j", ({ visible, ...authority }) => { + const { channel, ...flags } = authority; + const resolved = resolveHeartbeatRunResponse({ ...flags, nativeConversationFinalResponse: true, + resultJson: { summary: "Unqualified legacy summary", nativeResult: { schema: "paperclip.run_result.v1", + reportedWorkDisposition: "yielded", summary: "Waiting for the next message." } }, + finalAgentMessage: { text: "Which project should own this?", sourceEventId: "chat-final-1", channel }, + }); + expect(resolved.text).toBe(visible ? "Which project should own this?" : null); + expect(resolved.decision.commentAction).toBe(visible ? "create" : "none"); + }); + it("persists a final assistant reply when the server completed a conversation turn", () => { expect(resolveHeartbeatRunResponse({ conversationTurnFinished: true, diff --git a/server/src/services/heartbeat-run-summary.ts b/server/src/services/heartbeat-run-summary.ts index 42aedfad3da..ad4276f7e3f 100644 --- a/server/src/services/heartbeat-run-summary.ts +++ b/server/src/services/heartbeat-run-summary.ts @@ -572,11 +572,17 @@ export function resolveHeartbeatRunResponse(input: { const existingText = readCommentText(input.existingComment?.body); if (input.nativeConversationFinalResponse) { - if (input.interactionOwnsResponse || hasYieldedSemanticResult(resultJson)) return { + if (input.interactionOwnsResponse || (hasYieldedSemanticResult(resultJson) && !input.conversationTurnFinished)) return { text: null, decision: decision("none", { commentAction: "none", reasonCodes: ["interaction_owns_conversation_response"] }), }; - const final = resolveCompletedUpstreamResponse(false); + const final = resolveCompletedUpstreamResponse(false) + ?? (input.conversationTurnFinished && compatibleTerminalAssistant && finalAgentText + && !isStructuredSemanticResultText(finalAgentText) ? { + text: finalAgentText, + decision: decision("final_agent_message", { sourceEventId: input.finalAgentMessage?.sourceEventId, + commentAction: "create", reasonCodes: ["server_finished_conversation_terminal_message"] }), + } : null); if (final) { const reuse = !!input.existingComment && final.text === existingText; return { ...final, decision: { ...final.decision, 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 1c2cc4399a3..f0ab0a4dbaa 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 @@ -2,7 +2,7 @@ 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"; + nativeRunFinalizations, nativeRunResults, statusDecisions, 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"; @@ -20,7 +20,7 @@ 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) { + async function fixture(conversation = true, options: { disposition?: "done" | "yielded"; channel?: "final" | "unknown" } = {}) { 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`, @@ -45,12 +45,16 @@ describe.skipIf(!support.supported)("committed Agent Chat reply recovery", () => 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 } }; + payload: { kind: "agentMessage", channel: options.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" }); + const disposition = options.disposition ?? "done"; + await port.completeRun({ result: { schema: "paperclip.run_result.v1", reportedWorkDisposition: disposition, + summary: "A completed hire, with its final message preserved separately.", + ...(disposition === "yielded" ? { continuation: { kind: "response_wake" as const, summary: "Wait for the next human message", idempotencyKey: "conversation-next-message" } } : {}), + completionClaim: { contractRevision: "test", + objectiveSatisfied: true, criteria: [], remainingWork: [] }, evidence: [], verification: [], + attentionRequests: [], artifacts: [] }, + terminal: { schema: "paperclip.prp.terminal.v1", runTerminalState: "succeeded", turnTerminalState: "completed", reportedWorkDisposition: disposition }, 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"); @@ -73,6 +77,17 @@ describe.skipIf(!support.supported)("committed Agent Chat reply recovery", () => await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(false); expect(await comments(f.runId)).toHaveLength(1); }); + it("recovers a server-finished yielded conversation from its completed unknown-channel reply", async () => { + const f = await fixture(true, { disposition: "yielded", channel: "unknown" }); + expect((await db.select().from(statusDecisions).where(eq(statusDecisions.runId, f.runId)))[0]) + .toMatchObject({ reasonCode: "conversation_turn_finished" }); + await expect(recoverCommittedNativeChatResponse(db, f.input)).resolves.toBe(true); + expect(await comments(f.runId)).toEqual([expect.objectContaining({ body: f.reply, createdByRunId: f.runId })]); + expect(await issue(f.issueId)).toMatchObject({ status: "in_review", conversationState: "waiting" }); + 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", diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index ec4606dd7e0..40a3fdfe3f9 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -926,10 +926,11 @@ async function recoverCommittedNativeChatResponseInScope( run.status === "succeeded" && record(run.contextSnapshot).skipIssueComment !== true && !isExternalChatPresentationContext(run.contextSnapshot) && - result.reportedWorkDisposition === "done" && terminal.runTerminalState === "succeeded" && terminal.turnTerminalState === "completed" && - terminal.reportedWorkDisposition === "done" && + ((result.reportedWorkDisposition === "done" && terminal.reportedWorkDisposition === "done") || + (currentConversation && result.reportedWorkDisposition === "yielded" && terminal.reportedWorkDisposition === "yielded" + && !await hasChatRunOwnedProviderInteraction(tx, input))) && (await resolveChatRunPresentationAuthorizationReason(tx, input)) === "internal_agent_write" ) { const rows = await tx.select({ seq: heartbeatRunEvents.seq, payload: heartbeatRunEvents.payload }) @@ -942,7 +943,9 @@ async function recoverCommittedNativeChatResponseInScope( 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'`, + currentConversation + ? inArray(sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,channel}'`, ["final", "unknown"]) + : sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,channel}' = 'final'`, )).orderBy(desc(heartbeatRunEvents.seq)).limit(200); const finalAgentMessage = selectHeartbeatRunFinalAgentMessage({ candidates: rows.flatMap((row) => { @@ -960,6 +963,8 @@ async function recoverCommittedNativeChatResponseInScope( const resolved = resolveHeartbeatRunResponse({ resultJson: { ...record(run.resultJson), nativeResult: result }, nativeConversationFinalResponse: conversation, + conversationTurnFinished: currentConversation, + interactionOwnsResponse: conversation && await hasChatRunOwnedProviderInteraction(tx, input), existingComment: findHeartbeatRunCompletionComment(comments, run.resultJson), finalAgentMessage, });