mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-12 14:19:00 +01:00
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - People talk to agents in Agent Chat. The agent's final message becomes the chat reply. > - A chat turn can end while its task stays held: a pending approval, an execution stage or a dependency blocker keeps the status. > - #20179 made live run completion post that held turn's answer. It added `conversationTurnAnswered()`. > - Background presentation recovery still required a `conversation_turn_finished` decision and a `done` result. > - If the server restarts between the status commit and the live presentation, recovery skips the held turn. The person never gets an answer. > - Restarts are product failures. A committed answer must not be lost. > - This pull request makes recovery use the same answered rule as the live path, through one shared helper. > - The benefit is that a held turn's answer reaches the chat exactly once, with or without a restart. ## Linked Issues or Issue Description Refs #20179 (live fix; this PR closes the "Known gap" in its Risks section). **What happened** In an Agent Chat, the agent resubmitted a hire approval and ended its turn with a yielded result. The finalizer committed `governed_response_waiting` and moved the task to `in_review`. The server restarted before run completion posted the reply. Background presentation recovery found the run as a candidate, but declined it, because it accepted only `conversation_turn_finished` with a `done` result. No reply was posted. The run stayed a pending presentation candidate. **Expected behavior** After a restart, recovery posts the held turn's final answer exactly once. It never posts a second copy when the live path already presented the answer. When the turn's own question or confirmation card presents it, recovery posts nothing and records that decision, as the live path does. **Steps to reproduce** 1. Open an Agent Chat. Link a pending hire approval to the chat task. 2. Let the agent end the turn with a yielded result and a `response_wake` continuation. The status decision commits as `governed_response_waiting`. 3. Stop the server after the status commit and before run completion presents the reply. 4. Start the server. Presentation recovery runs. 5. See that the chat shows no reply. ## What Changed - `server/src/services/native-runtime/native-governance.ts`: add `conversationTurnPresentsAnswer()`. It is the one rule for whether a successful conversation turn's prose is its answer: `conversation_turn_finished`, or a committed decision with no pending card of the turn (`conversationTurnAnswered()`). - `server/src/services/heartbeat/run-completion.ts`: the live path calls the shared helper. No change in behavior. - `server/src/services/native-runtime/native-run-finalizer.ts` (`recoverCommittedNativeChatResponse`): - A current conversation turn no longer needs `conversation_turn_finished`. The turn must still be the unsettled current turn: same agent, user, session generation, `conversationState: active`, and no other execution run. A `conversation_turn_finished` turn still needs `in_progress`. - A conversation turn with a `yielded` result is accepted next to `done`. - The resolver gets `conversationTurnFinished` from the shared helper. - When the turn's own card presents it, recovery records `presentationDecision: none` (reason `committed_conversation_interaction_presented`) and settles the conversation turn, like the live path. The run then leaves the pending presentation set. Before, such a run stayed a candidate. - `server/src/services/native-runtime/native-response-presentation-query.ts`: the candidate prefilter matches the same turns: any committed reason for an active conversation turn, and `yielded` conversation results. - `server/src/services/native-runtime/agent-chat-response-recovery.test.ts`: restart regressions with the real finalizer and a real database. - `doc/execution-semantics.md`: state that recovery applies the same rule. ## Verification - Failing first. With only the new tests applied, 5 of 7 new cases failed: pending approval, revision-requested approval, execution stage (all yielded), pending approval with `done`, and the own-card case. The run was a candidate, but recovery posted nothing and did not settle it. - After the fix: `cd server && npx vitest run src/services/native-runtime/agent-chat-response-recovery.test.ts src/services/heartbeat/run-completion.test.ts src/__tests__/idle-native-finalization.test.ts src/services/presentation-ownership-read.test.ts src/services/chat-monitor-continuation.test.ts src/services/native-runtime/native-preserved-monitor-wait.test.ts src/services/native-runtime/native-response-publication-retry.test.ts src/services/heartbeat/run-lifecycle.test.ts src/__tests__/scoped-deadline-work.test.ts src/services/native-runtime/paperclip-control-plane-port.test.ts src/__tests__/heartbeat-run-summary.test.ts src/__tests__/agent-conversations.test.ts`: 12 files, 627 tests passed. - The new cases check: - Held turns: the run is a candidate, the check posts the exact final answer once, the presentation decision is recorded, the run leaves the candidate set, and a second recovery returns false with one comment. - Live already presented: no candidate, no second comment. - Own question card: no comment, `presentationDecision: none`, the card stays pending, the conversation settles to `waiting`, and the run leaves the candidate set. - Previous-release live decision `none` (`yielded_control_plane_wait`): kept, not re-presented. - `tsc --noEmit -p server`: no errors in changed files. Four unrelated errors remain in this local worktree (`esbuild` not linked in `native-harness-prestage.test.ts`, and a linked-package type mismatch in `run-dispatch/adapters/postgres.ts`). - Turn start: no change. The helper runs only in run completion (after the provider turn ends) and in background presentation recovery. Nothing new runs before `turn.start`, and no files move. The live path keeps the same queries: it short-circuits on `conversation_turn_finished`, and otherwise runs the same card read as before. ## Risks - Low to medium. Recovery can now post the answer of a conversation turn whose task is held. This matches the live path since #20179. - A card-presented turn found by recovery is now settled with `presentationDecision: none`. Before, it stayed a pending candidate and kept the presentation worker (and idle-sleep checks) busy. - A `conversation_turn_finished` turn keeps the `in_progress` check. Held turns accept any status their gate committed. The unsettled-turn checks (active conversation state, session generation, execution run) still block stale or superseded turns. - Previous-release state: no persisted shape changes. A run committed by the previous release with no saved presentation decision (the restart window) is recovered by the new code. The test fixture uses the unchanged finalizer, so it saves exactly the previous release's shape. A run whose live path already recorded a decision keeps it. A held turn that the previous release recorded as `yielded_control_plane_wait` is not re-presented, so no late duplicate or stale reply appears. - Not changed: generic stranded-work recovery in `recovery/service.ts` still skips only `conversation_turn_finished` runs during the publication window. > 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 - Anthropic Claude Opus 5.5 (`claude-opus-5-5`), with tool use (shell, file edits, test runs) in Claude Code. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] 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>
1924 lines
91 KiB
TypeScript
1924 lines
91 KiB
TypeScript
import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js";
|
|
import { isContextCompactionRun } from "../context-compaction.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 { nativeLeaseHolder, notifyNativeTokenLeaseClaimed } from "./native-token-lease-settlement.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 { conversationTurnPresentsAnswer, 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, readNativeSupersededMonitorWait } 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,
|
|
completedPlanProposalIds,
|
|
readCompletedAssistantMessageCandidate,
|
|
resolveHeartbeatRunResponse,
|
|
selectHeartbeatRunFinalAgentMessage,
|
|
} from "../heartbeat-run-summary.js";
|
|
import { resolveChatRunPresentationAuthorizationReason } from "../chat-run-publications.js";
|
|
import { settledResponsePublicationSql } from "./native-response-publication-receipt.js";
|
|
import { hasChatRunOwnedProviderInteraction } from "../chat-interaction-arbitration.js";
|
|
import {
|
|
authorizeCommittedChatResponse,
|
|
CommittedChatResponseAuthorizationError,
|
|
} from "../durable-chat-wakeup.js";
|
|
import {
|
|
isNativeRunnerOwnershipHeld,
|
|
} from "./native-runner-ownership.js";
|
|
import { crashPoint } from "@paperclipai/shared/crash-points";
|
|
|
|
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);
|
|
await notifyNativeTokenLeaseClaimed(tx, coordinator);
|
|
const [updated] = await tx
|
|
.update(nativeRunFinalizations)
|
|
.set({
|
|
leaseOwner,
|
|
leaseExpiresAt: new Date(now.getTime() + 5 * 60_000),
|
|
// This process, bound to the token, so a hard kill mid-finalization is
|
|
// proven and settled on the shared ownership path. The controller_*
|
|
// columns stay the executor's.
|
|
leaseHolder: nativeLeaseHolder(leaseOwner),
|
|
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;
|
|
let settledWithoutReply = 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);
|
|
// The turn is still the conversation's unsettled current turn. A held
|
|
// turn keeps whatever status its gate committed; a finished turn
|
|
// preserves in_progress.
|
|
const currentConversation = conversation
|
|
&& issue.conversationAgentId === run.agentId && issue.conversationUserId === run.responsibleUserId
|
|
&& (decision.reasonCode !== "conversation_turn_finished" || 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") ||
|
|
// A conversation turn can yield while an approval, an execution
|
|
// stage or a blocker holds its task; it still answered the person.
|
|
(conversation && result.reportedWorkDisposition === "yielded" && terminal.reportedWorkDisposition === "yielded")) &&
|
|
(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 proposedItemIds = completedPlanProposalIds(await tx.select({ payload: heartbeatRunEvents.payload })
|
|
.from(heartbeatRunEvents).where(and(
|
|
eq(heartbeatRunEvents.companyId, input.companyId),
|
|
eq(heartbeatRunEvents.runId, run.id),
|
|
eq(heartbeatRunEvents.eventType, "plan.proposed"),
|
|
)).limit(200));
|
|
const finalAgentMessage = selectHeartbeatRunFinalAgentMessage({
|
|
candidates: rows.flatMap((row) => {
|
|
const candidate = readCompletedAssistantMessageCandidate({ seq: row.seq, prpEvent: record(row.payload).prpEvent, proposedItemIds });
|
|
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 },
|
|
// The same answered rule as live run completion.
|
|
conversationTurnFinished: conversation && await conversationTurnPresentsAnswer({
|
|
db: tx, ...input, finalizationPhase: coordinator.phase, finalizationReasonCode: decision.reasonCode,
|
|
}),
|
|
existingComment: findHeartbeatRunCompletionComment(comments, run.resultJson),
|
|
finalAgentMessage,
|
|
});
|
|
if (resolved.decision.commentAction === "reuse" && typeof resolved.decision.commentId === "string") {
|
|
// The reply is already on the task (for example, the live path wrote
|
|
// it but did not record its decision). Record it; never post again.
|
|
await tx.update(heartbeatRuns).set({
|
|
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({
|
|
presentationDecision: { ...resolved.decision,
|
|
reasonCodes: [...resolved.decision.reasonCodes, "committed_task_response_reused"] },
|
|
})}::jsonb || ${settledResponsePublicationSql()}`, updatedAt: new Date(),
|
|
}).where(eq(heartbeatRuns.id, run.id));
|
|
return false;
|
|
}
|
|
if (conversation && resolved.decision.commentAction === "none") {
|
|
// The turn's own question or confirmation card presents it, as on
|
|
// the live path. Record that decision once so the turn settles and
|
|
// leaves the pending presentation set.
|
|
await tx.update(heartbeatRuns).set({
|
|
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({
|
|
presentationDecision: { ...resolved.decision,
|
|
reasonCodes: [...resolved.decision.reasonCodes, "committed_conversation_interaction_presented"] },
|
|
})}::jsonb`, updatedAt: new Date(),
|
|
}).where(eq(heartbeatRuns.id, run.id));
|
|
settledWithoutReply = true;
|
|
return false;
|
|
}
|
|
if (!resolved.text || resolved.decision.commentAction !== "create") return false;
|
|
await notifyChatRunMilestones(tx, input);
|
|
const comment = await issueService(db).addComment(input.issueId, resolved.text,
|
|
{ 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 || ${settledResponsePublicationSql()}`, 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 (settledWithoutReply) await settleConversationTurnForRun(db, input.companyId, input.runId);
|
|
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 supersededMonitorWait = terminalState === "succeeded" && input.workspaceFinalizeStatus === "succeeded" &&
|
|
assessment.reportedDisposition === "yielded" && assessment.continuation?.kind === "monitor"
|
|
? await readNativeSupersededMonitorWait(input.db, {
|
|
companyId: run.companyId, issueId: authoritativeIssue.id, runId: run.id, agentId: run.agentId,
|
|
}, authoritativeIssue) : null;
|
|
if (supersededMonitorWait && supersededMonitorWait.resultId !== resultRow.id) throw new NativeStatusRaceError();
|
|
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,
|
|
monitorWaitSuperseded: supersededMonitorWait !== 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,
|
|
contextCompactionRun: isContextCompactionRun(run.contextSnapshot),
|
|
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,
|
|
supersededMonitorWait,
|
|
});
|
|
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,
|
|
requireSupersededMonitorWait: decision.reasonCode === "scheduled_monitor_superseded"
|
|
? supersededMonitorWait ?? 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,
|
|
});
|
|
crashPoint("native.finalize.after_decision_before_projection");
|
|
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");
|
|
}
|