fix(server): guard scheduled native status reassessment commits

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-26 02:17:42 -05:00
1 parent 46c410e7f6
commit 5eb9ae59a6
5 files changed
+134 -1

No files matched your search

@@ -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.
@@ -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.
@@ -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.
@@ -30,7 +30,11 @@ function record(value: unknown): Record<string, unknown> {
: {};
}
export async function reassessNativeStatusFromSource(db: Db, row: NativeStatusReassessmentSource): Promise<Outcome> {
export async function reassessNativeStatusFromSource(
db: Db,
row: NativeStatusReassessmentSource,
options: { onlyWhenDue?: boolean } = {},
): Promise<Outcome> {
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
@@ -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") {