diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 6541f4088b1..2478035988a 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2241,6 +2241,13 @@ apply. 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. +In an Agent Chat conversation, a successful turn whose status decision committed +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. + A delegated-task completion callback can finish native execution before its accepted final reply is saved. The completion outbox must retain that callback's queued target and attempt while the existing presentation selector finds pending diff --git a/server/src/services/heartbeat/run-completion.test.ts b/server/src/services/heartbeat/run-completion.test.ts index 6ff985ed53d..42c34facc34 100644 --- a/server/src/services/heartbeat/run-completion.test.ts +++ b/server/src/services/heartbeat/run-completion.test.ts @@ -16,7 +16,7 @@ import { createHash, randomUUID } from "node:crypto"; import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; import { createNativeRunLogCoverageVerifier, formatNativeRunLogEvent } from "../native-runtime/native-run-log-export.js"; import type { Db } from "@paperclipai/db"; -import { activityLog, agents, agentWakeupRequests, budgetReservations, caseHandoffs, cases, companies, completionContracts, costEvents, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, heartbeatRunEvents, issueComments, issueQuestionResponseDeliveries, issueRecoveryActions, issues, nativeRunFinalizations } from "@paperclipai/db"; +import { activityLog, agents, agentWakeupRequests, approvals, issueApprovals, budgetReservations, caseHandoffs, cases, companies, completionContracts, costEvents, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, heartbeatRunEvents, issueComments, issueQuestionResponseDeliveries, issueRecoveryActions, issues, nativeRunFinalizations } from "@paperclipai/db"; import { and, eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, beforeEach, describe, expect, it, onTestFinished, vi } from "vitest"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; @@ -469,6 +469,67 @@ describe.skipIf(!support.supported && !process.env.PAPERCLIP_TEST_DATABASE_URL)( expect(f.deps.finalizeAgentStatus).toHaveBeenCalledWith(f.agent.id, "succeeded", null, expect.any(Object)); }); + // A chat turn can finish while its task keeps waiting: a resubmitted hire + // approval, an execution stage or a dependency blocker holds the status. + // The accepted final answer must still reach the conversation. Only a card + // the run itself left pending in the thread presents the turn instead. + it.each([ + { hold: "pending approval", reasonCode: "governed_response_waiting", replies: true }, + { hold: "approval sent back for revision", reasonCode: "governed_response_waiting", replies: true }, + { hold: "execution stage", reasonCode: "governed_response_waiting", replies: true }, + { hold: "dependency blocker", reasonCode: "durable_dependency_blocker_bound", replies: true }, + { hold: "finished turn", reasonCode: "conversation_turn_finished", replies: true }, + { hold: "own question card", reasonCode: "governed_response_waiting", replies: false }, + { hold: "execution stage and own question card", reasonCode: "governed_response_waiting", replies: false }, + ] as const)("posts a waiting conversation turn's accepted answer: $hold", async ({ hold, reasonCode, replies }) => { + const f = await fixture({ runtimeMode: "native" }, `CNV${randomUUID().slice(0, 6)}`); + const [issue] = await db.insert(issues).values({ companyId: f.run.companyId, title: "Chat with CEO", + status: "in_progress", assigneeAgentId: f.agent.id, responsibleUserId: "local-board", + conversationAgentId: f.agent.id, conversationUserId: "local-board", conversationState: "active", + ...(hold.startsWith("execution stage") ? { executionState: { status: "pending" } } : {}), + }).returning(); + await db.update(heartbeatRuns).set({ nativeIssueId: issue.id, responsibleUserId: "local-board", + contextSnapshot: { issueId: issue.id, conversationSessionGeneration: 0 } }).where(eq(heartbeatRuns.id, f.run.id)); + if (hold.includes("approval")) { + const [approval] = await db.insert(approvals).values({ companyId: f.run.companyId, type: "hire_agent", + status: hold === "pending approval" ? "pending" : "revision_requested", payload: { name: "Researcher" } }).returning(); + await db.insert(issueApprovals).values({ companyId: f.run.companyId, issueId: issue.id, approvalId: approval.id }); + } + if (hold.endsWith("own question card")) { + await issueThreadInteractionService(db).create(issue, { kind: "ask_user_questions", sourceRunId: f.run.id, + continuationPolicy: "wake_assignee", payload: { version: 1, questions: [{ id: "model", prompt: "Which model?", + selectionMode: "single", required: true, options: [{ id: "flash", label: "Flash" }] }] } }, + { agentId: f.agent.id, runId: f.run.id }); + } + const answer = "Resubmitted the hire with the cheaper model. [Review the pending approval](/CNV/approvals/a1)."; + f.input.issueId = issue.id; + f.input.issueRef = issue; + f.input.issueContext = await createHeartbeatRunPreparation(db).getIssueExecutionContext(f.run.companyId, issue.id); + // The finalizer's saved shape, unchanged from the previous release. + f.input.adapterResult.resultJson = { + nativeResult: { schema: "paperclip.run_result.v1", reportedWorkDisposition: "yielded", + summary: "Waiting for the approval decision.", + continuation: { kind: "response_wake", summary: "Wait for the decision.", idempotencyKey: "hire-wait" } }, + finalizationPhase: "committed", finalizationReasonCode: reasonCode, + }; + f.deps.findLatestCompletedFinalAgentMessage.mockResolvedValue({ seq: 1, text: answer, sourceEventId: "final-event", channel: "final", + reasonCode: "latest_non_empty_completed_final_agent_message" }); + f.deps.issuesSvc.addComment.mockImplementation(async (issueId, body) => { + const [comment] = await db.insert(issueComments).values({ companyId: f.run.companyId, issueId, + authorAgentId: f.agent.id, authorType: "agent", createdByRunId: f.run.id, body }).returning(); + return { ...comment, authorType: "agent" as const }; + }); + + await f.completion.completeRun(f.input); + + const posted = (await db.select().from(issueComments).where(eq(issueComments.issueId, issue.id))) + .filter(comment => comment.createdByRunId === f.run.id); + expect(posted.map(comment => comment.body)).toEqual(replies ? [answer] : []); + expect((await read(f.run.id)).resultJson).toMatchObject({ presentationDecision: replies + ? { chosenSource: "final_agent_message", commentAction: "create", commentId: posted[0]?.id } + : { chosenSource: "none", commentAction: "none" } }); + }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); + it("settles the accepted native exact rate-card price without replacing it with zero", async () => { const f = await fixture({ runtimeMode: "native" }); const runnerInstanceId = randomUUID(); diff --git a/server/src/services/heartbeat/run-completion.ts b/server/src/services/heartbeat/run-completion.ts index 2a24bbe3006..425e681f025 100644 --- a/server/src/services/heartbeat/run-completion.ts +++ b/server/src/services/heartbeat/run-completion.ts @@ -45,6 +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 { 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"; @@ -845,8 +846,15 @@ export function createHeartbeatRunCompletion(db: Db, dependencies: HeartbeatRunC resultJson: persistedResultJson, interactionOwnsResponse: Boolean(issueId && readAcceptedTaskContinuationSummary(persistedResultJson) && await hasChatRunOwnedProviderInteraction(db, { companyId: livenessRun.companyId, issueId, runId: livenessRun.id })), - conversationTurnFinished: isConversation(issueContext) && livenessRun.status === "succeeded" - && persistedResultJson?.finalizationReasonCode === "conversation_turn_finished", + 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, + })))), existingComment: existingRunComment, finalAgentMessage, preferFinalResponseOverExistingComment: diff --git a/server/src/services/native-runtime/native-governance.ts b/server/src/services/native-runtime/native-governance.ts index dd8f2a27835..84b7c5c5b80 100644 --- a/server/src/services/native-runtime/native-governance.ts +++ b/server/src/services/native-runtime/native-governance.ts @@ -24,6 +24,32 @@ export async function lockNativeGovernanceSources(input: { db: Db; companyId: st eq(approvals.companyId, input.companyId))).for("share", { noWait: true }); } +/** The pending thread interaction (question or confirmation card) that gates this run, if any. */ +async function pendingNativeInteraction(input: { + db: Db; + companyId: string; + issueId: string; + runId: string; + ignoredInteractionIds?: string[]; +}): Promise<{ id: string } | null> { + const [issue] = await input.db.select({ conversationAgentId: issues.conversationAgentId, conversationUserId: issues.conversationUserId }) + .from(issues).where(and(eq(issues.id, input.issueId), eq(issues.companyId, input.companyId))); + return input.db + .select({ id: issueThreadInteractions.id }) + .from(issueThreadInteractions) + .where( + and( + eq(issueThreadInteractions.companyId, input.companyId), + eq(issueThreadInteractions.issueId, input.issueId), + eq(issueThreadInteractions.status, "pending"), + ...(input.ignoredInteractionIds?.length ? [notInArray(issueThreadInteractions.id, input.ignoredInteractionIds)] : []), + activeIssueInteractionCondition({ runId: input.runId, conversationMode: isConversation(issue) }), + ), + ) + .limit(1) + .then((rows) => rows[0] ?? null); +} + export async function pendingNativeGovernance(input: { db: Db; companyId: string; @@ -37,23 +63,8 @@ export async function pendingNativeGovernance(input: { if (record(input.executionState).status === "pending") { return { kind: "execution_stage", id: input.runId }; } - const [issue] = await input.db.select({ conversationAgentId: issues.conversationAgentId, conversationUserId: issues.conversationUserId }) - .from(issues).where(and(eq(issues.id, input.issueId), eq(issues.companyId, input.companyId))); const [pendingInteraction, pendingApproval] = await Promise.all([ - input.db - .select({ id: issueThreadInteractions.id }) - .from(issueThreadInteractions) - .where( - and( - eq(issueThreadInteractions.companyId, input.companyId), - eq(issueThreadInteractions.issueId, input.issueId), - eq(issueThreadInteractions.status, "pending"), - ...(input.ignoredInteractionIds?.length ? [notInArray(issueThreadInteractions.id, input.ignoredInteractionIds)] : []), - activeIssueInteractionCondition({ runId: input.runId, conversationMode: isConversation(issue) }), - ), - ) - .limit(1) - .then((rows) => rows[0] ?? null), + pendingNativeInteraction(input), input.db .select({ id: approvals.id }) .from(issueApprovals) @@ -75,3 +86,20 @@ export async function pendingNativeGovernance(input: { return null; } + +/** + * A committed conversation turn has answered the person, whatever status its + * task keeps. An approval or an execution stage can hold the task, but neither + * presents that answer in the conversation. Only a pending thread interaction + * (a question or confirmation card) is the turn's visible response. + */ +export async function conversationTurnAnswered(input: { + db: Db; + companyId: string; + issueId: string; + runId: string; +}): Promise { + // Check cards on their own: an execution stage outranks a card as the + // status gate, but it does not present the turn. + return !(await pendingNativeInteraction(input)); +}