PaperClipAI/server/src/services/native-runtime/native-run-finalizer.ts

import { reconcileMuseNativeTerminalRun } from "./muse-terminal-reconciliation.js";
import { assertNativeCaseHandoffResult } from "./native-case-handoff.js";
import { scheduledIssueMonitorWait } from "../issue-monitors.js";
import { readNativeMonitorDelivery } from "./native-monitor-delivery.js";
import { isNativePlanWaitResult, readNativePlanWait } from "./native-plan-wait.js";
import { activeIssueInteractionCondition } from "../issue-question-context.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 { dismissAutomaticCompletionReviews } from "./automatic-completion-reviews.js";
import { getNativeReviewAssignment, readNativeReviewAssignmentContext } from "./native-review-participant.js";
import { conversationNativeDecision, isConversation } 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 {
  approvals,
  agentWakeupRequests,
  completionContracts,
  heartbeatRuns,
  heartbeatRunEvents,
  issueApprovals,
  issueComments,
  issueRecoveryActions,
  issueThreadInteractions,
  issues,
  nativeRunFinalizations,
  nativeRunResults,
  statusDecisions,
  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,
  type NativeAuthoritativeIssueStatus,
  type NativeGovernanceGate,
} 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 { issueService } from "../issues.js";
import { publishChatPublicationCommitSignal } from "../chat-publication-reconciliation.js";
import { nativeSha256 } from "./canonical.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 { 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,
  nativeRunnerOwnershipNotHeldCondition,
} 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
      );
    })
  );
}

/** Server-owned terminal conversion used by live finalization and read models. */
export function projectNativeTerminalRunStatus(
  terminalState: "succeeded" | "failed" | "cancelled" | "active",
) {
  return terminalState === "active" ? ("running" as const) : terminalState;
}

/** Fact-based arbitration at the production finalization authority seam. */
export function resolveNativeFinalizerStatus(
  input: Parameters<typeof arbitrateNativeStatus>[0],
) {
  return arbitrateNativeStatus(input);
}

export async function pendingNativeGovernance(input: {
  db: Db;
  companyId: string;
  issueId: string;
  runId: string;
  executionState: Record<string, unknown> | null;
}): Promise<NativeGovernanceGate | null> {
  // Execution policy is issue-owned state and takes priority over the two
  // durable review surfaces. Completion must mediate all three authorities.
  if (record(input.executionState).status === "pending") {
    return { kind: "execution_stage", id: input.runId };
  }
  const [issue] = await input.db.select({ conversationAgentId: issues.conversationAgentId, conversationUserId: issues.conversationUserId })
    .from(issues).where(and(eq(issues.id, input.issueId), eq(issues.companyId, input.companyId)));
  const [pendingInteraction, pendingApproval] = await Promise.all([
    input.db
      .select({ id: issueThreadInteractions.id })
      .from(issueThreadInteractions)
      .where(
        and(
          eq(issueThreadInteractions.companyId, input.companyId),
          eq(issueThreadInteractions.issueId, input.issueId),
          eq(issueThreadInteractions.status, "pending"),
          activeIssueInteractionCondition({ runId: input.runId, conversationMode: isConversation(issue) }),
        ),
      )
      .limit(1)
      .then((rows) => rows[0] ?? null),
    input.db
      .select({ id: approvals.id })
      .from(issueApprovals)
      .innerJoin(approvals, eq(issueApprovals.approvalId, approvals.id))
      .where(
        and(
          eq(issueApprovals.companyId, input.companyId),
          eq(issueApprovals.issueId, input.issueId),
          eq(approvals.companyId, input.companyId),
          inArray(approvals.status, ["pending", "revision_requested"]),
        ),
      )
      .limit(1)
      .then((rows) => rows[0] ?? null),
  ]);
  if (pendingInteraction)
    return { kind: "interaction", id: pendingInteraction.id };
  if (pendingApproval) return { kind: "approval", id: pendingApproval.id };
  return null;
}

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);
}

async function claimCoordinator(input: {
  db: Db;
  runId: string;
  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 (
      coordinator.leaseOwner &&
      coordinator.leaseExpiresAt &&
      coordinator.leaseExpiresAt > now &&
      coordinator.leaseOwner !== leaseOwner
    )
      throw new Error("native_finalization_lease_busy");
    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,
        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;
}

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;
}) {
  const now = new Date();
  const nextAttemptAt = new Date(now.getTime() + 30_000);
  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,
      })
      .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;
    if (phase === "terminal_failure") await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.run.companyId);
    await tx
      .update(nativeRunFinalizations)
      .set({
        phase,
        leaseOwner: null,
        leaseExpiresAt: null,
        failureCode,
        failureDetail: {
          message: input.message.slice(0, 2_000),
          originalFailureCode: input.failureCode,
          ...(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);
    }
    const [updatedRun] = await tx
      .update(heartbeatRuns)
      .set({
        executionStatusDeliveryId: randomUUID(),
        ...(projectsTerminalStatus
          ? {
              status:
                input.failureScope === "workspace"
                  ? "failed"
                  : exhaustedRunStatus,
              finishedAt: input.run.finishedAt ?? now,
            }
          : {}),
        nativePhase: phase,
        nativePhaseUpdatedAt: now,
        resultJson: {
          ...record(input.run.resultJson),
          finalizationPhase: phase,
          failureCode,
          originalFailureCode: input.failureCode,
          nextAttemptAt:
            supersededByNewerRun || exhausted
              ? null
              : nextAttemptAt.toISOString(),
        },
        updatedAt: now,
      })
      .where(eq(heartbeatRuns.id, input.run.id))
      .returning();
    if (projectsTerminalStatus) terminalRunToEmit = updatedRun ?? null;
    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) {
      await issueService(tx as unknown as Db).update(
        input.coordinator.issueId,
        {
          status:
            input.failureScope === "workspace"
              ? "blocked"
              : "in_review",
        },
        tx,
      );
    }
    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,
          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;
}) {
  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"
        ? 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,
  });
}

async function resolveCommittedFinalizationRecovery(db: Db, run: typeof heartbeatRuns.$inferSelect, issueId: string) {
  const actions = await db.select().from(issueRecoveryActions).where(and(
    eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, issueId),
    eq(issueRecoveryActions.kind, "active_run_watchdog"), inArray(issueRecoveryActions.status, ["active", "escalated"]),
    sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`,
    or(
      sql`${issueRecoveryActions.wakePolicy}->>'kind' = 'resume_native_run'`,
      eq(issueRecoveryActions.cause, "native_workspace_finalization_owner_unverified"),
    ),
  ));
  for (const action of actions) await issueRecoveryActionService(db).resolveActiveForIssue({
    companyId: run.companyId, sourceIssueId: issueId, actionId: action.id,
    status: "resolved", outcome: "restored", resolutionNote: "The accepted native result and workspace finalization committed successfully; this run needs no further finalization retry.",
  });
}

function recoveredExecutionFailureMetadata() {
  // Read the row being updated so a concurrent cleanup diagnostic is retained.
  return sql`case when ${heartbeatRuns.error} is not null or ${heartbeatRuns.errorCode} is not null
    then jsonb_build_object('recoveredExecutionFailure', jsonb_build_object(
      'schema', 'paperclip.recovered_execution_failure.v1',
      'errorCode', ${heartbeatRuns.errorCode}, 'error', ${heartbeatRuns.error},
      'observedAt', ${heartbeatRuns.updatedAt}
    )) else '{}'::jsonb end`;
}

async function projectCommittedRun(input: {
  db: Db;
  run: typeof heartbeatRuns.$inferSelect;
  coordinator: typeof nativeRunFinalizations.$inferSelect;
}) {
  if (!input.coordinator.resultId) return;
  const resultRow = await input.db
    .select({ resultJson: nativeRunResults.resultJson })
    .from(nativeRunResults)
    .where(
      and(
        eq(nativeRunResults.id, input.coordinator.resultId),
        eq(nativeRunResults.runId, input.run.id),
        eq(nativeRunResults.companyId, input.run.companyId),
      ),
    )
    .limit(1)
    .then((rows) => rows[0] ?? null);
  const terminalState = record(
    record(resultRow?.resultJson).terminal,
  ).runTerminalState;
  if (!["succeeded", "failed", "cancelled"].includes(String(terminalState))) {
    throw new Error("native_finalization_invalid");
  }
  const projectedStatus = projectNativeTerminalRunStatus(
    terminalState as "succeeded" | "failed" | "cancelled",
  );
  const needsRepair = and(
    eq(heartbeatRuns.id, input.run.id),
    eq(heartbeatRuns.runtimeMode, "native"),
    or(
      inArray(heartbeatRuns.status, ["queued", "running", "failed"]),
      and(
        eq(heartbeatRuns.status, "succeeded"),
        or(
          isNotNull(heartbeatRuns.error),
          isNotNull(heartbeatRuns.errorCode),
          isNull(heartbeatRuns.finishedAt),
          sql`${heartbeatRuns.nativePhase} is distinct from 'committed'`,
          sql`${heartbeatRuns.resultJson}->>'finalizationPhase' is distinct from 'committed'`,
          sql`${heartbeatRuns.resultJson}->>'nextAttemptAt' is not null`,
          sql`${heartbeatRuns.resultJson}->>'failureCode' is not null`,
          sql`${heartbeatRuns.resultJson}->>'originalFailureCode' is not null`,
        ),
      ),
    ),
    // Reconciliation revisits committed results periodically. Only repair
    // a changed projection; rewriting an unchanged failed run would mint a
    // fresh status delivery (and failure toast) on every sweep. Check the
    // current row so concurrent replays cannot both queue the same repair.
    or(
      sql`${heartbeatRuns.status} is distinct from ${projectedStatus}`,
      isNull(heartbeatRuns.finishedAt),
      sql`${heartbeatRuns.nativePhase} is distinct from 'committed'`,
      sql`${heartbeatRuns.resultJson}->>'finalizationPhase' is distinct from 'committed'`,
      sql`${heartbeatRuns.resultJson}->>'nextAttemptAt' is not null`,
      sql`${heartbeatRuns.resultJson}->>'failureCode' is not null`,
      sql`${heartbeatRuns.resultJson}->>'originalFailureCode' is not null`,
      ...(terminalState === "succeeded"
        ? [isNotNull(heartbeatRuns.error), isNotNull(heartbeatRuns.errorCode)]
        : []),
    ),
    nativeRunnerOwnershipNotHeldCondition(),
  );
  const [candidate] = await input.db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(needsRepair).limit(1);
  if (!candidate) return;
  const now = new Date();
  const [updatedRun] = await input.db.transaction(async transaction => {
    const tx = transaction as unknown as Db;
    await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, input.run.companyId);
    return tx
      .update(heartbeatRuns)
      .set({
        executionStatusDeliveryId: randomUUID(),
        status: projectedStatus,
        finishedAt: input.run.finishedAt ?? now,
        nativePhase: "committed",
        nativePhaseUpdatedAt: now,
        ...(terminalState === "succeeded"
          ? {
              error: null,
              errorCode: null,
              // Capture the row being updated, not the earlier admission read:
              // cleanup may have recorded a new diagnostic in the meantime.
              // Other result metadata and all physical-owner evidence stay put.
              resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb)
                || ${recoveredExecutionFailureMetadata()}
                || jsonb_build_object('finalizationPhase', 'committed', 'failureCode', null, 'originalFailureCode', null, 'nextAttemptAt', null)`,
            }
          : { resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || jsonb_build_object('finalizationPhase', 'committed', 'failureCode', null, 'originalFailureCode', null, 'nextAttemptAt', null)` }),
        updatedAt: now,
      })
      .where(needsRepair)
      .returning();
  });
  // Metadata repairs can preserve the terminal status. Only a genuine status
  // transition should emit another terminal event. A failed run that leaves a
  // pending retry still needs its first failure report.
  if (updatedRun && updatedRun.status !== input.run.status) {
    await emitAgentTaskRun(input.db, updatedRun);
    void reportRunFailure(input.db, updatedRun);
  } else if (updatedRun && isNewTerminalRunFailure(input.run, updatedRun)) {
    void reportRunFailure(input.db, updatedRun);
  }
}

async function materializeCommittedReviewResponse(db: Db, runId: string) {
  try {
    const [run] = await db
      .select()
      .from(heartbeatRuns)
      .where(eq(heartbeatRuns.id, runId))
      .limit(1);
    if (!run?.nativeIssueId || !run.resultJson?.externalChatReviewPresentation)
      return;
    const decisionId = record(
      run.resultJson.externalChatReviewPresentation,
    ).decisionId;
    if (
      typeof decisionId !== "string" ||
      (await hasMaterializedNativeReviewResponse(db, {
        companyId: run.companyId,
        issueId: run.nativeIssueId,
        runId,
        decisionId,
      }))
    )
      return;
    const summary = record(run.resultJson.nativeResult).summary;
    if (
      typeof summary !== "string" ||
      !(await authorizeNativeChatReviewPresentation(db, {
        companyId: run.companyId,
        issueId: run.nativeIssueId,
        runId,
        resultJson: run.resultJson,
      }))
    )
      return;
    // The coordinator transaction is already committed. addComment owns a
    // fresh issue transaction, rechecks the proof and carries selected files;
    // its same-run/text and publication keys make recovery replay idempotent.
    await issueService(db).addComment(
      run.nativeIssueId,
      summary,
      { agentId: run.agentId, runId },
      {
        authorizationReason: "allow_chat_run_presentation",
      },
    );
    publishChatPublicationCommitSignal({
      companyId: run.companyId,
      issueId: run.nativeIssueId,
      runId: run.id,
      agentId: run.agentId,
      eventType: "run.presentation.resolved",
    });
  } catch (error) {
    // The durable committed proof remains retryable by the next reconciler
    // sweep. Presentation failure must not reclassify a successful native run.
    logger.warn(
      { runId, error },
      "Committed chat review response is awaiting presentation retry",
    );
  }
}

async function acceptedResponseDigestMatches(
  db: Db,
  run: typeof heartbeatRuns.$inferSelect,
  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> {
  let agentId: string | null = null;
  let materialized = false;
  try {
    materialized = await db.transaction(async (transaction) => {
      const tx = transaction as unknown as Db;
      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()
        .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.
      if (
        typeof record(record(run.resultJson).presentationDecision).commentId ===
        "string"
      )
        return false;
      const [coordinator] = await tx
        .select()
        .from(nativeRunFinalizations)
        .where(
          and(
            eq(nativeRunFinalizations.runId, run.id),
            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 [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;
        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;
      }
      // Copyback can be owned by reconciliation before the live heartbeat
      // reaches its presentation step. Recover a completed internal task's
      // exact final reply from the same accepted turn, without rerunning work
      // or granting external-chat publication authority.
      if (
        !isConversation(issue) &&
        run.status === "succeeded" &&
        record(run.contextSnapshot).skipIssueComment !== true &&
        !isExternalChatPresentationContext(run.contextSnapshot) &&
        result.reportedWorkDisposition === "done" &&
        terminal.runTerminalState === "succeeded" &&
        terminal.turnTerminalState === "completed" &&
        terminal.reportedWorkDisposition === "done" &&
        (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"),
            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;
        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;
        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;
      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;
    // Keep transient failures visible and retryable without reclassifying the
    // committed run or acknowledging a publication that never committed.
    logger.warn(
      { runId: input.runId, error },
      "Committed chat response is awaiting presentation retry",
    );
    return false;
  }
  if (materialized && agentId)
    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 finalizeNativeRun(input: {
  db: Db;
  runId: string;
  workspaceFinalizeStatus: "succeeded" | "failed";
  /** Reconciliation owns terminal run projection; the live heartbeat does it afterward. */
  projectRunStatus?: boolean;
  /** Workspace-only replay must not consume the provider recovery budget. */
  preserveProviderAttempt?: boolean;
  failpoint?: NativeStatusCommitFailpoint;
}) {
  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 claim = await claimCoordinator({
    db: input.db,
    runId: input.runId,
    preserveProviderAttempt: input.preserveProviderAttempt,
  });
  const coordinator = claim.coordinator;
  if (!claim.leaseOwner && coordinator.phase === "committed") {
    if (isNativeRunnerOwnershipHeld(run)) return coordinator;
    const presentationAlreadyMaterialized = coordinator.decisionId
      ? await hasMaterializedNativeReviewResponse(input.db, {
          companyId: run.companyId,
          issueId: coordinator.issueId,
          runId: run.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: run.companyId,
          issueId: coordinator.issueId,
          runId: run.id,
          decisionId: coordinator.decisionId!,
          resultId: coordinator.resultId!,
          assessmentId: coordinator.assessmentId!,
        }),
      );
    }
    if (input.projectRunStatus)
      await projectCommittedRun({ db: input.db, run, coordinator });
    if (input.projectRunStatus && !presentationAlreadyMaterialized)
      await materializeCommittedReviewResponse(input.db, input.runId);
    if (input.projectRunStatus)
      await repairCommittedNativeChatResponse(input.db, {
        companyId: run.companyId,
        issueId: coordinator.issueId,
        runId: run.id,
      });
    if (input.projectRunStatus) {
      await settleExternalConversation(input.db, run.companyId, coordinator.issueId).catch((err) => {
        logger.warn({ err, runId: run.id }, "Slack conversation settlement deferred to reconciliation");
      });
    }
    await resolveCommittedFinalizationRecovery(input.db, run, coordinator.issueId);
    if (input.projectRunStatus) {
      const [projectedRun] = await input.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
      if (projectedRun) await reconcileMuseNativeTerminalRun(input.db, projectedRun);
    }
    return coordinator;
  }
  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 governanceGate = await pendingNativeGovernance({
      db: input.db,
      companyId: run.companyId,
      issueId: authoritativeIssue.id,
      runId: run.id,
      executionState: record(authoritativeIssue.executionState),
    });
    const externalChatResponseWaitAuthorization =
      assessment.reportedDisposition === "yielded" &&
      assessment.continuation?.kind === "response_wake"
        ? await resolveExternalChatResponseWaitAuthorization({
            db: input.db,
            binding: {
              companyId: run.companyId,
              issueId: authoritativeIssue.id,
              runId: run.id,
              agentId: run.agentId,
            },
          })
        : "not_applicable";
    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 } = 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 monitorWaitAt = scheduledIssueMonitorWait(authoritativeIssue, run.agentId);
    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 proposedDecision = resolveNativeFinalizerStatus({
      caseHandoffAuthorized: Boolean(await assertNativeCaseHandoffResult(input.db, run.id, authoritativeIssue.id, result)),
      monitorWaitDispatched: monitorDelivery !== 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,
      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),
    });
    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,
    });
    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,
        requireMonitorWait: decision.reasonCode === "scheduled_monitor_waiting" && monitorWaitAt
          ? { agentId: run.agentId, nextCheckAt: monitorWaitAt } : undefined,
        requireMonitorDelivery: decision.reasonCode === "scheduled_monitor_dispatched"
          ? monitorDelivery ?? 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:
          decision.reasonCode === "external_chat_response_waiting"
            ? { agentId: run.agentId }
            : 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 = {
        ...(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 projectRun = (tx: Db) => tx
        .update(heartbeatRuns)
        .set({
          executionStatusDeliveryId: randomUUID(),
          ...(input.projectRunStatus
            ? {
                status:
                  terminalState === "succeeded"
                    ? "succeeded"
                    : terminalState === "cancelled"
                      ? "cancelled"
                      : "failed",
                finishedAt: now,
              }
            : {}),
          nativePhase: finalizationPhase,
          nativePhaseUpdatedAt: now,
          ...(clearExecutionFailure ? { error: null, errorCode: null } : {}),
          ...(providerFailure ? {
            // Recovery can finalize before heartbeat saves the adapter result.
            // Retain any independently recorded execution/cleanup failure.
            error: sql`coalesce(${heartbeatRuns.error}, ${providerFailure.errorMessage})`,
            errorCode: sql`coalesce(${heartbeatRuns.errorCode}, ${providerFailure.errorCode})`,
          } : {}),
          resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb)
            || ${JSON.stringify(finalizationMetadata)}::jsonb
            || ${clearExecutionFailure ? recoveredExecutionFailureMetadata() : sql`'{}'::jsonb`}`,
          updatedAt: now,
        })
        .where(eq(heartbeatRuns.id, run.id))
        .returning();
      const [updatedRun] = input.projectRunStatus
        ? await input.db.transaction(async transaction => {
            const tx = transaction as unknown as Db;
            await notifyDeliveryWork(tx, DELIVERY_QUEUES.accounting, run.companyId);
            return projectRun(tx);
          })
        : await projectRun(input.db);
      if (
        input.projectRunStatus &&
        !alreadyEmittedByCommittedDecision &&
        updatedRun
      ) {
        await emitAgentTaskRun(input.db, updatedRun);
        void reportRunFailure(input.db, updatedRun);
      }
      if (input.projectRunStatus)
        await materializeCommittedReviewResponse(input.db, input.runId);
      if (input.projectRunStatus && finalizationPhase === "committed")
        await repairCommittedNativeChatResponse(input.db, {
          companyId: run.companyId,
          issueId: coordinator.issueId,
          runId: run.id,
        });
      if (input.projectRunStatus && finalizationPhase === "committed") {
        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") await resolveCommittedFinalizationRecovery(input.db, run, coordinator.issueId);
      if (input.projectRunStatus && updatedRun) 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");
}