diff --git a/doc/architecture/native-status-arbitration.md b/doc/architecture/native-status-arbitration.md index fdc7a077ed..912a90b313 100644 --- a/doc/architecture/native-status-arbitration.md +++ b/doc/architecture/native-status-arbitration.md @@ -220,6 +220,19 @@ Examples: - a materialization failure records the failed phase and next retry time rather than silently dropping the side effect. +## Policy upgrades + +The policy version on an assessment is audit metadata. New runs use the current +rules. A version change alone does not reassess an old run, change task status, +or ask a person to review completion. New evidence and explicit status changes +still use the existing reconciliation paths. + +Reconciliation also withdraws pending review cards created solely by the old +policy-version check. It restores the previous status only if that exact decision +and status version are still current and no other review gate is pending. A later +user or agent decision takes precedence. The old assessments and decisions remain +in the audit history; cleanup does not accept or reject the agent's work. + ## Diagnosing an unexpected status Start with the terminal heartbeat run and inspect: diff --git a/packages/paperclip-runner/spec/fixtures/status-authority-sdk.json b/packages/paperclip-runner/spec/fixtures/status-authority-sdk.json index 895813cd2f..3d3a3254e0 100644 --- a/packages/paperclip-runner/spec/fixtures/status-authority-sdk.json +++ b/packages/paperclip-runner/spec/fixtures/status-authority-sdk.json @@ -341,9 +341,9 @@ "mode": "native", "covers": { "decisionRows": [], "terminalRows": [], "attentionRows": [], "livenessRows": [], "reconciliationRows": ["REC-08"], "compatibilityRows": [], "migrationRows": [] }, "tags": ["supersession", "reconciliation", "deterministic_replay"], - "given": { "priorIssueStatus": "in_progress", "turnTerminalState": "completed", "runTerminalState": "succeeded", "reportedWorkDisposition": "done", "nativeFinalization": "present", "completionState": "new_policy_requires_review", "trigger": "authorized_agent" }, - "expected": { "runStatus": "succeeded", "statusAction": "in_review", "reasonCode": "completion_review_required", "requiredEffects": ["bind_reviewer", "append_superseding_assessment"], "forbiddenEffects": ["mutate_old_decision"], "livePathKind": "review", "preserveClaim": true, "nativeRecords": true, "decisionCount": 2, "maxWakeCount": 1, "maxNotificationCount": 1 }, - "replay": { "attempts": 2, "sameDecisionDigest": false, "maxSemanticDecisions": 2, "maxDomainEffectsPerKey": 1 } + "given": { "priorIssueStatus": "in_progress", "turnTerminalState": "completed", "runTerminalState": "succeeded", "reportedWorkDisposition": "done", "nativeFinalization": "present", "completionState": "policy_version_changed", "trigger": "authorized_agent" }, + "expected": { "runStatus": "succeeded", "statusAction": "preserve", "reasonCode": "prior_fixture_decision", "requiredEffects": [], "forbiddenEffects": ["mutate_old_decision", "bind_reviewer", "append_superseding_assessment"], "livePathKind": null, "preserveClaim": true, "nativeRecords": true, "decisionCount": 1, "maxWakeCount": 0, "maxNotificationCount": 0 }, + "replay": { "attempts": 2, "sameDecisionDigest": true, "maxSemanticDecisions": 1, "maxDomainEffectsPerKey": 1 } }, { "id": "legacy-adapter-unchanged", diff --git a/server/src/__tests__/native-status-arbiter-corpus.test.ts b/server/src/__tests__/native-status-arbiter-corpus.test.ts index 79479e4134..65a9596fc9 100644 --- a/server/src/__tests__/native-status-arbiter-corpus.test.ts +++ b/server/src/__tests__/native-status-arbiter-corpus.test.ts @@ -1,7 +1,7 @@ import { randomUUID } from "node:crypto"; import { readFileSync } from "node:fs"; import { fileURLToPath } from "node:url"; -import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { and, eq } from "drizzle-orm"; import { agents, @@ -63,6 +63,8 @@ import { reconcileNativeFinalizations, resolveNativeReconciliationStatus, } from "../services/native-runtime/native-finalization-reconciler.js"; +import * as activityLog from "../services/activity-log.js"; +import { dismissObsoleteNativePolicyReviews } from "../services/native-runtime/obsolete-policy-reviews.js"; import { issueService } from "../services/issues.js"; import { issueThreadInteractionService } from "../services/issue-thread-interactions.js"; import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; @@ -242,11 +244,11 @@ const nativeStatusEffectKinds = new Set([ const supersedingDecisionStates = new Set([ "new_evidence_satisfies_contract", "dependency_now_done", "explicit_resume_capability", - "board_cancelled_before_cas", "new_policy_requires_review", "authorized_writer_incremented_version", + "board_cancelled_before_cas", "authorized_writer_incremented_version", ]); const liveReconciliationStates = new Set([ - "board_cancelled_before_cas", "new_evidence_satisfies_contract", "new_policy_requires_review", + "board_cancelled_before_cas", "new_evidence_satisfies_contract", "policy_version_changed", ]); function initialRunStatus(fixture: Fixture) { @@ -319,7 +321,6 @@ function reconciliationFactsFor(completionState: string) { case "new_evidence_satisfies_contract": return { newEvidenceSatisfiesContract: true }; case "dependency_now_done": return { dependencyResolved: true }; case "explicit_resume_capability": return { authorizedResume: true }; - case "new_policy_requires_review": return { policyVersionChanged: true }; case "authorized_writer_incremented_version": return { statusVersionAdvanced: true }; default: return null; } @@ -529,7 +530,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { triggerActorCompanyId: companyId, priorIssueStatus: completionState === "board_cancelled_before_cas" ? "in_progress" : priorStatus, priorStatusVersion: 0, - policyVersion: completionState === "new_policy_requires_review" + policyVersion: completionState === "policy_version_changed" ? "phase6-v1" : NATIVE_STATUS_ARBITER_POLICY_VERSION, assessmentJson: { @@ -955,7 +956,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { issueId: seeded.issueId, assessmentId: seeded.assessmentId, decisionVersion: 1, - policyVersion: completionState === "new_policy_requires_review" + policyVersion: completionState === "policy_version_changed" ? "phase6-v1" : NATIVE_STATUS_ARBITER_POLICY_VERSION, fromStatus: completionState === "board_cancelled_before_cas" ? "in_progress" : priorIssueStatus, @@ -980,20 +981,43 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { updatedAt: new Date(Date.now() + 1_000), }).where(eq(issueWorkProducts.id, seeded.workProductId)); } - const [reconciled] = await reconcileNativeFinalizations(db, [seeded.runId]); - if (!reconciled?.reconciliationDecision || !reconciled.decisionId) { - throw new Error(`${fixture.id}: live reconciliation did not commit an authoritative decision`); + if (completionState === "policy_version_changed") { + await db.insert(workspaceOperations).values({ + companyId, heartbeatRunId: seeded.runId, issueId: seeded.issueId, + phase: "workspace_finalize", status: "succeeded", exitCode: 0, cwd: process.cwd(), finishedAt: new Date(), + }); + } + const [reconciled] = await reconcileNativeFinalizations(db, [seeded.runId]); + if (completionState === "policy_version_changed") { + const decisions = await db.select().from(statusDecisions).where(eq(statusDecisions.issueId, seeded.issueId)); + const assessments = await db.select().from(workAssessments).where(eq(workAssessments.issueId, seeded.issueId)); + expect(decisions).toHaveLength(1); + expect(decisions[0]!.id).toBe(priorDecision!.id); + expect(assessments).toHaveLength(1); + expect(assessments[0]!.policyVersion).toBe("phase6-v1"); + semanticConsumer = "native-reconciliation-consumer"; + consumerDecision = pushDecisionConsumer(semanticConsumer, { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "preserve", toStatus: decisions[0]!.toStatus as NativeStatusDecision["toStatus"], + reasonCode: decisions[0]!.reasonCode, unblockDescriptor: null, effects: [], + }); + liveEntrypointCommitted = true; + consumerExecutions.push({ consumer: "native-reconciliation-entrypoint", observed: { decisionId: priorDecision!.id } }); + } else { + if (!reconciled?.reconciliationDecision || !reconciled.decisionId) { + throw new Error(`${fixture.id}: live reconciliation did not commit an authoritative decision`); + } + semanticConsumer = "native-reconciliation-consumer"; + consumerDecision = pushDecisionConsumer(semanticConsumer, reconciled.reconciliationDecision); + liveEntrypointCommitted = true; + consumerExecutions.push({ + consumer: "native-reconciliation-entrypoint", + observed: { + action: reconciled.reconciliationAction, + decisionId: reconciled.decisionId, + }, + }); } - semanticConsumer = "native-reconciliation-consumer"; - consumerDecision = pushDecisionConsumer(semanticConsumer, reconciled.reconciliationDecision); - liveEntrypointCommitted = true; - consumerExecutions.push({ - consumer: "native-reconciliation-entrypoint", - observed: { - action: reconciled.reconciliationAction, - decisionId: reconciled.decisionId, - }, - }); } if ( seeded.nativeRecords @@ -1858,7 +1882,10 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { ]; for (const [field, expected] of mutations) { const mutated = { ...fixture, expected }; - expect(comparisonFailures(mutated, observed), `${fixture.id}:${field}`).not.toEqual([]); + const mutationObserved = field === "forbiddenEffects" && observed.effects.length === 0 + ? { ...observed, effects: [observedEffect] } + : observed; + expect(comparisonFailures(mutated, mutationObserved), `${fixture.id}:${field}`).not.toEqual([]); } } }, 60_000); @@ -1973,24 +2000,177 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { .rejects.toThrow("native_pending_effect_target_missing:enqueue_continuation"); }, 30_000); - it("preserves terminal issues when a newer reconciliation policy is available", () => { - expect(resolveNativeReconciliationStatus({ - facts: { policyVersionChanged: true }, - priorIssueStatus: "done", - agentId, - })).toMatchObject({ - statusAction: "preserve", - toStatus: "done", - reasonCode: "prior_status_terminal_preserved", - effects: [{ kind: "append_superseding_assessment" }], + async function seedPolicyReview(options: { genuine?: boolean; priorStatus?: "in_progress" | "blocked" | "in_review" } = {}) { + const template = corpus.fixtures.find((candidate) => candidate.mode === "native")!; + const priorStatus = options.priorStatus ?? "in_progress"; + const seeded = await seedFixture({ + ...template, id: `policy-review-${randomUUID()}`, + given: { ...template.given, priorIssueStatus: priorStatus, completionState: "policy_review_cleanup" }, }); + const [assessment] = await db.select().from(workAssessments).where(eq(workAssessments.id, seeded.assessmentId)); + const previousAssessmentId = randomUUID(); + await db.insert(workAssessments).values({ + ...assessment!, id: previousAssessmentId, policyVersion: "previous-policy", + inputDigest: `previous-assessment:${seeded.issueId}`, + }); + await db.update(workAssessments).set({ supersedesAssessmentId: previousAssessmentId }) + .where(eq(workAssessments.id, seeded.assessmentId)); + const committed = await commitNativeStatusDecision({ + db, companyId, issueId: seeded.issueId, runId: seeded.runId, + assessmentId: seeded.assessmentId, priorStatus, priorStatusVersion: 0, priorDecisionId: null, + decision: { + policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, + statusAction: "in_review", toStatus: "in_review", reasonCode: "completion_review_required", unblockDescriptor: null, + effects: options.genuine + ? [{ kind: "bind_reviewer", prompt: "Review the release before publishing.", ownerUserId: null }] + : [ + { kind: "bind_reviewer", prompt: "Review the superseding native policy assessment.", ownerUserId: null }, + { kind: "append_superseding_assessment" }, + ], + }, + }); + const [interaction] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.issueId, seeded.issueId)); + const [decision] = await db.select().from(statusDecisions).where(eq(statusDecisions.id, committed.decision.id)); + await db.insert(workspaceOperations).values({ + companyId, heartbeatRunId: seeded.runId, issueId: seeded.issueId, + phase: "workspace_finalize", status: "succeeded", exitCode: 0, cwd: process.cwd(), finishedAt: new Date(), + }); + return { ...seeded, decision: decision!, interaction: interaction! }; + } + + it("withdraws obsolete policy reviews, restores the prior status, and is idempotent", async () => { + const seeded = await seedPolicyReview(); + await reconcileNativeFinalizations(db, [seeded.runId]); + const [interaction] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, seeded.interaction.id)); + const [issue] = await db.select().from(issues).where(eq(issues.id, seeded.issueId)); + expect(interaction).toMatchObject({ status: "cancelled", result: { outcome: "withdrawn" } }); + expect(issue).toMatchObject({ status: "in_progress", statusVersion: 2, lastStatusDecisionId: null }); + const decisions = await db.select().from(statusDecisions).where(eq(statusDecisions.issueId, seeded.issueId)); + expect(decisions.find((row) => row.id === seeded.decision.id)).toEqual(seeded.decision); + await reconcileNativeFinalizations(db, [seeded.runId]); + expect(await db.select().from(issues).where(eq(issues.id, seeded.issueId))).toEqual([issue]); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.issueId, seeded.issueId))).toEqual(decisions); + }, 30_000); + + it.each(["accepted", "rejected"])("leaves an already %s review untouched", async (status) => { + const seeded = await seedPolicyReview(); + await db.update(issueThreadInteractions).set({ status }).where(eq(issueThreadInteractions.id, seeded.interaction.id)); + await dismissObsoleteNativePolicyReviews(db, [seeded.runId]); + const [issue] = await db.select().from(issues).where(eq(issues.id, seeded.issueId)); + expect(issue).toMatchObject({ status: "in_review", lastStatusDecisionId: seeded.decision.id }); + const [interaction] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, seeded.interaction.id)); + expect(interaction!.status).toBe(status); + }); + + it("preserves real completion reviews and limits cleanup to the requested runs", async () => { + const genuine = await seedPolicyReview({ genuine: true }); + const other = await seedPolicyReview(); + await dismissObsoleteNativePolicyReviews(db, [genuine.runId]); + for (const seeded of [genuine, other]) { + const [interaction] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, seeded.interaction.id)); + expect(interaction!.status).toBe("pending"); + } + }); + + it.each(["blocked", "done", "cancelled"])("dismisses the obsolete card without undoing a later %s status", async (status) => { + const seeded = await seedPolicyReview(); + await issueService(db).update(seeded.issueId, { status }); + const before = await db.select().from(issues).where(eq(issues.id, seeded.issueId)); + await dismissObsoleteNativePolicyReviews(db, [seeded.runId]); + expect(await db.select().from(issues).where(eq(issues.id, seeded.issueId))).toEqual(before); + const [interaction] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, seeded.interaction.id)); + expect(interaction!.status).toBe(status === "blocked" ? "cancelled" : "expired"); + }); + + it("does not restore status after a newer decision or a status change back to review", async () => { + for (const changedPointer of [false, true]) { + const seeded = await seedPolicyReview(); + if (changedPointer) { + await db.update(issues).set({ lastStatusDecisionId: null }).where(eq(issues.id, seeded.issueId)); + } else { + await issueService(db).update(seeded.issueId, { status: "in_progress" }); + await issueService(db).update(seeded.issueId, { status: "in_review" }); + } + const before = await db.select().from(issues).where(eq(issues.id, seeded.issueId)); + await dismissObsoleteNativePolicyReviews(db, [seeded.runId]); + expect(await db.select().from(issues).where(eq(issues.id, seeded.issueId))).toEqual(before); + } + }); + + it("keeps review status when a separate request still needs a response", async () => { + const seeded = await seedPolicyReview(); + await db.insert(issueThreadInteractions).values({ + companyId, issueId: seeded.issueId, kind: "request_confirmation", status: "pending", + payload: { version: 1, prompt: "Approve publishing the release." }, + }); + await dismissObsoleteNativePolicyReviews(db, [seeded.runId]); + const [issue] = await db.select().from(issues).where(eq(issues.id, seeded.issueId)); + expect(issue!.status).toBe("in_review"); + }); + + it("isolates a failed cleanup candidate and retries it on the next pass", async () => { + const first = await seedPolicyReview(); + const second = await seedPolicyReview(); + const runIds = [first.runId, second.runId]; + const transaction = vi.spyOn(db, "transaction").mockRejectedValueOnce(new Error("injected cleanup failure")); + try { + await expect(dismissObsoleteNativePolicyReviews(db, runIds)).resolves.toBeUndefined(); + } finally { + transaction.mockRestore(); + } + const statuses = await Promise.all([first, second].map(async (seeded) => { + const [interaction] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, seeded.interaction.id)); + return interaction!.status; + })); + expect(statuses.sort()).toEqual(["cancelled", "pending"]); + await dismissObsoleteNativePolicyReviews(db, runIds); + for (const seeded of [first, second]) { + const [issue] = await db.select().from(issues).where(eq(issues.id, seeded.issueId)); + expect(issue!.status).toBe("in_progress"); + } + }); + + it("keeps committed cleanup and continues publication after a live event fails", async () => { + const first = await seedPolicyReview(); + const second = await seedPolicyReview(); + const publish = vi.spyOn(activityLog, "publishActivity").mockImplementationOnce(() => { + throw new Error("injected live publication failure"); + }); + try { + await dismissObsoleteNativePolicyReviews(db, [first.runId, second.runId]); + expect(publish.mock.calls.length).toBeGreaterThan(1); + for (const seeded of [first, second]) { + const [issue] = await db.select().from(issues).where(eq(issues.id, seeded.issueId)); + const [interaction] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, seeded.interaction.id)); + expect(issue!.status).toBe("in_progress"); + expect(interaction!.status).toBe("cancelled"); + } + } finally { + publish.mockRestore(); + } + }); + + it("continues native finalization when the obsolete-card lookup fails", async () => { + const seeded = await seedPolicyReview({ genuine: true }); + const select = vi.spyOn(db, "select").mockImplementationOnce(() => { + throw new Error("injected cleanup lookup failure"); + }); + try { + const reconciled = await reconcileNativeFinalizations(db, [seeded.runId]); + expect(reconciled).toHaveLength(1); + expect(reconciled[0]!.phase).toBe("committed"); + } finally { + select.mockRestore(); + } + }); + + it("preserves a later authoritative status during reconciliation", () => { expect(resolveNativeReconciliationStatus({ - facts: { authoritativeStatusChanged: true, policyVersionChanged: true }, + facts: { authoritativeStatusChanged: true }, priorIssueStatus: "blocked", agentId, })).toMatchObject({ - statusAction: "preserve", - toStatus: "blocked", + statusAction: "preserve", toStatus: "blocked", reasonCode: "prior_status_terminal_preserved", }); }); diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index 4f891f8970..c8df927b56 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -1,3 +1,4 @@ +import { logger } from "../../middleware/logger.js"; import { createHash, randomUUID } from "node:crypto"; import { and, asc, desc, eq, gt, inArray, isNotNull, isNull, lte, notInArray, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -29,6 +30,7 @@ import { issueRecoveryActionService } from "../issue-recovery-actions.js"; import { issueService } from "../issues.js"; import { emitAgentTaskRun } from "../agent-task-run-telemetry.js"; import { resumeNativeWorkspaceFinalization } from "./native-workspace-finalizer.js"; +import { dismissObsoleteNativePolicyReviews } from "./obsolete-policy-reviews.js"; import { cleanupNativeWorkspaceSync, readNativeWorkspaceSyncReference, @@ -56,7 +58,6 @@ export type NativeReconciliationFacts = { newEvidenceSatisfiesContract?: boolean; dependencyResolved?: boolean; authorizedResume?: boolean; - policyVersionChanged?: boolean; statusVersionAdvanced?: boolean; }; @@ -110,22 +111,6 @@ export function resolveNativeReconciliationStatus(input: { if (input.facts.authoritativeStatusChanged) { return preserve("prior_status_terminal_preserved", [{ kind: "append_superseding_assessment" }]); } - if (input.facts.policyVersionChanged) { - if (["done", "cancelled"].includes(input.priorIssueStatus)) { - return preserve("prior_status_terminal_preserved", [{ kind: "append_superseding_assessment" }]); - } - return { - policyVersion: NATIVE_STATUS_ARBITER_POLICY_VERSION, - statusAction: "in_review", - toStatus: "in_review", - reasonCode: "completion_review_required", - unblockDescriptor: null, - effects: [ - { kind: "bind_reviewer", prompt: "Review the superseding native policy assessment.", ownerUserId: null }, - { kind: "append_superseding_assessment" }, - ], - }; - } if (input.facts.statusVersionAdvanced) { return preserve("arbitration_conflict_reloaded", [ { kind: "increment_status_version" }, @@ -552,6 +537,9 @@ export async function reconcileNativeFinalizations( }) => Promise; } = {}, ) { + await dismissObsoleteNativePolicyReviews(db, runIds).catch((err) => { + logger.warn({ err }, "Obsolete native policy review lookup failed; continuing native reconciliation"); + }); const rows = await db .select({ runId: heartbeatRuns.id, @@ -643,7 +631,6 @@ export async function reconcileNativeFinalizations( ), )).limit(1).then((entries) => entries[0] ?? null) : null; - const policyVersionChanged = assessment?.policyVersion !== NATIVE_STATUS_ARBITER_POLICY_VERSION; const currentDecision = row.decisionId ? await db.select({ assessmentId: statusDecisions.assessmentId, @@ -715,7 +702,7 @@ export async function reconcileNativeFinalizations( let reassessment = null; let resultRow = null; let contractRow = null; - if (assessment && (policyVersionChanged || authoritativeStatusChanged || changedEvidence)) { + if (assessment && (authoritativeStatusChanged || changedEvidence)) { [resultRow, contractRow] = await Promise.all([ db.select().from(nativeRunResults).where(and( eq(nativeRunResults.id, assessment.resultId), @@ -746,11 +733,9 @@ export async function reconcileNativeFinalizations( && reassessment.verificationPassed === true; const facts: NativeReconciliationFacts = authoritativeStatusChanged ? { authoritativeStatusChanged: true } - : policyVersionChanged - ? { policyVersionChanged: true } - : newEvidenceSatisfiesContract - ? { newEvidenceSatisfiesContract: true } - : {}; + : newEvidenceSatisfiesContract + ? { newEvidenceSatisfiesContract: true } + : {}; if (Object.keys(facts).length > 0) { if (!assessment || !reassessment || !resultRow || !contractRow) { throw new Error("native_reconciliation_reassessment_missing"); diff --git a/server/src/services/native-runtime/obsolete-policy-reviews.ts b/server/src/services/native-runtime/obsolete-policy-reviews.ts new file mode 100644 index 0000000000..e43262875e --- /dev/null +++ b/server/src/services/native-runtime/obsolete-policy-reviews.ts @@ -0,0 +1,150 @@ +import { and, eq, inArray, isNull, sql } from "drizzle-orm"; +import { + approvals, issueApprovals, issueThreadInteractions, issues, + nativeRunFinalizations, statusDecisionEffects, statusDecisions, workAssessments, + type Db, +} from "@paperclipai/db"; +import { logger } from "../../middleware/logger.js"; +import { issueService } from "../issues.js"; +import { issueThreadInteractionService } from "../issue-thread-interactions.js"; +import { enqueueTerminalIssueInteractionChatPublications } from "../chat-interaction-publications.js"; +import { persistActivity, publishActivity, type ActivityPublication } from "../activity-log.js"; + +const obsoletePrompt = "Review the superseding native policy assessment."; + +/** Retire only the old version-change gate, never a real completion review. */ +export async function dismissObsoleteNativePolicyReviews(db: Db, runIds?: string[]) { + const candidates = await db.select({ + interaction: issueThreadInteractions, + decision: statusDecisions, + priorDecisionId: workAssessments.priorDecisionId, + }).from(issueThreadInteractions) + .innerJoin(statusDecisionEffects, and( + eq(statusDecisionEffects.companyId, issueThreadInteractions.companyId), + eq(statusDecisionEffects.issueId, issueThreadInteractions.issueId), + sql`${statusDecisionEffects.targetId} = ${issueThreadInteractions.id}::text`, + eq(statusDecisionEffects.targetType, "issue_thread_interaction"), + eq(statusDecisionEffects.effectKind, "bind_reviewer"), + )) + .innerJoin(statusDecisions, and( + eq(statusDecisions.id, statusDecisionEffects.decisionId), + eq(statusDecisions.companyId, issueThreadInteractions.companyId), + eq(statusDecisions.issueId, issueThreadInteractions.issueId), + eq(statusDecisions.runId, issueThreadInteractions.sourceRunId), + )) + .innerJoin(workAssessments, and( + eq(workAssessments.id, statusDecisions.assessmentId), + eq(workAssessments.companyId, statusDecisions.companyId), + eq(workAssessments.issueId, statusDecisions.issueId), + )) + .where(and( + eq(issueThreadInteractions.status, "pending"), + eq(issueThreadInteractions.kind, "request_confirmation"), + isNull(issueThreadInteractions.createdByAgentId), + isNull(issueThreadInteractions.createdByUserId), + eq(statusDecisions.applicationState, "applied"), + eq(statusDecisions.reasonCode, "completion_review_required"), + eq(statusDecisions.toStatus, "in_review"), + sql`${issueThreadInteractions.idempotencyKey} = 'native-review:' || ${statusDecisions.id}::text`, + sql`${issueThreadInteractions.payload}->>'prompt' = ${obsoletePrompt}`, + // Match the complete old decision, including its unique assessment-only effect. + sql`${statusDecisions.decisionJson}->'effects' = ${JSON.stringify([ + { kind: "bind_reviewer", prompt: obsoletePrompt, ownerUserId: null }, + { kind: "append_superseding_assessment" }, + ])}::jsonb`, + ...(runIds?.length ? [inArray(statusDecisions.runId, runIds)] : []), + )).limit(100); + + for (const { interaction, decision, priorDecisionId } of candidates) { + const publications: ActivityPublication[] = []; + try { + await db.transaction(async (tx) => { + // Same lock order as status commits: coordinator, issue, interaction. + await tx.select({ runId: nativeRunFinalizations.runId }).from(nativeRunFinalizations) + .where(and(eq(nativeRunFinalizations.runId, decision.runId), + eq(nativeRunFinalizations.companyId, decision.companyId))) + .for("update"); + const issue = await tx.select().from(issues).where(and( + eq(issues.id, decision.issueId), eq(issues.companyId, decision.companyId), + )).for("update").then((rows) => rows[0]); + if (!issue) return; + const now = new Date(); + const [cancelled] = await tx.update(issueThreadInteractions).set({ + status: "cancelled", + result: { version: 1, outcome: "withdrawn", reason: "A Paperclip upgrade does not require completion review." }, + resolvedAt: now, + updatedAt: now, + }).where(and( + eq(issueThreadInteractions.id, interaction.id), + eq(issueThreadInteractions.companyId, decision.companyId), + eq(issueThreadInteractions.status, "pending"), + )).returning({ id: issueThreadInteractions.id }); + if (!cancelled) return; + const terminalInteraction = await issueThreadInteractionService(tx as unknown as Db).getById(cancelled.id); + if (terminalInteraction) { + await enqueueTerminalIssueInteractionChatPublications(tx as unknown as Db, terminalInteraction); + } + + const pendingInteraction = await tx.select({ id: issueThreadInteractions.id }) + .from(issueThreadInteractions).where(and( + eq(issueThreadInteractions.companyId, issue.companyId), + eq(issueThreadInteractions.issueId, issue.id), + eq(issueThreadInteractions.status, "pending"), + )).limit(1); + const pendingApproval = await tx.select({ id: approvals.id }).from(issueApprovals) + .innerJoin(approvals, and(eq(approvals.id, issueApprovals.approvalId), + eq(approvals.companyId, issue.companyId))) + .where(and(eq(issueApprovals.companyId, issue.companyId), eq(issueApprovals.issueId, issue.id), + inArray(approvals.status, ["pending", "revision_requested"]))).limit(1); + const restoreStatus = issue.status === "in_review" + && issue.lastStatusDecisionId === decision.id + && issue.statusVersion === Number(decision.decisionJson.projectedStatusVersion ?? decision.decisionVersion) + && ["backlog", "todo", "in_progress", "blocked"].includes(decision.fromStatus) + && pendingInteraction.length === 0 && pendingApproval.length === 0 + && issue.executionState?.status !== "pending"; + if (restoreStatus) { + const priorDecision = priorDecisionId ? await tx.select().from(statusDecisions).where(and( + eq(statusDecisions.id, priorDecisionId), eq(statusDecisions.companyId, issue.companyId), + eq(statusDecisions.issueId, issue.id), + )).then((rows) => rows[0]) : null; + await issueService(tx as unknown as Db).update(issue.id, { + status: decision.fromStatus, + // This is an administrative correction, not a replay of the old decision. + lastStatusDecisionId: null, + unblockDescriptor: priorDecision?.decisionJson.unblockDescriptor as typeof issue.unblockDescriptor ?? null, + }, tx, publications); + } + const { publication } = await persistActivity(tx as unknown as Db, { + companyId: issue.companyId, + actorType: "system", + actorId: "native-policy-review-cleanup", + action: restoreStatus ? "issue.updated" : "issue.interaction_cancelled", + entityType: "issue", + entityId: issue.id, + issueId: issue.id, + runId: decision.runId, + details: { + source: "obsolete_native_policy_review", + interactionId: interaction.id, + decisionId: decision.id, + fromStatus: issue.status, + toStatus: restoreStatus ? decision.fromStatus : issue.status, + }, + }); + publications.push(publication); + }); + } catch (err) { + logger.warn({ err, interactionId: interaction.id, issueId: decision.issueId }, + "Failed to withdraw obsolete native policy review; will retry on the next pass"); + continue; + } + for (const publication of publications) { + try { + publishActivity(publication); + } catch (err) { + logger.warn({ err, interactionId: interaction.id, issueId: decision.issueId }, + "Obsolete native policy review cleanup committed; live activity publication failed, history is preserved"); + } + } + } +}