From 46c410e7f6eb4b1460e4bd248a934d449dcb56c7 Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 26 Oct 2026 02:04:43 -0500 Subject: [PATCH] refactor(server): isolate native status reassessment ownership Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 9 + ...-native-status-reassessment-sleep-audit.md | 28 ++ .../native-finalization-reconciler.ts | 326 +----------------- .../native-reconciliation-status.ts | 114 ++++++ .../native-status-reassessment.ts | 222 ++++++++++++ 5 files changed, 385 insertions(+), 314 deletions(-) create mode 100644 server/src/services/native-runtime/native-reconciliation-status.ts create mode 100644 server/src/services/native-runtime/native-status-reassessment.ts diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index e8fb236b2be..763e24cd9fd 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2452,3 +2452,12 @@ draft and unmerged. /tmp/sleep49-final-gates.json, /tmp/sleep49-comments.json and /tmp/sleep49-review-threads.json preserve the observations. The PR body records the passing remote gates and keeps the local embedded-database failure visible. Slice 50 and all other remaining original obligations are incomplete. + + +Slice 50 extraction checkpoint: reassessment and its pure policy now have native +runtime owners, with the original reconciler delegating to them. All 94 status +corpus cases pass before and after extraction; all 187 governed-response cases +and server types pass afterward. No polling duty has been removed. The audit +records the critical transaction boundary: scheduled guards belong inside the +status committer, which publishes after its own transaction. Producer coverage, +worker selection, lifecycle/cost proof and a new PR remain outstanding. diff --git a/doc/plans/2026-10-26-native-status-reassessment-sleep-audit.md b/doc/plans/2026-10-26-native-status-reassessment-sleep-audit.md index 5bec68fc4cb..e7dbb3e7cce 100644 --- a/doc/plans/2026-10-26-native-status-reassessment-sleep-audit.md +++ b/doc/plans/2026-10-26-native-status-reassessment-sleep-audit.md @@ -96,3 +96,31 @@ broader scopes; ordinary writes must not reload every company run. This is an audit, not an implemented migration. No polling entrance is removed by this document. No Cloud alarm coverage or host sleep eligibility changes. + + +## Owner extraction checkpoint + +The status-authority corpus passes all 94 cases before extraction on a fresh +PostgreSQL database. Reassessment now lives in native-status-reassessment.ts; +its pure arbitration policy lives in native-reconciliation-status.ts. Existing +imports through the reconciler remain compatible. The combined reconciler +delegates to the extracted owner and retains all periodic behavior for now. +It avoids the legacy review-response lookup when that repair is disabled. + +The extracted owner distinguishes unchanged, superseded, deferred and committed +outcomes. Operational errors still propagate; the existing status-race behavior +remains deferred. All 94 status-authority cases and 187 governed-response cases +pass after extraction, as does server typecheck. Evidence: /tmp/sleep50-baseline.log, +/tmp/sleep50-extract-corpus.log, /tmp/sleep50-extract-responses.log and +/tmp/sleep50-extract-types.log. Each corpus run uses a fresh database because its +company prefixes persist beyond the suite. + +The scheduler entry point is not ready yet. The status committer locks the +coordinator before the task and publishes activity after its own transaction. +Wrapping it in a new outer transaction would publish before that outer commit; +do not do that. Add any scheduled source/lease/deadline guard at the committer's +existing transaction boundary, and exercise competing source/claim changes there. +The new worker must reload source facts and preserve deferred races/retries; it +must not treat an extracted snapshot-based helper as fresh scheduling authority. +Producer notifications, discovery cost, lifecycle tests and periodic removal +remain outstanding. No slice 50 PR has been published. diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index 50faf1722d9..07bbeaabbec 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -1,42 +1,37 @@ +import { reassessNativeStatusFromSource } from "./native-status-reassessment.js"; +import { resolveNativeReconciliationStatus } from "./native-reconciliation-status.js"; +export { resolveNativeReconciliationStatus, type NativeReconciliationFacts } from "./native-reconciliation-status.js"; import { pendingNativeStatusEffectsCondition } from "./native-pending-effect-work.js"; import { notifyUnsafeWorkspaceExportRunChange } from "../unsafe-workspace-export-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import type { WorkspaceWorkers } from "../plugin-workspaces.js"; import { recoverLegacyUnsafeWorkspaceExports } from "./native-workspace-export-recovery.js"; -import { hasPendingNativeChildCompletion } from "./native-child-completion-delivery.js"; -import { dismissAutomaticCompletionReviews, decisionHasRetiredAutomaticReview } from "./automatic-completion-reviews.js"; +import { dismissAutomaticCompletionReviews } from "./automatic-completion-reviews.js"; import { logger } from "../../middleware/logger.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "../delivery-work-notifications.js"; import { createHash, randomUUID } from "node:crypto"; import { and, asc, desc, eq, gt, inArray, isNotNull, isNull, lt, lte, ne, not, notInArray, or, sql, type SQL } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { - completionContracts, agents, heartbeatRunEvents, heartbeatRuns, - issueWorkProducts, issues, nativeRunFinalizations, nativeRunResults, statusDecisionEffects, statusDecisions, - workAssessments, workspaceOperations, } from "@paperclipai/db"; import { finalizeNativeRun, - pendingNativeGovernance, - resolveNativeFinalizerStatus, recordNativeFinalizationFailure, repairCommittedNativeReviewResponse, repairCommittedNativeChatResponse, } from "./native-run-finalizer.js"; import { - commitNativeStatusDecision, dispatchPendingNativeStatusEffects, - NativeStatusRaceError, } from "./status-decision-committer.js"; import { issueRecoveryActionService } from "../issue-recovery-actions.js"; import { issueService } from "../issues.js"; @@ -49,17 +44,13 @@ import { readNativeWorkspaceSyncReference, } from "./native-workspace-sync.js"; import type { EnvironmentRuntimeService } from "../environment-runtime.js"; -import { classifyNativeEvidence } from "./evidence-classifier.js"; import { createNativeReconciliationPassTiming } from "./native-reconciliation-timing.js"; -import { recordNativeWorkAssessment } from "./work-assessments.js"; import { isNativeRunnerOwnershipHeld, nativeRunnerOwnershipNotHeldCondition, } from "./native-runner-ownership.js"; import { - NATIVE_STATUS_ARBITER_POLICY_VERSION, type NativeAuthoritativeIssueStatus, - type NativeStatusDecision, } from "./status-arbiter.js"; import { getNativeReviewAssignment } from "./native-review-participant.js"; @@ -68,125 +59,12 @@ import { isAutomaticRecoverySuppressedByPauseHold } from "../recovery/pause-hold import { readRunCancellation } from "../run-cancellation.js"; import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; -export type NativeReconciliationFacts = { - equivalentAttentionFamily?: boolean; - canonicalReplay?: boolean; - callerMaterialConflict?: boolean; - workspaceOperationPending?: boolean; - undeliveredEffectCount?: number; - authoritativeStatusChanged?: boolean; - newEvidenceSatisfiesContract?: boolean; - hasPendingChildCompletion?: boolean; - dependencyResolved?: boolean; - authorizedResume?: boolean; - statusVersionAdvanced?: boolean; -}; - function record(value: unknown): Record { return value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; } -/** Append-only recovery/reconciliation re-arbitration from durable facts. */ -export function resolveNativeReconciliationStatus(input: { - facts: NativeReconciliationFacts; - priorIssueStatus: NativeAuthoritativeIssueStatus; - agentId: string; -}): NativeStatusDecision { - const preserve = (reasonCode: string, effects: NativeStatusDecision["effects"]): NativeStatusDecision => ({ - policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, - statusAction: "preserve", - toStatus: input.priorIssueStatus, - reasonCode, - unblockDescriptor: null, - effects, - }); - const continuation = (reasonCode: string, effects: NativeStatusDecision["effects"] = []): NativeStatusDecision => ({ - policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, - statusAction: "in_progress", - toStatus: "in_progress", - reasonCode, - unblockDescriptor: null, - effects: [{ - kind: "enqueue_continuation", - continuationKind: "same_agent", - summary: `Continue native reconciliation after ${reasonCode}.`, - idempotencyKey: `native-reconciliation:${reasonCode}`, - agentId: input.agentId, - }, ...effects], - }); - - if (input.facts.equivalentAttentionFamily) { - return preserve("attention_duplicate_suppressed", [{ kind: "link_canonical_request" }]); - } - - if (input.facts.callerMaterialConflict) { - return preserve("structured_result_replay_conflict", [{ - kind: "record_finalization_error", - cause: "structured_result_replay_conflict", - nextAction: "Use a fresh canonical result identity.", - agentId: input.agentId, - }]); - } - if (input.facts.authoritativeStatusChanged) { - return preserve("prior_status_terminal_preserved", [{ kind: "append_superseding_assessment" }]); - } - if (input.facts.statusVersionAdvanced) { - return preserve("arbitration_conflict_reloaded", [ - { kind: "increment_status_version" }, - { kind: "schedule_reconciliation" }, - ]); - } - if (input.facts.newEvidenceSatisfiesContract) { - if (input.facts.hasPendingChildCompletion) { - return { - policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, - statusAction: "in_progress", - toStatus: "in_progress", - reasonCode: "native_child_completion_pending", - unblockDescriptor: null, - // The existing child-result wake owns the continuation. - effects: [{ kind: "release_checkout" }], - }; - } - return { - policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, - statusAction: "done", - toStatus: "done", - reasonCode: "decision_superseded_by_new_evidence", - unblockDescriptor: null, - effects: [{ kind: "release_checkout" }], - }; - } - if (input.facts.dependencyResolved) return continuation("dependency_resolved"); - if (input.facts.authorizedResume) return continuation("authorized_resume"); - if ((input.facts.undeliveredEffectCount ?? 0) > 0) { - return continuation("live_continuation_registered", [{ kind: "dispatch_pending_effect" }]); - } - if (input.facts.workspaceOperationPending) { - return { - policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, - statusAction: "done", - toStatus: "done", - reasonCode: "completion_contract_satisfied", - unblockDescriptor: null, - effects: [{ kind: "resume_workspace_operation" }, { kind: "release_checkout" }], - }; - } - if (input.facts.canonicalReplay) { - return { - policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, - statusAction: "done", - toStatus: "done", - reasonCode: "completion_contract_satisfied", - unblockDescriptor: null, - effects: [{ kind: "release_checkout" }], - }; - } - throw new Error("native_reconciliation_facts_invalid"); -} - export type NativeSessionResumeClaim = { runId: string; leaseOwner: string }; type NativeCleanupOutcome = { @@ -824,39 +702,12 @@ async function reconcileNativeFinalizationsPass( continue; } if (row.assessmentId) { - const assessment = await db.select({ - id: workAssessments.id, - contractId: workAssessments.contractId, - resultId: workAssessments.resultId, - policyVersion: workAssessments.policyVersion, - priorIssueStatus: workAssessments.priorIssueStatus, - priorStatusVersion: workAssessments.priorStatusVersion, - priorDecisionId: workAssessments.priorDecisionId, - assessmentJson: workAssessments.assessmentJson, - createdAt: workAssessments.createdAt, - }).from(workAssessments).where(and( - eq(workAssessments.id, row.assessmentId), - eq(workAssessments.companyId, row.companyId), - eq(workAssessments.issueId, row.issueId), - )).limit(1).then((entries) => entries[0] ?? null); - const changedEvidence = assessment - ? await db.select({ id: issueWorkProducts.id }).from(issueWorkProducts).where(and( - eq(issueWorkProducts.companyId, row.companyId), - eq(issueWorkProducts.issueId, row.issueId), - or( - gt(issueWorkProducts.createdAt, assessment.createdAt), - gt(issueWorkProducts.updatedAt, assessment.createdAt), - ), - )).limit(1).then((entries) => entries[0] ?? null) - : null; - const currentDecision = row.decisionId + const currentDecision = !options.skipNativeReviewResponse && row.coordinatorPhase === "committed" && row.decisionId ? await db.select().from(statusDecisions).where(and( - eq(statusDecisions.id, row.decisionId), - eq(statusDecisions.companyId, row.companyId), + eq(statusDecisions.id, row.decisionId), eq(statusDecisions.companyId, row.companyId), eq(statusDecisions.issueId, row.issueId), - )).limit(1).then((entries) => entries[0] ?? null) + )).limit(1).then(entries => entries[0] ?? null) : null; - const currentDecisionJson = record(currentDecision?.decisionJson); if (row.coordinatorPhase === "committed" && !options.skipNativeResponsePresentation) { // A later decision can supersede this run's task status, but cannot // erase its accepted, still-authorized response. Repair presentation @@ -866,12 +717,12 @@ async function reconcileNativeFinalizationsPass( }); } if ( + !options.skipNativeReviewResponse && row.coordinatorPhase === "committed" && row.decisionId && row.resultId && row.assessmentId && - currentDecisionJson.externalChatReviewPresentation && - !options.skipNativeReviewResponse + record(currentDecision?.decisionJson).externalChatReviewPresentation ) { // A later chat turn may retain the same pending review. Recover the // earlier response independently before status reconciliation skips @@ -885,162 +736,9 @@ async function reconcileNativeFinalizationsPass( assessmentId: row.assessmentId, }); } - if (row.coordinatorPhase === "committed" && row.decisionId && !currentDecision) { - throw new Error("native_committed_decision_missing"); - } - const supersededCommittedDecision = row.coordinatorPhase === "committed" - && row.decisionId !== null - && row.issueDecisionId !== null - && row.issueDecisionId !== row.decisionId - && !( - currentDecisionJson.statusAction === "preserve" - && assessment?.priorDecisionId === row.issueDecisionId - ); - if (supersededCommittedDecision) continue; - const persistedProjectedVersion = Number(currentDecisionJson.projectedStatusVersion); - const expectedProjectedVersion = currentDecision - ? Number.isFinite(persistedProjectedVersion) - ? persistedProjectedVersion - : Number(currentDecision.decisionVersion) - (currentDecisionJson.statusAction === "preserve" ? 1 : 0) - : null; - const issueMatchesCurrentDecision = !!assessment - && !!currentDecision - && row.issueDecisionId === row.decisionId - && currentDecision.assessmentId === assessment.id - && currentDecision.toStatus === row.issueStatus - && expectedProjectedVersion === Number(row.issueStatusVersion); - const authoritativeStatusChanged = !!assessment && !issueMatchesCurrentDecision && ( - assessment.priorIssueStatus !== row.issueStatus - || Number(assessment.priorStatusVersion) !== Number(row.issueStatusVersion) - ); - const retiredAutomaticReview = issueMatchesCurrentDecision && currentDecision - ? await decisionHasRetiredAutomaticReview(db, currentDecision) : false; - let reassessment = null; - let resultRow = null; - let contractRow = null; - if (assessment && (authoritativeStatusChanged || changedEvidence || retiredAutomaticReview)) { - [resultRow, contractRow] = await Promise.all([ - db.select().from(nativeRunResults).where(and( - eq(nativeRunResults.id, assessment.resultId), - eq(nativeRunResults.companyId, row.companyId), - eq(nativeRunResults.issueId, row.issueId), - eq(nativeRunResults.runId, row.runId), - )).limit(1).then((entries) => entries[0] ?? null), - db.select().from(completionContracts).where(and( - eq(completionContracts.id, assessment.contractId), - eq(completionContracts.companyId, row.companyId), - eq(completionContracts.issueId, row.issueId), - )).limit(1).then((entries) => entries[0] ?? null), - ]); - if (!resultRow || !contractRow) throw new Error("native_reconciliation_evidence_binding_missing"); - reassessment = await classifyNativeEvidence({ - db, - companyId: row.companyId, - issueId: row.issueId, - runId: row.runId, - contract: record(contractRow.contractJson), - result: record(record(resultRow.resultJson).result), - }); - } - const previousAssessment = record(assessment?.assessmentJson); - const newEvidenceSatisfiesContract = !!changedEvidence - && previousAssessment.allCriteriaSatisfied !== true - && reassessment?.allCriteriaSatisfied === true - && reassessment.verificationPassed === true; - const facts: NativeReconciliationFacts = authoritativeStatusChanged - ? { authoritativeStatusChanged: true } - : newEvidenceSatisfiesContract - ? { newEvidenceSatisfiesContract: true } - : {}; - if (Object.keys(facts).length > 0 || retiredAutomaticReview) { - if (!assessment || !reassessment || !resultRow || !contractRow) { - throw new Error("native_reconciliation_reassessment_missing"); - } - const currentIssue = retiredAutomaticReview - ? await db.select().from(issues).where(and(eq(issues.id, row.issueId), eq(issues.companyId, row.companyId))).then((entries) => entries[0]) - : null; - const readiness = retiredAutomaticReview ? await issueService(db).getDependencyReadiness(row.issueId, db) : null; - const currentRun = retiredAutomaticReview ? await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, row.runId)).then((entries) => entries[0]) : null; - // Never replay an old result over a new run or a later task contract. - const latestContract = retiredAutomaticReview ? await db.select({ id: completionContracts.id }).from(completionContracts) - .where(and(eq(completionContracts.companyId, row.companyId), eq(completionContracts.issueId, row.issueId))) - .orderBy(desc(completionContracts.revision)).limit(1).then((entries) => entries[0]) : null; - if (retiredAutomaticReview && (!currentIssue || currentRun?.status !== "succeeded" - || latestContract?.id !== contractRow.id - || (currentIssue.executionRunId && currentIssue.executionRunId !== row.runId))) continue; - // A committed decision proves the original barrier passed. If a later - // workspace operation exists, do not ignore a pending or failed retry. - const reviewBarrier = retiredAutomaticReview ? await db.select({ status: workspaceOperations.status }) - .from(workspaceOperations).where(and(eq(workspaceOperations.companyId, row.companyId), - eq(workspaceOperations.heartbeatRunId, row.runId), eq(workspaceOperations.phase, "workspace_finalize"))) - .orderBy(desc(workspaceOperations.createdAt)).limit(1).then((entries) => entries[0]) : null; - if (reviewBarrier && reviewBarrier.status !== "succeeded") continue; - const childCompletionRecipient = { - companyId: row.companyId, issueId: row.issueId, agentId: row.agentId, runId: row.runId, - sourceIntentId: typeof record(row.contextSnapshot).nativeStatusWakeIntentId === "string" - ? record(row.contextSnapshot).nativeStatusWakeIntentId as string : null, - }; - const hasPendingChildCompletion = await hasPendingNativeChildCompletion(db, childCompletionRecipient); - const reassessmentRow = await recordNativeWorkAssessment({ - db, - companyId: row.companyId, - issueId: row.issueId, - runId: row.runId, - turnId: resultRow.turnId, - contractId: contractRow.id, - contractCanonicalSha256: contractRow.canonicalSha256, - resultId: resultRow.id, - resultCanonicalSha256: resultRow.canonicalSha256, - priorIssueStatus: row.issueStatus, - priorStatusVersion: Number(row.issueStatusVersion), - priorDecisionId: row.issueDecisionId, - policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, - assessment: reassessment, - supersedesAssessmentId: assessment.id, - }); - const decision = retiredAutomaticReview && currentIssue - ? resolveNativeFinalizerStatus({ - hasPendingChildCompletion, - assessment: reassessment, terminalState: "succeeded", workspaceFinalizeStatus: "succeeded", - governanceGate: await pendingNativeGovernance({ db, companyId: row.companyId, issueId: row.issueId, - runId: row.runId, executionState: record(currentIssue.executionState) }), - completionClaimPolicyAccepted: contractRow.risk === "low" && contractRow.completionAuthority === "agent_claim_policy", - hasUnresolvedIssueBlockers: (readiness?.unresolvedBlockerCount ?? 0) > 0, - reviewOwnerUserId: currentIssue.responsibleUserId ?? currentIssue.createdByUserId, - priorIssueStatus: row.issueStatus as NativeAuthoritativeIssueStatus, agentId: row.agentId, - }) - : resolveNativeReconciliationStatus({ - facts: { ...facts, hasPendingChildCompletion }, - priorIssueStatus: row.issueStatus as NativeAuthoritativeIssueStatus, agentId: row.agentId, - }); - let committed: Awaited>; - try { - committed = await commitNativeStatusDecision({ - db, - companyId: row.companyId, - issueId: row.issueId, - runId: row.runId, - assessmentId: reassessmentRow.id, - priorStatus: row.issueStatus, - priorStatusVersion: Number(row.issueStatusVersion), - priorDecisionId: row.issueDecisionId, - decision, - requireNoPendingChildCompletion: decision.statusAction === "done" ? childCompletionRecipient : undefined, - supersedesCommittedDecisionId: row.coordinatorPhase === "committed" - ? row.decisionId ?? undefined - : undefined, - }); - } catch (error) { - if (error instanceof NativeStatusRaceError) continue; - throw error; - } - results.push({ - runId: row.runId, - phase: "committed", - reconciliationAction: decision.reasonCode, - decisionId: committed.decision.id, - reconciliationDecision: decision, - }); + const reassessed = await reassessNativeStatusFromSource(db, { ...row, assessmentId: row.assessmentId }); + if (reassessed.kind !== "unchanged") { + if (reassessed.kind === "committed") results.push(reassessed.result); continue; } } diff --git a/server/src/services/native-runtime/native-reconciliation-status.ts b/server/src/services/native-runtime/native-reconciliation-status.ts new file mode 100644 index 00000000000..6942b12f943 --- /dev/null +++ b/server/src/services/native-runtime/native-reconciliation-status.ts @@ -0,0 +1,114 @@ +import { NATIVE_STATUS_ARBITER_POLICY_VERSION, type NativeAuthoritativeIssueStatus, type NativeStatusDecision } from "./status-arbiter.js"; + +export type NativeReconciliationFacts = { + equivalentAttentionFamily?: boolean; + canonicalReplay?: boolean; + callerMaterialConflict?: boolean; + workspaceOperationPending?: boolean; + undeliveredEffectCount?: number; + authoritativeStatusChanged?: boolean; + newEvidenceSatisfiesContract?: boolean; + hasPendingChildCompletion?: boolean; + dependencyResolved?: boolean; + authorizedResume?: boolean; + statusVersionAdvanced?: boolean; +}; + + +export function resolveNativeReconciliationStatus(input: { + facts: NativeReconciliationFacts; + priorIssueStatus: NativeAuthoritativeIssueStatus; + agentId: string; +}): NativeStatusDecision { + const preserve = (reasonCode: string, effects: NativeStatusDecision["effects"]): NativeStatusDecision => ({ + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "preserve", + toStatus: input.priorIssueStatus, + reasonCode, + unblockDescriptor: null, + effects, + }); + const continuation = (reasonCode: string, effects: NativeStatusDecision["effects"] = []): NativeStatusDecision => ({ + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "in_progress", + toStatus: "in_progress", + reasonCode, + unblockDescriptor: null, + effects: [{ + kind: "enqueue_continuation", + continuationKind: "same_agent", + summary: `Continue native reconciliation after ${reasonCode}.`, + idempotencyKey: `native-reconciliation:${reasonCode}`, + agentId: input.agentId, + }, ...effects], + }); + + if (input.facts.equivalentAttentionFamily) { + return preserve("attention_duplicate_suppressed", [{ kind: "link_canonical_request" }]); + } + + if (input.facts.callerMaterialConflict) { + return preserve("structured_result_replay_conflict", [{ + kind: "record_finalization_error", + cause: "structured_result_replay_conflict", + nextAction: "Use a fresh canonical result identity.", + agentId: input.agentId, + }]); + } + if (input.facts.authoritativeStatusChanged) { + return preserve("prior_status_terminal_preserved", [{ kind: "append_superseding_assessment" }]); + } + if (input.facts.statusVersionAdvanced) { + return preserve("arbitration_conflict_reloaded", [ + { kind: "increment_status_version" }, + { kind: "schedule_reconciliation" }, + ]); + } + if (input.facts.newEvidenceSatisfiesContract) { + if (input.facts.hasPendingChildCompletion) { + return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "in_progress", + toStatus: "in_progress", + reasonCode: "native_child_completion_pending", + unblockDescriptor: null, + effects: [{ kind: "release_checkout" }], + }; + } + return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "done", + toStatus: "done", + reasonCode: "decision_superseded_by_new_evidence", + unblockDescriptor: null, + effects: [{ kind: "release_checkout" }], + }; + } + if (input.facts.dependencyResolved) return continuation("dependency_resolved"); + if (input.facts.authorizedResume) return continuation("authorized_resume"); + if ((input.facts.undeliveredEffectCount ?? 0) > 0) { + return continuation("live_continuation_registered", [{ kind: "dispatch_pending_effect" }]); + } + if (input.facts.workspaceOperationPending) { + return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "done", + toStatus: "done", + reasonCode: "completion_contract_satisfied", + unblockDescriptor: null, + effects: [{ kind: "resume_workspace_operation" }, { kind: "release_checkout" }], + }; + } + if (input.facts.canonicalReplay) { + return { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "done", + toStatus: "done", + reasonCode: "completion_contract_satisfied", + unblockDescriptor: null, + effects: [{ kind: "release_checkout" }], + }; + } + throw new Error("native_reconciliation_facts_invalid"); +} + diff --git a/server/src/services/native-runtime/native-status-reassessment.ts b/server/src/services/native-runtime/native-status-reassessment.ts new file mode 100644 index 00000000000..44b801a72ca --- /dev/null +++ b/server/src/services/native-runtime/native-status-reassessment.ts @@ -0,0 +1,222 @@ +import { and, desc, eq, gt, or } from "drizzle-orm"; +import { completionContracts, heartbeatRuns, issueWorkProducts, issues, nativeRunResults, + statusDecisions, workAssessments, workspaceOperations, type Db } from "@paperclipai/db"; +import { issueService } from "../issues.js"; +import { decisionHasRetiredAutomaticReview } from "./automatic-completion-reviews.js"; +import { classifyNativeEvidence } from "./evidence-classifier.js"; +import { hasPendingNativeChildCompletion } from "./native-child-completion-delivery.js"; +import { resolveNativeReconciliationStatus, type NativeReconciliationFacts } from "./native-reconciliation-status.js"; +import { pendingNativeGovernance, resolveNativeFinalizerStatus } from "./native-run-finalizer.js"; +import { NATIVE_STATUS_ARBITER_POLICY_VERSION, type NativeAuthoritativeIssueStatus, type NativeStatusDecision } from "./status-arbiter.js"; +import { commitNativeStatusDecision, NativeStatusRaceError } from "./status-decision-committer.js"; +import { recordNativeWorkAssessment } from "./work-assessments.js"; + +export type NativeStatusReassessmentSource = { + runId: string; companyId: string; agentId: string; issueId: string; + issueStatus: string; issueStatusVersion: number; issueDecisionId: string | null; + coordinatorPhase: string; assessmentId: string; decisionId: string | null; + contextSnapshot: Record | null; +}; + +type Outcome = { kind: "unchanged" | "superseded" | "deferred" } | { + kind: "committed"; + result: { runId: string; phase: "committed"; reconciliationAction: string | null; + decisionId: string; reconciliationDecision: NativeStatusDecision }; +}; + +function record(value: unknown): Record { + return value && typeof value === "object" && !Array.isArray(value) + ? value as Record + : {}; +} + +export async function reassessNativeStatusFromSource(db: Db, row: NativeStatusReassessmentSource): Promise { + const assessment = await db.select({ + id: workAssessments.id, + contractId: workAssessments.contractId, + resultId: workAssessments.resultId, + policyVersion: workAssessments.policyVersion, + priorIssueStatus: workAssessments.priorIssueStatus, + priorStatusVersion: workAssessments.priorStatusVersion, + priorDecisionId: workAssessments.priorDecisionId, + assessmentJson: workAssessments.assessmentJson, + createdAt: workAssessments.createdAt, + }).from(workAssessments).where(and( + eq(workAssessments.id, row.assessmentId), + eq(workAssessments.companyId, row.companyId), + eq(workAssessments.issueId, row.issueId), + )).limit(1).then((entries) => entries[0] ?? null); + const changedEvidence = assessment + ? await db.select({ id: issueWorkProducts.id }).from(issueWorkProducts).where(and( + eq(issueWorkProducts.companyId, row.companyId), + eq(issueWorkProducts.issueId, row.issueId), + or( + gt(issueWorkProducts.createdAt, assessment.createdAt), + gt(issueWorkProducts.updatedAt, assessment.createdAt), + ), + )).limit(1).then((entries) => entries[0] ?? null) + : null; + const currentDecision = row.decisionId + ? await db.select().from(statusDecisions).where(and( + eq(statusDecisions.id, row.decisionId), + eq(statusDecisions.companyId, row.companyId), + eq(statusDecisions.issueId, row.issueId), + )).limit(1).then((entries) => entries[0] ?? null) + : null; + const currentDecisionJson = record(currentDecision?.decisionJson); + if (row.coordinatorPhase === "committed" && row.decisionId && !currentDecision) { + throw new Error("native_committed_decision_missing"); + } + const supersededCommittedDecision = row.coordinatorPhase === "committed" + && row.decisionId !== null + && row.issueDecisionId !== null + && row.issueDecisionId !== row.decisionId + && !( + currentDecisionJson.statusAction === "preserve" + && assessment?.priorDecisionId === row.issueDecisionId + ); + if (supersededCommittedDecision) return { kind: "superseded" }; + const persistedProjectedVersion = Number(currentDecisionJson.projectedStatusVersion); + const expectedProjectedVersion = currentDecision + ? Number.isFinite(persistedProjectedVersion) + ? persistedProjectedVersion + : Number(currentDecision.decisionVersion) - (currentDecisionJson.statusAction === "preserve" ? 1 : 0) + : null; + const issueMatchesCurrentDecision = !!assessment + && !!currentDecision + && row.issueDecisionId === row.decisionId + && currentDecision.assessmentId === assessment.id + && currentDecision.toStatus === row.issueStatus + && expectedProjectedVersion === Number(row.issueStatusVersion); + const authoritativeStatusChanged = !!assessment && !issueMatchesCurrentDecision && ( + assessment.priorIssueStatus !== row.issueStatus + || Number(assessment.priorStatusVersion) !== Number(row.issueStatusVersion) + ); + const retiredAutomaticReview = issueMatchesCurrentDecision && currentDecision + ? await decisionHasRetiredAutomaticReview(db, currentDecision) : false; + let reassessment = null; + let resultRow = null; + let contractRow = null; + if (assessment && (authoritativeStatusChanged || changedEvidence || retiredAutomaticReview)) { + [resultRow, contractRow] = await Promise.all([ + db.select().from(nativeRunResults).where(and( + eq(nativeRunResults.id, assessment.resultId), + eq(nativeRunResults.companyId, row.companyId), + eq(nativeRunResults.issueId, row.issueId), + eq(nativeRunResults.runId, row.runId), + )).limit(1).then((entries) => entries[0] ?? null), + db.select().from(completionContracts).where(and( + eq(completionContracts.id, assessment.contractId), + eq(completionContracts.companyId, row.companyId), + eq(completionContracts.issueId, row.issueId), + )).limit(1).then((entries) => entries[0] ?? null), + ]); + if (!resultRow || !contractRow) throw new Error("native_reconciliation_evidence_binding_missing"); + reassessment = await classifyNativeEvidence({ + db, + companyId: row.companyId, + issueId: row.issueId, + runId: row.runId, + contract: record(contractRow.contractJson), + result: record(record(resultRow.resultJson).result), + }); + } + const previousAssessment = record(assessment?.assessmentJson); + const newEvidenceSatisfiesContract = !!changedEvidence + && previousAssessment.allCriteriaSatisfied !== true + && reassessment?.allCriteriaSatisfied === true + && reassessment.verificationPassed === true; + const facts: NativeReconciliationFacts = authoritativeStatusChanged + ? { authoritativeStatusChanged: true } + : newEvidenceSatisfiesContract + ? { newEvidenceSatisfiesContract: true } + : {}; + if (Object.keys(facts).length > 0 || retiredAutomaticReview) { + if (!assessment || !reassessment || !resultRow || !contractRow) { + throw new Error("native_reconciliation_reassessment_missing"); + } + const currentIssue = retiredAutomaticReview + ? await db.select().from(issues).where(and(eq(issues.id, row.issueId), eq(issues.companyId, row.companyId))).then((entries) => entries[0]) + : null; + const readiness = retiredAutomaticReview ? await issueService(db).getDependencyReadiness(row.issueId, db) : null; + const currentRun = retiredAutomaticReview ? await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, row.runId)).then((entries) => entries[0]) : null; + const latestContract = retiredAutomaticReview ? await db.select({ id: completionContracts.id }).from(completionContracts) + .where(and(eq(completionContracts.companyId, row.companyId), eq(completionContracts.issueId, row.issueId))) + .orderBy(desc(completionContracts.revision)).limit(1).then((entries) => entries[0]) : null; + if (retiredAutomaticReview && (!currentIssue || currentRun?.status !== "succeeded" + || latestContract?.id !== contractRow.id + || (currentIssue.executionRunId && currentIssue.executionRunId !== row.runId))) return { kind: "deferred" }; + const reviewBarrier = retiredAutomaticReview ? await db.select({ status: workspaceOperations.status }) + .from(workspaceOperations).where(and(eq(workspaceOperations.companyId, row.companyId), + eq(workspaceOperations.heartbeatRunId, row.runId), eq(workspaceOperations.phase, "workspace_finalize"))) + .orderBy(desc(workspaceOperations.createdAt)).limit(1).then((entries) => entries[0]) : null; + if (reviewBarrier && reviewBarrier.status !== "succeeded") return { kind: "deferred" }; + const childCompletionRecipient = { + companyId: row.companyId, issueId: row.issueId, agentId: row.agentId, runId: row.runId, + sourceIntentId: typeof record(row.contextSnapshot).nativeStatusWakeIntentId === "string" + ? record(row.contextSnapshot).nativeStatusWakeIntentId as string : null, + }; + const hasPendingChildCompletion = await hasPendingNativeChildCompletion(db, childCompletionRecipient); + const reassessmentRow = await recordNativeWorkAssessment({ + db, + companyId: row.companyId, + issueId: row.issueId, + runId: row.runId, + turnId: resultRow.turnId, + contractId: contractRow.id, + contractCanonicalSha256: contractRow.canonicalSha256, + resultId: resultRow.id, + resultCanonicalSha256: resultRow.canonicalSha256, + priorIssueStatus: row.issueStatus, + priorStatusVersion: Number(row.issueStatusVersion), + priorDecisionId: row.issueDecisionId, + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + assessment: reassessment, + supersedesAssessmentId: assessment.id, + }); + const decision = retiredAutomaticReview && currentIssue + ? resolveNativeFinalizerStatus({ + hasPendingChildCompletion, + assessment: reassessment, terminalState: "succeeded", workspaceFinalizeStatus: "succeeded", + governanceGate: await pendingNativeGovernance({ db, companyId: row.companyId, issueId: row.issueId, + runId: row.runId, executionState: record(currentIssue.executionState) }), + completionClaimPolicyAccepted: contractRow.risk === "low" && contractRow.completionAuthority === "agent_claim_policy", + hasUnresolvedIssueBlockers: (readiness?.unresolvedBlockerCount ?? 0) > 0, + reviewOwnerUserId: currentIssue.responsibleUserId ?? currentIssue.createdByUserId, + priorIssueStatus: row.issueStatus as NativeAuthoritativeIssueStatus, agentId: row.agentId, + }) + : resolveNativeReconciliationStatus({ + facts: { ...facts, hasPendingChildCompletion }, + priorIssueStatus: row.issueStatus as NativeAuthoritativeIssueStatus, agentId: row.agentId, + }); + let committed: Awaited>; + try { + committed = await commitNativeStatusDecision({ + db, + companyId: row.companyId, + issueId: row.issueId, + runId: row.runId, + assessmentId: reassessmentRow.id, + priorStatus: row.issueStatus, + priorStatusVersion: Number(row.issueStatusVersion), + priorDecisionId: row.issueDecisionId, + decision, + requireNoPendingChildCompletion: decision.statusAction === "done" ? childCompletionRecipient : undefined, + supersedesCommittedDecisionId: row.coordinatorPhase === "committed" + ? row.decisionId ?? undefined + : undefined, + }); + } catch (error) { + if (error instanceof NativeStatusRaceError) return { kind: "deferred" }; + throw error; + } + return { kind: "committed", result: { + runId: row.runId, + phase: "committed", + reconciliationAction: decision.reasonCode, + decisionId: committed.decision.id, + reconciliationDecision: decision, + } }; + } + return { kind: "unchanged" }; +}