From 87dfba6818dca415f4e350de2af9e4655faf132e Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Thu, 5 Nov 2026 01:19:04 -0600 Subject: [PATCH] fix(native): retry an unpublished accepted task reply durably instead of dropping it (#19200) 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 > - A native run commits its status decision and its final reply comment in separate steps > - Background presentation recovery publishes an accepted reply when the live path does not record one > - That worker dropped the run after one refused attempt and only threw on an error, with no record on the run > - A task could then show done while the user never saw the answer, until an unrelated wake reloaded the queue > - This pull request makes the publication of an accepted ordinary task reply durable, bounded, and recorded > - The benefit is that a task cannot close silently without its answer ## Linked Issues or Issue Description Refs #19130 (verify requested publication at finalization, for documents), #19183 (deliver messages saved while a native run finalizes), #19186 (stale owner done). This change does not duplicate them. It covers the final reply comment. **What happened?** A native run accepted a result with disposition `done`. The status decision set the task to done. The accepted final reply was not published as a task comment. The run had no publication record (no `run.presentation.resolved` event and no `presentationDecision`). The reply was published by background recovery 58 minutes later, after an unrelated wake reloaded the presentation queue. **Expected behavior** The accepted reply is published promptly. A refused or failed attempt is retried with backoff and never posts twice. If publication fails permanently, the run and the task show a named failure with an owner. **Steps to reproduce** 1. Commit an accepted native task result with disposition `done`, with the final reply not yet visible to presentation recovery. 2. Run the native response presentation worker check once. The attempt is refused. 3. On `master`, the worker returns `null` and drops the run. Nothing records the attempt. Nothing retries it until another notification for that scope. **Paperclip version or commit** `master` at 3b17b5ce15. **Deployment mode** Cloud staging, native `paperclip_runner` agent. Root cause notes: the staging evidence shows that the live completion path did not reach its presentation step for the affected run (no `lifecycle run succeeded` event). Staging server logs for that window are not available, so the exact first refusal is not proven. The worker defect is proven: one refusal makes `check()` return `null`, which deletes the scheduled entry. One thrown error escapes with only in-memory backoff and no record. The fix closes that gap for every refusal or error cause. ## What Changed - `native-response-presentation-work.ts`: when an accepted ordinary task reply (`done` or `needs_review`, not an Agent Chat conversation) is still unpublished after a due attempt, the worker records the attempt and keeps the run scheduled. A thrown error for this class is recorded instead of escaping. When the reply is published, an open retry receipt is set to `resolved`. - New `native-response-publication-receipt.ts`: stores the receipt in `heartbeat_runs.result_json.responsePublication` (attempts, last reason, last error, next attempt). Backoff is 30 s doubling to at most 5 min, for 6 attempts. Each retry appends a `run.response_publication.retrying` run event. After the last attempt it records `native_response_publication_failed` with the run's agent as owner, a `run.response_publication.failed` run event, and an `issue.response_publication_failed` task activity. - `native-response-presentation-query.ts`: the candidate query exposes the saved retry deadline, so a restarted worker reconstructs the schedule. It excludes a run whose publication is recorded as failed, so retries stay bounded. - An open retry receipt is settled to `resolved` in the same write that records the reply's `commentId` (background recovery and the live `recordRunPresentationDecision`). A candidate that disappears for another reason leaves the receipt unchanged, so the bound cannot reset. - When the reply comment already exists but its decision was not recorded, recovery now records it as `reuse` with that `commentId`. It does not post again and does not count a refusal. - Under the run lock, an attempt is not counted again when another worker already saved a future deadline. The worker also honors that saved deadline after recovery. - Idempotency is unchanged: the comment is written only while the presentation decision has no `commentId`, under the existing coordinator lock. The receipt write never overwrites a run that already has a `commentId`. - `doc/execution-semantics.md`: documents the contract. ## Verification - `cd server && npx vitest run src/services/native-runtime/native-response-publication-retry.test.ts` - Red on `master` (3 failed): after one refusal `check()` returned `null` (dropped); one injected transaction error escaped with no record; the refusal loop never stopped (more than 10 attempts, no failure recorded). - Green with this change (5 passed, including an already posted reply and two concurrent workers): one refusal keeps the run scheduled and recorded, a reloaded worker keeps the same deadline, the reply posts exactly once after it becomes visible, and repeated checks do not double-post; one injected error is recorded and retried; permanent refusal stops after bounded attempts with the named, owned failure on the run event and task activity. - Related suites pass: `agent-chat-response-recovery.test.ts`, `presentation-ownership-read.test.ts`, `idle-native-finalization.test.ts`, `chat-monitor-continuation.test.ts`, `native-preserved-monitor-wait.test.ts`, `paperclip-control-plane-port.test.ts`, `chat-channels.integration.test.ts`, `photon/channel.integration.test.ts` (2047 of 2049 passed in one combined run). One local failure needs a bundled Slack skill file that is missing in this worktree. The other ("materializes direct external-chat finals once") failed once under load and then passed three times in isolation and in the combined set with this change. - Not run: a live staging run. The change has not been deployed. ## Risks - Behavior change: an ordinary task reply that recovery keeps refusing now gets a recorded failure and a task activity after about 12 minutes. Before, it was dropped silently. A persistent integrity decline (for example a digest mismatch) now becomes visible this way. This is intended. - The status decision still commits before the reply. This PR does not hold `done` on the comment. It makes a missing reply visible with an owner. - Agent Chat conversations and yielded chat responses keep their existing behavior and their completion outbox owner. - No schema migration. The receipt lives in the existing `result_json` column. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use and code execution 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 - [x] All Paperclip CI gates are green - [x] 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 | 13 ++ .../src/services/heartbeat/run-completion.ts | 4 +- .../native-response-presentation-query.ts | 8 + .../native-response-presentation-work.ts | 36 +++- .../native-response-publication-receipt.ts | 136 +++++++++++++ .../native-response-publication-retry.test.ts | 178 ++++++++++++++++++ .../native-runtime/native-run-finalizer.ts | 14 +- 7 files changed, 382 insertions(+), 7 deletions(-) create mode 100644 server/src/services/native-runtime/native-response-publication-receipt.ts create mode 100644 server/src/services/native-runtime/native-response-publication-retry.test.ts diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index bb85e011b98..aae58c6b2a7 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -242,6 +242,19 @@ and the matching server-issued scheduling receipt on the source run. Company, task, agent, requester, schedule identity, and current delivery authorization remain enforced. Provider-supplied source IDs do not establish that lineage. +An accepted ordinary task result (`done` or `needs_review`) commits its status +decision independently of its final reply comment. If the reply is not yet +published when background presentation recovery is due, a refused or failed +attempt is recorded on the run as `resultJson.responsePublication` and retried +with bounded backoff (six attempts, at most five minutes apart). The reply stays +idempotent through the presentation decision's `commentId`, so a retry never +posts twice. Exhausted retries record the named failure +`native_response_publication_failed` with the run's agent as owner, a +`run.response_publication.failed` run event, and an +`issue.response_publication_failed` task activity, so the task cannot close +silently without its answer. Agent Chat conversations keep their completion +outbox as the bounded owner. + An accepted same-agent continuation retains the same presentation contract. Its successor inherits conversation provenance only through the applied status decision, delivered continuation effect, and admitted wake that name the same diff --git a/server/src/services/heartbeat/run-completion.ts b/server/src/services/heartbeat/run-completion.ts index 12dde035df8..fa2fbf051c9 100644 --- a/server/src/services/heartbeat/run-completion.ts +++ b/server/src/services/heartbeat/run-completion.ts @@ -1,6 +1,7 @@ import { preserveAcceptedRunUsageProjectionSql } from "../usage-receipts.js"; import { providerFailureFromError } from "../../vendor/paperclip-runner/index.js"; import { nativeTaskSessionPublicationHandled } from "../native-runtime/native-task-session-publication.js"; +import { settledResponsePublicationSql } from "../native-runtime/native-response-publication-receipt.js"; import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; import { measureRunPhase } from "../execution-preparation-timing.js"; import { PROJECT_ACCESS_DENIED, ProjectAccessDeniedError } from "../project-access-denial.js"; @@ -359,7 +360,8 @@ export async function recordRunPresentationDecision(db: Db, input: { await db.transaction(async tx => { await notifyChatRunMilestones(tx, { companyId: input.companyId, runId: input.runId, issueId: input.issueId }); await tx.update(heartbeatRuns).set({ - resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({ presentationDecision: input.decision })}::jsonb`, + resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({ presentationDecision: input.decision })}::jsonb + || ${typeof input.decision.commentId === "string" ? settledResponsePublicationSql() : sql`'{}'::jsonb`}`, updatedAt: new Date(), }).where(eq(heartbeatRuns.id, input.runId)); }); 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 bb2bbf129c7..87da12c4ff9 100644 --- a/server/src/services/native-runtime/native-response-presentation-query.ts +++ b/server/src/services/native-runtime/native-response-presentation-query.ts @@ -8,6 +8,11 @@ export function nativeResponsePresentationCandidates(scope?: NativeResponsePrese return db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, issueId: nativeRunFinalizations.issueId, agentId: heartbeatRuns.agentId, nextAttemptAt: nativeRunFinalizations.nextAttemptAt, endpointId: chatActions.endpointId, principalId: chatActions.principalId, + // Durable publication retry schedule for an ordinary final task reply. + publicationNextAttemptAt: sql`${heartbeatRuns.resultJson} #>> '{responsePublication,nextAttemptAt}'`.as("publication_next_attempt_at"), + // Agent Chat conversations keep their completion outbox as the bounded owner. + taskReply: sql`coalesce(${nativeRunResults.resultJson} #>> '{result,reportedWorkDisposition}' in ('done', 'needs_review') + and (coalesce(${issues.conversationAgentId}::text, '') = '' or coalesce(${issues.conversationUserId}, '') = ''), false)`.as("task_reply"), }).from(nativeRunFinalizations) .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId), eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.nativeIssueId, nativeRunFinalizations.issueId))) @@ -26,6 +31,9 @@ export function nativeResponsePresentationCandidates(scope?: NativeResponsePrese eq(nativeRunResults.schemaStatus, "accepted"), sql`jsonb_typeof(${heartbeatRuns.resultJson} #> '{presentationDecision,commentId}') is distinct from 'string'`, sql`coalesce(${heartbeatRuns.resultJson} #> '{presentationDecision,commentAction}', '"create"'::jsonb) = '"create"'::jsonb`, + // A reply whose bounded publication retries are exhausted stays visible + // as a named failure instead of being retried forever. + sql`${heartbeatRuns.resultJson} #>> '{responsePublication,state}' is distinct from 'failed'`, sql`${heartbeatRuns.contextSnapshot}->'skipIssueComment' is distinct from 'true'::jsonb`, sql`coalesce(${statusDecisions.decisionJson}->'externalChatReviewPresentation', 'null'::jsonb) in ('null'::jsonb, 'false'::jsonb, '0'::jsonb, '""'::jsonb)`, sql`${nativeRunResults.resultJson} #>> '{result,schema}' = 'paperclip.run_result.v1'`, diff --git a/server/src/services/native-runtime/native-response-presentation-work.ts b/server/src/services/native-runtime/native-response-presentation-work.ts index 50c89162c32..3eab73135ff 100644 --- a/server/src/services/native-runtime/native-response-presentation-work.ts +++ b/server/src/services/native-runtime/native-response-presentation-work.ts @@ -8,6 +8,7 @@ import { onCommittedChatResponseAuthorityRegistered } from "../durable-chat-wake import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; import { recoverCommittedNativeChatResponse } from "./native-run-finalizer.js"; import { measureRunPhase, withRunResponsePresentationTiming } from "../execution-preparation-timing.js"; +import { recordNativeResponsePublicationAttempt } from "./native-response-publication-receipt.js"; export function registerNativeResponsePresentationWork(db: Db, coordinator: ReturnType, canRun: () => boolean) { @@ -16,6 +17,12 @@ export function registerNativeResponsePresentationWork(db: Db, type Entry = { id: string; companyId: string; issueId: string; agentId: string; endpointId?: string | null; principalId?: string | null; dueAt: number }; +type Candidate = { nextAttemptAt: Date | null; publicationNextAttemptAt: string | null }; +function dueAt(row: Candidate) { + const published = row.publicationNextAttemptAt ? Date.parse(row.publicationNextAttemptAt) : 0; + return Math.max(row.nextAttemptAt?.getTime() ?? 0, Number.isFinite(published) ? published : 0); +} + export function nativeResponsePresentationWork(db: Db) { const candidates = (scope?: NativeResponsePresentationScope, after?: string, runId?: string) => { const query = nativeResponsePresentationCandidates(scope, after, runId).as("pending_native_response"); @@ -31,7 +38,7 @@ export function nativeResponsePresentationWork(db: Db) { let after: string | undefined; while (canRun() && !signal?.aborted) { const rows = await measureRunPhase("response.resolve", () => candidates(scope, after)); - for (const row of rows) entries.push({ id: row.id, companyId: row.companyId, issueId: row.issueId, agentId: row.agentId, endpointId: row.endpointId, principalId: row.principalId, dueAt: row.nextAttemptAt?.getTime() ?? 0 }); + for (const row of rows) entries.push({ id: row.id, companyId: row.companyId, issueId: row.issueId, agentId: row.agentId, endpointId: row.endpointId, principalId: row.principalId, dueAt: dueAt(row) }); if (rows.length < 100) return entries; after = rows[rows.length - 1].id; } @@ -45,13 +52,32 @@ export function nativeResponsePresentationWork(db: Db) { const scope = { companyId: entry.companyId, issueId: entry.issueId }; const [current] = await measureRunPhase("response.resolve", () => candidates(scope, undefined, entry.id)); if (!current) return null; - if (current.nextAttemptAt && current.nextAttemptAt.getTime() > Date.now()) return { ...entry, dueAt: current.nextAttemptAt.getTime() }; + if (dueAt(current) > Date.now()) return { ...entry, dueAt: dueAt(current) }; if (signal?.aborted) throw new Error("Native response presentation recovery stopped"); const recoveryStartedAt = Date.now(); - await recoverCommittedNativeChatResponse(db, { ...scope, runId: entry.id, onlyWhenDue: true }); + let failure: unknown = null; + try { + await recoverCommittedNativeChatResponse(db, { ...scope, runId: entry.id, onlyWhenDue: true }); + } catch (error) { + // Ordinary task replies own a durable, bounded retry below. Other + // presentation classes keep the coordinator's in-memory backoff. + if (!current.taskReply) throw error; + failure = error; + } const [remaining] = await measureRunPhase("response.resolve", () => candidates(scope, undefined, entry.id)); - return remaining?.nextAttemptAt && remaining.nextAttemptAt.getTime() > recoveryStartedAt - ? { ...entry, dueAt: remaining.nextAttemptAt.getTime() } : null; + // A published reply settles its retry receipt in the same write that + // records the comment, so a vanished candidate needs no cleanup here. + if (!remaining) return null; + // The coordinator, or another worker's saved retry, already moved the deadline. + if (dueAt(remaining) > recoveryStartedAt) return { ...entry, dueAt: dueAt(remaining) }; + if (!remaining.taskReply) return null; + // The accepted reply is still unpublished after a due attempt. Never drop + // it silently: schedule a bounded retry or record the named failure. + const attempt = await recordNativeResponsePublicationAttempt(db, { + companyId: entry.companyId, issueId: entry.issueId, runId: entry.id, + reason: failure ? "publication_error" : "publication_refused", error: failure, + }); + return attempt?.state === "retrying" ? { ...entry, dueAt: attempt.nextAttemptAt } : null; } function register(coordinator: ReturnType, canRun: () => boolean) { const worker = registerScopedDeadlineWork(coordinator, { diff --git a/server/src/services/native-runtime/native-response-publication-receipt.ts b/server/src/services/native-runtime/native-response-publication-receipt.ts new file mode 100644 index 00000000000..b3e3095f04f --- /dev/null +++ b/server/src/services/native-runtime/native-response-publication-receipt.ts @@ -0,0 +1,136 @@ +import { and, eq, sql } from "drizzle-orm"; +import { heartbeatRuns, type Db } from "@paperclipai/db"; +import { logActivity } from "../activity-log.js"; +import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; + +/** + * Durable, bounded publication of an accepted native task reply. + * + * The status decision for an accepted result commits independently of the + * final reply comment. When publication is refused or fails, the attempt is + * recorded on the run and retried with backoff. Exhausted retries leave a + * named failure with an owner on the run and the task instead of a silent + * done. The comment write itself stays idempotent through the presentation + * decision's `commentId`, so retries never double-post. + */ +export const NATIVE_RESPONSE_PUBLICATION_SCHEMA = "paperclip.native_response_publication.v1"; +export const NATIVE_RESPONSE_PUBLICATION_FAILED = "native_response_publication_failed"; +export const NATIVE_RESPONSE_PUBLICATION_MAX_ATTEMPTS = 6; +const BASE_DELAY_MS = 30_000; +const MAX_DELAY_MS = 300_000; + +export type NativeResponsePublicationReason = "publication_refused" | "publication_error"; + +type Receipt = { + schema: typeof NATIVE_RESPONSE_PUBLICATION_SCHEMA; + state: "retrying" | "failed" | "resolved"; + attempts: number; + firstAttemptAt: string; + lastAttemptAt: string; + lastReason: NativeResponsePublicationReason; + lastError?: string | null; + nextAttemptAt?: string | null; + reasonCode?: string; + owner?: { kind: "agent"; agentId: string }; + failedAt?: string; + resolvedAt?: string; +}; + +export function nativeResponsePublicationDelayMs(attempts: number) { + return Math.min(MAX_DELAY_MS, BASE_DELAY_MS * 2 ** Math.max(0, attempts - 1)); +} + +function readReceipt(resultJson: unknown): Receipt | null { + if (!resultJson || typeof resultJson !== "object") return null; + const receipt = (resultJson as Record).responsePublication; + return receipt && typeof receipt === "object" ? receipt as Receipt : null; +} + +function errorText(error: unknown) { + if (!error) return null; + const text = error instanceof Error ? error.message : String(error); + return text.slice(0, 500); +} + +function patchReceipt(tx: Db, runId: string, receipt: Receipt) { + // Never overwrite a materialized reply's run record with retry bookkeeping. + return tx.update(heartbeatRuns).set({ + resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) + || jsonb_build_object('responsePublication', ${JSON.stringify(receipt)}::jsonb)`, + updatedAt: new Date(), + }).where(and(eq(heartbeatRuns.id, runId), + sql`jsonb_typeof(${heartbeatRuns.resultJson} #> '{presentationDecision,commentId}') is distinct from 'string'`)) + .returning({ id: heartbeatRuns.id }); +} + +/** Record one unsuccessful attempt. Returns the next due time, or null once + * the bounded retry is exhausted and the failure has been recorded. */ +export async function recordNativeResponsePublicationAttempt(db: Db, input: { + companyId: string; issueId: string; runId: string; + reason: NativeResponsePublicationReason; error?: unknown; +}): Promise<{ state: "retrying"; nextAttemptAt: number } | { state: "failed" } | null> { + return db.transaction(async transaction => { + const tx = transaction as unknown as Db; + const [run] = await tx.select({ id: heartbeatRuns.id, agentId: heartbeatRuns.agentId, resultJson: heartbeatRuns.resultJson }) + .from(heartbeatRuns) + .where(and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId))) + .for("update"); + if (!run) return null; + const previous = readReceipt(run.resultJson); + if (previous?.state === "failed") return { state: "failed" as const }; + const now = new Date(); + // Another worker already counted this due attempt and saved the next + // deadline. Keep its schedule instead of counting the attempt twice. + const scheduled = previous?.state === "retrying" && previous.nextAttemptAt ? Date.parse(previous.nextAttemptAt) : NaN; + if (Number.isFinite(scheduled) && scheduled > now.getTime()) return { state: "retrying" as const, nextAttemptAt: scheduled }; + const attempts = (previous?.state === "retrying" ? previous.attempts : 0) + 1; + const base = { + schema: NATIVE_RESPONSE_PUBLICATION_SCHEMA as typeof NATIVE_RESPONSE_PUBLICATION_SCHEMA, + attempts, + firstAttemptAt: previous?.state === "retrying" ? previous.firstAttemptAt : now.toISOString(), + lastAttemptAt: now.toISOString(), + lastReason: input.reason, + lastError: errorText(input.error), + }; + if (attempts < NATIVE_RESPONSE_PUBLICATION_MAX_ATTEMPTS) { + const nextAttemptAt = new Date(now.getTime() + nativeResponsePublicationDelayMs(attempts)); + const [updated] = await patchReceipt(tx, run.id, { ...base, state: "retrying", nextAttemptAt: nextAttemptAt.toISOString() }); + if (!updated) return null; + await appendHeartbeatRunEvent(tx, { + companyId: input.companyId, runId: run.id, agentId: run.agentId, + eventType: "run.response_publication.retrying", stream: "system", level: "warn", + message: `Final reply publication ${input.reason === "publication_error" ? "failed" : "was refused"}; retry ${attempts + 1} of ${NATIVE_RESPONSE_PUBLICATION_MAX_ATTEMPTS} is scheduled`, + payload: { attempts, reason: input.reason, nextAttemptAt: nextAttemptAt.toISOString() }, + }); + return { state: "retrying" as const, nextAttemptAt: nextAttemptAt.getTime() }; + } + const owner = { kind: "agent" as const, agentId: run.agentId }; + const [updated] = await patchReceipt(tx, run.id, { ...base, state: "failed", nextAttemptAt: null, + reasonCode: NATIVE_RESPONSE_PUBLICATION_FAILED, owner, failedAt: now.toISOString() }); + if (!updated) return null; + await appendHeartbeatRunEvent(tx, { + companyId: input.companyId, runId: run.id, agentId: run.agentId, + eventType: "run.response_publication.failed", stream: "system", level: "error", + message: `The accepted final reply could not be published to the task after ${attempts} attempts`, + payload: { reasonCode: NATIVE_RESPONSE_PUBLICATION_FAILED, attempts, lastReason: input.reason, owner }, + }); + await logActivity(tx, { + companyId: input.companyId, actorType: "system", actorId: "native-response-publication", + action: "issue.response_publication_failed", entityType: "issue", entityId: input.issueId, + agentId: run.agentId, runId: run.id, + details: { runId: run.id, reasonCode: NATIVE_RESPONSE_PUBLICATION_FAILED, attempts, lastReason: input.reason, + lastError: base.lastError, owner }, + }); + return { state: "failed" as const }; + }); +} + +/** JSON object to merge into `result_json` in the same write that records a + * published reply's `commentId`. It settles an open retry receipt atomically + * with publication, and is empty when no retry is open. */ +export function settledResponsePublicationSql() { + return sql`(case when ${heartbeatRuns.resultJson} #>> '{responsePublication,state}' = 'retrying' + then jsonb_build_object('responsePublication', ((${heartbeatRuns.resultJson} -> 'responsePublication') - 'nextAttemptAt') + || jsonb_build_object('state', 'resolved', 'resolvedAt', ${new Date().toISOString()}::text)) + else '{}'::jsonb end)`; +} diff --git a/server/src/services/native-runtime/native-response-publication-retry.test.ts b/server/src/services/native-runtime/native-response-publication-retry.test.ts new file mode 100644 index 00000000000..56d63c13f06 --- /dev/null +++ b/server/src/services/native-runtime/native-response-publication-retry.test.ts @@ -0,0 +1,178 @@ +import { randomUUID } from "node:crypto"; +import { and, eq, sql } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { activityLog, agents, companies, completionContracts, createDb, heartbeatRunEvents, heartbeatRuns, issueComments, issues, + nativeRunFinalizations, 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 } from "./native-run-finalizer.js"; +import { nativeResponsePresentationWork } from "./native-response-presentation-work.js"; + +// An accepted native task result whose final reply was never published must +// not be abandoned after one refused or failed publication attempt. Retries +// are durable and bounded, never double-post, and an exhausted retry leaves a +// named failure with an owner on the run and the task. +const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_RESPONSE_PUBLICATION_TEST_DATABASE_URL"); +const support = external ? { supported: true } : await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("native final reply publication retry", () => { + let database: Awaited> | undefined; + let db: Db; + beforeAll(async () => { + if (!external) database = await startEmbeddedPostgresTestDatabase("native-response-publication-"); + db = createDb(external ?? database!.connectionString); + }, 60_000); + afterAll(async () => { await db?.$client.end(); await database?.cleanup(); }); + + const reply = "Keyboard shortcuts are now available on every plan."; + async function fixture() { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); + const sessionId = randomUUID(), runnerId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Reply publication fixture", issuePrefix: companyId.slice(0, 8) }); + await db.insert(agents).values({ id: agentId, companyId, name: "Writer", adapterType: "paperclip_runner" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Short changelog blurb", status: "in_progress", assigneeAgentId: agentId }); + 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: "Write the blurb", criteria: [] }, + canonicalSha256: "fixture-publication-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", contextSnapshot: { issueId, source: "assignment" } }); + 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: [] }); + await port.completeRun({ result: { schema: "paperclip.run_result.v1", reportedWorkDisposition: "done", + summary: "Wrote the blurb.", completionClaim: { contractRevision: "test", objectiveSatisfied: true, criteria: [], remainingWork: [] }, + evidence: [], verification: [], attentionRequests: [], artifacts: [] }, + terminal: { schema: "paperclip.prp.terminal.v1", runTerminalState: "succeeded", turnTerminalState: "completed", reportedWorkDisposition: "done" }, + turnId: "blurb-turn" }); + await finalizeNativeRun({ db, runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: false }); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)))[0].phase).toBe("committed"); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }).where(eq(heartbeatRuns.id, runId)); + const prpEvent = { runId, normalizedSessionId: sessionId, turnId: "blurb-turn", sourceEventId: "fixture-final-reply", + payload: { kind: "agentMessage", channel: "final", providerPhase: "final", text: reply } }; + const saveFinalReply = () => db.insert(heartbeatRunEvents).values({ companyId, runId, agentId, seq: 170, eventType: "item.completed", + payload: { prpEvent } }); + return { companyId, issueId, runId, agentId, saveFinalReply }; + } + const comments = (runId: string) => db.select().from(issueComments).where(eq(issueComments.createdByRunId, runId)); + const run = async (runId: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0]; + const receipt = async (runId: string) => ((await run(runId)).resultJson as Record | null)?.responsePublication; + // Advance the saved deadline instead of the clock so the database and the + // worker agree on "due" without faking timers inside the driver. + const makeDue = (runId: string) => db.execute(sql`update heartbeat_runs set result_json = jsonb_set(result_json, + '{responsePublication,nextAttemptAt}', to_jsonb('2000-01-01T00:00:00.000Z'::text)) where id = ${runId}`); + + it("keeps an accepted reply scheduled after a single refused publication and posts it exactly once", async () => { + const f = await fixture(); + const work = nativeResponsePresentationWork(db); + const [entry] = await work.load({ companyId: f.companyId, issueId: f.issueId }); + expect(entry).toMatchObject({ id: f.runId }); + + // The final reply is not visible yet, so publication is refused once. + const before = Date.now(); + const rescheduled = await work.check(entry); + expect(await comments(f.runId)).toEqual([]); + expect(rescheduled).not.toBeNull(); + expect(rescheduled!.dueAt).toBeGreaterThan(before); + expect(await receipt(f.runId)).toMatchObject({ state: "retrying", attempts: 1, lastReason: "publication_refused" }); + // A restarted worker reconstructs the saved schedule instead of losing it. + const [reloaded] = await nativeResponsePresentationWork(db).load({ companyId: f.companyId, issueId: f.issueId }); + expect(reloaded).toMatchObject({ id: f.runId, dueAt: rescheduled!.dueAt }); + // Not yet due: no second attempt. + await expect(work.check(reloaded)).resolves.toMatchObject({ dueAt: rescheduled!.dueAt }); + expect((await receipt(f.runId)).attempts).toBe(1); + + await f.saveFinalReply(); + await makeDue(f.runId); + const [due] = await work.load({ companyId: f.companyId, issueId: f.issueId }); + await expect(work.check(due)).resolves.toBeNull(); + expect(await comments(f.runId)).toEqual([expect.objectContaining({ body: reply, authorAgentId: f.agentId })]); + expect(await receipt(f.runId)).toMatchObject({ state: "resolved", attempts: 1 }); + expect(((await run(f.runId)).resultJson as Record).presentationDecision.commentId).toEqual(expect.any(String)); + // Idempotent: nothing left to publish and a repeated check never double-posts. + expect(await work.load({ companyId: f.companyId, issueId: f.issueId })).toEqual([]); + await expect(work.check(due)).resolves.toBeNull(); + expect(await comments(f.runId)).toHaveLength(1); + }); + + it("records a single publication error durably and retries it", async () => { + const f = await fixture(); + await f.saveFinalReply(); + const failure = new Error("fixture-comment-write-failed"); + let injected = false; + const failing = new Proxy(db, { get(target, key) { + if (key === "transaction" && !injected) { + return (callback: Parameters[0]) => db.transaction(async tx => { + // The publication transaction is the only one that locks the + // finalization coordinator; fail exactly that one attempt. + injected = true; await callback(tx); throw failure; + }); + } + return Reflect.get(target, key); + } }); + const work = nativeResponsePresentationWork(failing as Db); + const [entry] = await work.load({ companyId: f.companyId, issueId: f.issueId }); + const rescheduled = await work.check(entry); + expect(rescheduled).not.toBeNull(); + expect(await comments(f.runId)).toEqual([]); + expect(await receipt(f.runId)).toMatchObject({ state: "retrying", attempts: 1, lastReason: "publication_error", + lastError: "fixture-comment-write-failed" }); + await makeDue(f.runId); + await expect(work.check(entry)).resolves.toBeNull(); + expect(await comments(f.runId)).toEqual([expect.objectContaining({ body: reply })]); + }); + + it("records an already posted reply without posting again or reporting a failure", async () => { + const f = await fixture(); + await f.saveFinalReply(); + // The live path wrote the reply but stopped before recording its decision. + const [posted] = await db.insert(issueComments).values({ companyId: f.companyId, issueId: f.issueId, authorAgentId: f.agentId, + createdByRunId: f.runId, body: reply }).returning(); + const work = nativeResponsePresentationWork(db); + const [entry] = await work.load({ companyId: f.companyId, issueId: f.issueId }); + await expect(work.check(entry)).resolves.toBeNull(); + expect(await comments(f.runId)).toEqual([posted]); + expect(((await run(f.runId)).resultJson as Record).presentationDecision).toMatchObject({ + commentAction: "reuse", commentId: posted.id }); + expect(await receipt(f.runId)).toBeUndefined(); + expect(await work.load({ companyId: f.companyId, issueId: f.issueId })).toEqual([]); + }); + + it("counts one due attempt once when two workers check it concurrently", async () => { + const f = await fixture(); + const [entry] = await nativeResponsePresentationWork(db).load({ companyId: f.companyId, issueId: f.issueId }); + const results = await Promise.all([nativeResponsePresentationWork(db).check(entry), nativeResponsePresentationWork(db).check(entry)]); + expect(results.every(result => result !== null)).toBe(true); + expect(await receipt(f.runId)).toMatchObject({ state: "retrying", attempts: 1 }); + expect(await comments(f.runId)).toEqual([]); + }); + + it("stops after bounded attempts with a named, owned failure on the run and the task", async () => { + const f = await fixture(); + const work = nativeResponsePresentationWork(db); + let attempts = 0; + for (;;) { + const [entry] = await work.load({ companyId: f.companyId, issueId: f.issueId }); + if (!entry) break; + attempts += 1; + expect(attempts).toBeLessThanOrEqual(10); + await makeDue(f.runId).catch(() => undefined); + await work.check({ ...entry, dueAt: 0 }); + } + expect(await comments(f.runId)).toEqual([]); + const failed = await receipt(f.runId); + expect(failed).toMatchObject({ state: "failed", reasonCode: "native_response_publication_failed", + owner: { kind: "agent", agentId: f.agentId } }); + expect(failed.attempts).toBeGreaterThan(1); + const events = await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, f.runId), + eq(heartbeatRunEvents.eventType, "run.response_publication.failed"))); + expect(events).toEqual([expect.objectContaining({ level: "error" })]); + const activity = await db.select().from(activityLog).where(and(eq(activityLog.entityId, f.issueId), + eq(activityLog.action, "issue.response_publication_failed"))); + expect(activity).toEqual([expect.objectContaining({ runId: f.runId, agentId: f.agentId })]); + // Bounded: the failed reply is no longer rescheduled. + expect(await work.load({ companyId: f.companyId, issueId: f.issueId })).toEqual([]); + }); +}); diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index 740e1908026..1b218fa10ad 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -96,6 +96,7 @@ import { selectHeartbeatRunFinalAgentMessage, } from "../heartbeat-run-summary.js"; import { resolveChatRunPresentationAuthorizationReason } from "../chat-run-publications.js"; +import { settledResponsePublicationSql } from "./native-response-publication-receipt.js"; import { hasChatRunOwnedProviderInteraction } from "../chat-interaction-arbitration.js"; import { authorizeCommittedChatResponse, @@ -1036,6 +1037,17 @@ async function recoverCommittedNativeChatResponseInScope( existingComment: findHeartbeatRunCompletionComment(comments, run.resultJson), finalAgentMessage, }); + if (resolved.decision.commentAction === "reuse" && typeof resolved.decision.commentId === "string") { + // The reply is already on the task (for example, the live path wrote + // it but did not record its decision). Record it; never post again. + await tx.update(heartbeatRuns).set({ + resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({ + presentationDecision: { ...resolved.decision, + reasonCodes: [...resolved.decision.reasonCodes, "committed_task_response_reused"] }, + })}::jsonb || ${settledResponsePublicationSql()}`, updatedAt: new Date(), + }).where(eq(heartbeatRuns.id, run.id)); + 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, @@ -1045,7 +1057,7 @@ async function recoverCommittedNativeChatResponseInScope( resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({ presentationDecision: { ...resolved.decision, commentId: comment.id, reasonCodes: [...resolved.decision.reasonCodes, "committed_task_response_recovered"] }, - })}::jsonb`, updatedAt: new Date(), + })}::jsonb || ${settledResponsePublicationSql()}`, updatedAt: new Date(), }).where(eq(heartbeatRuns.id, run.id)); agentId = run.agentId; conversationReply = conversation;