mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-30 21:49:18 +01:00
refactor(server): isolate native status reassessment ownership
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
cf6f318b0d
commit
46c410e7f6
5 files changed
+385
-314
No files matched your search
@@ -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.
|
||||
@@ -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.
|
||||
@@ -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<string, unknown> {
|
||||
return value && typeof value === "object" && !Array.isArray(value)
|
||||
? value as Record<string, unknown>
|
||||
: {};
|
||||
}
|
||||
|
||||
/** 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<ReturnType<typeof commitNativeStatusDecision>>;
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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");
|
||||
}
|
||||
|
||||
@@ -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<string, unknown> | 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<string, unknown> {
|
||||
return value && typeof value === "object" && !Array.isArray(value)
|
||||
? value as Record<string, unknown>
|
||||
: {};
|
||||
}
|
||||
|
||||
export async function reassessNativeStatusFromSource(db: Db, row: NativeStatusReassessmentSource): Promise<Outcome> {
|
||||
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<ReturnType<typeof commitNativeStatusDecision>>;
|
||||
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" };
|
||||
}
|
||||
Reference in new issue
Block a user