From 5eb9ae59a615a533bfbf62e4aa4748698778efa4 Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 26 Oct 2026 02:17:42 -0500 Subject: [PATCH] fix(server): guard scheduled native status reassessment commits Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 9 +++ ...-native-status-reassessment-sleep-audit.md | 25 ++++++++ .../native-status-arbiter-corpus.test.ts | 63 +++++++++++++++++++ .../native-status-reassessment.ts | 10 ++- .../status-decision-committer.ts | 28 +++++++++ 5 files changed, 134 insertions(+), 1 deletion(-) 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 763e24cd9fd..b756dea4aca 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2461,3 +2461,12 @@ 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. + +Slice 50 scheduled-commit checkpoint: nine competing source/lease/deadline/effect/ +ownership cases reproduce unsafe scheduled reassessment before the new guard. +The final 103-case status corpus and server types pass. The audit preserves two +initial invalid fixtures and the before-fix constraint rejection. The guard is +opt-in at the existing committer transaction; no outer transaction or early +publication is added. Worker discovery, producer/lifecycle/cost evidence and +periodic removal remain outstanding. No new PR or merge is claimed; all original +remaining duties and older stack gates remain open. 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 e7dbb3e7cce..52e5c6b8ba5 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 @@ -124,3 +124,28 @@ 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. + +## Scheduled commit guard + +Scheduled reassessment now opts into a source guard inside the existing status +committer transaction. After coordinator/task locking it compares the selected +phase/result/assessment/decision, respects a future retry and live or unknown +lease, locks and checks native run ownership, and defers while the old decision +has pending or failed effects. Explicit callers retain their existing behavior. +The lock order remains coordinator, task, then run. Publications still occur only +after the committer's transaction commits. + +The first seven-case reproduction had five behavioral failures and two invalid +fixtures: clearing the assessment/result without clearing dependent references +violated schema constraints before reaching the committer. Corrected fixtures +plus pending-effect and ownership races failed all nine before the guard: eight +incorrect commits and one attempted commit rejected by the result constraint. +After the guard, all 103 status-authority corpus cases and server typecheck pass. +Every deferred case preserves the competing state and successfully commits once +that state is released. Logs: /tmp/sleep50-guard-before.log, +/tmp/sleep50-guard-before2.log, /tmp/sleep50-guard-after.log, +/tmp/sleep50-guard-types.log; statuses are saved beside the test logs. + +This is not worker acceptance. Candidate selection, dormant-versus-retry outcome +handling, producer coverage, lifecycle/cost tests and periodic removal remain. +No slice 50 PR has been published and no additional polling has been removed. diff --git a/server/src/__tests__/native-status-arbiter-corpus.test.ts b/server/src/__tests__/native-status-arbiter-corpus.test.ts index 7789dcc0d97..faa03542dc1 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 { reassessNativeStatusFromSource } from "../services/native-runtime/native-status-reassessment.js"; import { NativeRunCoordinatorStore } from "../services/native-runtime/native-run-coordinator-store.js"; import { registerNativePendingEffectWork } from "../services/native-runtime/native-pending-effect-work.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; @@ -2451,6 +2452,68 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { return seeded; } + it.each(["future", "claim", "unknown claim", "assessment", "result", "phase", "decision", "effects", "ownership"] as const)( + "guards scheduled reassessment against a changed %s at the commit boundary", async changed => { + const seeded = await seedAutomaticReview(); + await dismissAutomaticCompletionReviews(db, seeded.issueId); + const [source] = await db.select({ + runId: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + issueId: issues.id, issueStatus: issues.status, issueStatusVersion: issues.statusVersion, + issueDecisionId: issues.lastStatusDecisionId, coordinatorPhase: nativeRunFinalizations.phase, + assessmentId: nativeRunFinalizations.assessmentId, decisionId: nativeRunFinalizations.decisionId, + contextSnapshot: heartbeatRuns.contextSnapshot, + }).from(nativeRunFinalizations).innerJoin(heartbeatRuns, eq(heartbeatRuns.id, nativeRunFinalizations.runId)) + .innerJoin(issues, eq(issues.id, nativeRunFinalizations.issueId)) + .where(eq(nativeRunFinalizations.runId, seeded.runId)); + const selected = { ...source, assessmentId: source.assessmentId! }; + const future = new Date(Date.now() + 60_000); + const change = changed === "future" ? { nextAttemptAt: future } + : changed === "claim" ? { leaseOwner: "other-finalizer", leaseExpiresAt: future } + : changed === "unknown claim" ? { leaseOwner: "unknown-finalizer", leaseExpiresAt: null } + : changed === "assessment" ? { assessmentId: null, decisionId: null } + : changed === "result" ? { resultId: null, assessmentId: null, decisionId: null } + : changed === "phase" ? { phase: "terminal_failure" } + : changed === "decision" ? { decisionId: null } : null; + const [originalRun] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)); + const originalEffects = await db.select().from(statusDecisionEffects).where(eq(statusDecisionEffects.decisionId, selected.decisionId!)); + expect(originalEffects.length).toBeGreaterThan(0); + const transaction = db.transaction.bind(db); + 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" }) + .where(eq(statusDecisionEffects.decisionId, selected.decisionId!)); + if (changed === "ownership") await db.update(heartbeatRuns).set({ status: "running", + runtimeMode: "native", errorCode: "native_execution_ownership_unverified", nativePhase: "terminal_failure" }) + .where(eq(heartbeatRuns.id, seeded.runId)); + return transaction(callback, options); + }) as typeof db.transaction); + try { + expect(await reassessNativeStatusFromSource(db, selected, { onlyWhenDue: true })).toEqual({ kind: "deferred" }); + } finally { intercept.mockRestore(); } + 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); + if (changed === "effects") { + expect(await db.select().from(statusDecisionEffects).where(eq(statusDecisionEffects.decisionId, selected.decisionId!))) + .toEqual(originalEffects.map(effect => ({ ...effect, deliveryState: "pending" }))); + for (const effect of originalEffects) await db.update(statusDecisionEffects).set({ deliveryState: effect.deliveryState }) + .where(eq(statusDecisionEffects.id, effect.id)); + } + if (changed === "ownership") { + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)))[0]) + .toMatchObject({ status: "running", errorCode: "native_execution_ownership_unverified" }); + await db.update(heartbeatRuns).set({ status: originalRun.status, runtimeMode: originalRun.runtimeMode, + errorCode: originalRun.errorCode, nativePhase: originalRun.nativePhase }).where(eq(heartbeatRuns.id, seeded.runId)); + } + 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)); + expect(await reassessNativeStatusFromSource(db, selected, { onlyWhenDue: true })).toMatchObject({ kind: "committed" }); + expect((await issueService(db).getById(seeded.issueId))!.status).toBe("done"); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.issueId, seeded.issueId))).toHaveLength(2); + }, 30_000, + ); + it("retires a proven automatic review and applies the current successful completion exactly once", async () => { const seeded = await seedAutomaticReview(); // The persisted result is done, while its old assessment/decision required a review. diff --git a/server/src/services/native-runtime/native-status-reassessment.ts b/server/src/services/native-runtime/native-status-reassessment.ts index 44b801a72ca..66697202336 100644 --- a/server/src/services/native-runtime/native-status-reassessment.ts +++ b/server/src/services/native-runtime/native-status-reassessment.ts @@ -30,7 +30,11 @@ function record(value: unknown): Record { : {}; } -export async function reassessNativeStatusFromSource(db: Db, row: NativeStatusReassessmentSource): Promise { +export async function reassessNativeStatusFromSource( + db: Db, + row: NativeStatusReassessmentSource, + options: { onlyWhenDue?: boolean } = {}, +): Promise { const assessment = await db.select({ id: workAssessments.id, contractId: workAssessments.contractId, @@ -201,6 +205,10 @@ export async function reassessNativeStatusFromSource(db: Db, row: NativeStatusRe priorStatusVersion: Number(row.issueStatusVersion), priorDecisionId: row.issueDecisionId, decision, + requireReassessmentSource: options.onlyWhenDue ? { + phase: row.coordinatorPhase, resultId: resultRow.id, assessmentId: assessment.id, + decisionId: row.decisionId, agentId: row.agentId, + } : undefined, requireNoPendingChildCompletion: decision.statusAction === "done" ? childCompletionRecipient : undefined, supersedesCommittedDecisionId: row.coordinatorPhase === "committed" ? row.decisionId ?? undefined diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index 92002eed72b..c379e490059 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 { isNativeRunnerOwnershipHeld } from "./native-runner-ownership.js"; import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; import { notifyUnsafeWorkspaceExportRecovery } from "../unsafe-workspace-export-notifications.js"; import { notifyAutomaticCompletionReviews } from "../automatic-completion-review-notifications.js"; @@ -1599,6 +1600,9 @@ export async function commitNativeStatusDecision(input: { requireMonitorWait?: { agentId: string; nextCheckAt: string }; requireMonitorDelivery?: NativeMonitorDelivery; requireNoPendingChildCompletion?: NativeChildCompletionRecipient; + requireReassessmentSource?: { + phase: string; resultId: string; assessmentId: string; decisionId: string | null; agentId: string; + }; requireModelRejectionOwner?: { agentId: string; reviewContext: NativeReviewAssignmentContext | null }; requireProviderFailureOwner?: { agentId: string; reviewContext: NativeReviewAssignmentContext | null }; requireBoardResponseWaitSource?: NativeBoardResponseWaitSource; @@ -1650,6 +1654,30 @@ export async function commitNativeStatusDecision(input: { .limit(1) .then((rows) => rows[0] ?? null); if (!issue) throw new NativeStatusRaceError(); + if (input.requireReassessmentSource) { + const expected = input.requireReassessmentSource; + const now = Date.now(); + if (coordinator.phase === "terminal_failure" || coordinator.phase !== expected.phase || + coordinator.resultId !== expected.resultId || coordinator.assessmentId !== expected.assessmentId || + coordinator.decisionId !== expected.decisionId || + (coordinator.nextAttemptAt !== null && coordinator.nextAttemptAt.getTime() > now) || + (coordinator.leaseOwner !== null && + (coordinator.leaseExpiresAt === null || coordinator.leaseExpiresAt.getTime() > now))) { + throw new NativeStatusRaceError(); + } + const run = await tx.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId), + eq(heartbeatRuns.nativeIssueId, input.issueId), eq(heartbeatRuns.agentId, expected.agentId), + )).for("update").limit(1).then(rows => rows[0]); + if (!run || run.runtimeMode !== "native" || isNativeRunnerOwnershipHeld(run)) throw new NativeStatusRaceError(); + const pendingEffect = expected.decisionId ? await tx.select({ id: statusDecisionEffects.id }) + .from(statusDecisionEffects).where(and( + eq(statusDecisionEffects.companyId, input.companyId), eq(statusDecisionEffects.issueId, input.issueId), + eq(statusDecisionEffects.decisionId, expected.decisionId), + inArray(statusDecisionEffects.deliveryState, ["pending", "failed"]), + )).limit(1).then(rows => rows[0]) : null; + if (pendingEffect) throw new NativeStatusRaceError(); + } // A completed model turn cannot close a persistent conversation. Preserve // the task here; the response finalizer records its durable waiting state. if (issue.conversationAgentId && input.decision.statusAction === "done") {