mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-26 18:31:34 +01:00
DottaandPaperclip
dff63551fe
## Thinking Path
> - Paperclip manages agents that perform work for a company.
> - Task repositories are prepared at normal run admission.
> - A running native task can stage a repository while its work remains
unfinished.
> - The controller did not offer that saved setup as a continuation.
> - The agent could add a timer or wait for a user restart.
> - This change uses the existing same-agent continuation and status
ledger.
> - The accepted explanation and proven conversation remain available
after setup.
## Linked Issues or Issue Description
Refs #17045. That prerequisite has merged. This PR targets master.
**What happened?**
A projectless native task can request a repository for the next
admission. An unfinished task has no server-selected repository
continuation. An explicit same-agent continuation also loses chat
provenance.
**Expected behavior**
An unfinished standard task can yield and resume through the existing
admission queue. Its explanation and final result retain the authorized
conversation. A task that only stages a repository can complete without
another run.
**Steps to reproduce**
1. Start a native standard task in a task directory.
2. Request an authorized repository through prepare_repository.
3. Yield with unfinished work and no explicit continuation.
4. Observe that the former driver rejects the yield, although the
pending repository is saved.
## What Changed
- Offer a saved same-agent continuation for the current task's pending
repository.
- Accept that current server continuation in the shared provider driver.
- Resolve chat provenance through applied status decisions, delivered
effects, and admitted wakes.
- Preserve accepted setup explanations without repeating an owned
question card.
- Document optional reproduction details and configured repository
discovery.
- Extend database regressions and workspace access maps.
- Preserve Copilot profile v52 and bind the shared continuation source
in v53. Include the active identity in the QA image fingerprint.
## Verification
- Five focused server suites: 100 checks pass.
- Shared continuation driver: 12 checks pass.
- Server typecheck passes.
- Map contract and whitespace checks pass.
- Existing API and driver assertions fail on the former implementation.
- Real dispatcher and queue cohort: 43 checks pass. The original
reply-loss regression fails before the dispatch receipt fix.
- Continuation driver and installation identity cohort: 62 checks pass.
- Regenerated capability inventory and contract checks pass.
- Final response precedence cohort: 109 checks pass. The stale-comment
case fails on the former implementation.
- All required CI checks, build, and both source reviews pass on
b9c8677635.
- Live provider journeys and deployed acceptance remain pending. No live
qualification is claimed.
## Risks
A pending repository now offers a next admission when unfinished
standard work yields. Incorrect eligibility could create unwanted work.
The lookup requires the current workspace and task. Explicit completion
does not use it. Existing Stop, admission, approval, and transport
authorization remain authoritative. No schema or public task-status
change is included. The shared continuation change requires a new
Copilot source identity. This PR preserves v52 history and does not
claim fresh Copilot live qualification.
## Model Used
OpenAI Codex, GPT-6 family, with repository tools and code execution.
The runtime does not expose its exact deployment ID or context-window
size. Those values are not inferred.
## Checklist
- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge
---------
Co-authored-by: Paperclip <noreply@paperclip.ing>
1667 lines
67 KiB
TypeScript
1667 lines
67 KiB
TypeScript
import { reconcileMuseNativeTerminalRun } from "./muse-terminal-reconciliation.js";
|
|
import { assertNativeCaseHandoffResult } from "./native-case-handoff.js";
|
|
import { scheduledIssueMonitorWait } from "../issue-monitors.js";
|
|
import { readNativeMonitorDelivery } from "./native-monitor-delivery.js";
|
|
import { isNativePlanWaitResult, readNativePlanWait } from "./native-plan-wait.js";
|
|
import { activeIssueInteractionCondition } from "../issue-question-context.js";
|
|
import { hasPendingNativeChildCompletion } from "./native-child-completion-delivery.js";
|
|
import { settleExternalConversation } from "../slack-conversation-lifecycle.js";
|
|
import { executionFailureRetryCount } from "../execution-recovery-attempt.js";
|
|
import { readPersistedNativeFailureEvidence } from "./native-provider-failure-evidence.js";
|
|
import { dismissAutomaticCompletionReviews } from "./automatic-completion-reviews.js";
|
|
import { getNativeReviewAssignment, readNativeReviewAssignmentContext } from "./native-review-participant.js";
|
|
import { conversationNativeDecision, isConversation } from "../agent-conversations.js";
|
|
import { issueTreeControlService } from "../issue-tree-control.js";
|
|
import { randomUUID } from "node:crypto";
|
|
import { preserveNativeWorkspaceExportLease } from "./native-workspace-export-resume.js";
|
|
import { and, desc, eq, inArray, isNotNull, isNull, or, sql } from "drizzle-orm";
|
|
import type { Db } from "@paperclipai/db";
|
|
import {
|
|
approvals,
|
|
agentWakeupRequests,
|
|
completionContracts,
|
|
heartbeatRuns,
|
|
heartbeatRunEvents,
|
|
issueApprovals,
|
|
issueComments,
|
|
issueRecoveryActions,
|
|
issueThreadInteractions,
|
|
issues,
|
|
nativeRunFinalizations,
|
|
nativeRunResults,
|
|
statusDecisions,
|
|
workAssessments,
|
|
workspaceOperations,
|
|
} from "@paperclipai/db";
|
|
import { classifyNativeEvidence } from "./evidence-classifier.js";
|
|
import type { PrpIgnoredAttentionRequest, PrpTerminalState } from "@paperclipai/paperclip-runner";
|
|
import {
|
|
arbitrateNativeStatus,
|
|
NATIVE_STATUS_ARBITER_POLICY_VERSION,
|
|
type NativeAuthoritativeIssueStatus,
|
|
type NativeGovernanceGate,
|
|
} from "./status-arbiter.js";
|
|
import { recordNativeWorkAssessment } from "./work-assessments.js";
|
|
import {
|
|
commitNativeStatusDecision,
|
|
NativeStatusRaceError,
|
|
type NativeStatusCommitFailpoint,
|
|
} from "./status-decision-committer.js";
|
|
import { issueRecoveryActionService } from "../issue-recovery-actions.js";
|
|
import { issueService } from "../issues.js";
|
|
import { publishChatPublicationCommitSignal } from "../chat-publication-reconciliation.js";
|
|
import { nativeSha256 } from "./canonical.js";
|
|
import {
|
|
readNativeBoardResponseWaitOrigin,
|
|
readNativeBoardResponseWaitSource,
|
|
} from "./native-board-response-wait.js";
|
|
import { emitAgentTaskRun } from "../agent-task-run-telemetry.js";
|
|
import { isNewTerminalRunFailure, reportRunFailure } from "../run-failure-report.js";
|
|
import { resolveExternalChatResponseWaitAuthorization } from "./chat-attachment-reuse.js";
|
|
import {
|
|
authorizeNativeChatReviewPresentation,
|
|
hasMaterializedNativeReviewResponse,
|
|
restoreNativeChatReviewPresentationInTransaction,
|
|
} from "./native-chat-review-presentation.js";
|
|
import { logger } from "../../middleware/logger.js";
|
|
import {
|
|
CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON,
|
|
findHeartbeatRunCompletionComment,
|
|
isExternalChatPresentationContext,
|
|
readAcceptedTaskContinuationSummary,
|
|
readCompletedAssistantMessageCandidate,
|
|
resolveHeartbeatRunResponse,
|
|
selectHeartbeatRunFinalAgentMessage,
|
|
} from "../heartbeat-run-summary.js";
|
|
import { resolveChatRunPresentationAuthorizationReason } from "../chat-run-publications.js";
|
|
import { hasChatRunOwnedProviderInteraction } from "../chat-interaction-arbitration.js";
|
|
import {
|
|
authorizeCommittedChatResponse,
|
|
CommittedChatResponseAuthorizationError,
|
|
} from "../durable-chat-wakeup.js";
|
|
import {
|
|
isNativeRunnerOwnershipHeld,
|
|
nativeRunnerOwnershipNotHeldCondition,
|
|
} from "./native-runner-ownership.js";
|
|
|
|
function record(value: unknown): Record<string, unknown> {
|
|
return typeof value === "object" && value !== null && !Array.isArray(value)
|
|
? (value as Record<string, unknown>)
|
|
: {};
|
|
}
|
|
|
|
function authoritativeStatus(value: string): NativeAuthoritativeIssueStatus {
|
|
if (
|
|
![
|
|
"backlog",
|
|
"todo",
|
|
"in_progress",
|
|
"in_review",
|
|
"blocked",
|
|
"done",
|
|
"cancelled",
|
|
].includes(value)
|
|
) {
|
|
throw new Error("native_issue_status_invalid");
|
|
}
|
|
return value as NativeAuthoritativeIssueStatus;
|
|
}
|
|
|
|
function isCommittedAuditOnlyRouting(value: unknown) {
|
|
const receipts = record(value).nativeAttentionRouting;
|
|
return (
|
|
Array.isArray(receipts) &&
|
|
receipts.length > 0 &&
|
|
receipts.every((candidate) => {
|
|
const receipt = record(candidate);
|
|
const targets = receipt.materializedTargets;
|
|
return (
|
|
receipt.decisionId === null &&
|
|
receipt.reasonCode === "attention_duplicate_suppressed" &&
|
|
Array.isArray(targets) &&
|
|
targets.length > 0
|
|
);
|
|
})
|
|
);
|
|
}
|
|
|
|
/** Server-owned terminal conversion used by live finalization and read models. */
|
|
export function projectNativeTerminalRunStatus(
|
|
terminalState: "succeeded" | "failed" | "cancelled" | "active",
|
|
) {
|
|
return terminalState === "active" ? ("running" as const) : terminalState;
|
|
}
|
|
|
|
/** Fact-based arbitration at the production finalization authority seam. */
|
|
export function resolveNativeFinalizerStatus(
|
|
input: Parameters<typeof arbitrateNativeStatus>[0],
|
|
) {
|
|
return arbitrateNativeStatus(input);
|
|
}
|
|
|
|
export async function pendingNativeGovernance(input: {
|
|
db: Db;
|
|
companyId: string;
|
|
issueId: string;
|
|
runId: string;
|
|
executionState: Record<string, unknown> | null;
|
|
}): Promise<NativeGovernanceGate | null> {
|
|
// Execution policy is issue-owned state and takes priority over the two
|
|
// durable review surfaces. Completion must mediate all three authorities.
|
|
if (record(input.executionState).status === "pending") {
|
|
return { kind: "execution_stage", id: input.runId };
|
|
}
|
|
const [issue] = await input.db.select({ conversationAgentId: issues.conversationAgentId, conversationUserId: issues.conversationUserId })
|
|
.from(issues).where(and(eq(issues.id, input.issueId), eq(issues.companyId, input.companyId)));
|
|
const [pendingInteraction, pendingApproval] = await Promise.all([
|
|
input.db
|
|
.select({ id: issueThreadInteractions.id })
|
|
.from(issueThreadInteractions)
|
|
.where(
|
|
and(
|
|
eq(issueThreadInteractions.companyId, input.companyId),
|
|
eq(issueThreadInteractions.issueId, input.issueId),
|
|
eq(issueThreadInteractions.status, "pending"),
|
|
activeIssueInteractionCondition({ runId: input.runId, conversationMode: isConversation(issue) }),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null),
|
|
input.db
|
|
.select({ id: approvals.id })
|
|
.from(issueApprovals)
|
|
.innerJoin(approvals, eq(issueApprovals.approvalId, approvals.id))
|
|
.where(
|
|
and(
|
|
eq(issueApprovals.companyId, input.companyId),
|
|
eq(issueApprovals.issueId, input.issueId),
|
|
eq(approvals.companyId, input.companyId),
|
|
inArray(approvals.status, ["pending", "revision_requested"]),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null),
|
|
]);
|
|
if (pendingInteraction)
|
|
return { kind: "interaction", id: pendingInteraction.id };
|
|
if (pendingApproval) return { kind: "approval", id: pendingApproval.id };
|
|
return null;
|
|
}
|
|
|
|
async function acceptedInteractionFromRun(input: {
|
|
db: Db;
|
|
companyId: string;
|
|
issueId: string;
|
|
runId: string;
|
|
}) {
|
|
return input.db
|
|
.select({ id: issueThreadInteractions.id })
|
|
.from(issueThreadInteractions)
|
|
.where(
|
|
and(
|
|
eq(issueThreadInteractions.companyId, input.companyId),
|
|
eq(issueThreadInteractions.issueId, input.issueId),
|
|
eq(issueThreadInteractions.sourceRunId, input.runId),
|
|
eq(issueThreadInteractions.status, "accepted"),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
}
|
|
|
|
async function claimCoordinator(input: {
|
|
db: Db;
|
|
runId: string;
|
|
preserveProviderAttempt?: boolean;
|
|
}) {
|
|
const leaseOwner = `native-finalizer:${randomUUID()}`;
|
|
const now = new Date();
|
|
const claimed = await input.db.transaction(async (tx) => {
|
|
const coordinator = await tx
|
|
.select()
|
|
.from(nativeRunFinalizations)
|
|
.where(eq(nativeRunFinalizations.runId, input.runId))
|
|
.for("update")
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!coordinator?.resultId) throw new Error("native_finalization_missing");
|
|
if (coordinator.phase === "committed") {
|
|
if (!coordinator.decisionId) {
|
|
const run = await tx
|
|
.select({ resultJson: heartbeatRuns.resultJson })
|
|
.from(heartbeatRuns)
|
|
.where(
|
|
and(
|
|
eq(heartbeatRuns.id, coordinator.runId),
|
|
eq(heartbeatRuns.companyId, coordinator.companyId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!run || !isCommittedAuditOnlyRouting(run.resultJson)) {
|
|
throw new Error("native_finalization_committed_outcome_missing");
|
|
}
|
|
}
|
|
return { coordinator, leaseOwner: null };
|
|
}
|
|
if (coordinator.phase === "terminal_failure")
|
|
throw new Error("native_finalization_terminal_failure");
|
|
if (
|
|
coordinator.leaseOwner &&
|
|
coordinator.leaseExpiresAt &&
|
|
coordinator.leaseExpiresAt > now &&
|
|
coordinator.leaseOwner !== leaseOwner
|
|
)
|
|
throw new Error("native_finalization_lease_busy");
|
|
const [updated] = await tx
|
|
.update(nativeRunFinalizations)
|
|
.set({
|
|
leaseOwner,
|
|
leaseExpiresAt: new Date(now.getTime() + 5 * 60_000),
|
|
attempt:
|
|
input.preserveProviderAttempt || coordinator.controllerBootId !== null
|
|
? coordinator.attempt
|
|
: coordinator.attempt + 1,
|
|
phase:
|
|
coordinator.phase === "retryable_failure"
|
|
? coordinator.assessmentId
|
|
? "arbitrating"
|
|
: "workspace_finalizing"
|
|
: coordinator.phase,
|
|
failureCode: null,
|
|
failureDetail: null,
|
|
nextAttemptAt: null,
|
|
updatedAt: now,
|
|
})
|
|
.where(eq(nativeRunFinalizations.runId, input.runId))
|
|
.returning();
|
|
if (!updated) throw new Error("native_finalization_claim_failed");
|
|
return { coordinator: updated, leaseOwner };
|
|
});
|
|
return claimed;
|
|
}
|
|
|
|
async function recordRetryableFailure(input: {
|
|
db: Db;
|
|
run: typeof heartbeatRuns.$inferSelect;
|
|
coordinator: typeof nativeRunFinalizations.$inferSelect;
|
|
failureCode: string;
|
|
message: string;
|
|
nextAction: string;
|
|
projectRunStatus?: boolean;
|
|
failureScope?: "provider" | "workspace";
|
|
permanent?: boolean;
|
|
}) {
|
|
const now = new Date();
|
|
const nextAttemptAt = new Date(now.getTime() + 30_000);
|
|
const acceptedRunTerminalState = record(
|
|
input.run.resultJson,
|
|
).prpRunTerminalState;
|
|
const exhaustedRunStatus =
|
|
acceptedRunTerminalState === "succeeded"
|
|
? ("succeeded" as const)
|
|
: acceptedRunTerminalState === "cancelled"
|
|
? ("cancelled" as const)
|
|
: ("failed" as const);
|
|
let terminalRunToEmit: typeof heartbeatRuns.$inferSelect | null = null;
|
|
const outcome = await input.db.transaction(async (tx) => {
|
|
// Admission snapshots cannot authorize a failure write after a different
|
|
// owner committed success. Match status-committer lock ordering.
|
|
const current = await tx.select().from(nativeRunFinalizations).where(and(
|
|
eq(nativeRunFinalizations.runId, input.run.id),
|
|
eq(nativeRunFinalizations.companyId, input.run.companyId),
|
|
)).for("update").limit(1).then((rows) => rows[0] ?? null);
|
|
if (!current) throw new Error("native_finalization_missing");
|
|
// Audit-only attention first records an agent-owned invalid-result outcome;
|
|
// its caller still needs to materialize the normal bounded recovery action.
|
|
// Board-owned terminal repairs and late snapshots remain settled.
|
|
const currentExportRetry = record(record(current.failureDetail).workspaceExportRetry).requestId;
|
|
if (currentExportRetry && currentExportRetry !== record(record(input.coordinator.failureDetail).workspaceExportRetry).requestId) return current;
|
|
const recoveryOwner = record(record(current.failureDetail).recoveryOwner);
|
|
const pendingAgentRecovery = current.phase === "terminal_failure"
|
|
&& input.coordinator.phase === "terminal_failure"
|
|
&& recoveryOwner.kind === "agent" && recoveryOwner.agentId === input.run.agentId;
|
|
if (current.phase === "committed"
|
|
|| (current.phase === "terminal_failure" && !pendingAgentRecovery)
|
|
|| current.leaseOwner !== input.coordinator.leaseOwner) return current;
|
|
const [currentRun] = await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.run.id)).limit(1);
|
|
if (!currentRun) throw new Error("native_finalization_run_missing");
|
|
if (input.failureScope === "workspace") {
|
|
const exported = await tx.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and(
|
|
eq(workspaceOperations.companyId, input.run.companyId),
|
|
eq(workspaceOperations.heartbeatRunId, input.run.id),
|
|
eq(workspaceOperations.phase, "workspace_finalize"),
|
|
eq(workspaceOperations.status, "succeeded"),
|
|
)).limit(1);
|
|
// The live exporter may have acquired ownership since the stale error
|
|
// was raised, or already durably published a successful copyback.
|
|
if (exported.length || record(currentRun.runnerProfileJson).nativeWorkspaceFinalizationOwner) return current;
|
|
}
|
|
input = { ...input, coordinator: current, run: currentRun };
|
|
const issue = await tx
|
|
.select({
|
|
lastStatusDecisionId: issues.lastStatusDecisionId,
|
|
})
|
|
.from(issues)
|
|
.where(
|
|
and(
|
|
eq(issues.id, input.coordinator.issueId),
|
|
eq(issues.companyId, input.run.companyId),
|
|
),
|
|
)
|
|
.for("update")
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
const latestDecision = issue?.lastStatusDecisionId
|
|
? await tx
|
|
.select({
|
|
runId: workAssessments.runId,
|
|
})
|
|
.from(statusDecisions)
|
|
.innerJoin(
|
|
workAssessments,
|
|
eq(statusDecisions.assessmentId, workAssessments.id),
|
|
)
|
|
.where(
|
|
and(
|
|
eq(statusDecisions.id, issue.lastStatusDecisionId),
|
|
eq(statusDecisions.companyId, input.run.companyId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null)
|
|
: null;
|
|
const latestDecisionRun =
|
|
latestDecision?.runId && latestDecision.runId !== input.run.id
|
|
? await tx
|
|
.select({ createdAt: heartbeatRuns.createdAt })
|
|
.from(heartbeatRuns)
|
|
.where(
|
|
and(
|
|
eq(heartbeatRuns.id, latestDecision.runId),
|
|
eq(heartbeatRuns.companyId, input.run.companyId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null)
|
|
: null;
|
|
// A finalizer belongs to one immutable run result. If a later run already
|
|
// committed the issue's authoritative decision, this older coordinator is
|
|
// audit-only: retrying or escalating it must never reopen the newer result.
|
|
const supersededByNewerRun = Boolean(
|
|
latestDecisionRun && latestDecisionRun.createdAt > input.run.createdAt,
|
|
);
|
|
const priorFailureDetail = record(input.coordinator.failureDetail);
|
|
const workspaceFinalizeAttempt =
|
|
input.failureScope === "workspace"
|
|
? (typeof priorFailureDetail.workspaceFinalizeAttempt === "number" &&
|
|
Number.isInteger(priorFailureDetail.workspaceFinalizeAttempt) &&
|
|
priorFailureDetail.workspaceFinalizeAttempt >= 0
|
|
? priorFailureDetail.workspaceFinalizeAttempt
|
|
: 0) + 1
|
|
: null;
|
|
const exhausted =
|
|
input.permanent === true ||
|
|
(workspaceFinalizeAttempt !== null
|
|
? workspaceFinalizeAttempt >= 3
|
|
: input.coordinator.attempt >= 3);
|
|
const phase =
|
|
supersededByNewerRun || exhausted
|
|
? ("terminal_failure" as const)
|
|
: ("retryable_failure" as const);
|
|
const failureCode = supersededByNewerRun
|
|
? "native_finalization_superseded"
|
|
: input.permanent
|
|
? input.failureCode
|
|
: exhausted
|
|
? input.failureScope === "workspace"
|
|
? "native_workspace_sync_out_retry_exhausted"
|
|
: "native_finalization_retry_exhausted"
|
|
: input.failureCode;
|
|
await tx
|
|
.update(nativeRunFinalizations)
|
|
.set({
|
|
phase,
|
|
leaseOwner: null,
|
|
leaseExpiresAt: null,
|
|
failureCode,
|
|
failureDetail: {
|
|
message: input.message.slice(0, 2_000),
|
|
originalFailureCode: input.failureCode,
|
|
...(priorFailureDetail.workspaceExportRetry ? { workspaceExportRetry: priorFailureDetail.workspaceExportRetry } : {}),
|
|
...(workspaceFinalizeAttempt === null
|
|
? {}
|
|
: { workspaceFinalizeAttempt }),
|
|
recoveryOwner: supersededByNewerRun
|
|
? { kind: "none", reason: "newer_native_decision" }
|
|
: exhausted
|
|
? { kind: "board" }
|
|
: { kind: "agent", agentId: input.run.agentId },
|
|
nextAction: input.nextAction,
|
|
},
|
|
nextAttemptAt: supersededByNewerRun || exhausted ? null : nextAttemptAt,
|
|
updatedAt: now,
|
|
})
|
|
.where(eq(nativeRunFinalizations.runId, input.run.id));
|
|
const projectsTerminalStatus =
|
|
exhausted && !supersededByNewerRun && input.projectRunStatus;
|
|
if (projectsTerminalStatus && input.failureScope === "workspace") {
|
|
await preserveNativeWorkspaceExportLease(tx as unknown as Db, input.run, input.coordinator.resultId);
|
|
}
|
|
const [updatedRun] = await tx
|
|
.update(heartbeatRuns)
|
|
.set({
|
|
executionStatusDeliveryId: randomUUID(),
|
|
...(projectsTerminalStatus
|
|
? {
|
|
status:
|
|
input.failureScope === "workspace"
|
|
? "failed"
|
|
: exhaustedRunStatus,
|
|
finishedAt: input.run.finishedAt ?? now,
|
|
}
|
|
: {}),
|
|
nativePhase: phase,
|
|
nativePhaseUpdatedAt: now,
|
|
resultJson: {
|
|
...record(input.run.resultJson),
|
|
finalizationPhase: phase,
|
|
failureCode,
|
|
originalFailureCode: input.failureCode,
|
|
nextAttemptAt:
|
|
supersededByNewerRun || exhausted
|
|
? null
|
|
: nextAttemptAt.toISOString(),
|
|
},
|
|
updatedAt: now,
|
|
})
|
|
.where(eq(heartbeatRuns.id, input.run.id))
|
|
.returning();
|
|
if (projectsTerminalStatus) terminalRunToEmit = updatedRun ?? null;
|
|
if (supersededByNewerRun) {
|
|
await issueRecoveryActionService(
|
|
tx as unknown as Db,
|
|
).resolveActiveForIssue(
|
|
{
|
|
companyId: input.run.companyId,
|
|
sourceIssueId: input.coordinator.issueId,
|
|
cause: input.failureCode,
|
|
fingerprint: nativeSha256({
|
|
runId: input.run.id,
|
|
failureCode: input.failureCode,
|
|
}),
|
|
status: "resolved",
|
|
outcome: "false_positive",
|
|
resolutionNote:
|
|
"A newer native run already committed the authoritative issue decision; the stale finalizer was retired without changing issue state.",
|
|
},
|
|
tx,
|
|
);
|
|
} else if (exhausted) {
|
|
await issueService(tx as unknown as Db).update(
|
|
input.coordinator.issueId,
|
|
{
|
|
status:
|
|
input.failureScope === "workspace"
|
|
? "blocked"
|
|
: "in_review",
|
|
},
|
|
tx,
|
|
);
|
|
}
|
|
if (!supersededByNewerRun) {
|
|
await issueRecoveryActionService(tx as unknown as Db).upsertSourceScoped({
|
|
companyId: input.run.companyId,
|
|
sourceIssueId: input.coordinator.issueId,
|
|
kind: "active_run_watchdog",
|
|
ownerType: exhausted ? "board" : "agent",
|
|
ownerAgentId: exhausted ? null : input.run.agentId,
|
|
returnOwnerAgentId: input.run.agentId,
|
|
cause: failureCode,
|
|
fingerprint: nativeSha256({ runId: input.run.id, failureCode }),
|
|
evidence: {
|
|
runId: input.run.id,
|
|
coordinatorAttempt: input.coordinator.attempt,
|
|
...(workspaceFinalizeAttempt === null
|
|
? {}
|
|
: { workspaceFinalizeAttempt }),
|
|
originalFailureCode: input.failureCode,
|
|
},
|
|
nextAction: exhausted
|
|
? input.permanent
|
|
? input.nextAction
|
|
: `Finalization retry budget exhausted. ${input.nextAction}`
|
|
: input.nextAction,
|
|
wakePolicy: exhausted
|
|
? null
|
|
: {
|
|
kind: "resume_native_run",
|
|
runId: input.run.id,
|
|
notBefore: nextAttemptAt.toISOString(),
|
|
},
|
|
maxAttempts: 3,
|
|
supersedeOnIdentityChange: true,
|
|
});
|
|
}
|
|
return {
|
|
phase,
|
|
failureCode,
|
|
nextAttemptAt: supersededByNewerRun || exhausted ? null : nextAttemptAt,
|
|
};
|
|
});
|
|
if (terminalRunToEmit) {
|
|
await emitAgentTaskRun(input.db, terminalRunToEmit);
|
|
void reportRunFailure(input.db, terminalRunToEmit);
|
|
}
|
|
return {
|
|
...input.coordinator,
|
|
...outcome,
|
|
};
|
|
}
|
|
|
|
/** Converts malformed or incomplete persisted finalization state into named recovery. */
|
|
export async function recordNativeFinalizationFailure(input: {
|
|
db: Db;
|
|
runId: string;
|
|
error: unknown;
|
|
projectRunStatus?: boolean;
|
|
failureScope?: "provider" | "workspace";
|
|
permanent?: boolean;
|
|
}) {
|
|
const [run, coordinator] = await Promise.all([
|
|
input.db
|
|
.select()
|
|
.from(heartbeatRuns)
|
|
.where(eq(heartbeatRuns.id, input.runId))
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null),
|
|
input.db
|
|
.select()
|
|
.from(nativeRunFinalizations)
|
|
.where(eq(nativeRunFinalizations.runId, input.runId))
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null),
|
|
]);
|
|
if (!run || run.runtimeMode !== "native" || !coordinator) throw input.error;
|
|
const message =
|
|
input.error instanceof Error ? input.error.message : String(input.error);
|
|
if (message === "native_finalization_lease_busy") return coordinator;
|
|
const failureCode = message.startsWith("native_")
|
|
? message
|
|
: "native_finalization_invalid";
|
|
return recordRetryableFailure({
|
|
db: input.db,
|
|
run,
|
|
coordinator,
|
|
failureCode,
|
|
message,
|
|
nextAction:
|
|
input.failureScope === "workspace"
|
|
? input.permanent
|
|
? "Restore the exact sandbox containing the unexported workspace changes, or resolve the task manually from durable evidence."
|
|
: "Retry workspace export and merge from the retained sandbox; do not submit another provider turn."
|
|
: "Repair the persisted native result or contract discriminator, then resume finalization from the coordinator.",
|
|
projectRunStatus: input.projectRunStatus,
|
|
failureScope: input.failureScope,
|
|
permanent: input.permanent,
|
|
});
|
|
}
|
|
|
|
async function resolveCommittedFinalizationRecovery(db: Db, run: typeof heartbeatRuns.$inferSelect, issueId: string) {
|
|
const actions = await db.select().from(issueRecoveryActions).where(and(
|
|
eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, issueId),
|
|
eq(issueRecoveryActions.kind, "active_run_watchdog"), inArray(issueRecoveryActions.status, ["active", "escalated"]),
|
|
sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`,
|
|
or(
|
|
sql`${issueRecoveryActions.wakePolicy}->>'kind' = 'resume_native_run'`,
|
|
eq(issueRecoveryActions.cause, "native_workspace_finalization_owner_unverified"),
|
|
),
|
|
));
|
|
for (const action of actions) await issueRecoveryActionService(db).resolveActiveForIssue({
|
|
companyId: run.companyId, sourceIssueId: issueId, actionId: action.id,
|
|
status: "resolved", outcome: "restored", resolutionNote: "The accepted native result and workspace finalization committed successfully; this run needs no further finalization retry.",
|
|
});
|
|
}
|
|
|
|
function recoveredExecutionFailureMetadata() {
|
|
// Read the row being updated so a concurrent cleanup diagnostic is retained.
|
|
return sql`case when ${heartbeatRuns.error} is not null or ${heartbeatRuns.errorCode} is not null
|
|
then jsonb_build_object('recoveredExecutionFailure', jsonb_build_object(
|
|
'schema', 'paperclip.recovered_execution_failure.v1',
|
|
'errorCode', ${heartbeatRuns.errorCode}, 'error', ${heartbeatRuns.error},
|
|
'observedAt', ${heartbeatRuns.updatedAt}
|
|
)) else '{}'::jsonb end`;
|
|
}
|
|
|
|
async function projectCommittedRun(input: {
|
|
db: Db;
|
|
run: typeof heartbeatRuns.$inferSelect;
|
|
coordinator: typeof nativeRunFinalizations.$inferSelect;
|
|
}) {
|
|
if (!input.coordinator.resultId) return;
|
|
const resultRow = await input.db
|
|
.select({ resultJson: nativeRunResults.resultJson })
|
|
.from(nativeRunResults)
|
|
.where(
|
|
and(
|
|
eq(nativeRunResults.id, input.coordinator.resultId),
|
|
eq(nativeRunResults.runId, input.run.id),
|
|
eq(nativeRunResults.companyId, input.run.companyId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
const terminalState = record(
|
|
record(resultRow?.resultJson).terminal,
|
|
).runTerminalState;
|
|
if (!["succeeded", "failed", "cancelled"].includes(String(terminalState))) {
|
|
throw new Error("native_finalization_invalid");
|
|
}
|
|
const projectedStatus = projectNativeTerminalRunStatus(
|
|
terminalState as "succeeded" | "failed" | "cancelled",
|
|
);
|
|
const now = new Date();
|
|
const [updatedRun] = await input.db
|
|
.update(heartbeatRuns)
|
|
.set({
|
|
executionStatusDeliveryId: randomUUID(),
|
|
status: projectedStatus,
|
|
finishedAt: input.run.finishedAt ?? now,
|
|
nativePhase: "committed",
|
|
nativePhaseUpdatedAt: now,
|
|
...(terminalState === "succeeded"
|
|
? {
|
|
error: null,
|
|
errorCode: null,
|
|
// Capture the row being updated, not the earlier admission read:
|
|
// cleanup may have recorded a new diagnostic in the meantime.
|
|
// Other result metadata and all physical-owner evidence stay put.
|
|
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb)
|
|
|| ${recoveredExecutionFailureMetadata()}
|
|
|| jsonb_build_object('finalizationPhase', 'committed', 'failureCode', null, 'originalFailureCode', null, 'nextAttemptAt', null)`,
|
|
}
|
|
: { resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || jsonb_build_object('finalizationPhase', 'committed', 'failureCode', null, 'originalFailureCode', null, 'nextAttemptAt', null)` }),
|
|
updatedAt: now,
|
|
})
|
|
.where(
|
|
and(
|
|
eq(heartbeatRuns.id, input.run.id),
|
|
eq(heartbeatRuns.runtimeMode, "native"),
|
|
or(
|
|
inArray(heartbeatRuns.status, ["queued", "running", "failed"]),
|
|
and(
|
|
eq(heartbeatRuns.status, "succeeded"),
|
|
or(
|
|
isNotNull(heartbeatRuns.error),
|
|
isNotNull(heartbeatRuns.errorCode),
|
|
isNull(heartbeatRuns.finishedAt),
|
|
sql`${heartbeatRuns.nativePhase} is distinct from 'committed'`,
|
|
sql`${heartbeatRuns.resultJson}->>'finalizationPhase' is distinct from 'committed'`,
|
|
sql`${heartbeatRuns.resultJson}->>'nextAttemptAt' is not null`,
|
|
sql`${heartbeatRuns.resultJson}->>'failureCode' is not null`,
|
|
sql`${heartbeatRuns.resultJson}->>'originalFailureCode' is not null`,
|
|
),
|
|
),
|
|
),
|
|
// Reconciliation revisits committed results periodically. Only repair
|
|
// a changed projection; rewriting an unchanged failed run would mint a
|
|
// fresh status delivery (and failure toast) on every sweep. Check the
|
|
// current row so concurrent replays cannot both queue the same repair.
|
|
or(
|
|
sql`${heartbeatRuns.status} is distinct from ${projectedStatus}`,
|
|
isNull(heartbeatRuns.finishedAt),
|
|
sql`${heartbeatRuns.nativePhase} is distinct from 'committed'`,
|
|
sql`${heartbeatRuns.resultJson}->>'finalizationPhase' is distinct from 'committed'`,
|
|
sql`${heartbeatRuns.resultJson}->>'nextAttemptAt' is not null`,
|
|
sql`${heartbeatRuns.resultJson}->>'failureCode' is not null`,
|
|
sql`${heartbeatRuns.resultJson}->>'originalFailureCode' is not null`,
|
|
...(terminalState === "succeeded"
|
|
? [isNotNull(heartbeatRuns.error), isNotNull(heartbeatRuns.errorCode)]
|
|
: []),
|
|
),
|
|
nativeRunnerOwnershipNotHeldCondition(),
|
|
),
|
|
)
|
|
.returning();
|
|
// Metadata repairs can preserve the terminal status. Only a genuine status
|
|
// transition should emit another terminal event. A failed run that leaves a
|
|
// pending retry still needs its first failure report.
|
|
if (updatedRun && updatedRun.status !== input.run.status) {
|
|
await emitAgentTaskRun(input.db, updatedRun);
|
|
void reportRunFailure(input.db, updatedRun);
|
|
} else if (updatedRun && isNewTerminalRunFailure(input.run, updatedRun)) {
|
|
void reportRunFailure(input.db, updatedRun);
|
|
}
|
|
}
|
|
|
|
async function materializeCommittedReviewResponse(db: Db, runId: string) {
|
|
try {
|
|
const [run] = await db
|
|
.select()
|
|
.from(heartbeatRuns)
|
|
.where(eq(heartbeatRuns.id, runId))
|
|
.limit(1);
|
|
if (!run?.nativeIssueId || !run.resultJson?.externalChatReviewPresentation)
|
|
return;
|
|
const decisionId = record(
|
|
run.resultJson.externalChatReviewPresentation,
|
|
).decisionId;
|
|
if (
|
|
typeof decisionId !== "string" ||
|
|
(await hasMaterializedNativeReviewResponse(db, {
|
|
companyId: run.companyId,
|
|
issueId: run.nativeIssueId,
|
|
runId,
|
|
decisionId,
|
|
}))
|
|
)
|
|
return;
|
|
const summary = record(run.resultJson.nativeResult).summary;
|
|
if (
|
|
typeof summary !== "string" ||
|
|
!(await authorizeNativeChatReviewPresentation(db, {
|
|
companyId: run.companyId,
|
|
issueId: run.nativeIssueId,
|
|
runId,
|
|
resultJson: run.resultJson,
|
|
}))
|
|
)
|
|
return;
|
|
// The coordinator transaction is already committed. addComment owns a
|
|
// fresh issue transaction, rechecks the proof and carries selected files;
|
|
// its same-run/text and publication keys make recovery replay idempotent.
|
|
await issueService(db).addComment(
|
|
run.nativeIssueId,
|
|
summary,
|
|
{ agentId: run.agentId, runId },
|
|
{
|
|
authorizationReason: "allow_chat_run_presentation",
|
|
},
|
|
);
|
|
publishChatPublicationCommitSignal({
|
|
companyId: run.companyId,
|
|
issueId: run.nativeIssueId,
|
|
runId: run.id,
|
|
agentId: run.agentId,
|
|
eventType: "run.presentation.resolved",
|
|
});
|
|
} catch (error) {
|
|
// The durable committed proof remains retryable by the next reconciler
|
|
// sweep. Presentation failure must not reclassify a successful native run.
|
|
logger.warn(
|
|
{ runId, error },
|
|
"Committed chat review response is awaiting presentation retry",
|
|
);
|
|
}
|
|
}
|
|
|
|
async function acceptedResponseDigestMatches(
|
|
db: Db,
|
|
run: typeof heartbeatRuns.$inferSelect,
|
|
accepted: typeof nativeRunResults.$inferSelect,
|
|
): Promise<boolean> {
|
|
const envelope = record(accepted.resultJson);
|
|
const canonical = {
|
|
result: envelope.result,
|
|
terminal: envelope.terminal,
|
|
turnId: accepted.turnId,
|
|
};
|
|
const fingerprint = nativeSha256({
|
|
runId: run.id,
|
|
completionContractSha256: run.completionContractSha256,
|
|
canonicalSha256: accepted.canonicalSha256,
|
|
});
|
|
// The durable coordinator stores this canonical shape directly. The older
|
|
// ControlPlanePort also hashes its exact binding, which must be rebuilt from
|
|
// persisted run identity and the actual authenticated control-plane journal.
|
|
if (
|
|
`sha256:${nativeSha256(canonical)}` === accepted.canonicalSha256 &&
|
|
`sha256:${fingerprint}` === accepted.serverFingerprint
|
|
)
|
|
return true;
|
|
if (accepted.serverFingerprint !== fingerprint) return false;
|
|
const sources = await db
|
|
.selectDistinct({ sourceInstanceId: heartbeatRunEvents.sourceInstanceId })
|
|
.from(heartbeatRunEvents)
|
|
.where(
|
|
and(
|
|
eq(heartbeatRunEvents.companyId, run.companyId),
|
|
eq(heartbeatRunEvents.runId, run.id),
|
|
sql`${heartbeatRunEvents.payload}->'prpEvent'->>'sourceKind' = 'control_plane'`,
|
|
),
|
|
)
|
|
.limit(17);
|
|
if (sources.length > 16) return false;
|
|
return sources.some(
|
|
({ sourceInstanceId }) =>
|
|
typeof sourceInstanceId === "string" &&
|
|
nativeSha256({
|
|
binding: {
|
|
companyId: run.companyId,
|
|
issueId: run.nativeIssueId,
|
|
runId: run.id,
|
|
agentId: run.agentId,
|
|
sessionId: run.nativeSessionId,
|
|
completionContractId: run.completionContractId,
|
|
completionContractSha256: run.completionContractSha256,
|
|
sourceInstanceId: run.runnerInstanceId,
|
|
controlPlaneSourceInstanceId: sourceInstanceId,
|
|
},
|
|
...canonical,
|
|
}) === accepted.canonicalSha256,
|
|
);
|
|
}
|
|
|
|
/** Recover only the already accepted answer. Neither issue disposition nor
|
|
* provider state is repaired here; an unsafe native session stays quarantined. */
|
|
export async function repairCommittedNativeChatResponse(
|
|
db: Db,
|
|
input: { companyId: string; issueId: string; runId: string },
|
|
): Promise<boolean> {
|
|
let agentId: string | null = null;
|
|
let materialized = false;
|
|
try {
|
|
materialized = await db.transaction(async (transaction) => {
|
|
const tx = transaction as unknown as Db;
|
|
const [issue] = await tx
|
|
.select()
|
|
.from(issues)
|
|
.where(
|
|
and(
|
|
eq(issues.id, input.issueId),
|
|
eq(issues.companyId, input.companyId),
|
|
),
|
|
)
|
|
.for("update");
|
|
if (!issue) return false;
|
|
const [run] = await tx
|
|
.select()
|
|
.from(heartbeatRuns)
|
|
.where(
|
|
and(
|
|
eq(heartbeatRuns.id, input.runId),
|
|
eq(heartbeatRuns.companyId, input.companyId),
|
|
eq(heartbeatRuns.nativeIssueId, input.issueId),
|
|
eq(heartbeatRuns.runtimeMode, "native"),
|
|
),
|
|
)
|
|
.for("update");
|
|
if (
|
|
!run ||
|
|
isNativeRunnerOwnershipHeld(run) ||
|
|
!["succeeded", "failed", "timed_out"].includes(run.status) ||
|
|
!run.finishedAt
|
|
)
|
|
return false;
|
|
// An existing selected answer, including a subsequently deleted one, must
|
|
// never be replaced or resurrected by a background presentation repair.
|
|
if (
|
|
typeof record(record(run.resultJson).presentationDecision).commentId ===
|
|
"string"
|
|
)
|
|
return false;
|
|
const [coordinator] = await tx
|
|
.select()
|
|
.from(nativeRunFinalizations)
|
|
.where(
|
|
and(
|
|
eq(nativeRunFinalizations.runId, run.id),
|
|
eq(nativeRunFinalizations.companyId, input.companyId),
|
|
eq(nativeRunFinalizations.issueId, input.issueId),
|
|
eq(nativeRunFinalizations.phase, "committed"),
|
|
isNull(nativeRunFinalizations.leaseOwner),
|
|
),
|
|
)
|
|
.for("update");
|
|
if (
|
|
!coordinator?.resultId ||
|
|
!coordinator.decisionId ||
|
|
!coordinator.assessmentId
|
|
)
|
|
return false;
|
|
const [accepted] = await tx
|
|
.select()
|
|
.from(nativeRunResults)
|
|
.where(
|
|
and(
|
|
eq(nativeRunResults.id, coordinator.resultId),
|
|
eq(nativeRunResults.runId, run.id),
|
|
eq(nativeRunResults.companyId, input.companyId),
|
|
eq(nativeRunResults.issueId, input.issueId),
|
|
eq(
|
|
nativeRunResults.completionContractId,
|
|
run.completionContractId!,
|
|
),
|
|
eq(nativeRunResults.schemaStatus, "accepted"),
|
|
),
|
|
)
|
|
.for("share");
|
|
const [decision] = await tx
|
|
.select()
|
|
.from(statusDecisions)
|
|
.where(
|
|
and(
|
|
eq(statusDecisions.id, coordinator.decisionId),
|
|
eq(statusDecisions.runId, run.id),
|
|
eq(statusDecisions.companyId, input.companyId),
|
|
eq(statusDecisions.issueId, input.issueId),
|
|
eq(statusDecisions.assessmentId, coordinator.assessmentId),
|
|
),
|
|
)
|
|
.for("share");
|
|
const envelope = record(accepted?.resultJson);
|
|
const result = record(envelope.result);
|
|
const terminal = record(envelope.terminal);
|
|
if (
|
|
!accepted ||
|
|
!decision ||
|
|
result.schema !== "paperclip.run_result.v1" ||
|
|
!(await acceptedResponseDigestMatches(tx, run, accepted))
|
|
)
|
|
return false;
|
|
// Existing governed-review responses have their own stricter gate-bound
|
|
// presentation contract, including selected attachments. Do not bypass it.
|
|
if (record(decision.decisionJson).externalChatReviewPresentation)
|
|
return false;
|
|
const monitorResult = { nativeResult: result, finalizationPhase: "committed", finalizationReasonCode: decision.reasonCode };
|
|
if (readAcceptedTaskContinuationSummary(monitorResult) && run.status === "succeeded" &&
|
|
terminal.runTerminalState === "succeeded" && terminal.turnTerminalState === "completed" &&
|
|
terminal.reportedWorkDisposition === "yielded" && record(run.contextSnapshot).skipIssueComment !== true &&
|
|
!await hasChatRunOwnedProviderInteraction(tx, input)) {
|
|
const authorizationReason = await resolveChatRunPresentationAuthorizationReason(tx, input);
|
|
const resolved = resolveHeartbeatRunResponse({ resultJson: monitorResult });
|
|
if (!resolved.text) return false;
|
|
const comment = await issueService(db).addComment(input.issueId, resolved.text,
|
|
{ agentId: run.agentId, runId: run.id }, { authorizationReason, completionReply: true }, tx);
|
|
await tx.update(heartbeatRuns).set({
|
|
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({
|
|
presentationDecision: { ...resolved.decision, commentId: comment.id,
|
|
reasonCodes: [...resolved.decision.reasonCodes, "committed_monitor_response_recovered"] },
|
|
})}::jsonb`, updatedAt: new Date(),
|
|
}).where(eq(heartbeatRuns.id, run.id));
|
|
agentId = run.agentId;
|
|
return true;
|
|
}
|
|
// Copyback can be owned by reconciliation before the live heartbeat
|
|
// reaches its presentation step. Recover a completed internal task's
|
|
// exact final reply from the same accepted turn, without rerunning work
|
|
// or granting external-chat publication authority.
|
|
if (
|
|
!isConversation(issue) &&
|
|
run.status === "succeeded" &&
|
|
record(run.contextSnapshot).skipIssueComment !== true &&
|
|
!isExternalChatPresentationContext(run.contextSnapshot) &&
|
|
result.reportedWorkDisposition === "done" &&
|
|
terminal.runTerminalState === "succeeded" &&
|
|
terminal.turnTerminalState === "completed" &&
|
|
terminal.reportedWorkDisposition === "done" &&
|
|
(await resolveChatRunPresentationAuthorizationReason(tx, input)) === "internal_agent_write"
|
|
) {
|
|
const rows = await tx.select({ seq: heartbeatRunEvents.seq, payload: heartbeatRunEvents.payload })
|
|
.from(heartbeatRunEvents).where(and(
|
|
eq(heartbeatRunEvents.companyId, input.companyId),
|
|
eq(heartbeatRunEvents.runId, run.id),
|
|
eq(heartbeatRunEvents.eventType, "item.completed"),
|
|
sql`${heartbeatRunEvents.payload}->'prpEvent'->>'turnId' = ${accepted.turnId}`,
|
|
sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,kind}' = 'agentMessage'`,
|
|
sql`${heartbeatRunEvents.payload} #>> '{prpEvent,payload,channel}' = 'final'`,
|
|
)).orderBy(desc(heartbeatRunEvents.seq)).limit(200);
|
|
const finalAgentMessage = selectHeartbeatRunFinalAgentMessage({
|
|
candidates: rows.flatMap((row) => {
|
|
const candidate = readCompletedAssistantMessageCandidate({ seq: row.seq, prpEvent: record(row.payload).prpEvent });
|
|
return candidate ? [candidate] : [];
|
|
}),
|
|
});
|
|
if (!finalAgentMessage) return false;
|
|
const comments = await tx.select({ id: issueComments.id, body: issueComments.body, presentation: issueComments.presentation })
|
|
.from(issueComments).where(and(
|
|
eq(issueComments.companyId, input.companyId),
|
|
eq(issueComments.issueId, input.issueId),
|
|
eq(issueComments.createdByRunId, run.id),
|
|
)).orderBy(desc(issueComments.createdAt), desc(issueComments.id));
|
|
const resolved = resolveHeartbeatRunResponse({
|
|
resultJson: { ...record(run.resultJson), nativeResult: result },
|
|
existingComment: findHeartbeatRunCompletionComment(comments, run.resultJson),
|
|
finalAgentMessage,
|
|
});
|
|
if (!resolved.text || resolved.decision.commentAction !== "create") return false;
|
|
const comment = await issueService(db).addComment(input.issueId, resolved.text,
|
|
{ agentId: run.agentId, runId: run.id },
|
|
{ authorizationReason: "internal_agent_write", completionReply: true }, tx);
|
|
await tx.update(heartbeatRuns).set({
|
|
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify({
|
|
presentationDecision: { ...resolved.decision, commentId: comment.id,
|
|
reasonCodes: [...resolved.decision.reasonCodes, "committed_task_response_recovered"] },
|
|
})}::jsonb`, updatedAt: new Date(),
|
|
}).where(eq(heartbeatRuns.id, run.id));
|
|
agentId = run.agentId;
|
|
return true;
|
|
}
|
|
if (
|
|
result.reportedWorkDisposition !== "yielded" ||
|
|
record(result.continuation).kind !== "response_wake" ||
|
|
!Array.isArray(result.attentionRequests) ||
|
|
result.attentionRequests.length > 0 ||
|
|
terminal.runTerminalState !== "succeeded" ||
|
|
terminal.turnTerminalState !== "completed" ||
|
|
terminal.reportedWorkDisposition !== "yielded"
|
|
) return false;
|
|
await authorizeCommittedChatResponse(db, tx, {
|
|
...input,
|
|
agentId: run.agentId,
|
|
resultId: accepted.id,
|
|
});
|
|
if (
|
|
(await resolveChatRunPresentationAuthorizationReason(tx, input)) !==
|
|
CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON
|
|
)
|
|
return false;
|
|
const resolved = resolveHeartbeatRunResponse({
|
|
resultJson: {
|
|
nativeResult: result,
|
|
finalizationPhase: "committed",
|
|
finalizationReasonCode: decision.reasonCode,
|
|
},
|
|
preferFinalResponseOverExistingComment: true,
|
|
externalChatResponseWakeSummaryAuthorized: true,
|
|
externalChatCommittedResponseWakeSummaryAuthorized: true,
|
|
});
|
|
if (!resolved.text || resolved.decision.commentAction !== "create")
|
|
return false;
|
|
const comment = await issueService(db).addComment(
|
|
input.issueId,
|
|
resolved.text,
|
|
{ agentId: run.agentId, runId: run.id },
|
|
{ authorizationReason: CHAT_RUN_PRESENTATION_AUTHORIZATION_REASON, completionReply: true },
|
|
tx,
|
|
);
|
|
const presentationDecision = {
|
|
...resolved.decision,
|
|
commentId: comment.id,
|
|
reasonCodes: [
|
|
...resolved.decision.reasonCodes,
|
|
"committed_response_recovered",
|
|
],
|
|
};
|
|
await tx
|
|
.update(heartbeatRuns)
|
|
.set({
|
|
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || ${JSON.stringify(
|
|
{
|
|
presentationDecision,
|
|
nativeCommittedChatResponse: {
|
|
schema: "paperclip.native_committed_chat_response.v1",
|
|
resultId: accepted.id,
|
|
canonicalSha256: accepted.canonicalSha256,
|
|
decisionId: decision.id,
|
|
},
|
|
},
|
|
)}::jsonb`,
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(eq(heartbeatRuns.id, run.id));
|
|
await projectCommittedRun({ db: tx, run, coordinator });
|
|
agentId = run.agentId;
|
|
return true;
|
|
});
|
|
} catch (error) {
|
|
if (error instanceof CommittedChatResponseAuthorizationError) return false;
|
|
// Keep transient failures visible and retryable without reclassifying the
|
|
// committed run or acknowledging a publication that never committed.
|
|
logger.warn(
|
|
{ runId: input.runId, error },
|
|
"Committed chat response is awaiting presentation retry",
|
|
);
|
|
return false;
|
|
}
|
|
if (materialized && agentId)
|
|
publishChatPublicationCommitSignal({
|
|
...input,
|
|
agentId,
|
|
eventType: "run.presentation.resolved",
|
|
});
|
|
return materialized;
|
|
}
|
|
|
|
/** Presentation recovery is independent from re-arbitrating issue status. */
|
|
export async function repairCommittedNativeReviewResponse(
|
|
db: Db,
|
|
input: {
|
|
companyId: string;
|
|
issueId: string;
|
|
runId: string;
|
|
decisionId: string;
|
|
resultId: string;
|
|
assessmentId: string;
|
|
},
|
|
) {
|
|
if (await hasMaterializedNativeReviewResponse(db, input)) return;
|
|
const restored = await db.transaction((tx) =>
|
|
restoreNativeChatReviewPresentationInTransaction(
|
|
tx as unknown as Db,
|
|
input,
|
|
),
|
|
);
|
|
if (restored) await materializeCommittedReviewResponse(db, input.runId);
|
|
}
|
|
export async function finalizeNativeRun(input: {
|
|
db: Db;
|
|
runId: string;
|
|
workspaceFinalizeStatus: "succeeded" | "failed";
|
|
/** Reconciliation owns terminal run projection; the live heartbeat does it afterward. */
|
|
projectRunStatus?: boolean;
|
|
/** Workspace-only replay must not consume the provider recovery budget. */
|
|
preserveProviderAttempt?: boolean;
|
|
failpoint?: NativeStatusCommitFailpoint;
|
|
}) {
|
|
const run = await input.db
|
|
.select()
|
|
.from(heartbeatRuns)
|
|
.where(eq(heartbeatRuns.id, input.runId))
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!run || run.runtimeMode !== "native")
|
|
throw new Error("native_finalization_run_missing");
|
|
const claim = await claimCoordinator({
|
|
db: input.db,
|
|
runId: input.runId,
|
|
preserveProviderAttempt: input.preserveProviderAttempt,
|
|
});
|
|
const coordinator = claim.coordinator;
|
|
if (!claim.leaseOwner && coordinator.phase === "committed") {
|
|
if (isNativeRunnerOwnershipHeld(run)) return coordinator;
|
|
const presentationAlreadyMaterialized = coordinator.decisionId
|
|
? await hasMaterializedNativeReviewResponse(input.db, {
|
|
companyId: run.companyId,
|
|
issueId: coordinator.issueId,
|
|
runId: run.id,
|
|
decisionId: coordinator.decisionId,
|
|
})
|
|
: false;
|
|
if (
|
|
!presentationAlreadyMaterialized &&
|
|
coordinator.decisionId &&
|
|
coordinator.resultId &&
|
|
coordinator.assessmentId
|
|
) {
|
|
await input.db.transaction((tx) =>
|
|
restoreNativeChatReviewPresentationInTransaction(tx as unknown as Db, {
|
|
companyId: run.companyId,
|
|
issueId: coordinator.issueId,
|
|
runId: run.id,
|
|
decisionId: coordinator.decisionId!,
|
|
resultId: coordinator.resultId!,
|
|
assessmentId: coordinator.assessmentId!,
|
|
}),
|
|
);
|
|
}
|
|
if (input.projectRunStatus)
|
|
await projectCommittedRun({ db: input.db, run, coordinator });
|
|
if (input.projectRunStatus && !presentationAlreadyMaterialized)
|
|
await materializeCommittedReviewResponse(input.db, input.runId);
|
|
if (input.projectRunStatus)
|
|
await repairCommittedNativeChatResponse(input.db, {
|
|
companyId: run.companyId,
|
|
issueId: coordinator.issueId,
|
|
runId: run.id,
|
|
});
|
|
if (input.projectRunStatus) {
|
|
await settleExternalConversation(input.db, run.companyId, coordinator.issueId).catch((err) => {
|
|
logger.warn({ err, runId: run.id }, "Slack conversation settlement deferred to reconciliation");
|
|
});
|
|
}
|
|
await resolveCommittedFinalizationRecovery(input.db, run, coordinator.issueId);
|
|
if (input.projectRunStatus) {
|
|
const [projectedRun] = await input.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run.id));
|
|
if (projectedRun) await reconcileMuseNativeTerminalRun(input.db, projectedRun);
|
|
}
|
|
return coordinator;
|
|
}
|
|
const [resultRow, contractRow] = await Promise.all([
|
|
input.db
|
|
.select()
|
|
.from(nativeRunResults)
|
|
.where(
|
|
and(
|
|
eq(nativeRunResults.id, coordinator.resultId!),
|
|
eq(nativeRunResults.companyId, run.companyId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null),
|
|
input.db
|
|
.select()
|
|
.from(completionContracts)
|
|
.where(
|
|
and(
|
|
eq(completionContracts.id, run.completionContractId!),
|
|
eq(completionContracts.companyId, run.companyId),
|
|
eq(completionContracts.issueId, coordinator.issueId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null),
|
|
]);
|
|
if (!resultRow) throw new Error("native_result_missing");
|
|
if (!contractRow) throw new Error("native_completion_contract_missing");
|
|
const envelope = record(resultRow.resultJson);
|
|
const result = record(envelope.result);
|
|
const terminal = record(envelope.terminal);
|
|
const terminalState = terminal.runTerminalState;
|
|
if (!["succeeded", "failed", "cancelled"].includes(String(terminalState))) {
|
|
throw new Error("native_finalization_invalid");
|
|
}
|
|
|
|
await input.db
|
|
.update(nativeRunFinalizations)
|
|
.set({ phase: "ready_for_assessment", updatedAt: new Date() })
|
|
.where(
|
|
and(
|
|
eq(nativeRunFinalizations.runId, input.runId),
|
|
eq(nativeRunFinalizations.leaseOwner, claim.leaseOwner!),
|
|
),
|
|
);
|
|
const classifiedAssessment = await classifyNativeEvidence({
|
|
db: input.db,
|
|
companyId: run.companyId,
|
|
issueId: coordinator.issueId,
|
|
runId: run.id,
|
|
contract: record(contractRow.contractJson),
|
|
result,
|
|
});
|
|
const persistedIgnoredAttention = record(
|
|
envelope.normalizationDiagnostics,
|
|
).ignoredAttentionRequests;
|
|
const assessment = {
|
|
...classifiedAssessment,
|
|
ignoredAttentionRequests: [
|
|
...(Array.isArray(persistedIgnoredAttention)
|
|
? (persistedIgnoredAttention as PrpIgnoredAttentionRequest[])
|
|
: []),
|
|
...classifiedAssessment.ignoredAttentionRequests,
|
|
],
|
|
};
|
|
|
|
await dismissAutomaticCompletionReviews(input.db, coordinator.issueId);
|
|
const sourceWake = run.wakeupRequestId ? await input.db.select({ payload: agentWakeupRequests.payload })
|
|
.from(agentWakeupRequests).where(and(eq(agentWakeupRequests.id, run.wakeupRequestId),
|
|
eq(agentWakeupRequests.companyId, run.companyId))).then((rows) => rows[0]) : null;
|
|
// One follow-up may repair an incomplete report. Repeated incomplete results
|
|
// require a visible recovery action instead of an unbounded wake loop.
|
|
const allowIncompleteContinuation = record(sourceWake?.payload).continuationIdempotencyKey !== "native-completion-incomplete";
|
|
let supersedesAssessmentId: string | null = coordinator.assessmentId;
|
|
for (let attempt = 0; attempt < 3; attempt += 1) {
|
|
const authoritativeIssue = await input.db
|
|
.select()
|
|
.from(issues)
|
|
.where(
|
|
and(
|
|
eq(issues.id, coordinator.issueId),
|
|
eq(issues.companyId, run.companyId),
|
|
),
|
|
)
|
|
.limit(1)
|
|
.then((rows) => rows[0] ?? null);
|
|
if (!authoritativeIssue)
|
|
throw new Error("native_finalization_issue_missing");
|
|
const governanceGate = await pendingNativeGovernance({
|
|
db: input.db,
|
|
companyId: run.companyId,
|
|
issueId: authoritativeIssue.id,
|
|
runId: run.id,
|
|
executionState: record(authoritativeIssue.executionState),
|
|
});
|
|
const externalChatResponseWaitAuthorization =
|
|
assessment.reportedDisposition === "yielded" &&
|
|
assessment.continuation?.kind === "response_wake"
|
|
? await resolveExternalChatResponseWaitAuthorization({
|
|
db: input.db,
|
|
binding: {
|
|
companyId: run.companyId,
|
|
issueId: authoritativeIssue.id,
|
|
runId: run.id,
|
|
agentId: run.agentId,
|
|
},
|
|
})
|
|
: "not_applicable";
|
|
const boardResponseWaitOrigin =
|
|
externalChatResponseWaitAuthorization === "not_applicable" &&
|
|
assessment.reportedDisposition === "yielded" &&
|
|
assessment.continuation?.kind === "response_wake" &&
|
|
!governanceGate
|
|
? await readNativeBoardResponseWaitOrigin(input.db, {
|
|
companyId: run.companyId,
|
|
issueId: authoritativeIssue.id,
|
|
runId: run.id,
|
|
agentId: run.agentId,
|
|
})
|
|
: null;
|
|
const boardResponseWait = boardResponseWaitOrigin
|
|
? await readNativeBoardResponseWaitSource(
|
|
input.db,
|
|
boardResponseWaitOrigin,
|
|
)
|
|
: null;
|
|
const [dependencyReadiness, resolvedInteraction, pauseHold] = await Promise.all([
|
|
issueService(input.db).getDependencyReadiness(
|
|
authoritativeIssue.id,
|
|
input.db,
|
|
),
|
|
acceptedInteractionFromRun({
|
|
db: input.db,
|
|
companyId: run.companyId,
|
|
issueId: authoritativeIssue.id,
|
|
runId: run.id,
|
|
}),
|
|
(terminalState === "failed" || (assessment.reportedDisposition === "yielded" &&
|
|
assessment.continuation?.kind === "response_wake" &&
|
|
assessment.hasBlockingRemainingWork))
|
|
? issueTreeControlService(input.db).getActivePauseHoldGate(
|
|
run.companyId, authoritativeIssue.id,
|
|
)
|
|
: Promise.resolve(null),
|
|
]);
|
|
const reviewContext = readNativeReviewAssignmentContext(run.contextSnapshot);
|
|
const nativeReview = reviewContext ? await getNativeReviewAssignment(input.db, {
|
|
companyId: run.companyId, issueId: authoritativeIssue.id, agentId: run.agentId,
|
|
contextSnapshot: reviewContext, allowResolvedByRunId: run.id,
|
|
}) : null;
|
|
const planWait = isNativePlanWaitResult(result)
|
|
? await readNativePlanWait(input.db, { companyId: run.companyId, issueId: authoritativeIssue.id, runId: run.id, agentId: run.agentId })
|
|
: null;
|
|
// Loss of the authority behind this server-issued wait must never fall
|
|
// through to the generic response_wake auto-continuation branch.
|
|
if (isNativePlanWaitResult(result) &&
|
|
(!planWait || nativeSha256(planWait.result) !== nativeSha256(result))) {
|
|
throw new Error("native_plan_wait_authority_lost");
|
|
}
|
|
const { providerFailure, turnFailure } = await readPersistedNativeFailureEvidence(
|
|
input.db, run, resultRow.turnId, envelope.terminal as PrpTerminalState,
|
|
);
|
|
const ownsProviderFailureDecision =
|
|
(!authoritativeIssue.executionRunId || authoritativeIssue.executionRunId === run.id) &&
|
|
(reviewContext ? nativeReview?.interaction.status === "pending"
|
|
: authoritativeIssue.assigneeAgentId === run.agentId && !authoritativeIssue.assigneeUserId);
|
|
const childCompletionRecipient = {
|
|
companyId: run.companyId, issueId: authoritativeIssue.id, agentId: run.agentId, runId: run.id,
|
|
sourceIntentId: typeof record(run.contextSnapshot).nativeStatusWakeIntentId === "string"
|
|
? record(run.contextSnapshot).nativeStatusWakeIntentId as string : null,
|
|
};
|
|
const hasPendingChildCompletion = !reviewContext &&
|
|
await hasPendingNativeChildCompletion(input.db, childCompletionRecipient);
|
|
const monitorWaitAt = scheduledIssueMonitorWait(authoritativeIssue, run.agentId);
|
|
const monitorDelivery = assessment.reportedDisposition === "yielded" &&
|
|
assessment.continuation?.kind === "monitor" &&
|
|
authoritativeIssue.executionRunId !== run.id && authoritativeIssue.checkoutRunId !== run.id
|
|
? await readNativeMonitorDelivery(input.db, { companyId: run.companyId, issueId: coordinator.issueId, runId: run.id })
|
|
: null;
|
|
const proposedDecision = resolveNativeFinalizerStatus({
|
|
caseHandoffAuthorized: Boolean(await assertNativeCaseHandoffResult(input.db, run.id, authoritativeIssue.id, result)),
|
|
monitorWaitDispatched: monitorDelivery !== null,
|
|
monitorWaitAuthorized: authoritativeIssue.workMode === "standard" && authoritativeIssue.executionRunId === run.id && monitorWaitAt !== null,
|
|
planWaitAuthorized: planWait !== null,
|
|
hasPendingChildCompletion,
|
|
providerModelRejected: providerFailure?.errorCode === "native_provider_model_rejected" && ownsProviderFailureDecision,
|
|
providerOverloaded: providerFailure?.errorCode === "native_provider_overloaded" && ownsProviderFailureDecision,
|
|
providerFailureSuperseded: providerFailure?.errorCode === "native_provider_overloaded" && !ownsProviderFailureDecision,
|
|
failureRetryCount: executionFailureRetryCount(run),
|
|
...(reviewContext ? { nativeReviewOutcome: nativeReview
|
|
? nativeReview.interaction.status === "pending" ? "pending" as const : "resolved" as const
|
|
: "stale" as const } : {}),
|
|
assessment,
|
|
terminalState: terminalState as "succeeded" | "failed" | "cancelled",
|
|
workspaceFinalizeStatus: input.workspaceFinalizeStatus,
|
|
governanceGate,
|
|
allowIncompleteContinuation,
|
|
completionClaimPolicyAccepted:
|
|
contractRow.risk === "low" &&
|
|
contractRow.completionAuthority === "agent_claim_policy",
|
|
hasUnresolvedIssueBlockers:
|
|
dependencyReadiness.unresolvedBlockerCount > 0,
|
|
governanceResolvedForRun: resolvedInteraction !== null,
|
|
externalChatResponseWaitAuthorization,
|
|
boardResponseWaitAuthorized: boardResponseWait !== null,
|
|
boardResponseWaitOrigin: boardResponseWaitOrigin !== null,
|
|
isConversation: isConversation(authoritativeIssue),
|
|
hasActivePauseHold: pauseHold !== null,
|
|
reviewOwnerUserId:
|
|
authoritativeIssue.responsibleUserId ??
|
|
authoritativeIssue.createdByUserId ??
|
|
null,
|
|
agentId: run.agentId,
|
|
priorIssueStatus: authoritativeStatus(authoritativeIssue.status),
|
|
});
|
|
const decision = conversationNativeDecision({
|
|
conversation: isConversation(authoritativeIssue), terminalState,
|
|
workspaceFinalizeStatus: input.workspaceFinalizeStatus, hasGovernanceGate: !!governanceGate,
|
|
priorStatus: authoritativeStatus(authoritativeIssue.status), decision: proposedDecision,
|
|
});
|
|
const assessmentRow = await recordNativeWorkAssessment({
|
|
db: input.db,
|
|
companyId: run.companyId,
|
|
issueId: authoritativeIssue.id,
|
|
runId: run.id,
|
|
turnId: resultRow.turnId,
|
|
contractId: resultRow.completionContractId,
|
|
contractCanonicalSha256: contractRow.canonicalSha256,
|
|
resultId: resultRow.id,
|
|
resultCanonicalSha256: resultRow.canonicalSha256,
|
|
priorIssueStatus: authoritativeIssue.status,
|
|
priorStatusVersion: Number(authoritativeIssue.statusVersion),
|
|
priorDecisionId: authoritativeIssue.lastStatusDecisionId,
|
|
policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION,
|
|
assessment,
|
|
supersedesAssessmentId,
|
|
workspaceFinalizeStatus: input.workspaceFinalizeStatus,
|
|
monitorDelivery,
|
|
monitorWaitAt: assessment.continuation?.kind === "monitor" ? monitorWaitAt : null,
|
|
});
|
|
await input.db
|
|
.update(nativeRunFinalizations)
|
|
.set({
|
|
phase: "arbitrating",
|
|
// The old decision refers to its original assessment through a
|
|
// composite foreign key. Replace this pair together in the status
|
|
// committer, after the new decision and its effects are durable.
|
|
// Before any decision exists, retain the assessment for crash recovery.
|
|
assessmentId: sql`case when ${nativeRunFinalizations.decisionId} is null
|
|
then ${assessmentRow.id}::uuid else ${nativeRunFinalizations.assessmentId} end`,
|
|
updatedAt: new Date(),
|
|
})
|
|
.where(
|
|
and(
|
|
eq(nativeRunFinalizations.runId, input.runId),
|
|
eq(nativeRunFinalizations.leaseOwner, claim.leaseOwner!),
|
|
),
|
|
);
|
|
try {
|
|
const repairBoardResponseWait =
|
|
boardResponseWait !== null &&
|
|
assessment.hasBlockingRemainingWork &&
|
|
["completion_evidence_incomplete", "prior_status_preserved_no_live_path"].includes(decision.reasonCode ?? "");
|
|
const committed = await commitNativeStatusDecision({
|
|
db: input.db,
|
|
companyId: run.companyId,
|
|
issueId: authoritativeIssue.id,
|
|
runId: run.id,
|
|
assessmentId: assessmentRow.id,
|
|
priorStatus: authoritativeIssue.status,
|
|
priorStatusVersion: Number(authoritativeIssue.statusVersion),
|
|
priorDecisionId: authoritativeIssue.lastStatusDecisionId,
|
|
decision,
|
|
requireMonitorWait: decision.reasonCode === "scheduled_monitor_waiting" && monitorWaitAt
|
|
? { agentId: run.agentId, nextCheckAt: monitorWaitAt } : undefined,
|
|
requireMonitorDelivery: decision.reasonCode === "scheduled_monitor_dispatched"
|
|
? monitorDelivery ?? undefined : undefined,
|
|
requirePlanWaitSource:
|
|
decision.reasonCode === "native_plan_accepted_waiting_for_continuation"
|
|
? planWait?.source : undefined,
|
|
requireNoPendingChildCompletion: decision.statusAction === "done" && !reviewContext
|
|
? childCompletionRecipient : undefined,
|
|
requireModelRejectionOwner: decision.reasonCode === "native_provider_model_rejected"
|
|
? { agentId: run.agentId, reviewContext }
|
|
: undefined,
|
|
requireProviderFailureOwner: decision.reasonCode?.startsWith("native_provider_overloaded")
|
|
? { agentId: run.agentId, reviewContext } : undefined,
|
|
requireBoardResponseWaitSource:
|
|
decision.reasonCode === "board_response_waiting" || repairBoardResponseWait
|
|
? boardResponseWait?.source
|
|
: undefined,
|
|
requireBoardResponseWaitOrigin:
|
|
decision.reasonCode === "board_response_waiting" ||
|
|
decision.reasonCode === "board_response_wait_superseded" ||
|
|
repairBoardResponseWait
|
|
? (boardResponseWaitOrigin ?? undefined)
|
|
: undefined,
|
|
requireExternalChatResponseWaitAuthorization:
|
|
decision.reasonCode === "external_chat_response_waiting"
|
|
? { agentId: run.agentId }
|
|
: undefined,
|
|
reviewResponsePresentation:
|
|
decision.reasonCode === "governed_response_waiting" &&
|
|
governanceGate?.kind === "interaction" &&
|
|
externalChatResponseWaitAuthorization === "authorized"
|
|
? {
|
|
agentId: run.agentId,
|
|
resultId: resultRow.id,
|
|
gateId: governanceGate.id,
|
|
}
|
|
: undefined,
|
|
failpoint: input.failpoint,
|
|
});
|
|
const now = new Date();
|
|
const finalizationFailed = decision.effects.some(
|
|
(effect) => effect.kind === "record_finalization_error",
|
|
);
|
|
const finalizationPhase = finalizationFailed
|
|
? "retryable_failure"
|
|
: "committed";
|
|
// commitNativeStatusDecision() already committed its own transaction above.
|
|
// Its "cancel_continuations" effect (status-decision-committer.ts) writes a
|
|
// terminal status to this same run and emits for it. Skip the emit here in
|
|
// that case so one run reaching a terminal state emits exactly one event.
|
|
const alreadyEmittedByCommittedDecision = decision.effects.some(
|
|
(effect) => effect.kind === "cancel_continuations",
|
|
);
|
|
const clearExecutionFailure = input.projectRunStatus && finalizationPhase === "committed" && terminalState === "succeeded";
|
|
const finalizationMetadata = {
|
|
...(providerFailure ? { nativeProviderFailure: providerFailure.diagnostic } : {}),
|
|
...(turnFailure ? { nativeTurnFailure: turnFailure } : {}),
|
|
finalizationPhase,
|
|
...(finalizationPhase === "committed" ? { failureCode: null, originalFailureCode: null, nextAttemptAt: null } : {}),
|
|
assessmentId: assessmentRow.id,
|
|
decisionId: committed.decision.id,
|
|
authoritativeDecision: decision.toStatus,
|
|
finalizationPolicyVersion: decision.policyVersion,
|
|
finalizationReasonCode: decision.reasonCode,
|
|
// Only the locked status transaction can mint this presentation
|
|
// proof. Never carry a runner-provided marker forward.
|
|
externalChatReviewPresentation: record(
|
|
committed.decision.decisionJson,
|
|
).externalChatReviewPresentation
|
|
? {
|
|
...record(
|
|
record(committed.decision.decisionJson)
|
|
.externalChatReviewPresentation,
|
|
),
|
|
decisionId: committed.decision.id,
|
|
}
|
|
: null,
|
|
...(record(committed.decision.decisionJson)
|
|
.externalChatReviewPresentation
|
|
? { nativeResult: result }
|
|
: {}),
|
|
verificationCaveats: assessment.verificationCaveats,
|
|
ignoredAttentionRequests: assessment.ignoredAttentionRequests,
|
|
issueStatusBefore: authoritativeIssue.status,
|
|
issueStatusAfter: committed.issue.status,
|
|
statusVersionBefore: Number(authoritativeIssue.statusVersion),
|
|
statusVersionAfter: Number(committed.issue.statusVersion),
|
|
workspaceFinalizeStatus: input.workspaceFinalizeStatus,
|
|
};
|
|
const [updatedRun] = await input.db
|
|
.update(heartbeatRuns)
|
|
.set({
|
|
executionStatusDeliveryId: randomUUID(),
|
|
...(input.projectRunStatus
|
|
? {
|
|
status:
|
|
terminalState === "succeeded"
|
|
? "succeeded"
|
|
: terminalState === "cancelled"
|
|
? "cancelled"
|
|
: "failed",
|
|
finishedAt: now,
|
|
}
|
|
: {}),
|
|
nativePhase: finalizationPhase,
|
|
nativePhaseUpdatedAt: now,
|
|
...(clearExecutionFailure ? { error: null, errorCode: null } : {}),
|
|
...(providerFailure ? {
|
|
// Recovery can finalize before heartbeat saves the adapter result.
|
|
// Retain any independently recorded execution/cleanup failure.
|
|
error: sql`coalesce(${heartbeatRuns.error}, ${providerFailure.errorMessage})`,
|
|
errorCode: sql`coalesce(${heartbeatRuns.errorCode}, ${providerFailure.errorCode})`,
|
|
} : {}),
|
|
resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb)
|
|
|| ${JSON.stringify(finalizationMetadata)}::jsonb
|
|
|| ${clearExecutionFailure ? recoveredExecutionFailureMetadata() : sql`'{}'::jsonb`}`,
|
|
updatedAt: now,
|
|
})
|
|
.where(eq(heartbeatRuns.id, run.id))
|
|
.returning();
|
|
if (
|
|
input.projectRunStatus &&
|
|
!alreadyEmittedByCommittedDecision &&
|
|
updatedRun
|
|
) {
|
|
await emitAgentTaskRun(input.db, updatedRun);
|
|
void reportRunFailure(input.db, updatedRun);
|
|
}
|
|
if (input.projectRunStatus)
|
|
await materializeCommittedReviewResponse(input.db, input.runId);
|
|
if (input.projectRunStatus && finalizationPhase === "committed")
|
|
await repairCommittedNativeChatResponse(input.db, {
|
|
companyId: run.companyId,
|
|
issueId: coordinator.issueId,
|
|
runId: run.id,
|
|
});
|
|
if (input.projectRunStatus && finalizationPhase === "committed") {
|
|
await settleExternalConversation(input.db, run.companyId, coordinator.issueId).catch((err) => {
|
|
logger.warn({ err, runId: run.id }, "Slack conversation settlement deferred to reconciliation");
|
|
});
|
|
}
|
|
if (finalizationPhase === "committed") await resolveCommittedFinalizationRecovery(input.db, run, coordinator.issueId);
|
|
if (input.projectRunStatus && updatedRun) await reconcileMuseNativeTerminalRun(input.db, updatedRun);
|
|
return {
|
|
...coordinator,
|
|
phase: finalizationPhase,
|
|
assessmentId: assessmentRow.id,
|
|
decisionId: committed.decision.id,
|
|
};
|
|
} catch (error) {
|
|
if (error instanceof NativeStatusRaceError && attempt < 2) {
|
|
supersedesAssessmentId = assessmentRow.id;
|
|
await new Promise((resolve) => setTimeout(resolve, 25 * (attempt + 1)));
|
|
continue;
|
|
}
|
|
return recordRetryableFailure({
|
|
db: input.db,
|
|
run,
|
|
coordinator,
|
|
failureCode:
|
|
error instanceof NativeStatusRaceError
|
|
? "status_cas_exhausted"
|
|
: "side_effect_planning_failed",
|
|
message: error instanceof Error ? error.message : String(error),
|
|
nextAction:
|
|
error instanceof NativeStatusRaceError
|
|
? "Reassess against the latest authoritative issue status version."
|
|
: "Retry atomic status and liveness materialization from the persisted assessment.",
|
|
projectRunStatus: input.projectRunStatus,
|
|
});
|
|
}
|
|
}
|
|
throw new Error("native_finalization_status_race");
|
|
}
|