From a4a38b8a9dde748823232e837adc38a7f1fc3bc6 Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 26 Oct 2026 03:40:45 -0500 Subject: [PATCH] fix(server): serialize reassessment with completion contract revisions Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 9 ++++ ...-native-status-reassessment-sleep-audit.md | 44 +++++++++++++++++++ .../native-status-arbiter-corpus.test.ts | 40 +++++++++++++++-- .../native-runtime/completion-contracts.ts | 23 +++++++--- .../native-status-reassessment.ts | 2 +- .../status-decision-committer.ts | 7 ++- 6 files changed, 114 insertions(+), 11 deletions(-) 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 5bc775c6621..e0bab01aa55 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2500,3 +2500,12 @@ saved retry deadline, all 28 lifecycle cases and server types pass. The owner audit preserves failures and handoff/contract producer findings. Remaining business-fact races, producer coverage, cost, wiring, periodic removal and PR acceptance are still open. Nothing merged; the full stack goal is incomplete. + +Slice 50 contract checkpoint: a newer committed completion contract reproduced +stale completion; a real uncommitted writer reproduced blocking on the task lock. +The contract owner now exposes a nonblocking current-revision guard, acquired +before status row locks and held through commit. All 145 corpus/lifecycle tests +and server types pass. Failure logs and the lock-order rationale are recorded in +the owner audit. Workspace/governance races, producer coverage, cost, startup +integration, periodic removal, remote acceptance and the rest of the full stack +remain open. Nothing merged. 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 3cea63b4317..49a0985414f 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 @@ -511,3 +511,47 @@ Latest-contract, workspace-barrier, governance, dependency and other business-fa races remain open, along with remaining producers, query cost/backlog proof, startup wiring, periodic removal, publication and PR gates. The full original stack scope, older unresolved gates and master reconciliation remain outstanding. + +## Completion-contract serialization checkpoint + +The completion-contract writer is `ensureNativeCompletionContract`. Production +callers are native run preparation and heartbeat runtime selection; both select +or ensure the contract before their later run-state mutation. The writer owns a +per-company/task advisory transaction lock, and revisions are append-only. The +scheduled retired-review commit now calls the owner's +`lockCurrentNativeCompletionContract` before taking coordinator/task/run row +locks. The helper tries that same advisory lock without waiting and verifies the +expected contract is still the latest revision. A busy writer or replaced +contract returns the existing reassessment race outcome. The lock remains held +through the status transaction. Explicit legacy reconciliation is unchanged. + +This order matters: inserting a contract acquires a foreign-key lock on its task. +Trying to lock the task before recognizing an in-flight contract writer can wait +behind that writer. The before-fix tests reproduced a stale `done` decision after +a newer contract committed, and a 30-second timeout while a real contract writer +held an uncommitted revision. The concurrency fixture now also applies a two-second +statement timeout to the status transaction so a future blocking regression can +unwind and release its competing writer rather than waiting for suite timeout. +The writer uses the real service inside a caller-owned transaction; it is rolled +back only after reassessment has returned its expected retry outcome. + +The updated tests prove no status decision commits against either race, a newer +committed contract leaves the old result dormant on subsequent inspection, a +rolled-back revision leaves no row, and restoring the original authoritative +contract allows completion. The contract key is factored privately in its owning +module so the writer and guard cannot diverge. No scheduler-specific SQL lock key +or provider policy is introduced into shared scheduling. + +All 117 status-corpus and 28 worker lifecycle cases pass in one combined run +(145 total), including the retained-history startup deadline regression. Server +typecheck passes. Evidence: `/tmp/sleep50-contract-before.log` and `.status` +(stale-completion failure and writer timeout), `/tmp/sleep50-contract-after.log` +and `.status` (145 passes), `/tmp/sleep50-contract-types.log` and `.status` +(pass). All commands have terminal status. + +Workspace-barrier and governance/dependency races, remaining producer coverage, +query-cost/backlog proof, startup registration, periodic removal and full PR +acceptance remain outstanding. The existing native workspace finalization owner +already has a separate physical/durable ownership protocol; audit and reuse that +boundary before adding any barrier fence. No slice 50 PR, merge, deployed sleep +or completion of the overall stack is claimed by this checkpoint. diff --git a/server/src/__tests__/native-status-arbiter-corpus.test.ts b/server/src/__tests__/native-status-arbiter-corpus.test.ts index 9de419aee3c..93e34b6c7d2 100644 --- a/server/src/__tests__/native-status-arbiter-corpus.test.ts +++ b/server/src/__tests__/native-status-arbiter-corpus.test.ts @@ -1,3 +1,4 @@ +import { ensureNativeCompletionContract } from "../services/native-runtime/completion-contracts.js"; import { nativeStatusReassessmentWork } from "../services/native-runtime/native-status-reassessment-work.js"; import { reassessNativeStatusFromSource } from "../services/native-runtime/native-status-reassessment.js"; import { NativeRunCoordinatorStore } from "../services/native-runtime/native-run-coordinator-store.js"; @@ -11,7 +12,7 @@ import { randomUUID } from "node:crypto"; import { readFileSync } from "node:fs"; import { fileURLToPath } from "node:url"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { and, eq } from "drizzle-orm"; +import { and, eq, sql } from "drizzle-orm"; import { agents, agentWakeupRequests, @@ -2453,7 +2454,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { return seeded; } - it.each(["future", "claim", "unknown claim", "assessment", "result", "phase", "decision", "effects", "ownership", "run outcome", "task execution"] as const)( + it.each(["future", "claim", "unknown claim", "assessment", "result", "phase", "decision", "effects", "ownership", "run outcome", "task execution", "contract revision", "contract writer"] as const)( "guards scheduled reassessment against a changed %s at the commit boundary", async changed => { const seeded = await seedAutomaticReview(); await dismissAutomaticCompletionReviews(db, seeded.issueId); @@ -2485,6 +2486,14 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { const originalEffects = await db.select().from(statusDecisionEffects).where(eq(statusDecisionEffects.decisionId, selected.decisionId!)); expect(originalEffects.length).toBeGreaterThan(0); const transaction = db.transaction.bind(db); + let newContractId: string | undefined; + let releaseWriter: (() => void) | undefined; + let writer: Promise | undefined; + const writeContract = async (owner: typeof db) => { + const saved = await ensureNativeCompletionContract({ db: owner, companyId, + issue: { id: seeded.issueId, title: "New direction", description: "Verify the revised objective" }, actorId: agentId }); + newContractId = saved.row.id; + }; const intercept = vi.spyOn(db, "transaction").mockImplementationOnce((async (callback, options) => { if (change) await db.update(nativeRunFinalizations).set(change).where(eq(nativeRunFinalizations.runId, seeded.runId)); if (changed === "effects") await db.update(statusDecisionEffects).set({ deliveryState: "pending" }) @@ -2496,11 +2505,28 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { .where(eq(heartbeatRuns.id, seeded.runId)); if (changed === "task execution") await db.update(issues).set({ executionRunId: replacementRunId }) .where(eq(issues.id, seeded.issueId)); - return transaction(callback, options); + if (changed === "contract revision") await writeContract(db); + if (changed === "contract writer") { + let writerReady!: () => void; + const ready = new Promise(resolve => { writerReady = resolve; }); + const released = new Promise(resolve => { releaseWriter = resolve; }); + const rollback = new Error("rollback competing contract"); + writer = transaction(async tx => { + await writeContract(tx as unknown as typeof db); + writerReady(); + await released; + throw rollback; + }).catch(error => { if (error !== rollback) throw error; }); + await Promise.race([ready, writer]); + } + return transaction(async tx => { + if (changed === "contract writer") await tx.execute(sql`set local statement_timeout = '2s'`); + return callback(tx); + }, options); }) as typeof db.transaction); try { expect(await reassessNativeStatusFromSource(db, selected, { onlyWhenDue: true })).toEqual({ kind: "deferred", retry: true }); - } finally { intercept.mockRestore(); } + } finally { intercept.mockRestore(); releaseWriter?.(); await writer; } expect((await issueService(db).getById(seeded.issueId))!.status).toBe("in_review"); expect(await db.select().from(statusDecisions).where(eq(statusDecisions.issueId, seeded.issueId))).toHaveLength(1); if (change) expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, seeded.runId)))[0]).toMatchObject(change); @@ -2525,6 +2551,12 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { await db.update(issues).set({ executionRunId: originalIssue.executionRunId }).where(eq(issues.id, seeded.issueId)); await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, replacementRunId)); } + if (changed === "contract revision") { + expect(await reassessNativeStatusFromSource(db, selected, { onlyWhenDue: true })).toEqual({ kind: "deferred" }); + await db.delete(completionContracts).where(eq(completionContracts.id, newContractId!)); + } + if (changed === "contract writer") expect(await db.select().from(completionContracts) + .where(eq(completionContracts.id, newContractId!))).toHaveLength(0); await db.update(nativeRunFinalizations).set({ phase: selected.coordinatorPhase, resultId: seeded.resultId, assessmentId: selected.assessmentId, decisionId: selected.decisionId, nextAttemptAt: null, leaseOwner: null, leaseExpiresAt: null }).where(eq(nativeRunFinalizations.runId, seeded.runId)); diff --git a/server/src/services/native-runtime/completion-contracts.ts b/server/src/services/native-runtime/completion-contracts.ts index f8a642769db..7a4fc728bfd 100644 --- a/server/src/services/native-runtime/completion-contracts.ts +++ b/server/src/services/native-runtime/completion-contracts.ts @@ -133,6 +133,22 @@ export function buildNativeCompletionContract( }; } +function nativeCompletionContractLockKey(companyId: string, issueId: string): string { + return `paperclip:native-completion-contract:${companyId}:${issueId}`; +} + +export async function lockCurrentNativeCompletionContract(tx: Db, input: { + companyId: string; issueId: string; contractId: string; +}): Promise { + const [lock] = await tx.execute(sql`select pg_try_advisory_xact_lock(hashtextextended( + ${nativeCompletionContractLockKey(input.companyId, input.issueId)}, 0)) as acquired`) as unknown as Array<{ acquired: boolean }>; + if (!lock.acquired) return false; + const [latest] = await tx.select({ id: completionContracts.id }).from(completionContracts) + .where(and(eq(completionContracts.companyId, input.companyId), eq(completionContracts.issueId, input.issueId))) + .orderBy(desc(completionContracts.revision)).limit(1); + return latest?.id === input.contractId; +} + export async function ensureNativeCompletionContract(input: { db: Db; companyId: string; @@ -151,11 +167,8 @@ export async function ensureNativeCompletionContract(input: { humanResponseId?: string | null; }) { return input.db.transaction(async (tx) => { - await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${[ - "paperclip:native-completion-contract", - input.companyId, - input.issue.id, - ].join(":")}, 0))`); + await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended( + ${nativeCompletionContractLockKey(input.companyId, input.issue.id)}, 0))`); const policy = resolveNativeCompletionPolicy(input.issue); const latest = await tx .select() diff --git a/server/src/services/native-runtime/native-status-reassessment.ts b/server/src/services/native-runtime/native-status-reassessment.ts index c83d63de6ac..cb46318727e 100644 --- a/server/src/services/native-runtime/native-status-reassessment.ts +++ b/server/src/services/native-runtime/native-status-reassessment.ts @@ -207,7 +207,7 @@ export async function reassessNativeStatusFromSource( decision, requireReassessmentSource: options.onlyWhenDue ? { phase: row.coordinatorPhase, resultId: resultRow.id, assessmentId: assessment.id, - decisionId: row.decisionId, agentId: row.agentId, retiredAutomaticReview: !!retiredAutomaticReview, + decisionId: row.decisionId, agentId: row.agentId, retiredAutomaticReview: retiredAutomaticReview ? { contractId: contractRow.id } : null, } : undefined, requireNoPendingChildCompletion: decision.statusAction === "done" ? childCompletionRecipient : undefined, supersedesCommittedDecisionId: row.coordinatorPhase === "committed" diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index 72a2cce10f6..f85bd21ef09 100644 --- a/server/src/services/native-runtime/status-decision-committer.ts +++ b/server/src/services/native-runtime/status-decision-committer.ts @@ -1,3 +1,4 @@ +import { lockCurrentNativeCompletionContract } from "./completion-contracts.js"; import { isNativeRunnerOwnershipHeld } from "./native-runner-ownership.js"; import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; @@ -1602,7 +1603,7 @@ export async function commitNativeStatusDecision(input: { requireNoPendingChildCompletion?: NativeChildCompletionRecipient; requireReassessmentSource?: { phase: string; resultId: string; assessmentId: string; decisionId: string | null; agentId: string; - retiredAutomaticReview: boolean; + retiredAutomaticReview: { contractId: string } | null; }; requireModelRejectionOwner?: { agentId: string; reviewContext: NativeReviewAssignmentContext | null }; requireProviderFailureOwner?: { agentId: string; reviewContext: NativeReviewAssignmentContext | null }; @@ -1627,6 +1628,10 @@ export async function commitNativeStatusDecision(input: { const publications: ActivityPublication[] = []; const terminalRunsToEmit: (typeof heartbeatRuns.$inferSelect)[] = []; const committed = await input.db.transaction(async (tx) => { + const retiredReview = input.requireReassessmentSource?.retiredAutomaticReview; + if (retiredReview && !await lockCurrentNativeCompletionContract(tx as unknown as Db, { + companyId: input.companyId, issueId: input.issueId, contractId: retiredReview.contractId, + })) throw new NativeStatusRaceError(); const coordinator = await tx .select() .from(nativeRunFinalizations)