fix(server): serialize reassessment with completion contract revisions

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-26 03:40:45 -05:00
1 parent 97719a8ac5
commit a4a38b8a9d
6 files changed
+114 -11

No files matched your search

@@ -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.
@@ -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.
@@ -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<void> | 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<void>(resolve => { writerReady = resolve; });
const released = new Promise<void>(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));
@@ -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<boolean> {
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()
@@ -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"
@@ -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)