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/doc/execution-semantics.md b/doc/execution-semantics.md index 9409d0d608e..17ee6c4f445 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -330,6 +330,10 @@ persisted cancellations of parked retries through the normal guarded cancellatio path if the original owner stops or its final database write fails. They preserve the original cancellation request and never start a provider turn for that work. +Replaying an audited native Stop also fences a later same-run retry. It preserves +the original intent and acknowledgement. That retry fence does not prove that +the provider stopped or that workspace cleanup finished. + Recovery instructions in a wake describe only an active or escalated recovery action. Resolving or cancelling that action removes its repair instructions from subsequent wakes, so the original worker can resume the task. This prompt change diff --git a/packages/paperclip-runner/README.md b/packages/paperclip-runner/README.md index 24808690a98..02271ebc583 100644 --- a/packages/paperclip-runner/README.md +++ b/packages/paperclip-runner/README.md @@ -485,6 +485,12 @@ stable free-text option IDs. Legacy callers remain supported. Calls that supply both forms must describe the same complete form; partial forms remain invalid. The native recovery bridge uses the same projection for answer delivery. +OpenCode 1.18.34 native questions allow a written answer by default when `custom` +is omitted. The driver preserves that default and explicit `custom:false`; the +native flag takes precedence over the legacy `allowCustom` alias. Choice options +remain intact. Questions without options remain text-only, without a separate +custom-answer field. See the [pinned native question contract](https://github.com/anomalyco/opencode/blob/v1.18.34/packages/schema/src/v1/question.ts). + The server validates canonical answer constraints before persistence. Regex matching runs in isolated workers with a one-second deadline and at most four active workers. A timeout or capacity error leaves the question pending. The diff --git a/packages/paperclip-runner/src/drivers/opencode/opencode-question-normalization.test.ts b/packages/paperclip-runner/src/drivers/opencode/opencode-question-normalization.test.ts new file mode 100644 index 00000000000..9969fc30c0b --- /dev/null +++ b/packages/paperclip-runner/src/drivers/opencode/opencode-question-normalization.test.ts @@ -0,0 +1,37 @@ +import { describe, expect, it } from "vitest"; +import { parsePaperclipQuestionResponse } from "../../contracts/question-set.js"; +import { normalizeOpenCodeQuestionSet } from "./opencode-server-driver.js"; + +const question = { id: "reference", header: "Reference", question: "Which reference should appear in the note?", + options: [{ label: "No reference", description: "Do not include a reference" }] }; +const response = { schema: "paperclip.question_response.v1", answers: { reference: { customText: "Use AMBERexact." } } }; + +describe("OpenCode native written-answer defaults", () => { + it.each([ + { name: "omitted native default", fields: {}, enabled: true }, + { name: "explicit native true", fields: { custom: true }, enabled: true }, + { name: "explicit native false", fields: { custom: false }, enabled: false }, + { name: "legacy alias true", fields: { allowCustom: true }, enabled: true }, + { name: "legacy alias false", fields: { allowCustom: false }, enabled: false }, + { name: "native opt-out takes precedence", fields: { custom: false, allowCustom: true }, enabled: false }, + ])("preserves $name without changing the supplied choice", ({ fields, enabled }) => { + const set = normalizeOpenCodeQuestionSet([{ ...question, ...fields }]); + expect(set.questions[0]).toMatchObject({ id: "reference", prompt: question.question, answerMode: "single_select", + options: [{ id: "option-1", label: "No reference", description: "Do not include a reference" }] }); + if (enabled) { + expect(set.questions[0].customAnswer).toEqual({ enabled: true, label: "Other", placeholder: "Enter another answer" }); + expect(parsePaperclipQuestionResponse(set, response)).toEqual(response); + } else { + expect(set.questions[0]).not.toHaveProperty("customAnswer"); + expect(() => parsePaperclipQuestionResponse(set, response)).toThrow("custom answers are not enabled"); + } + }); + it.each([undefined, true, false])("keeps a question without choices text-only (custom=%s)", custom => { + const set = normalizeOpenCodeQuestionSet([{ ...question, options: [], ...(custom === undefined ? {} : { custom }) }]); + expect(set.questions[0]).toMatchObject({ id: "reference", answerMode: "text" }); + expect(set.questions[0]).not.toHaveProperty("options"); + expect(set.questions[0]).not.toHaveProperty("customAnswer"); + expect(parsePaperclipQuestionResponse(set, { ...response, answers: { reference: { text: "Use AMBERexact." } } })) + .toEqual({ ...response, answers: { reference: { text: "Use AMBERexact." } } }); + }); +}); diff --git a/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.test.ts b/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.test.ts index f89dc89059c..58af69bad4b 100644 --- a/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.test.ts +++ b/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.test.ts @@ -1799,7 +1799,7 @@ describe("OpenCodeServerDriver", () => { await recovered!.session!.close({ reason: "recovery-test" }); }); - it("normalizes native question.asked events and replies with ordered OpenCode answer arrays", async () => { + it("preserves omitted native custom defaults and replies with ordered OpenCode answer arrays", async () => { await chmod(fixture, 0o755); const root = await mkdtemp(join(tmpdir(), "paperclip-opencode-question-")); const workspace = await mkdtemp( diff --git a/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.ts b/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.ts index 45fb78ccefc..6576fe894e7 100644 --- a/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.ts +++ b/packages/paperclip-runner/src/drivers/opencode/opencode-server-driver.ts @@ -2654,6 +2654,7 @@ export function normalizeOpenCodeQuestionSet( ? { description: text(option.description).slice(0, 4_000) } : {}), })); + const custom = question.custom ?? question.allowCustom ?? true; return { id: openCodeQuestionId(question, index), ...(text(question.header) @@ -2674,7 +2675,7 @@ export function normalizeOpenCodeQuestionSet( ? "multi_select" : "single_select", ...(options.length > 0 ? { options } : {}), - ...(question.custom === true || question.allowCustom === true + ...(options.length > 0 && custom === true ? { customAnswer: { enabled: true, diff --git a/packages/paperclip-runner/test/fixtures/fake-opencode-server.mjs b/packages/paperclip-runner/test/fixtures/fake-opencode-server.mjs index 3b03d4d52e2..38da64e96a1 100644 --- a/packages/paperclip-runner/test/fixtures/fake-opencode-server.mjs +++ b/packages/paperclip-runner/test/fixtures/fake-opencode-server.mjs @@ -41,7 +41,6 @@ function nativeQuestion() { description: "Deploy directly to production.", }, ], - custom: true, }, { id: "regions", 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; } diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index fea3271f66d..59d7a94e408 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -99,6 +99,7 @@ import { buildNativeHeartbeatPreparationSpans, createNativeRunTrace } from "./na import { NativeRunnerOwnershipUnverifiedError } from "./native-runner-ownership.js"; import type { AdapterRuntimeEvent } from "../../adapters/index.js"; import type { AdapterUsageCheckpoint } from "@paperclipai/adapter-utils"; +import { acknowledgedNativeStopExecutionHasStopped, isAcknowledgedNativeStop } from "../acknowledged-native-stop.js"; vi.mock("node:fs", async (importOriginal) => { const actual = await importOriginal(); @@ -7852,6 +7853,27 @@ describe("native session cancellation", () => { expect(state.persistActivity).not.toHaveBeenCalled(); }); + it.each([ + { companyId: "another-company" }, { runId: "another-run" }, { issueId: "another-issue" }, + { scope: "issue" }, { effects: ["cancel_issue"] }, { intentAuditId: null }, + ])("does not fence a failed retry for a foreign or stale Stop receipt %j", async changed => { + const intent = { schema: "paperclip.native-cancellation.v1", intentId: "native-cancellation:earlier-stop", + ...execution.binding, scope: "run", reasonCode: "cancellation_run_only", effects: ["release_run_resources"], + dispatchState: "acknowledged", dispatched: true, intentAuditId: "earlier-intent-audit", + acknowledgementAuditId: "earlier-ack-audit", ...changed }; + const coordinator = { runId: execution.binding.runId, assessmentId: null, phase: "retryable_failure" }; + const persistence = cancellationDb({ status: "failed", coordinator, resultJson: { nativeCancellation: intent } }); + const before = structuredClone(persistence.getResultJson()); + state.cancel.mockClear(); + state.persistActivity.mockClear(); + await expect(cancelNativeSession(execution.binding.runId, "retry Stop", { db: persistence.db, scope: "run" })).rejects.toThrow(); + expect(persistence.updates).toEqual([]); + expect(coordinator.phase).toBe("retryable_failure"); + expect(persistence.getResultJson()).toEqual(before); + expect(state.cancel).not.toHaveBeenCalled(); + expect(state.persistActivity).not.toHaveBeenCalled(); + }); + it("does not mistake a recovered running attempt for the failed retry that reserved Stop", async () => { const id = "11111111-1111-4111-8111-111111111111"; const persistence = cancellationDb({ status: "running", @@ -7877,6 +7899,39 @@ describe("native session cancellation", () => { expect(persistence.getResultJsonUpdateCount()).toBe(writes); }); + it("fences a failed retry whose earlier Stop is already acknowledged without changing its receipts", async () => { + const intent = { schema: "paperclip.native-cancellation.v1", intentId: "native-cancellation:earlier-stop", + ...execution.binding, scope: "run", reasonCode: "cancellation_run_only", + effects: ["release_run_resources"], dispatchState: "acknowledged", dispatched: true, + intentAuditId: "earlier-intent-audit", acknowledgementAuditId: "earlier-ack-audit", + acknowledgedAt: "2026-01-01T00:00:00.000Z", decisionId: null, priorCoordinatorDecisionId: null }; + const coordinator = { runId: execution.binding.runId, assessmentId: null, phase: "retryable_failure", + nextAttemptAt: new Date(Date.now() + 30_000) }; + const persistence = cancellationDb({ status: "failed", coordinator, + resultJson: { nativeCancellation: intent, executionCancellation: { state: "requested", requestedAt: "later-stop" } } }); + const before = structuredClone(persistence.getResultJson()); + state.cancel.mockClear(); + state.persistActivity.mockClear(); + for (let attempt = 0; attempt < 2; attempt++) { + await expect(cancelNativeSession(execution.binding.runId, "retry Stop", { db: persistence.db, scope: "run" })) + .resolves.toMatchObject({ dispatched: true, auditId: intent.intentAuditId }); + expect(coordinator).toMatchObject({ phase: "terminal_failure", failureCode: "native_retry_cancelled", nextAttemptAt: null }); + expect(persistence.getResultJson()).toEqual(before); + } + expect(persistence.updates.filter(update => update.table === nativeRunFinalizations)).toHaveLength(1); + expect(persistence.getResultJsonUpdateCount()).toBe(0); + expect(state.cancel).not.toHaveBeenCalled(); + expect(state.persistActivity).not.toHaveBeenCalled(); + const stopped = { id: execution.binding.runId, companyId: execution.binding.companyId, + nativeIssueId: execution.binding.issueId, status: "cancelled", processPid: null, processGroupId: null, + resultJson: { ...persistence.getResultJson(), cancelledByActorType: "user", cancelledByUserId: "operator" } }; + const lease = { provider: "daytona", providerLeaseId: "unconfirmed-provider-resource", status: "active", + releasedAt: null, cleanupStatus: null, metadata: {} }; + const leaseDb = { select: () => ({ from: () => ({ where: async () => [lease] }) }) } as unknown as Db; + expect(isAcknowledgedNativeStop(stopped)).toBe(true); + await expect(acknowledgedNativeStopExecutionHasStopped(leaseDb, stopped)).resolves.toBe(false); + }); + it("uses the reserved caller UUID and permits same-intent and default retries", async () => { const id = "11111111-1111-4111-8111-111111111111"; const persistence = cancellationDb({ resultJson: { startupCancellation: { cancellationRequestId: id } } }); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 65ea9f0c089..9cec1870142 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -7541,6 +7541,27 @@ export async function cancelNativeSession( runId, companyId: cancellationContext.companyId, issueId: cancellationContext.issueId, resultJson, requestId: requestedId, scope: options.scope ?? "run", }, coordinator)) throw conflict("Native retry is no longer cancellable"); + const fenceRetryCancellation = async () => { + if (coordinator.phase !== "retryable_failure") return; + await tx + .update(nativeRunFinalizations) + .set({ + phase: "terminal_failure", + nextAttemptAt: null, + leaseOwner: null, + leaseExpiresAt: null, + recoveryState: "blocked", + failureCode: "native_retry_cancelled", + updatedAt: new Date(), + }) + .where( + and( + eq(nativeRunFinalizations.runId, runId), + eq(nativeRunFinalizations.companyId, cancellationContext.companyId), + eq(nativeRunFinalizations.phase, "retryable_failure"), + ), + ); + }; const existing = record(resultJson.nativeCancellation); const existingIntentId = typeof existing.intentId === "string" && existing.intentId.length > 0 @@ -7564,6 +7585,10 @@ export async function cancelNativeSession( : null; if (!existingAuditId) throw new Error("native_cancellation_intent_audit_missing"); + if (retryCancellation && coordinator.phase === "retryable_failure") { + await notifyNativeResultCommit(tx, { companyId: cancellationContext.companyId, runId }); + await fenceRetryCancellation(); + } return { intentId: existingIntentId, auditId: existingAuditId, @@ -7643,24 +7668,7 @@ export async function cancelNativeSession( // Cancellation is also an authority fence for a durable retry whose // preceding provider has already failed. No in-memory session is needed. await notifyNativeResultCommit(tx, { companyId: cancellationContext.companyId, runId }); - await tx - .update(nativeRunFinalizations) - .set({ - phase: "terminal_failure", - nextAttemptAt: null, - leaseOwner: null, - leaseExpiresAt: null, - recoveryState: "blocked", - failureCode: "native_retry_cancelled", - updatedAt: new Date(), - }) - .where( - and( - eq(nativeRunFinalizations.runId, runId), - eq(nativeRunFinalizations.companyId, cancellationContext.companyId), - eq(nativeRunFinalizations.phase, "retryable_failure"), - ), - ); + await fenceRetryCancellation(); intentPublication = activity.publication; return { intentId, diff --git a/tests/runner-e2e/chat-feature-testops-api.mjs b/tests/runner-e2e/chat-feature-testops-api.mjs index 704dd5842a7..9b2b1ddfc3f 100644 --- a/tests/runner-e2e/chat-feature-testops-api.mjs +++ b/tests/runner-e2e/chat-feature-testops-api.mjs @@ -379,9 +379,13 @@ export async function verifyChatUpload(api, env, selected, exported, { wait = ms const result = await get(`/api/testresult/${row.id}`); if (result.id !== row.id || result.projectId !== 1 || result.launchId !== selected.launchId || result.testCaseId !== selected.caseId || result.fullName !== selected.fullName || result.jobRun?.id !== Number(selected.allureJobRunId) - || result.jobRun?.job?.id !== selected.jobId || result.jobRun?.launchId !== selected.launchId || result.jobRun?.url !== runUrl - || !Array.isArray(result.links) || !result.links.some(link => link.url === runUrl) - || !result.links.some(link => link.url === sourceUrl)) fail('chat_testops_upload_mismatch'); + || result.jobRun?.launchId !== selected.launchId || result.jobRun?.url !== runUrl + || result.jobRun.job != null && (result.jobRun.job.id !== run.job.id + || result.jobRun.job.type != null && result.jobRun.job.type !== run.job.type + || result.jobRun.job.url != null && !publicJobUrl(result.jobRun.job.url, env)) + || !Array.isArray(result.links) || result.links.some(link => typeof link.url === 'string' + && (link.url.startsWith('https://github.com/paperclipai/paperclip/actions/runs/') && link.url !== runUrl + || link.url.startsWith('https://github.com/paperclipai/paperclip/commit/') && link.url !== sourceUrl))) fail('chat_testops_upload_mismatch'); if (!['passed', 'failed', 'broken', 'skipped', 'unknown'].includes(result.status)) fail('chat_testops_upload_mismatch'); if (result.status !== 'unknown' && result.status !== exported.status) fail('chat_testops_upload_mismatch'); const attached = await get(`/api/testresult/attachment?${new URLSearchParams({ testResultId: String(row.id), page: '0', size: '100' })}`); diff --git a/tests/runner-e2e/chat-feature-testops-api.test.mjs b/tests/runner-e2e/chat-feature-testops-api.test.mjs index 4422a6e0d0b..9b2bc7c74f7 100644 --- a/tests/runner-e2e/chat-feature-testops-api.test.mjs +++ b/tests/runner-e2e/chat-feature-testops-api.test.mjs @@ -322,11 +322,11 @@ const uploadReceipt = { status: 'passed', cleanup: 'passed', caseId: 91, campaig { name: 'Execution provenance and checks', contentType: 'application/json', contentLength: 300 }, { name: 'Sanitized Product E2E record and accounting', contentType: 'application/json', contentLength: 800 }] }; const uploadRun = { id: 9, job: { id: 7, type: 'github' }, launchId: 10, url: 'https://github.com/paperclipai/paperclip/actions/runs/123', stage: 'finished' }; -function uploadApi(change = {}, absent = 0) { +function uploadApi(change = {}, absent = 0, runChange = {}) { let polls = 0; return async (route, method) => { assert.equal(method, 'GET'); - if (route === '/api/jobrun/9') return uploadRun; + if (route === '/api/jobrun/9') return { ...uploadRun, ...runChange }; if (route === '/api/launch/10') return { id: 10, projectId: 1 }; if (route.startsWith('/api/testresult?')) { polls++; return { content: polls <= absent ? [] : [{ id: 45, testCaseId: 91 }], totalElements: polls <= absent ? 0 : 1, totalPages: polls <= absent ? 0 : 1 }; } if (route === '/api/testresult/45') return { id: 45, projectId: 1, launchId: 10, testCaseId: 91, fullName, jobRun: uploadRun, status: 'passed', @@ -348,11 +348,45 @@ test('readback verifies exact current launch/case/source/status and safe attachm test('readback rejects wrong launch, case, current jobrun, source and contradictory result status', async () => { const { verifyChatUpload } = await import('./chat-feature-testops-api.mjs'); - for (const change of [{ launchId: 11 }, { testCaseId: 92 }, { jobRun: { ...uploadRun, id: 11 } }, { links: [] }, { status: 'failed' }]) { + for (const change of [{ launchId: 11 }, { testCaseId: 92 }, { jobRun: { ...uploadRun, id: 11 } }, { links: [{ url: 'https://github.com/paperclipai/paperclip/commit/' + 'b'.repeat(40) }] }, { status: 'failed' }]) { await assert.rejects(verifyChatUpload(uploadApi(change), uploadContext, uploadSelection, uploadReceipt), /upload_mismatch/); } }); +test('readback joins an unexpanded result jobrun to the authoritative jobrun response', async () => { + const { verifyChatUpload } = await import('./chat-feature-testops-api.mjs'); + const { job, ...embedded } = uploadRun; + const result = await verifyChatUpload(uploadApi({ jobRun: embedded }), uploadContext, uploadSelection, uploadReceipt); + assert.equal(result.resultId, 45); assert.equal(result.jobId, 7); assert.equal(result.verification, 'VERIFIED'); +}); + +test('readback accepts protected empty presentation links while retaining the source-bound current run', async () => { + const { verifyChatUpload } = await import('./chat-feature-testops-api.mjs'); + const { job, ...embedded } = uploadRun; + const result = await verifyChatUpload(uploadApi({ jobRun: embedded, links: [] }), uploadContext, uploadSelection, uploadReceipt); + assert.equal(result.targetSha, uploadContext.GITHUB_SHA); assert.equal(result.runId, '123'); + assert.equal(result.resultId, 45); assert.equal(result.verification, 'VERIFIED'); +}); + +test('readback rejects authoritative or embedded identity contradictions and a foreign source selection', async () => { + const { verifyChatUpload } = await import('./chat-feature-testops-api.mjs'); + const { job, ...embedded } = uploadRun; + for (const runChange of [{ job: { id: 8, type: 'github' } }, { job: { id: 7, type: 'gitlab' } }, + { url: 'https://github.com/paperclipai/paperclip/actions/runs/124' }, { launchId: 11 }]) { + await assert.rejects(verifyChatUpload(uploadApi({ jobRun: embedded, links: [] }, 0, runChange), uploadContext, uploadSelection, uploadReceipt), /jobrun_mismatch/); + } + for (const change of [{ jobRun: { ...embedded, id: 11 } }, { jobRun: { ...embedded, job: { id: 8 } } }, + { jobRun: { ...embedded, job: { id: 7, type: 'gitlab' } } }, { jobRun: { ...embedded, job: { id: 7, url: 'https://github.com/other/repo' } } }, + { jobRun: { ...embedded, launchId: 11 } }, { jobRun: { ...embedded, url: 'https://github.com/paperclipai/paperclip/actions/runs/124' } }, + { links: [{ url: 'https://github.com/paperclipai/paperclip/actions/runs/124' }] }, + { links: [{ url: 'https://github.com/paperclipai/paperclip/commit/' + 'b'.repeat(40) }] }]) { + await assert.rejects(verifyChatUpload(uploadApi(change), uploadContext, uploadSelection, uploadReceipt), /upload_mismatch/); + } + for (const selection of [{ ...uploadSelection, targetSha: 'b'.repeat(40) }, { ...uploadSelection, workflowSha: 'b'.repeat(40) }]) { + await assert.rejects(verifyChatUpload(uploadApi({ jobRun: embedded, links: [] }), uploadContext, selection, uploadReceipt), /upload_mismatch/); + } +}); + test('readback missing result or screenshot metadata is bounded unverified, never a fabricated pass', async () => { const { verifyChatUpload } = await import('./chat-feature-testops-api.mjs'); let waits = 0; diff --git a/ui/src/components/task-chat/TaskChatComposerDock.test.tsx b/ui/src/components/task-chat/TaskChatComposerDock.test.tsx new file mode 100644 index 00000000000..68ba2e5623c --- /dev/null +++ b/ui/src/components/task-chat/TaskChatComposerDock.test.tsx @@ -0,0 +1,117 @@ +// @vitest-environment jsdom + +import { flushSync } from "react-dom"; +import { createRoot, type Root } from "react-dom/client"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { TaskChatComposerDock } from "./TaskChatComposerDock"; + +describe("TaskChatComposerDock", () => { + let container: HTMLDivElement; + let root: Root; + + beforeEach(() => { + container = document.createElement("div"); + document.body.appendChild(container); + root = createRoot(container); + }); + + afterEach(() => { + flushSync(() => root.unmount()); + container.remove(); + vi.restoreAllMocks(); + }); + + function render(coordinator = true) { + const onTranscriptWheel = vi.fn(); + flushSync(() => root.render( +
+
+
+ +