Files
PaperClipAI/server/src/services/native-runtime/native-run-finalizer.ts
T
DottaandClaude Opus 5.5 3b17b5ce15 fix(native): keep a reassigned-away run from closing the new owner's task (#19186)
## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work
> - A native agent run reports "done" in an accepted result. A finalizer
then decides the task status.
> - The board can reassign a task while the run is in flight. The old
run is cancelled, but its accepted result still goes to the finalizer.
> - The finalizer did not check who owns the task now. The old owner's
"done" closed the task of the new owner.
> - The new owner never did the work, and the task shows done with no
reply from the current owner.
> - This pull request makes a done report apply only when the reporting
run's agent still owns the task.
> - The benefit is that reassignment keeps its meaning: the new owner
keeps the open task, and the old result stays as history.

## Linked Issues or Issue Description

**What happened?**

On a hosted staging instance, agent A started a native run on a task.
Agent A finished the work and submitted a "done" result. Then the board
reassigned the task to agent B, and Paperclip cancelled agent A's run
with `issue_reassigned`. About 45 seconds later, the native finalizer
committed agent A's done decision with reason
`completion_claim_policy_accepted`. The task became done. Agent B's runs
were then cancelled because the task was terminal. Agent B never worked
on the task.

**Expected behavior**

A done report from a run that no longer owns the task must not close the
task. The accepted result stays as history. The task status stays as it
is for the current owner.

**Steps to reproduce**

1. Assign a low-risk native task to agent A and start a run.
2. Submit a done result for the run.
3. Reassign the task to agent B before finalization.
4. Finalize the run. Before this change, the task becomes done.

**Paperclip version or commit**

Observed on staging at `b9fbbf8d30`. Reproduced on `master` at
`06826915d7`.

Related: #19130 adds a publication check for requested documents at
finalization. It does not check task ownership.

## What Changed

- `status-arbiter.ts`: add `ownsNativeTaskCompletion` and
`supersedeStaleOwnerCompletion`. A done decision from an agent that is
not the current assignee (another agent, a user, or nobody) becomes
`preserve` with reason `completion_owner_superseded` and no effects.
Other decisions do not change. This is the same ownership rule that the
provider-failure path already uses (`ownsProviderFailureDecision`).
- `native-run-finalizer.ts`: apply the rule after arbitration and before
the deliverable check. Conversation tasks keep their existing path.
- `native-status-reassessment.ts`: apply the same rule, so new evidence
from a reassigned-away run cannot close the task later.
- The status committer does not change. A reassignment during
finalization already increments `statusVersion` (migration 0358). The
committer then replans, and the retry applies the new rule.
- `doc/execution-semantics.md`: describe the rule next to the existing
"a late finalizer cannot change a reassigned task" statement.

## Verification

New file
`server/src/services/native-runtime/native-stale-owner-completion.test.ts`.
It uses embedded PostgreSQL, the real coordinator store and the real
finalizer.

- Control: the owning run's done result closes the task with
`completion_claim_policy_accepted`.
- Reassigned before finalization: the task stays `in_progress` with
agent B. The decision is `completion_owner_superseded`. The accepted
result row is unchanged.
- Reassigned after arbitration and before the status commit: the commit
replans and the task stays `in_progress`.
- Reassessment control: later approved evidence closes the owner's task
with `decision_superseded_by_new_evidence`.
- Reassessment after reassignment: later approved evidence from the old
run does not close the task. The decision is
`completion_owner_superseded`. This case fails when the reassessment
rule is disabled.
- Pure policy cases: another agent, a board user and no assignee all
preserve status. A non-done decision passes through.

Red before the fix (same tests, fix not applied):

```sh
cd server && npx vitest run src/services/native-runtime/native-stale-owner-completion.test.ts
# Tests 2 failed | 1 passed — both reassignment cases: expected "in_progress", received "done"
```

Green after the fix:

```sh
cd server && npx vitest run src/services/native-runtime/native-stale-owner-completion.test.ts
# Tests 10 passed

cd server && npx vitest run src/services/native-runtime/paperclip-control-plane-port.test.ts \
  src/services/native-runtime/native-completion-feedback.test.ts src/services/native-runtime/native-deliverable-feedback.test.ts \
  src/services/native-runtime/native-skill-test-status.test.ts src/services/native-runtime/native-routine-status.test.ts \
  src/__tests__/native-status-arbiter-corpus.test.ts src/__tests__/native-status-reassessment-work.test.ts \
  src/services/native-runtime/agent-chat-response-recovery.test.ts src/__tests__/native-run-projection-work.test.ts \
  src/services/native-runtime/task-reassignment.test.ts src/services/native-runtime/native-provider-failure.integration.test.ts
# 11 files, 614 tests passed
```

I did not run the full `pnpm test:run`, `pnpm -r typecheck` or `pnpm
build` locally. CI runs them.

## Risks

- A native run that reassigns its own task to a user or to another
agent, and then reports done, no longer closes the task through
arbitration. The status that the agent set through the API stays. This
matches the existing provider-failure ownership rule.
- A native run on a task with no assignee cannot close it through
arbitration. Native task runs start from an assignment, so this is not
expected in normal use.
- Review runs and conversation tasks keep their existing paths. Review
outcomes already preserve status before this rule.
- No schema change and no migration.

> For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and
discuss it in `#dev` before opening the PR. Feature PRs that overlap
with planned core work may need to be redirected — check the roadmap
first. See `CONTRIBUTING.md`.

## Model Used

- Provider: Anthropic. Model: Claude Opus 5.5 (`claude-opus-5-5`), in
the Claude Code agent harness.
- Capabilities used: tool use, shell and code execution, repository
search, browser reads of staging, and extended reasoning.
- The model wrote the change, the tests and this description. A human
requested the fix.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [ ] All Paperclip CI gates are green
- [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-05 00:02:06 -06:00

1859 lines
87 KiB
TypeScript

import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js";
import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js";
import { notifyNativeRecoveryNotices } from "../native-recovery-notice-notifications.js";
import { resolveCommittedNativeRecoveryNotices } from "./native-recovery-notices.js";
import { notifyNativeRunProjection } from "../native-run-projection-notifications.js";
import { projectCommittedRun, recoveredExecutionFailureMetadata, settleNativeTerminalWake } from "./native-run-projection.js";
import { notifyNativeResultCommit } from "../native-result-commit-notifications.js";
import { notifyRetainedNativeSessionCleanup } from "../retained-native-session-cleanup-notifications.js";
import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js";
import { notifyUnsafeWorkspaceExportRunChange } from "../unsafe-workspace-export-notifications.js";
import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js";
import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js";
import { notifyInstructionCollection } from "../instruction-collection-notifications.js";
import { readRequiredQualityReview } from "../required-quality-review.js";
import { NativeDeliverableEvidenceError, validateNativeDeliverableEvidence } from "./native-deliverable-feedback.js";
import { pendingNativeGovernance } from "./native-governance.js";
import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js";
import { onboardingFirstTaskCompletionAllowed } from "../onboarding-first-task-admission.js";
import { notifyExecutionRunWork } from "../execution-control-notifications.js";
import { reconcileMuseNativeTerminalRun } from "./muse-terminal-reconciliation.js";
import { assertNativeCaseHandoffResult } from "./native-case-handoff.js";
import { scheduledIssueMonitorWait } from "../issue-monitors.js";
import { hasPublishedGitHubReview, readInheritedGitHubReviewWait } from "../chat-channels/github/review-wait.js";
import { readNativeMonitorDelivery, readNativePreservedMonitorWait } from "./native-monitor-delivery.js";
import { isNativePlanWaitResult, readNativePlanWait } from "./native-plan-wait.js";
import { hasPendingNativeChildCompletion } from "./native-child-completion-delivery.js";
import { settleExternalConversation } from "../slack-conversation-lifecycle.js";
import { executionFailureRetryCount } from "../execution-recovery-attempt.js";
import { readPersistedNativeFailureEvidence } from "./native-provider-failure-evidence.js";
import { explainFailedRun, NATIVE_RUN_FAILED_UNEXPLAINED } from "./native-run-failure-explanation.js";
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, 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";
import { and, desc, eq, inArray, isNotNull, isNull, or, sql } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
import {
agentWakeupRequests,
completionContracts,
heartbeatRuns,
heartbeatRunEvents,
issueComments,
issueThreadInteractions,
issues,
nativeRunFinalizations,
nativeRunResults,
statusDecisions,
statusDecisionEffects,
workAssessments,
workspaceOperations,
} from "@paperclipai/db";
import { classifyNativeEvidence } from "./evidence-classifier.js";
import type { PrpIgnoredAttentionRequest, PrpTerminalState } from "@paperclipai/paperclip-runner";
import {
arbitrateNativeStatus,
NATIVE_STATUS_ARBITER_POLICY_VERSION,
supersedeStaleOwnerCompletion,
type NativeAuthoritativeIssueStatus,
} from "./status-arbiter.js";
import { recordNativeWorkAssessment } from "./work-assessments.js";
import {
commitNativeStatusDecision,
NativeStatusRaceError,
type NativeStatusCommitFailpoint,
} from "./status-decision-committer.js";
import { issueRecoveryActionService } from "../issue-recovery-actions.js";
import { preservedResultHasStop, readPreservedNativeResultFinalization } from "./preserved-native-result-finalization.js";
import { issueService } from "../issues.js";
import { publishChatPublicationCommitSignal } from "../chat-publication-reconciliation.js";
import { nativeSha256 } from "./canonical.js";
import { beginRunPhase, measureRunPhase, withRunResponsePresentationTiming } from "../execution-preparation-timing.js";
import {
readNativeBoardResponseWaitOrigin,
readNativeBoardResponseWaitSource,
} from "./native-board-response-wait.js";
import { emitAgentTaskRun } from "../agent-task-run-telemetry.js";
import { isNewTerminalRunFailure, reportRunFailure } from "../run-failure-report.js";
import { isPublishedGitHubReviewCompletionContinuation, readPublishedGitHubReviewCompletionContinuation, resolveExternalChatResponseWaitAuthorization } from "./chat-attachment-reuse.js";
import {
authorizeNativeChatReviewPresentation,
hasMaterializedNativeReviewResponse,
restoreNativeChatReviewPresentationInTransaction,
} from "./native-chat-review-presentation.js";
import { logger } from "../../middleware/logger.js";
import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js";
import {
CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON,
findHeartbeatRunCompletionComment,
isExternalChatPresentationContext,
readAcceptedTaskContinuationSummary,
readCompletedAssistantMessageCandidate,
resolveHeartbeatRunResponse,
selectHeartbeatRunFinalAgentMessage,
} from "../heartbeat-run-summary.js";
import { resolveChatRunPresentationAuthorizationReason } from "../chat-run-publications.js";
import { hasChatRunOwnedProviderInteraction } from "../chat-interaction-arbitration.js";
import {
authorizeCommittedChatResponse,
CommittedChatResponseAuthorizationError,
} from "../durable-chat-wakeup.js";
import {
isNativeRunnerOwnershipHeld,
} from "./native-runner-ownership.js";
function record(value: unknown): Record<string, unknown> {
return typeof value === "object" && value !== null && !Array.isArray(value)
? (value as Record<string, unknown>)
: {};
}
function authoritativeStatus(value: string): NativeAuthoritativeIssueStatus {
if (
![
"backlog",
"todo",
"in_progress",
"in_review",
"blocked",
"done",
"cancelled",
].includes(value)
) {
throw new Error("native_issue_status_invalid");
}
return value as NativeAuthoritativeIssueStatus;
}
function isCommittedAuditOnlyRouting(value: unknown) {
const receipts = record(value).nativeAttentionRouting;
return (
Array.isArray(receipts) &&
receipts.length > 0 &&
receipts.every((candidate) => {
const receipt = record(candidate);
const targets = receipt.materializedTargets;
return (
receipt.decisionId === null &&
receipt.reasonCode === "attention_duplicate_suppressed" &&
Array.isArray(targets) &&
targets.length > 0
);
})
);
}
export { projectNativeTerminalRunStatus } from "./native-run-projection.js";
/** Fact-based arbitration at the production finalization authority seam. */
export function resolveNativeFinalizerStatus(
input: Parameters<typeof arbitrateNativeStatus>[0],
) {
return arbitrateNativeStatus(input);
}
export { pendingNativeGovernance } from "./native-governance.js";
async function acceptedInteractionFromRun(input: {
db: Db;
companyId: string;
issueId: string;
runId: string;
}) {
return input.db
.select({ id: issueThreadInteractions.id })
.from(issueThreadInteractions)
.where(
and(
eq(issueThreadInteractions.companyId, input.companyId),
eq(issueThreadInteractions.issueId, input.issueId),
eq(issueThreadInteractions.sourceRunId, input.runId),
eq(issueThreadInteractions.status, "accepted"),
),
)
.limit(1)
.then((rows) => rows[0] ?? null);
}
export class NativeFinalizationDeferredError extends Error {
constructor(readonly reason: "lease_busy" | "retry_not_due") {
super(`native_finalization_${reason}`);
}
}
async function claimCoordinator(input: {
db: Db;
runId: string;
onlyWhenDue?: boolean;
preserveProviderAttempt?: boolean;
}) {
const leaseOwner = `native-finalizer:${randomUUID()}`;
const now = new Date();
const claimed = await input.db.transaction(async (tx) => {
const coordinator = await tx
.select()
.from(nativeRunFinalizations)
.where(eq(nativeRunFinalizations.runId, input.runId))
.for("update")
.limit(1)
.then((rows) => rows[0] ?? null);
if (!coordinator?.resultId) throw new Error("native_finalization_missing");
if (coordinator.phase === "committed") {
if (!coordinator.decisionId) {
const run = await tx
.select({ resultJson: heartbeatRuns.resultJson })
.from(heartbeatRuns)
.where(
and(
eq(heartbeatRuns.id, coordinator.runId),
eq(heartbeatRuns.companyId, coordinator.companyId),
),
)
.limit(1)
.then((rows) => rows[0] ?? null);
if (!run || !isCommittedAuditOnlyRouting(run.resultJson)) {
throw new Error("native_finalization_committed_outcome_missing");
}
}
return { coordinator, leaseOwner: null };
}
if (coordinator.phase === "terminal_failure")
throw new Error("native_finalization_terminal_failure");
if (input.onlyWhenDue && coordinator.nextAttemptAt && coordinator.nextAttemptAt > now)
throw new NativeFinalizationDeferredError("retry_not_due");
if (
coordinator.leaseOwner &&
coordinator.leaseExpiresAt &&
coordinator.leaseExpiresAt > now &&
coordinator.leaseOwner !== leaseOwner
)
throw new NativeFinalizationDeferredError("lease_busy");
if (coordinator.decisionId) await notifyNativeFinalizationWork(tx, coordinator);
else await notifyNativeResultCommit(tx, coordinator);
const [updated] = await tx
.update(nativeRunFinalizations)
.set({
leaseOwner,
leaseExpiresAt: new Date(now.getTime() + 5 * 60_000),
attempt:
input.preserveProviderAttempt || coordinator.controllerBootId !== null
? coordinator.attempt
: coordinator.attempt + 1,
phase:
coordinator.phase === "retryable_failure"
? coordinator.assessmentId
? "arbitrating"
: "workspace_finalizing"
: coordinator.phase,
recoveryState: null,
recoveryRequestId: null,
failureCode: null,
failureDetail: null,
nextAttemptAt: null,
updatedAt: now,
})
.where(eq(nativeRunFinalizations.runId, input.runId))
.returning();
if (!updated) throw new Error("native_finalization_claim_failed");
return { coordinator: updated, leaseOwner };
});
return claimed;
}
/**
* A workspace copy-back cause is server-built text in a fixed grammar (see
* remoteWorkspaceFailureCause and the native post-result diagnostic). Re-check the shape before it reaches the run:
* no paths, no line breaks, bounded length.
*/
export function readWorkspaceFailureCause(value: unknown): string | null {
return typeof value === "string" && value.length <= 1024
&& /^(?:remote_workspace|native_post_result)_[a-z_]+(?::[A-Za-z0-9_=;:,.() -]*)?$/.test(value) ? value : null;
}
function terminalFinalizationFailureMessage(failureCode: string, failureScope: "provider" | "workspace" | undefined,
failureCause?: string | null) {
if (failureCode === "native_workspace_sync_out_retry_exhausted") {
// This code also covers local-directory and plugin workspaces, so the
// message stays general and defers to the recovery action's next step.
return "Saving the workspace failed after 3 attempts. The agent's result is kept; follow the run's recovery action to repair the workspace."
+ (failureCause ? ` Last cause: ${failureCause}` : "");
}
if (failureCode === "native_workspace_sync_out_unrecoverable") {
return "The sandbox holding unexported workspace changes is unavailable, so its files could not be saved.";
}
if (failureCode === "native_finalization_retry_exhausted") {
return "Native run finalization failed after 3 attempts.";
}
return failureScope === "workspace"
? `Native workspace finalization failed (${failureCode}).`
: `Native run finalization failed (${failureCode}).`;
}
async function recordRetryableFailure(input: {
db: Db;
run: typeof heartbeatRuns.$inferSelect;
coordinator: typeof nativeRunFinalizations.$inferSelect;
failureCode: string;
message: string;
nextAction: string;
projectRunStatus?: boolean;
failureScope?: "provider" | "workspace";
permanent?: boolean;
failureCause?: string | null;
}) {
const now = new Date();
const nextAttemptAt = new Date(now.getTime() + 30_000);
const failureCause = readWorkspaceFailureCause(input.failureCause);
const acceptedRunTerminalState = record(
input.run.resultJson,
).prpRunTerminalState;
const exhaustedRunStatus =
acceptedRunTerminalState === "succeeded"
? ("succeeded" as const)
: acceptedRunTerminalState === "cancelled"
? ("cancelled" as const)
: ("failed" as const);
let terminalRunToEmit: typeof heartbeatRuns.$inferSelect | null = null;
const outcome = await input.db.transaction(async (tx) => {
// Admission snapshots cannot authorize a failure write after a different
// owner committed success. Match status-committer lock ordering.
const current = await tx.select().from(nativeRunFinalizations).where(and(
eq(nativeRunFinalizations.runId, input.run.id),
eq(nativeRunFinalizations.companyId, input.run.companyId),
)).for("update").limit(1).then((rows) => rows[0] ?? null);
if (!current) throw new Error("native_finalization_missing");
// Audit-only attention first records an agent-owned invalid-result outcome;
// its caller still needs to materialize the normal bounded recovery action.
// Board-owned terminal repairs and late snapshots remain settled.
const currentExportRetry = record(record(current.failureDetail).workspaceExportRetry).requestId;
if (currentExportRetry && currentExportRetry !== record(record(input.coordinator.failureDetail).workspaceExportRetry).requestId) return current;
const recoveryOwner = record(record(current.failureDetail).recoveryOwner);
const pendingAgentRecovery = current.phase === "terminal_failure"
&& input.coordinator.phase === "terminal_failure"
&& recoveryOwner.kind === "agent" && recoveryOwner.agentId === input.run.agentId;
if (current.phase === "committed"
|| (current.phase === "terminal_failure" && !pendingAgentRecovery)
|| current.leaseOwner !== input.coordinator.leaseOwner) return current;
const [currentRun] = await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.run.id)).limit(1);
if (!currentRun) throw new Error("native_finalization_run_missing");
if (input.failureScope === "workspace") {
const exported = await tx.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and(
eq(workspaceOperations.companyId, input.run.companyId),
eq(workspaceOperations.heartbeatRunId, input.run.id),
eq(workspaceOperations.phase, "workspace_finalize"),
eq(workspaceOperations.status, "succeeded"),
)).limit(1);
// The live exporter may have acquired ownership since the stale error
// was raised, or already durably published a successful copyback.
if (exported.length || record(currentRun.runnerProfileJson).nativeWorkspaceFinalizationOwner) return current;
}
input = { ...input, coordinator: current, run: currentRun };
const issue = await tx
.select({
lastStatusDecisionId: issues.lastStatusDecisionId,
status: issues.status, statusVersion: issues.statusVersion,
assigneeAgentId: issues.assigneeAgentId, assigneeUserId: issues.assigneeUserId,
})
.from(issues)
.where(
and(
eq(issues.id, input.coordinator.issueId),
eq(issues.companyId, input.run.companyId),
),
)
.for("update")
.limit(1)
.then((rows) => rows[0] ?? null);
const latestDecision = issue?.lastStatusDecisionId
? await tx
.select({
runId: workAssessments.runId,
})
.from(statusDecisions)
.innerJoin(
workAssessments,
eq(statusDecisions.assessmentId, workAssessments.id),
)
.where(
and(
eq(statusDecisions.id, issue.lastStatusDecisionId),
eq(statusDecisions.companyId, input.run.companyId),
),
)
.limit(1)
.then((rows) => rows[0] ?? null)
: null;
const latestDecisionRun =
latestDecision?.runId && latestDecision.runId !== input.run.id
? await tx
.select({ createdAt: heartbeatRuns.createdAt })
.from(heartbeatRuns)
.where(
and(
eq(heartbeatRuns.id, latestDecision.runId),
eq(heartbeatRuns.companyId, input.run.companyId),
),
)
.limit(1)
.then((rows) => rows[0] ?? null)
: null;
// A finalizer belongs to one immutable run result. If a later run already
// committed the issue's authoritative decision, this older coordinator is
// audit-only: retrying or escalating it must never reopen the newer result.
const supersededByNewerRun = Boolean(
latestDecisionRun && latestDecisionRun.createdAt > input.run.createdAt,
);
const priorFailureDetail = record(input.coordinator.failureDetail);
const workspaceFinalizeAttempt =
input.failureScope === "workspace"
? (typeof priorFailureDetail.workspaceFinalizeAttempt === "number" &&
Number.isInteger(priorFailureDetail.workspaceFinalizeAttempt) &&
priorFailureDetail.workspaceFinalizeAttempt >= 0
? priorFailureDetail.workspaceFinalizeAttempt
: 0) + 1
: null;
const exhausted =
input.permanent === true ||
(workspaceFinalizeAttempt !== null
? workspaceFinalizeAttempt >= 3
: input.coordinator.attempt >= 3);
const phase =
supersededByNewerRun || exhausted
? ("terminal_failure" as const)
: ("retryable_failure" as const);
const failureCode = supersededByNewerRun
? "native_finalization_superseded"
: input.permanent
? input.failureCode
: exhausted
? input.failureScope === "workspace"
? "native_workspace_sync_out_retry_exhausted"
: "native_finalization_retry_exhausted"
: input.failureCode;
await notifyNativeFinalizationWork(tx, { companyId: input.run.companyId, runId: input.run.id });
if (phase === "terminal_failure") await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.run.companyId);
if (phase === "terminal_failure" && failureCode === "native_workspace_sync_out_unsafe_archive") {
await notifyUnsafeWorkspaceExportRecovery(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId });
}
await notifyExecutionRunWork(tx, { companyId: input.run.companyId });
await notifyChatRunMilestones(tx, { companyId: input.run.companyId, issueId: input.coordinator.issueId });
await tx
.update(nativeRunFinalizations)
.set({
phase,
leaseOwner: null,
leaseExpiresAt: null,
failureCode,
failureDetail: {
message: input.message.slice(0, 2_000),
originalFailureCode: input.failureCode,
...(failureCause ? { failureCause } : {}),
...(priorFailureDetail.workspaceExportRetry ? { workspaceExportRetry: priorFailureDetail.workspaceExportRetry } : {}),
...(workspaceFinalizeAttempt === null
? {}
: { workspaceFinalizeAttempt }),
recoveryOwner: supersededByNewerRun
? { kind: "none", reason: "newer_native_decision" }
: exhausted
? { kind: "board" }
: { kind: "agent", agentId: input.run.agentId },
nextAction: input.nextAction,
},
nextAttemptAt: supersededByNewerRun || exhausted ? null : nextAttemptAt,
updatedAt: now,
})
.where(eq(nativeRunFinalizations.runId, input.run.id));
const projectsTerminalStatus =
exhausted && !supersededByNewerRun && input.projectRunStatus;
if (projectsTerminalStatus && input.failureScope === "workspace") {
await preserveNativeWorkspaceExportLease(tx as unknown as Db, input.run, input.coordinator.resultId);
}
if (projectsTerminalStatus) await notifyInstructionCollection(tx, { companyId: input.run.companyId, runId: input.run.id });
if (projectsTerminalStatus) await notifyCancelledWorkspacePreservation(tx, { companyId: input.run.companyId, runId: input.run.id });
if (projectsTerminalStatus) await notifyUnsafeWorkspaceExportRunChange(tx, input.run);
const terminalRunStatus = input.failureScope === "workspace" ? "failed" : exhaustedRunStatus;
const [updatedRun] = await tx
.update(heartbeatRuns)
.set({
executionStatusDeliveryId: randomUUID(),
...(projectsTerminalStatus
? {
// Stop can win after the failure's admission snapshot. The
// UPDATE locks and evaluates the current row, not that snapshot.
status: sql`case when ${heartbeatRuns.status} = 'cancelled' then ${heartbeatRuns.status}
else ${terminalRunStatus} end`,
finishedAt: sql`case when ${heartbeatRuns.status} = 'cancelled' then ${heartbeatRuns.finishedAt}
else coalesce(${heartbeatRuns.finishedAt}, ${now.toISOString()}::timestamptz) end`,
// A failed run names its cause. Keep an earlier root cause, and
// leave a run that Stop already cancelled without a failure.
...(terminalRunStatus === "failed"
? explainFailedRun({
errorCode: failureCode,
error: terminalFinalizationFailureMessage(failureCode, input.failureScope, failureCause),
})
: {}),
}
: {}),
nativePhase: phase,
nativePhaseUpdatedAt: now,
// Preserve concurrent cancellation and accounting receipts while
// recording the finalizer's own failure history.
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({
finalizationPhase: phase,
failureCode,
originalFailureCode: input.failureCode,
// The latest attempt's redacted cause; cleared when it named none.
failureCause,
nextAttemptAt: supersededByNewerRun || exhausted ? null : nextAttemptAt.toISOString(),
})}::jsonb`,
updatedAt: now,
})
.where(eq(heartbeatRuns.id, input.run.id))
.returning();
if (projectsTerminalStatus && updatedRun) await settleNativeTerminalWake(tx as unknown as Db, updatedRun);
if (projectsTerminalStatus) terminalRunToEmit = updatedRun ?? null;
let workspaceFinalizationBlock: Record<string, unknown> | undefined;
if (supersededByNewerRun) {
await issueRecoveryActionService(
tx as unknown as Db,
).resolveActiveForIssue(
{
companyId: input.run.companyId,
sourceIssueId: input.coordinator.issueId,
cause: input.failureCode,
fingerprint: nativeSha256({
runId: input.run.id,
failureCode: input.failureCode,
}),
status: "resolved",
outcome: "false_positive",
resolutionNote:
"A newer native run already committed the authoritative issue decision; the stale finalizer was retired without changing issue state.",
},
tx,
);
} else if (exhausted) {
const blocked = await issueService(tx as unknown as Db).update(
input.coordinator.issueId,
{
status:
input.failureScope === "workspace"
? "blocked"
: "in_review",
},
tx,
);
if (input.failureScope === "workspace" && issue && blocked?.status === "blocked"
&& !["blocked", "done", "cancelled"].includes(issue.status)
&& issue.assigneeAgentId === input.run.agentId && !issue.assigneeUserId
&& input.run.status !== "cancelled" && !preservedResultHasStop(input.run.resultJson)
&& input.coordinator.resultId && input.run.completionContractId && input.run.completionContractSha256) {
const [accepted] = await tx.select({ id: nativeRunResults.id, sha256: nativeRunResults.canonicalSha256 })
.from(nativeRunResults).where(and(eq(nativeRunResults.companyId, input.run.companyId),
eq(nativeRunResults.issueId, input.coordinator.issueId), eq(nativeRunResults.runId, input.run.id),
eq(nativeRunResults.id, input.coordinator.resultId), eq(nativeRunResults.schemaStatus, "accepted"),
eq(nativeRunResults.completionContractId, input.run.completionContractId))).limit(1);
if (accepted) workspaceFinalizationBlock = {
schema: "paperclip.native-workspace-finalization-block.v1",
companyId: input.run.companyId, issueId: input.coordinator.issueId,
runId: input.run.id, agentId: input.run.agentId, resultId: accepted.id, resultSha256: accepted.sha256,
contractId: input.run.completionContractId, contractSha256: input.run.completionContractSha256,
identityContextId: input.run.activeIdentityContextId, responsibleUserId: input.run.responsibleUserId,
statusVersion: Number(blocked.statusVersion), priorDecisionId: issue.lastStatusDecisionId,
};
}
}
if (!supersededByNewerRun) {
await issueRecoveryActionService(tx as unknown as Db).upsertSourceScoped({
companyId: input.run.companyId,
sourceIssueId: input.coordinator.issueId,
kind: "active_run_watchdog",
ownerType: exhausted ? "board" : "agent",
ownerAgentId: exhausted ? null : input.run.agentId,
returnOwnerAgentId: input.run.agentId,
cause: failureCode,
fingerprint: nativeSha256({ runId: input.run.id, failureCode }),
evidence: {
runId: input.run.id,
...(workspaceFinalizationBlock ? { workspaceFinalizationBlock } : {}),
coordinatorAttempt: input.coordinator.attempt,
...(workspaceFinalizeAttempt === null
? {}
: { workspaceFinalizeAttempt }),
originalFailureCode: input.failureCode,
},
nextAction: exhausted
? input.permanent
? input.nextAction
: `Finalization retry budget exhausted. ${input.nextAction}`
: input.nextAction,
wakePolicy: exhausted
? null
: {
kind: "resume_native_run",
runId: input.run.id,
notBefore: nextAttemptAt.toISOString(),
},
maxAttempts: 3,
supersedeOnIdentityChange: true,
});
}
return {
phase,
failureCode,
nextAttemptAt: supersededByNewerRun || exhausted ? null : nextAttemptAt,
};
});
if (terminalRunToEmit) {
await emitAgentTaskRun(input.db, terminalRunToEmit);
void reportRunFailure(input.db, terminalRunToEmit);
}
return {
...input.coordinator,
...outcome,
};
}
/** Converts malformed or incomplete persisted finalization state into named recovery. */
export async function recordNativeFinalizationFailure(input: {
db: Db;
runId: string;
error: unknown;
projectRunStatus?: boolean;
failureScope?: "provider" | "workspace";
permanent?: boolean;
failureCause?: string | null;
}) {
const [run, coordinator] = await Promise.all([
input.db
.select()
.from(heartbeatRuns)
.where(eq(heartbeatRuns.id, input.runId))
.limit(1)
.then((rows) => rows[0] ?? null),
input.db
.select()
.from(nativeRunFinalizations)
.where(eq(nativeRunFinalizations.runId, input.runId))
.limit(1)
.then((rows) => rows[0] ?? null),
]);
if (!run || run.runtimeMode !== "native" || !coordinator) throw input.error;
const message =
input.error instanceof Error ? input.error.message : String(input.error);
if (message === "native_finalization_lease_busy") return coordinator;
const failureCode = message.startsWith("native_")
? message
: "native_finalization_invalid";
return recordRetryableFailure({
db: input.db,
run,
coordinator,
failureCode,
message,
nextAction:
input.failureScope === "workspace"
&& (message === "native_plugin_workspace_branch_mismatch" || message === "native_plugin_workspace_owner_mismatch")
? "Repair the plugin workspace branch or owner, then resume finalization without another provider turn."
: input.failureScope === "workspace"
? input.permanent
? "Restore the exact sandbox containing the unexported workspace changes, or resolve the task manually from durable evidence."
: "Retry workspace export and merge from the retained sandbox; do not submit another provider turn."
: "Repair the persisted native result or contract discriminator, then resume finalization from the coordinator.",
projectRunStatus: input.projectRunStatus,
failureScope: input.failureScope,
permanent: input.permanent,
failureCause: input.failureCause,
});
}
async function materializeNativeReviewResponse(db: Db, runId: string, dbOrTx: Db = db) {
const [run] = await dbOrTx
.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
nativeIssueId: heartbeatRuns.nativeIssueId, resultJson: heartbeatRuns.resultJson })
.from(heartbeatRuns)
.where(eq(heartbeatRuns.id, runId))
.limit(1);
if (!run?.nativeIssueId || !run.resultJson?.externalChatReviewPresentation)
return null;
const decisionId = record(
run.resultJson.externalChatReviewPresentation,
).decisionId;
if (
typeof decisionId !== "string" ||
(await hasMaterializedNativeReviewResponse(dbOrTx, {
companyId: run.companyId,
issueId: run.nativeIssueId,
runId,
decisionId,
}))
)
return null;
const summary = record(run.resultJson.nativeResult).summary;
if (
typeof summary !== "string" ||
!(await authorizeNativeChatReviewPresentation(dbOrTx, {
companyId: run.companyId,
issueId: run.nativeIssueId,
runId,
resultJson: run.resultJson,
}))
)
return null;
await issueService(db).addComment(
run.nativeIssueId,
summary,
{ agentId: run.agentId, runId },
{
authorizationReason: "allow_chat_run_presentation",
},
dbOrTx,
);
return run;
}
function publishNativeReviewResponse(run: Pick<typeof heartbeatRuns.$inferSelect, "id" | "companyId" | "agentId" | "nativeIssueId">) {
publishChatPublicationCommitSignal({ companyId: run.companyId, issueId: run.nativeIssueId!,
runId: run.id, agentId: run.agentId, eventType: "run.presentation.resolved" });
}
async function materializeCommittedReviewResponse(db: Db, runId: string) {
try {
const run = await materializeNativeReviewResponse(db, runId);
if (run) publishNativeReviewResponse(run);
} catch (error) {
logger.warn({ runId, error }, "Committed chat review response is awaiting presentation retry");
}
}
async function acceptedResponseDigestMatches(
db: Db,
run: Pick<typeof heartbeatRuns.$inferSelect,
"id" | "companyId" | "agentId" | "nativeIssueId" | "nativeSessionId" |
"completionContractId" | "completionContractSha256" | "runnerInstanceId">,
accepted: typeof nativeRunResults.$inferSelect,
): Promise<boolean> {
const envelope = record(accepted.resultJson);
const canonical = {
result: envelope.result,
terminal: envelope.terminal,
turnId: accepted.turnId,
};
const fingerprint = nativeSha256({
runId: run.id,
completionContractSha256: run.completionContractSha256,
canonicalSha256: accepted.canonicalSha256,
});
// The durable coordinator stores this canonical shape directly. The older
// ControlPlanePort also hashes its exact binding, which must be rebuilt from
// persisted run identity and the actual authenticated control-plane journal.
if (
`sha256:${nativeSha256(canonical)}` === accepted.canonicalSha256 &&
`sha256:${fingerprint}` === accepted.serverFingerprint
)
return true;
if (accepted.serverFingerprint !== fingerprint) return false;
const sources = await db
.selectDistinct({ sourceInstanceId: heartbeatRunEvents.sourceInstanceId })
.from(heartbeatRunEvents)
.where(
and(
eq(heartbeatRunEvents.companyId, run.companyId),
eq(heartbeatRunEvents.runId, run.id),
sql`${heartbeatRunEvents.payload}->'prpEvent'->>'sourceKind' = 'control_plane'`,
),
)
.limit(17);
if (sources.length > 16) return false;
return sources.some(
({ sourceInstanceId }) =>
typeof sourceInstanceId === "string" &&
nativeSha256({
binding: {
companyId: run.companyId,
issueId: run.nativeIssueId,
runId: run.id,
agentId: run.agentId,
sessionId: run.nativeSessionId,
completionContractId: run.completionContractId,
completionContractSha256: run.completionContractSha256,
sourceInstanceId: run.runnerInstanceId,
controlPlaneSourceInstanceId: sourceInstanceId,
},
...canonical,
}) === accepted.canonicalSha256,
);
}
/** Recover only the already accepted answer. Neither issue disposition nor
* provider state is repaired here; an unsafe native session stays quarantined. */
export async function repairCommittedNativeChatResponse(
db: Db,
input: { companyId: string; issueId: string; runId: string },
): Promise<boolean> {
try {
return await recoverCommittedNativeChatResponse(db, input);
} catch (error) {
logger.warn(
{ runId: input.runId, error },
"Committed chat response is awaiting presentation retry",
);
return false;
}
}
export async function recoverCommittedNativeChatResponse(
db: Db,
input: { companyId: string; issueId: string; runId: string; onlyWhenDue?: boolean },
): Promise<boolean> {
return withRunResponsePresentationTiming(input.runId, () => recoverCommittedNativeChatResponseInScope(db, input));
}
async function recoverCommittedNativeChatResponseInScope(
db: Db,
input: { companyId: string; issueId: string; runId: string; onlyWhenDue?: boolean },
): Promise<boolean> {
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) => {
entered();
const tx = transaction as unknown as Db;
const [coordinator] = await tx
.select()
.from(nativeRunFinalizations)
.where(
and(
eq(nativeRunFinalizations.runId, input.runId),
eq(nativeRunFinalizations.companyId, input.companyId),
eq(nativeRunFinalizations.issueId, input.issueId),
eq(nativeRunFinalizations.phase, "committed"),
isNull(nativeRunFinalizations.leaseOwner),
),
)
.for("update");
if (
!coordinator?.resultId ||
!coordinator.decisionId ||
!coordinator.assessmentId
)
return false;
const [issue] = await tx
.select()
.from(issues)
.where(
and(
eq(issues.id, input.issueId),
eq(issues.companyId, input.companyId),
),
)
.for("update");
if (!issue) return false;
const [run] = await tx
.select({
id: heartbeatRuns.id,
companyId: heartbeatRuns.companyId,
agentId: heartbeatRuns.agentId,
responsibleUserId: heartbeatRuns.responsibleUserId,
nativeIssueId: heartbeatRuns.nativeIssueId,
nativeSessionId: heartbeatRuns.nativeSessionId,
completionContractId: heartbeatRuns.completionContractId,
completionContractSha256: heartbeatRuns.completionContractSha256,
runnerInstanceId: heartbeatRuns.runnerInstanceId,
runtimeMode: heartbeatRuns.runtimeMode,
status: heartbeatRuns.status,
errorCode: heartbeatRuns.errorCode,
nativePhase: heartbeatRuns.nativePhase,
finishedAt: heartbeatRuns.finishedAt,
resultJson: heartbeatRuns.resultJson,
contextSnapshot: sql<Record<string, unknown> | 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',
${heartbeatRuns.contextSnapshot} #> '{paperclipWake,externalInteractionContinuation}')
) end`,
})
.from(heartbeatRuns)
.where(
and(
eq(heartbeatRuns.id, input.runId),
eq(heartbeatRuns.companyId, input.companyId),
eq(heartbeatRuns.nativeIssueId, input.issueId),
eq(heartbeatRuns.runtimeMode, "native"),
),
)
.for("update");
if (
!run ||
isNativeRunnerOwnershipHeld(run) ||
!["succeeded", "failed", "timed_out"].includes(run.status) ||
!run.finishedAt
)
return false;
// An existing selected answer, including a subsequently deleted one, must
// never be replaced or resurrected by a background presentation repair.
const recordedPresentation = record(record(run.resultJson).presentationDecision);
if (
typeof recordedPresentation.commentId === "string" ||
(recordedPresentation.commentAction !== undefined && recordedPresentation.commentAction !== "create") ||
record(run.contextSnapshot).skipIssueComment === true
)
return false;
if (input.onlyWhenDue) {
if (coordinator.nextAttemptAt && coordinator.nextAttemptAt.getTime() > Date.now()) return false;
const [pendingEffect] = await tx.select({ id: statusDecisionEffects.id }).from(statusDecisionEffects).where(and(
eq(statusDecisionEffects.companyId, input.companyId),
eq(statusDecisionEffects.issueId, input.issueId),
eq(statusDecisionEffects.decisionId, coordinator.decisionId),
inArray(statusDecisionEffects.deliveryState, ["pending", "failed"]),
)).limit(1);
if (pendingEffect) return false;
}
const [accepted] = await tx
.select()
.from(nativeRunResults)
.where(
and(
eq(nativeRunResults.id, coordinator.resultId),
eq(nativeRunResults.runId, run.id),
eq(nativeRunResults.companyId, input.companyId),
eq(nativeRunResults.issueId, input.issueId),
eq(
nativeRunResults.completionContractId,
run.completionContractId!,
),
eq(nativeRunResults.schemaStatus, "accepted"),
),
)
.for("share");
const [decision] = await tx
.select()
.from(statusDecisions)
.where(
and(
eq(statusDecisions.id, coordinator.decisionId),
eq(statusDecisions.runId, run.id),
eq(statusDecisions.companyId, input.companyId),
eq(statusDecisions.issueId, input.issueId),
eq(statusDecisions.assessmentId, coordinator.assessmentId),
),
)
.for("share");
const envelope = record(accepted?.resultJson);
const result = record(envelope.result);
const terminal = record(envelope.terminal);
if (
!accepted ||
!decision ||
result.schema !== "paperclip.run_result.v1" ||
!(await acceptedResponseDigestMatches(tx, run, accepted))
)
return false;
// Existing governed-review responses have their own stricter gate-bound
// presentation contract, including selected attachments. Do not bypass it.
if (record(decision.decisionJson).externalChatReviewPresentation)
return false;
const monitorResult = { nativeResult: result, finalizationPhase: "committed", finalizationReasonCode: decision.reasonCode };
if (readAcceptedTaskContinuationSummary(monitorResult) && run.status === "succeeded" &&
terminal.runTerminalState === "succeeded" && terminal.turnTerminalState === "completed" &&
terminal.reportedWorkDisposition === "yielded" && record(run.contextSnapshot).skipIssueComment !== true &&
!await hasChatRunOwnedProviderInteraction(tx, input)) {
const authorizationReason = await resolveChatRunPresentationAuthorizationReason(tx, input);
const resolved = resolveHeartbeatRunResponse({ resultJson: monitorResult });
if (!resolved.text) return false;
await notifyChatRunMilestones(tx, input);
const comment = await issueService(db).addComment(input.issueId, resolved.text,
{ agentId: run.agentId, runId: run.id }, { authorizationReason, completionReply: true }, tx);
await tx.update(heartbeatRuns).set({
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({
presentationDecision: { ...resolved.decision, commentId: comment.id,
reasonCodes: [...resolved.decision.reasonCodes, "committed_monitor_response_recovered"] },
})}::jsonb`, updatedAt: new Date(),
}).where(eq(heartbeatRuns.id, run.id));
agentId = run.agentId;
return true;
}
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;
// A deferred ordinary-task review still needs its recorded final reply.
// Publishing that reply must not resolve the interaction or the task.
if (
(!conversation || currentConversation) &&
run.status === "succeeded" &&
record(run.contextSnapshot).skipIssueComment !== true &&
!isExternalChatPresentationContext(run.contextSnapshot) &&
terminal.runTerminalState === "succeeded" &&
terminal.turnTerminalState === "completed" &&
((result.reportedWorkDisposition === "done" && terminal.reportedWorkDisposition === "done") ||
(!conversation && result.reportedWorkDisposition === "needs_review" && terminal.reportedWorkDisposition === "needs_review")) &&
(await resolveChatRunPresentationAuthorizationReason(tx, input)) === "internal_agent_write"
) {
const rows = await tx.select({ seq: heartbeatRunEvents.seq, payload: heartbeatRunEvents.payload })
.from(heartbeatRunEvents).where(and(
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'`,
)).orderBy(desc(heartbeatRunEvents.seq)).limit(200);
const finalAgentMessage = selectHeartbeatRunFinalAgentMessage({
candidates: rows.flatMap((row) => {
const candidate = readCompletedAssistantMessageCandidate({ seq: row.seq, prpEvent: record(row.payload).prpEvent });
return candidate ? [candidate] : [];
}),
});
if (!finalAgentMessage) return false;
const comments = await tx.select({ id: issueComments.id, body: issueComments.body, presentation: issueComments.presentation })
.from(issueComments).where(and(
eq(issueComments.companyId, input.companyId),
eq(issueComments.issueId, input.issueId),
eq(issueComments.createdByRunId, run.id),
)).orderBy(desc(issueComments.createdAt), desc(issueComments.id));
const resolved = resolveHeartbeatRunResponse({
resultJson: { ...record(run.resultJson), nativeResult: result },
existingComment: findHeartbeatRunCompletionComment(comments, run.resultJson),
finalAgentMessage,
});
if (!resolved.text || resolved.decision.commentAction !== "create") return false;
await notifyChatRunMilestones(tx, input);
const comment = await issueService(db).addComment(input.issueId, resolved.text,
{ agentId: run.agentId, runId: run.id },
{ authorizationReason: "internal_agent_write", completionReply: true }, tx);
await tx.update(heartbeatRuns).set({
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(),
}).where(eq(heartbeatRuns.id, run.id));
agentId = run.agentId;
conversationReply = conversation;
return true;
}
if (
result.reportedWorkDisposition !== "yielded" ||
record(result.continuation).kind !== "response_wake" ||
!Array.isArray(result.attentionRequests) ||
result.attentionRequests.length > 0 ||
terminal.runTerminalState !== "succeeded" ||
terminal.turnTerminalState !== "completed" ||
terminal.reportedWorkDisposition !== "yielded"
) return false;
await authorizeCommittedChatResponse(db, tx, {
...input,
agentId: run.agentId,
resultId: accepted.id,
});
if (
(await resolveChatRunPresentationAuthorizationReason(tx, input)) !==
CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON
)
return false;
const resolved = resolveHeartbeatRunResponse({
resultJson: {
nativeResult: result,
finalizationPhase: "committed",
finalizationReasonCode: decision.reasonCode,
},
preferFinalResponseOverExistingComment: true,
externalChatResponseWakeSummaryAuthorized: true,
externalChatCommittedResponseWakeSummaryAuthorized: true,
});
if (!resolved.text || resolved.decision.commentAction !== "create")
return false;
await notifyChatRunMilestones(tx, input);
const comment = await issueService(db).addComment(
input.issueId,
resolved.text,
{ agentId: run.agentId, runId: run.id },
{ authorizationReason: CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON, completionReply: true },
tx,
);
const presentationDecision = {
...resolved.decision,
commentId: comment.id,
reasonCodes: [
...resolved.decision.reasonCodes,
"committed_response_recovered",
],
};
await tx
.update(heartbeatRuns)
.set({
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify(
{
presentationDecision,
nativeCommittedChatResponse: {
schema: "paperclip.native_committed_chat_response.v1",
resultId: accepted.id,
canonicalSha256: accepted.canonicalSha256,
decisionId: decision.id,
},
},
)}::jsonb`,
updatedAt: new Date(),
})
.where(eq(heartbeatRuns.id, run.id));
await projectCommittedRun({ db: tx, run, coordinator });
agentId = run.agentId;
return true;
}));
} catch (error) {
if (error instanceof CommittedChatResponseAuthorizationError) return false;
throw error;
} finally {
entered();
}
if (materialized && agentId) {
if (conversationReply) await settleConversationTurnForRun(db, input.companyId, input.runId);
publishChatPublicationCommitSignal({
...input,
agentId,
eventType: "run.presentation.resolved",
});
}
return materialized;
}
/** Presentation recovery is independent from re-arbitrating issue status. */
export async function repairCommittedNativeReviewResponse(
db: Db,
input: {
companyId: string;
issueId: string;
runId: string;
decisionId: string;
resultId: string;
assessmentId: string;
},
) {
if (await hasMaterializedNativeReviewResponse(db, input)) return;
const restored = await db.transaction((tx) =>
restoreNativeChatReviewPresentationInTransaction(
tx as unknown as Db,
input,
),
);
if (restored) await materializeCommittedReviewResponse(db, input.runId);
}
export async function recoverCommittedNativeReviewResponse(db: Db, input: {
companyId: string; issueId: string; runId: string; decisionId: string;
resultId: string; assessmentId: string; onlyWhenDue?: boolean;
}): Promise<boolean> {
const materialized = await db.transaction(async transaction => {
const tx = transaction as unknown as Db;
const [coordinator] = await tx.select({ id: nativeRunFinalizations.runId }).from(nativeRunFinalizations)
.where(and(eq(nativeRunFinalizations.companyId, input.companyId),
eq(nativeRunFinalizations.issueId, input.issueId), eq(nativeRunFinalizations.runId, input.runId))).for("update");
if (!coordinator) return null;
const [issue] = await tx.select({ id: issues.id }).from(issues)
.where(and(eq(issues.companyId, input.companyId), eq(issues.id, input.issueId))).for("update");
if (!issue || await hasMaterializedNativeReviewResponse(tx, input)) return null;
if (!await restoreNativeChatReviewPresentationInTransaction(tx, input)) return null;
return materializeNativeReviewResponse(db, input.runId, tx);
});
if (materialized) publishNativeReviewResponse(materialized);
return materialized !== null;
}
export async function finalizeNativeRun(input: {
db: Db;
runId: string;
workspaceFinalizeStatus: "succeeded" | "failed";
onlyWhenDue?: boolean;
/** Reconciliation owns terminal run projection; the live heartbeat does it afterward. */
projectRunStatus?: boolean;
skipNativeResponsePresentation?: boolean;
skipNativeReviewResponse?: boolean;
skipCommittedRunProjection?: boolean;
skipExternalConversationSettlement?: boolean;
skipExternalAgentSettlement?: boolean;
skipNativeRecoveryNotices?: boolean;
/** Workspace-only replay must not consume the provider recovery budget. */
preserveProviderAttempt?: boolean;
failpoint?: NativeStatusCommitFailpoint;
}) {
const admittedRun = await input.db
.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
runtimeMode: heartbeatRuns.runtimeMode, status: heartbeatRuns.status, errorCode: heartbeatRuns.errorCode,
nativeIssueId: heartbeatRuns.nativeIssueId,
contextSnapshot: sql<Record<string, unknown>>`jsonb_build_object('issueId', ${heartbeatRuns.contextSnapshot}->>'issueId')`,
nativePhase: heartbeatRuns.nativePhase, finishedAt: heartbeatRuns.finishedAt })
.from(heartbeatRuns)
.where(eq(heartbeatRuns.id, input.runId))
.limit(1)
.then((rows) => rows[0] ?? null);
if (!admittedRun || admittedRun.runtimeMode !== "native")
throw new Error("native_finalization_run_missing");
const claim = await claimCoordinator({
db: input.db,
runId: input.runId,
preserveProviderAttempt: input.preserveProviderAttempt,
onlyWhenDue: input.onlyWhenDue,
});
const coordinator = claim.coordinator;
if (!claim.leaseOwner && coordinator.phase === "committed") {
if (isNativeRunnerOwnershipHeld(admittedRun)) return coordinator;
const presentationAlreadyMaterialized = input.skipNativeReviewResponse || (coordinator.decisionId
? await hasMaterializedNativeReviewResponse(input.db, {
companyId: admittedRun.companyId,
issueId: coordinator.issueId,
runId: admittedRun.id,
decisionId: coordinator.decisionId,
})
: false);
if (
!presentationAlreadyMaterialized &&
coordinator.decisionId &&
coordinator.resultId &&
coordinator.assessmentId
) {
await input.db.transaction((tx) =>
restoreNativeChatReviewPresentationInTransaction(tx as unknown as Db, {
companyId: admittedRun.companyId,
issueId: coordinator.issueId,
runId: admittedRun.id,
decisionId: coordinator.decisionId!,
resultId: coordinator.resultId!,
assessmentId: coordinator.assessmentId!,
}),
);
}
if (input.projectRunStatus && !input.skipCommittedRunProjection)
await projectCommittedRun({ db: input.db, run: admittedRun, coordinator });
if (input.projectRunStatus && !presentationAlreadyMaterialized)
await materializeCommittedReviewResponse(input.db, input.runId);
if (input.projectRunStatus && !input.skipNativeResponsePresentation)
await repairCommittedNativeChatResponse(input.db, {
companyId: admittedRun.companyId,
issueId: coordinator.issueId,
runId: admittedRun.id,
});
if (input.projectRunStatus && !input.skipExternalConversationSettlement) {
await settleExternalConversation(input.db, admittedRun.companyId, coordinator.issueId).catch((err) => {
logger.warn({ err, runId: admittedRun.id }, "Slack conversation settlement deferred to reconciliation");
});
}
if (!input.skipNativeRecoveryNotices) await resolveCommittedNativeRecoveryNotices(input.db, { companyId: admittedRun.companyId, runId: admittedRun.id, issueId: coordinator.issueId });
if (input.projectRunStatus && !input.skipExternalAgentSettlement) {
const [projectedRun] = await input.db.select({
id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId,
runtimeMode: heartbeatRuns.runtimeMode, status: heartbeatRuns.status,
runnerProfileJson: sql<Record<string, unknown> | null>`case when ${heartbeatRuns.runnerProfileJson} is null then null
else jsonb_build_object('nativeExecutionInput', jsonb_build_object(
'schema', ${heartbeatRuns.runnerProfileJson} #> '{nativeExecutionInput,schema}',
'provider', jsonb_build_object(
'kind', ${heartbeatRuns.runnerProfileJson} #> '{nativeExecutionInput,provider,kind}',
'binding', jsonb_build_object(
'companyId', ${heartbeatRuns.runnerProfileJson} #> '{nativeExecutionInput,provider,binding,companyId}',
'agentId', ${heartbeatRuns.runnerProfileJson} #> '{nativeExecutionInput,provider,binding,agentId}',
'bindingId', ${heartbeatRuns.runnerProfileJson} #> '{nativeExecutionInput,provider,binding,bindingId}'
)
)
)) end`,
}).from(heartbeatRuns).where(eq(heartbeatRuns.id, admittedRun.id));
if (projectedRun) await reconcileMuseNativeTerminalRun(input.db, projectedRun);
}
return coordinator;
}
const run = await input.db.select().from(heartbeatRuns)
.where(eq(heartbeatRuns.id, input.runId)).limit(1).then(rows => rows[0] ?? null);
if (!run || run.runtimeMode !== "native") throw new Error("native_finalization_run_missing");
const [resultRow, contractRow] = await Promise.all([
input.db
.select()
.from(nativeRunResults)
.where(
and(
eq(nativeRunResults.id, coordinator.resultId!),
eq(nativeRunResults.companyId, run.companyId),
),
)
.limit(1)
.then((rows) => rows[0] ?? null),
input.db
.select()
.from(completionContracts)
.where(
and(
eq(completionContracts.id, run.completionContractId!),
eq(completionContracts.companyId, run.companyId),
eq(completionContracts.issueId, coordinator.issueId),
),
)
.limit(1)
.then((rows) => rows[0] ?? null),
]);
if (!resultRow) throw new Error("native_result_missing");
if (!contractRow) throw new Error("native_completion_contract_missing");
const envelope = record(resultRow.resultJson);
const result = record(envelope.result);
const terminal = record(envelope.terminal);
const terminalState = terminal.runTerminalState;
if (!["succeeded", "failed", "cancelled"].includes(String(terminalState))) {
throw new Error("native_finalization_invalid");
}
await input.db
.update(nativeRunFinalizations)
.set({ phase: "ready_for_assessment", updatedAt: new Date() })
.where(
and(
eq(nativeRunFinalizations.runId, input.runId),
eq(nativeRunFinalizations.leaseOwner, claim.leaseOwner!),
),
);
const classifiedAssessment = await classifyNativeEvidence({
db: input.db,
companyId: run.companyId,
issueId: coordinator.issueId,
runId: run.id,
contract: record(contractRow.contractJson),
result,
});
const persistedIgnoredAttention = record(
envelope.normalizationDiagnostics,
).ignoredAttentionRequests;
const assessment = {
...classifiedAssessment,
ignoredAttentionRequests: [
...(Array.isArray(persistedIgnoredAttention)
? (persistedIgnoredAttention as PrpIgnoredAttentionRequest[])
: []),
...classifiedAssessment.ignoredAttentionRequests,
],
};
await dismissAutomaticCompletionReviews(input.db, coordinator.issueId);
const sourceWake = run.wakeupRequestId ? await input.db.select({ payload: agentWakeupRequests.payload })
.from(agentWakeupRequests).where(and(eq(agentWakeupRequests.id, run.wakeupRequestId),
eq(agentWakeupRequests.companyId, run.companyId))).then((rows) => rows[0]) : null;
// One follow-up may repair an incomplete report. Repeated incomplete results
// require a visible recovery action instead of an unbounded wake loop.
const allowIncompleteContinuation = record(sourceWake?.payload).continuationIdempotencyKey !== "native-completion-incomplete";
let supersedesAssessmentId: string | null = coordinator.assessmentId;
for (let attempt = 0; attempt < 3; attempt += 1) {
const authoritativeIssue = await input.db
.select()
.from(issues)
.where(
and(
eq(issues.id, coordinator.issueId),
eq(issues.companyId, run.companyId),
),
)
.limit(1)
.then((rows) => rows[0] ?? null);
if (!authoritativeIssue)
throw new Error("native_finalization_issue_missing");
const preservedResultFinalization = run.status !== "running"
? await readPreservedNativeResultFinalization(input.db, {
companyId: run.companyId, issueId: authoritativeIssue.id, runId: run.id, agentId: run.agentId,
}) : null;
const publishedReviewContinuation = assessment.reportedDisposition === "yielded" &&
isPublishedGitHubReviewCompletionContinuation(assessment.continuation)
? await readPublishedGitHubReviewCompletionContinuation({ db: input.db,
binding: { companyId: run.companyId, issueId: authoritativeIssue.id, runId: run.id, agentId: run.agentId } })
: null;
const externalChatResponseWaitAuthorization =
assessment.reportedDisposition === "yielded" &&
assessment.continuation?.kind === "response_wake"
? isPublishedGitHubReviewCompletionContinuation(assessment.continuation)
? publishedReviewContinuation?.idempotencyKey === assessment.continuation.idempotencyKey ? "authorized" : "revoked"
: await resolveExternalChatResponseWaitAuthorization({
db: input.db,
binding: {
companyId: run.companyId,
issueId: authoritativeIssue.id,
runId: run.id,
agentId: run.agentId,
},
})
: "not_applicable";
const monitorWaitAt = scheduledIssueMonitorWait(authoritativeIssue, run.agentId);
const preservedMonitorWait = terminalState === "succeeded" && input.workspaceFinalizeStatus === "succeeded" &&
assessment.reportedDisposition === "yielded" && assessment.continuation?.kind === "monitor"
? await readNativePreservedMonitorWait(input.db, {
companyId: run.companyId, issueId: authoritativeIssue.id, runId: run.id, agentId: run.agentId,
}, authoritativeIssue) : null;
if (preservedMonitorWait && preservedMonitorWait.resultId !== resultRow.id) throw new NativeStatusRaceError();
const supersededReviews = terminalState === "succeeded" && input.workspaceFinalizeStatus === "succeeded" &&
assessment.attentionRequests.length === 0 && (assessment.reportedDisposition === "done" ||
(assessment.reportedDisposition === "yielded" && (externalChatResponseWaitAuthorization === "authorized" ||
(assessment.continuation?.kind === "monitor" && monitorWaitAt !== null && !preservedMonitorWait))))
? await findSupersededExternalReviewRequests(input.db, {
companyId: run.companyId, issueId: authoritativeIssue.id, runId: run.id, agentId: run.agentId,
}) : [];
const governanceGate = await pendingNativeGovernance({
db: input.db,
companyId: run.companyId,
issueId: authoritativeIssue.id,
runId: run.id,
executionState: record(authoritativeIssue.executionState),
ignoredInteractionIds: supersededReviews.map(({ interaction }) => interaction.id),
});
const boardResponseWaitOrigin =
externalChatResponseWaitAuthorization === "not_applicable" &&
assessment.reportedDisposition === "yielded" &&
assessment.continuation?.kind === "response_wake" &&
!governanceGate
? await readNativeBoardResponseWaitOrigin(input.db, {
companyId: run.companyId,
issueId: authoritativeIssue.id,
runId: run.id,
agentId: run.agentId,
})
: null;
const boardResponseWait = boardResponseWaitOrigin
? await readNativeBoardResponseWaitSource(
input.db,
boardResponseWaitOrigin,
)
: null;
const [dependencyReadiness, resolvedInteraction, pauseHold] = await Promise.all([
issueService(input.db).getDependencyReadiness(
authoritativeIssue.id,
input.db,
),
acceptedInteractionFromRun({
db: input.db,
companyId: run.companyId,
issueId: authoritativeIssue.id,
runId: run.id,
}),
(terminalState === "failed" || (assessment.reportedDisposition === "yielded" &&
assessment.continuation?.kind === "response_wake" &&
assessment.hasBlockingRemainingWork))
? issueTreeControlService(input.db).getActivePauseHoldGate(
run.companyId, authoritativeIssue.id,
)
: Promise.resolve(null),
]);
const reviewContext = readNativeReviewAssignmentContext(run.contextSnapshot);
const nativeReview = reviewContext ? await getNativeReviewAssignment(input.db, {
companyId: run.companyId, issueId: authoritativeIssue.id, agentId: run.agentId,
contextSnapshot: reviewContext, allowResolvedByRunId: run.id,
}) : null;
const planWait = isNativePlanWaitResult(result)
? await readNativePlanWait(input.db, { companyId: run.companyId, issueId: authoritativeIssue.id, runId: run.id, agentId: run.agentId })
: null;
// Loss of the authority behind this server-issued wait must never fall
// through to the generic response_wake auto-continuation branch.
if (isNativePlanWaitResult(result) &&
(!planWait || nativeSha256(planWait.result) !== nativeSha256(result))) {
throw new Error("native_plan_wait_authority_lost");
}
const { providerFailure, turnFailure, providerExplanation } = await readPersistedNativeFailureEvidence(
input.db, run, resultRow.turnId, envelope.terminal as PrpTerminalState,
);
const ownsProviderFailureDecision =
(!authoritativeIssue.executionRunId || authoritativeIssue.executionRunId === run.id) &&
(reviewContext ? nativeReview?.interaction.status === "pending"
: authoritativeIssue.assigneeAgentId === run.agentId && !authoritativeIssue.assigneeUserId);
const childCompletionRecipient = {
companyId: run.companyId, issueId: authoritativeIssue.id, agentId: run.agentId, runId: run.id,
sourceIntentId: typeof record(run.contextSnapshot).nativeStatusWakeIntentId === "string"
? record(run.contextSnapshot).nativeStatusWakeIntentId as string : null,
};
const hasPendingChildCompletion = !reviewContext &&
await hasPendingNativeChildCompletion(input.db, childCompletionRecipient);
const monitorDelivery = assessment.reportedDisposition === "yielded" &&
assessment.continuation?.kind === "monitor" &&
authoritativeIssue.executionRunId !== run.id && authoritativeIssue.checkoutRunId !== run.id
? await readNativeMonitorDelivery(input.db, { companyId: run.companyId, issueId: coordinator.issueId, runId: run.id })
: null;
const qualityReview = await readRequiredQualityReview(input.db, authoritativeIssue);
const statusInput: Parameters<typeof resolveNativeFinalizerStatus>[0] = {
requiredQualityReviewSatisfied: qualityReview.satisfied,
caseHandoffAuthorized: Boolean(await assertNativeCaseHandoffResult(input.db, run.id, authoritativeIssue.id, result)),
monitorWaitDispatched: monitorDelivery !== null,
monitorWaitPreserved: preservedMonitorWait !== null,
monitorWaitAuthorized: authoritativeIssue.workMode === "standard" && authoritativeIssue.executionRunId === run.id && monitorWaitAt !== null,
planWaitAuthorized: planWait !== null,
hasPendingChildCompletion,
providerModelRejected: providerFailure?.errorCode === "native_provider_model_rejected" && ownsProviderFailureDecision,
providerOverloaded: providerFailure?.errorCode === "native_provider_overloaded" && ownsProviderFailureDecision,
providerFailureSuperseded: providerFailure?.errorCode === "native_provider_overloaded" && !ownsProviderFailureDecision,
failureRetryCount: executionFailureRetryCount(run),
...(reviewContext ? { nativeReviewOutcome: nativeReview
? nativeReview.interaction.status === "pending" ? "pending" as const : "resolved" as const
: "stale" as const } : {}),
assessment,
terminalState: terminalState as "succeeded" | "failed" | "cancelled",
workspaceFinalizeStatus: input.workspaceFinalizeStatus,
governanceGate,
allowIncompleteContinuation,
completionClaimPolicyAccepted:
contractRow.risk === "low" &&
contractRow.completionAuthority === "agent_claim_policy",
hasUnresolvedIssueBlockers:
dependencyReadiness.unresolvedBlockerCount > 0,
governanceResolvedForRun: resolvedInteraction !== null,
externalChatResponseWaitAuthorization,
hasUnprovenPreservedResultStatus: preservedResultFinalization?.statusRestoration === "unproven",
publishedExternalReview: assessment.reportedDisposition === "yielded" &&
(externalChatResponseWaitAuthorization === "authorized" || monitorWaitAt !== null)
? await hasPublishedGitHubReview(input.db, {
companyId: run.companyId, issueId: authoritativeIssue.id, runId: run.id,
}) || (await readInheritedGitHubReviewWait(input.db, run))?.kind === "published" : false,
boardResponseWaitAuthorized: boardResponseWait !== null,
boardResponseWaitOrigin: boardResponseWaitOrigin !== null,
isConversation: isConversation(authoritativeIssue),
hasActivePauseHold: pauseHold !== null,
reviewOwnerUserId:
authoritativeIssue.responsibleUserId ??
authoritativeIssue.createdByUserId ??
null,
agentId: run.agentId,
priorIssueStatus: authoritativeStatus(authoritativeIssue.status),
};
let proposedDecision = resolveNativeFinalizerStatus(statusInput);
// Already accepted results still require the server-owned first-task decision.
// Keep the task open without inventing an interaction or rerunning the work.
if (proposedDecision.statusAction === "done" && !await onboardingFirstTaskCompletionAllowed(input.db, {
companyId: run.companyId, issueId: authoritativeIssue.id, runId: run.id,
})) {
proposedDecision = { ...proposedDecision, statusAction: "in_review", toStatus: "in_review",
reasonCode: "onboarding_first_task_decision_required", unblockDescriptor: null, effects: [{ kind: "release_checkout" }] };
}
// A reassigned-away run keeps its accepted result, but cannot close the
// new owner's task. The status committer rechecks this under the issue lock.
if (!isConversation(authoritativeIssue)) {
proposedDecision = supersedeStaleOwnerCompletion(proposedDecision, authoritativeIssue, run.agentId);
}
if (proposedDecision.statusAction === "done") {
const continuation = record(run.contextSnapshot?.executionContinuation);
const objective = typeof continuation.objective === "string" ? continuation.objective
: [authoritativeIssue.title, authoritativeIssue.description].filter(Boolean).join("\n");
try {
await validateNativeDeliverableEvidence(input.db, { companyId: run.companyId, issueId: authoritativeIssue.id,
runId: run.id, objective, semanticToolReceipts: run.resultJson?.semanticToolReceipts }, result);
} catch (error) {
if (!(error instanceof NativeDeliverableEvidenceError)) throw error;
const incomplete = resolveNativeFinalizerStatus({ ...statusInput,
assessment: { ...assessment, hasBlockingRemainingWork: true, continuation: null } });
proposedDecision = { ...incomplete, effects: incomplete.effects.map(effect =>
effect.kind === "enqueue_continuation" ? { ...effect, summary: error.message }
: effect.kind === "record_finalization_error" ? { ...effect, nextAction: error.message } : effect) };
}
}
const decision = conversationNativeDecision({
conversation: isConversation(authoritativeIssue), terminalState,
workspaceFinalizeStatus: input.workspaceFinalizeStatus, hasGovernanceGate: !!governanceGate,
priorStatus: authoritativeStatus(authoritativeIssue.status), decision: proposedDecision,
});
const assessmentRow = await recordNativeWorkAssessment({
db: input.db,
companyId: run.companyId,
issueId: authoritativeIssue.id,
runId: run.id,
turnId: resultRow.turnId,
contractId: resultRow.completionContractId,
contractCanonicalSha256: contractRow.canonicalSha256,
resultId: resultRow.id,
resultCanonicalSha256: resultRow.canonicalSha256,
priorIssueStatus: authoritativeIssue.status,
priorStatusVersion: Number(authoritativeIssue.statusVersion),
priorDecisionId: authoritativeIssue.lastStatusDecisionId,
policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION,
assessment,
supersedesAssessmentId,
workspaceFinalizeStatus: input.workspaceFinalizeStatus,
monitorDelivery,
monitorWaitAt: assessment.continuation?.kind === "monitor" ? monitorWaitAt : null,
preservedMonitorWait,
});
await input.db
.update(nativeRunFinalizations)
.set({
phase: "arbitrating",
// The old decision refers to its original assessment through a
// composite foreign key. Replace this pair together in the status
// committer, after the new decision and its effects are durable.
// Before any decision exists, retain the assessment for crash recovery.
assessmentId: sql`case when ${nativeRunFinalizations.decisionId} is null
then ${assessmentRow.id}::uuid else ${nativeRunFinalizations.assessmentId} end`,
updatedAt: new Date(),
})
.where(
and(
eq(nativeRunFinalizations.runId, input.runId),
eq(nativeRunFinalizations.leaseOwner, claim.leaseOwner!),
),
);
try {
const repairBoardResponseWait =
boardResponseWait !== null &&
assessment.hasBlockingRemainingWork &&
["completion_evidence_incomplete", "prior_status_preserved_no_live_path"].includes(decision.reasonCode ?? "");
const committed = await commitNativeStatusDecision({
db: input.db,
companyId: run.companyId,
issueId: authoritativeIssue.id,
runId: run.id,
assessmentId: assessmentRow.id,
priorStatus: authoritativeIssue.status,
priorStatusVersion: Number(authoritativeIssue.statusVersion),
priorDecisionId: authoritativeIssue.lastStatusDecisionId,
decision,
requirePreservedResultFinalization: preservedResultFinalization ?? undefined,
supersededExternalReviewRequests: supersededReviews.length && !governanceGate &&
["completion_claim_policy_accepted", "completion_contract_satisfied", "external_chat_response_waiting", "scheduled_monitor_waiting"].includes(decision.reasonCode ?? "")
? { agentId: run.agentId, resultId: resultRow.id, interactionIds: supersededReviews.map(({ interaction }) => interaction.id) } : undefined,
requireMonitorWait: decision.reasonCode === "scheduled_monitor_waiting" && monitorWaitAt
? { agentId: run.agentId, nextCheckAt: monitorWaitAt } : undefined,
requireMonitorDelivery: decision.reasonCode === "scheduled_monitor_dispatched"
? monitorDelivery ?? undefined : undefined,
requirePreservedMonitorWait: decision.reasonCode === "scheduled_monitor_preserved"
? preservedMonitorWait ?? undefined : undefined,
requirePlanWaitSource:
decision.reasonCode === "native_plan_accepted_waiting_for_continuation"
? planWait?.source : undefined,
requireNoPendingChildCompletion: decision.statusAction === "done" && !reviewContext
? childCompletionRecipient : undefined,
requireModelRejectionOwner: decision.reasonCode === "native_provider_model_rejected"
? { agentId: run.agentId, reviewContext }
: undefined,
requireProviderFailureOwner: decision.reasonCode?.startsWith("native_provider_overloaded")
? { agentId: run.agentId, reviewContext } : undefined,
requireBoardResponseWaitSource:
decision.reasonCode === "board_response_waiting" || repairBoardResponseWait
? boardResponseWait?.source
: undefined,
requireBoardResponseWaitOrigin:
decision.reasonCode === "board_response_waiting" ||
decision.reasonCode === "board_response_wait_superseded" ||
repairBoardResponseWait
? (boardResponseWaitOrigin ?? undefined)
: undefined,
requireExternalChatResponseWaitAuthorization:
["external_chat_response_waiting", "preserved_result_status_unproven"].includes(decision.reasonCode ?? "")
? { agentId: run.agentId, ...(publishedReviewContinuation ? { completionContinuationKey: publishedReviewContinuation.idempotencyKey } : {}) }
: undefined,
reviewResponsePresentation:
decision.reasonCode === "governed_response_waiting" &&
governanceGate?.kind === "interaction" &&
externalChatResponseWaitAuthorization === "authorized"
? {
agentId: run.agentId,
resultId: resultRow.id,
gateId: governanceGate.id,
}
: undefined,
failpoint: input.failpoint,
});
const now = new Date();
const finalizationFailed = decision.effects.some(
(effect) => effect.kind === "record_finalization_error",
);
const finalizationPhase = finalizationFailed
? "retryable_failure"
: "committed";
// commitNativeStatusDecision() already committed its own transaction above.
// Its "cancel_continuations" effect (status-decision-committer.ts) writes a
// terminal status to this same run and emits for it. Skip the emit here in
// that case so one run reaching a terminal state emits exactly one event.
const alreadyEmittedByCommittedDecision = decision.effects.some(
(effect) => effect.kind === "cancel_continuations",
);
const clearExecutionFailure = input.projectRunStatus && finalizationPhase === "committed" && terminalState === "succeeded";
const finalizationMetadata = {
...(providerExplanation ? { providerFailure: providerExplanation } : {}),
...(providerFailure ? { nativeProviderFailure: providerFailure.diagnostic } : {}),
...(turnFailure ? { nativeTurnFailure: turnFailure } : {}),
finalizationPhase,
...(finalizationPhase === "committed" ? { failureCode: null, originalFailureCode: null, nextAttemptAt: null } : {}),
assessmentId: assessmentRow.id,
decisionId: committed.decision.id,
authoritativeDecision: decision.toStatus,
finalizationPolicyVersion: decision.policyVersion,
finalizationReasonCode: decision.reasonCode,
// Only the locked status transaction can mint this presentation
// proof. Never carry a runner-provided marker forward.
externalChatReviewPresentation: record(
committed.decision.decisionJson,
).externalChatReviewPresentation
? {
...record(
record(committed.decision.decisionJson)
.externalChatReviewPresentation,
),
decisionId: committed.decision.id,
}
: null,
...(record(committed.decision.decisionJson)
.externalChatReviewPresentation
? { nativeResult: result }
: {}),
verificationCaveats: assessment.verificationCaveats,
ignoredAttentionRequests: assessment.ignoredAttentionRequests,
issueStatusBefore: authoritativeIssue.status,
issueStatusAfter: committed.issue.status,
statusVersionBefore: Number(authoritativeIssue.statusVersion),
statusVersionAfter: Number(committed.issue.statusVersion),
workspaceFinalizeStatus: input.workspaceFinalizeStatus,
};
const projectedStatus = terminalState === "succeeded" ? "succeeded" as const
: terminalState === "cancelled" ? "cancelled" as const : "failed" as const;
// Workspace persistence may finish after Stop has already cancelled the
// run. Preserve that terminal outcome while retaining the accepted result
// and finalization receipts. Evaluate against the locked current row,
// including a cancellation that won after this finalizer read the run.
const projectedRunStatus = sql`case when ${heartbeatRuns.status} = 'cancelled'
then ${heartbeatRuns.status} else ${projectedStatus} end`;
const projectedResult = sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb)
|| ${JSON.stringify(finalizationMetadata)}::jsonb
|| ${clearExecutionFailure ? sql`case when ${heartbeatRuns.status} = 'cancelled' then '{}'::jsonb
else ${recoveredExecutionFailureMetadata()} end` : sql`'{}'::jsonb`}`;
const nextError = providerFailure ? sql`coalesce(${heartbeatRuns.error}, ${providerFailure.errorMessage})`
: clearExecutionFailure ? sql`null` : sql`${heartbeatRuns.error}`;
const nextErrorCode = providerFailure ? sql`coalesce(${heartbeatRuns.errorCode}, ${providerFailure.errorCode})`
: clearExecutionFailure ? sql`null` : sql`${heartbeatRuns.errorCode}`;
// A run projected to failed always records why, even when the provider
// reported a failed terminal without a classified failure.
const projectsFailedStatus = input.projectRunStatus && projectedStatus === "failed";
const explained = projectsFailedStatus
? explainFailedRun(NATIVE_RUN_FAILED_UNEXPLAINED, { error: nextError, errorCode: nextErrorCode })
: null;
const projectedError = explained?.error ?? sql`case when ${heartbeatRuns.status} = 'cancelled' then ${heartbeatRuns.error}
else ${nextError} end`;
const projectedErrorCode = explained?.errorCode ?? sql`case when ${heartbeatRuns.status} = 'cancelled' then ${heartbeatRuns.errorCode}
else ${nextErrorCode} end`;
const projectionChanged = or(
sql`${heartbeatRuns.nativePhase} is distinct from ${finalizationPhase}`,
sql`${heartbeatRuns.resultJson} is distinct from ${projectedResult}`,
sql`${heartbeatRuns.error} is distinct from ${projectedError}`,
sql`${heartbeatRuns.errorCode} is distinct from ${projectedErrorCode}`,
...(input.projectRunStatus ? [
sql`${heartbeatRuns.status} is distinct from ${projectedRunStatus}`,
isNull(heartbeatRuns.finishedAt),
] : []),
);
const projectRun = (tx: Db) => tx
.update(heartbeatRuns)
.set({
executionStatusDeliveryId: sql`case when ${projectionChanged}
then ${randomUUID()}::uuid else ${heartbeatRuns.executionStatusDeliveryId} end`,
...(input.projectRunStatus
? {
status: projectedRunStatus,
finishedAt: sql`case when ${heartbeatRuns.status} = 'cancelled' then ${heartbeatRuns.finishedAt}
when ${heartbeatRuns.status} is distinct from ${projectedStatus} then ${now.toISOString()}::timestamptz else coalesce(${heartbeatRuns.finishedAt}, ${now.toISOString()}::timestamptz) end`,
}
: {}),
nativePhase: finalizationPhase,
nativePhaseUpdatedAt: now,
...(clearExecutionFailure || providerFailure || projectsFailedStatus ? { error: projectedError, errorCode: projectedErrorCode } : {}),
resultJson: projectedResult,
updatedAt: now,
})
.where(eq(heartbeatRuns.id, run.id))
.returning();
const projection = await input.db.transaction(async transaction => {
const tx = transaction as unknown as Db;
const [before] = await tx.select({ status: heartbeatRuns.status, runtimeMode: heartbeatRuns.runtimeMode,
nativePhase: heartbeatRuns.nativePhase, finishedAt: heartbeatRuns.finishedAt })
.from(heartbeatRuns).where(and(eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId)))
.for("update").limit(1);
if (!before) return null;
if (input.projectRunStatus) await notifyExternalAgentSettlement(tx, { companyId: run.companyId, runId: run.id });
await notifyNativeRecoveryNotices(tx, { companyId: run.companyId, runId: run.id });
await notifyNativeRunProjection(tx, { companyId: run.companyId, runId: run.id });
await notifyNativeEnvironmentSettlement(tx, { companyId: run.companyId, runId: run.id });
await notifyRetainedNativeSessionCleanup(tx, { companyId: run.companyId, runId: run.id });
if (input.projectRunStatus) await notifyInstructionCollection(tx, { companyId: run.companyId, runId: run.id });
if (input.projectRunStatus) await notifyCancelledWorkspacePreservation(tx, { companyId: run.companyId, runId: run.id });
if (input.projectRunStatus) await notifyUnsafeWorkspaceExportRunChange(tx, run);
if (input.projectRunStatus) await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, run.companyId);
await notifyExecutionRunWork(tx, { companyId: run.companyId });
await notifyChatRunMilestones(tx, { companyId: run.companyId, issueId: coordinator.issueId });
const [after] = await projectRun(tx);
if (after && input.projectRunStatus) await settleNativeTerminalWake(tx, after);
return after ? { before, after } : null;
});
const updatedRun = projection?.after;
if (
input.projectRunStatus &&
!alreadyEmittedByCommittedDecision &&
projection && projection.after.status !== projection.before.status
) {
await emitAgentTaskRun(input.db, projection.after);
void reportRunFailure(input.db, projection.after);
} else if (input.projectRunStatus && !alreadyEmittedByCommittedDecision && projection
&& isNewTerminalRunFailure(projection.before, projection.after)) {
void reportRunFailure(input.db, projection.after);
}
if (input.projectRunStatus)
await materializeCommittedReviewResponse(input.db, input.runId);
if (input.projectRunStatus && finalizationPhase === "committed" && !input.skipNativeResponsePresentation)
await repairCommittedNativeChatResponse(input.db, {
companyId: run.companyId,
issueId: coordinator.issueId,
runId: run.id,
});
if (input.projectRunStatus && finalizationPhase === "committed" && !input.skipExternalConversationSettlement) {
await settleExternalConversation(input.db, run.companyId, coordinator.issueId).catch((err) => {
logger.warn({ err, runId: run.id }, "Slack conversation settlement deferred to reconciliation");
});
}
if (finalizationPhase === "committed" && !input.skipNativeRecoveryNotices) await resolveCommittedNativeRecoveryNotices(input.db, { companyId: run.companyId, runId: run.id, issueId: coordinator.issueId });
if (input.projectRunStatus && updatedRun && !input.skipExternalAgentSettlement) await reconcileMuseNativeTerminalRun(input.db, updatedRun);
return {
...coordinator,
phase: finalizationPhase,
assessmentId: assessmentRow.id,
decisionId: committed.decision.id,
};
} catch (error) {
if (error instanceof NativeStatusRaceError && attempt < 2) {
supersedesAssessmentId = assessmentRow.id;
await new Promise((resolve) => setTimeout(resolve, 25 * (attempt + 1)));
continue;
}
return recordRetryableFailure({
db: input.db,
run,
coordinator,
failureCode:
error instanceof NativeStatusRaceError
? "status_cas_exhausted"
: "side_effect_planning_failed",
message: error instanceof Error ? error.message : String(error),
nextAction:
error instanceof NativeStatusRaceError
? "Reassess against the latest authoritative issue status version."
: "Retry atomic status and liveness materialization from the persisted assessment.",
projectRunStatus: input.projectRunStatus,
});
}
}
throw new Error("native_finalization_status_race");
}