From a23679a5f0722a1b8d371610e2c33fb7f9666a4b Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 26 Oct 2026 04:21:59 -0500 Subject: [PATCH] fix(server): fence reassessment against dependency changes Co-Authored-By: Paperclip --- ...2026-10-24-recurring-server-sleep-stack.md | 11 +++ ...-native-status-reassessment-sleep-audit.md | 49 +++++++++++++ server/src/__tests__/issues-service.test.ts | 10 ++- .../native-status-arbiter-corpus.test.ts | 70 ++++++++++++++++++- server/src/services/issues.ts | 29 +++++++- .../native-status-reassessment.ts | 5 +- .../status-decision-committer.ts | 6 +- 7 files changed, 169 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 54398f3081e..f188627ed3d 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2535,3 +2535,14 @@ question-history fixture passes on recheck, and server types pass. Dependency an remaining authority/producer audits, task-history/query costs, wiring, periodic removal and remote gates remain outstanding. The full stack remains incomplete; nothing merged. + + +Slice 50 dependency checkpoint: seven stale-completion races are reproduced and +fixed through the existing dependency-readiness owner. Scheduled status commit +locks/rechecks prerequisite and workspace-finalization sources with nonblocking +contention handling; ordinary readers retain their behavior. All 181 distinct +status/lifecycle/dependency consumer cases and server types pass. The audit keeps +the initial unordered assertion and misconfigured embedded-test invocation +failures and their repairs. Remaining authority/producer coverage, cost, startup +integration, periodic removal, remote gates, earlier stack findings and the full +original scope 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 38248a4793f..2988fad88a3 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 @@ -685,3 +685,52 @@ Dependency/readiness and remaining selected-authority/evidence races, remaining producers (including handoffs and OAuth reopening), discovery/backlog costs, startup wiring, periodic removal and complete PR gates remain outstanding. No slice 50 PR or merge is claimed; the original full stack scope remains open. + + +## Dependency readiness commit guard (2026-10-26) + +Seven real database races reproduced unsafe completion: a new blocker relation, +a reopened blocker, an uncommitted blocker status writer, a newly recorded +unattributed workspace operation, a changed finalize receipt, an uncommitted +receipt update, and an uncommitted unattributed operation insert. Before the +repair all seven committed a status decision despite stale dependency readiness. +Evidence: /tmp/sleep50-dependencies-before.log and .status (exit 1). + +Dependency semantics remain in issueService.getDependencyReadiness. Its opt-in +lockSources mode requires an existing transaction holding the dependent task +FOR UPDATE. It locks relation/blocker rows FOR UPDATE NOWAIT, then relevant +execution workspaces FOR UPDATE NOWAIT and their existing operation rows FOR +SHARE NOWAIT. The workspace lock fences new operations through the existing FK, +including operations without an issue attribution. Ordinary readiness consumers +retain their previous query and locking behavior. The attributed/unattributed +operation selection and later successful finalization rules are unchanged. + +The scheduled retired-review commit compares the readiness boolean used by the +policy with this locked read; contention or a changed predicate defers through +the existing NativeStatusRaceError retry. It does not create a queue/table or +introduce an independent timer. Locks only cover the current task's prerequisites +and relevant workspaces. History-dependent locking cost still needs measurement +as part of the outstanding discovery/commit cost audit. + +Initial combined verification passed 135 corpus cases (including all seven new +races), but an older effect-state assertion compared unordered SQL results and +failed when physical row order changed. Both reads now order by effect ordinal. +The lifecycle suite also failed before setup because the invocation used an +incorrect external DB variable and fell back to broken embedded PostgreSQL. +These failures remain in /tmp/sleep50-dependencies-after.log/.status; neither was +counted as passing. The dependency consumer selection passed 15 cases, and the +initial typecheck passed. The consumer suite now supports the standard validated +dedicated external test database contract and closes its pool. + +Final verification (all commands terminal): +- /tmp/sleep50-dependencies-final.log/.status: 136 status corpus plus 30 worker + lifecycle cases, all 166 passing. Fresh corpus DB and retained lifecycle history. +- /tmp/sleep50-dependencies-consumers-final.log/.status: all 15 existing dependency + readiness/blocker consumer cases passing; other suites deliberately not selected. +- /tmp/sleep50-dependencies-final-types.log/.status: server typecheck passing. + +The seven regressions also restore or roll back the competing state and prove +that reassessment can subsequently complete. Remaining authority/evidence races, +producer coverage (including handoffs and OAuth reopening), task-history/backlog +cost, startup wiring, periodic removal and remote PR acceptance remain open. +No new PR or merge, and no whole-instance sleep claim. diff --git a/server/src/__tests__/issues-service.test.ts b/server/src/__tests__/issues-service.test.ts index 737039c5463..bbf94717ff3 100644 --- a/server/src/__tests__/issues-service.test.ts +++ b/server/src/__tests__/issues-service.test.ts @@ -57,6 +57,8 @@ import { } from "../services/execution-workspace-policy.ts"; import { buildAgentMentionHref, buildProjectMentionHref, MAX_ISSUE_REQUEST_DEPTH, type IssueWorkMode } from "@paperclipai/shared"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; + const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; @@ -4111,14 +4113,15 @@ describeEmbeddedPostgres("issueService.create workspace inheritance", () => { }); }); -describeEmbeddedPostgres("issueService blockers and dependency wake readiness", () => { +const dependencyTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_ISSUE_DEPENDENCIES_TEST_DATABASE_URL"); +(dependencyTestDatabaseUrl ? describe : describeEmbeddedPostgres)("issueService blockers and dependency wake readiness", () => { let db!: ReturnType; let svc!: ReturnType; let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-issues-blockers-"); - db = createDb(tempDb.connectionString); + if (!dependencyTestDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-issues-blockers-"); + db = createDb(dependencyTestDatabaseUrl ?? tempDb!.connectionString); svc = issueService(db); await ensureIssueRelationsTable(db); }, 20_000); @@ -4139,6 +4142,7 @@ describeEmbeddedPostgres("issueService blockers and dependency wake readiness", }); afterAll(async () => { + await db?.$client.end({ timeout: 5 }); await tempDb?.cleanup(); }); diff --git a/server/src/__tests__/native-status-arbiter-corpus.test.ts b/server/src/__tests__/native-status-arbiter-corpus.test.ts index 981288d75c3..b658e4f32c5 100644 --- a/server/src/__tests__/native-status-arbiter-corpus.test.ts +++ b/server/src/__tests__/native-status-arbiter-corpus.test.ts @@ -23,6 +23,8 @@ import { agentWakeupRequests, companies, completionContracts, + executionWorkspaces, + issueRelations, createDb, heartbeatRuns, issueThreadInteractions, @@ -2491,7 +2493,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { id: replacementRunId, companyId, agentId, status: "running", runtimeMode: "native", nativeIssueId: seeded.issueId, contextSnapshot: { issueId: seeded.issueId }, }); - const originalEffects = await db.select().from(statusDecisionEffects).where(eq(statusDecisionEffects.decisionId, selected.decisionId!)); + const originalEffects = await db.select().from(statusDecisionEffects).where(eq(statusDecisionEffects.decisionId, selected.decisionId!)).orderBy(statusDecisionEffects.ordinal); expect(originalEffects.length).toBeGreaterThan(0); const transaction = db.transaction.bind(db); let newContractId: string | undefined; @@ -2562,7 +2564,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { 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!))) + expect(await db.select().from(statusDecisionEffects).where(eq(statusDecisionEffects.decisionId, selected.decisionId!)).orderBy(statusDecisionEffects.ordinal)) .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)); @@ -2688,6 +2690,70 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { }, 30_000, ); + it.each(["new blocker", "reopened blocker", "blocker writer", "workspace operation", "workspace status", "workspace writer", "workspace insert writer"] as const)( + "fences reassessment against changed dependency readiness: %s", 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 blockerId = randomUUID(), workspaceId = randomUUID(), operationId = randomUUID(); + const workspaceCase = changed.startsWith("workspace"); + if (workspaceCase) await db.insert(executionWorkspaces).values({ id: workspaceId, companyId, + mode: "isolated", strategyType: "git_worktree", name: "Dependency workspace" }); + await db.insert(issues).values({ id: blockerId, companyId, title: "Prerequisite", + status: changed === "new blocker" ? "todo" : "done", executionWorkspaceId: workspaceCase ? workspaceId : null }); + if (changed !== "new blocker") await db.insert(issueRelations).values({ companyId, issueId: blockerId, relatedIssueId: seeded.issueId, type: "blocks" }); + if (changed === "workspace status" || changed === "workspace writer") await db.insert(workspaceOperations).values({ + id: operationId, companyId, issueId: blockerId, executionWorkspaceId: workspaceId, phase: "workspace_finalize", status: "succeeded" }); + expect((await issueService(db).getDependencyReadiness(seeded.issueId)).isDependencyReady).toBe(true); + const transaction = db.transaction.bind(db); + let release!: () => void; + let writer: Promise | undefined; + const pendingOperation = { id: operationId, companyId, executionWorkspaceId: workspaceId, + phase: "workspace_finalize", status: "running" }; + const intercept = vi.spyOn(db, "transaction").mockImplementationOnce((async (callback, options) => { + if (changed === "new blocker") await db.insert(issueRelations).values({ companyId, issueId: blockerId, relatedIssueId: seeded.issueId, type: "blocks" }); + if (changed === "reopened blocker") await db.update(issues).set({ status: "todo" }).where(eq(issues.id, blockerId)); + if (changed === "workspace operation") await db.insert(workspaceOperations).values(pendingOperation); + if (changed === "workspace status") await db.update(workspaceOperations).set({ status: "failed" }).where(eq(workspaceOperations.id, operationId)); + if (changed.endsWith("writer")) { + let entered!: () => void; + const ready = new Promise(resolve => { entered = resolve; }); + const released = new Promise(resolve => { release = resolve; }); + const rollback = new Error("rollback dependency writer"); + writer = transaction(async tx => { + if (changed === "blocker writer") await tx.update(issues).set({ status: "todo" }).where(eq(issues.id, blockerId)); + if (changed === "workspace writer") await tx.update(workspaceOperations).set({ status: "failed" }).where(eq(workspaceOperations.id, operationId)); + if (changed === "workspace insert writer") await tx.insert(workspaceOperations).values(pendingOperation); + entered(); await released; throw rollback; + }).catch(error => { if (error !== rollback) throw error; }); + await Promise.race([ready, writer]); + } + return transaction(async tx => { + 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(); release?.(); 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 (changed === "new blocker" || changed === "reopened blocker") await db.update(issues).set({ status: "done" }).where(eq(issues.id, blockerId)); + if (changed === "workspace operation" || changed === "workspace status") await db.update(workspaceOperations).set({ status: "succeeded" }).where(eq(workspaceOperations.id, operationId)); + expect(await reassessNativeStatusFromSource(db, selected, { onlyWhenDue: true })).toMatchObject({ kind: "committed" }); + expect((await issueService(db).getById(seeded.issueId))!.status).toBe("done"); + }, 30_000, + ); + describe("native status reassessment work", () => { const selectedScope = (seeded: { issueId: string; runId: string }) => ({ companyId, issueId: seeded.issueId, runId: seeded.runId }); diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index d7cd1e556ee..3c631332d3c 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -2470,6 +2470,7 @@ async function listPendingFinalizeBlockerIssueIds( blockerIssueId: string; executionWorkspaceId: string; }>, + lockSources = false, ): Promise> { const pending = new Set(); const blockerIssueIds = [ @@ -2486,7 +2487,14 @@ async function listPendingFinalizeBlockerIssueIds( ), ); - const rows = await dbOrTx + if (lockSources) { + // Fence new operations, including ones without an issue attribution, through + // their workspace FK. Existing operation changes are locked below. + await dbOrTx.select({ id: executionWorkspaces.id }).from(executionWorkspaces) + .where(and(eq(executionWorkspaces.companyId, companyId), inArray(executionWorkspaces.id, executionWorkspaceIds))) + .for("update", { noWait: true }); + } + const query = dbOrTx .select({ issueId: workspaceOperations.issueId, executionWorkspaceId: workspaceOperations.executionWorkspaceId, @@ -2505,6 +2513,8 @@ async function listPendingFinalizeBlockerIssueIds( ), ); + const rows = await (lockSources ? query.for("share", { noWait: true }) : query); + const latestAttributedByBlockerWorkspace = new Map< string, { phase: string; status: string; startedAt: Date } @@ -2654,6 +2664,7 @@ async function listIssueDependencyReadinessMap( dbOrTx: Pick, companyId: string, issueIds: string[], + lockSources = false, ) { const uniqueIssueIds = [...new Set(issueIds.filter(Boolean))]; const readinessMap = new Map(); @@ -2662,7 +2673,7 @@ async function listIssueDependencyReadinessMap( } if (uniqueIssueIds.length === 0) return readinessMap; - const blockerRows = await dbOrTx + const blockerQuery = dbOrTx .select({ issueId: issueRelations.relatedIssueId, blockerIssueId: issueRelations.issueId, @@ -2679,6 +2690,10 @@ async function listIssueDependencyReadinessMap( ), ); + // Callers opting into locks already hold the dependent issue FOR UPDATE, + // which fences new relation FKs. NOWAIT avoids reversing graph-writer order. + const blockerRows = await (lockSources ? blockerQuery.for("update", { noWait: true }) : blockerQuery); + // Collect issue/workspace pairs of "done" blockers — these are the only ones // subject to the workspace-finalize barrier. Blockers that aren't done already // mark the dependent as not-ready and don't need a finalize check. @@ -2699,6 +2714,7 @@ async function listIssueDependencyReadinessMap( dbOrTx, companyId, doneBlockerWorkspacePairs, + lockSources, ); for (const row of blockerRows) { @@ -9193,7 +9209,13 @@ export function issueService(db: Db) { }; }, - getDependencyReadiness: async (issueId: string, dbOrTx: any = db) => { + // lockSources requires a transaction holding the dependent issue FOR UPDATE. + // The ordinary read path retains its existing query/locking behavior. + getDependencyReadiness: async ( + issueId: string, + dbOrTx: any = db, + options: { lockSources?: boolean } = {}, + ) => { const issue = await dbOrTx .select({ id: issues.id, companyId: issues.companyId }) .from(issues) @@ -9206,6 +9228,7 @@ export function issueService(db: Db) { dbOrTx, issue.companyId, [issueId], + options.lockSources, ); return readiness.get(issueId) ?? createIssueDependencyReadiness(issueId); }, diff --git a/server/src/services/native-runtime/native-status-reassessment.ts b/server/src/services/native-runtime/native-status-reassessment.ts index a2aebb2f47d..d28072468fa 100644 --- a/server/src/services/native-runtime/native-status-reassessment.ts +++ b/server/src/services/native-runtime/native-status-reassessment.ts @@ -180,6 +180,7 @@ export async function reassessNativeStatusFromSource( assessment: reassessment, supersedesAssessmentId: assessment.id, }); + const hasUnresolvedIssueBlockers = (readiness?.unresolvedBlockerCount ?? 0) > 0; const governanceGate = retiredAutomaticReview && currentIssue ? await pendingNativeGovernance({ db, companyId: row.companyId, issueId: row.issueId, runId: row.runId, executionState: record(currentIssue.executionState) }) : null; @@ -189,7 +190,7 @@ export async function reassessNativeStatusFromSource( assessment: reassessment, terminalState: "succeeded", workspaceFinalizeStatus: "succeeded", governanceGate, completionClaimPolicyAccepted: contractRow.risk === "low" && contractRow.completionAuthority === "agent_claim_policy", - hasUnresolvedIssueBlockers: (readiness?.unresolvedBlockerCount ?? 0) > 0, + hasUnresolvedIssueBlockers, reviewOwnerUserId: currentIssue.responsibleUserId ?? currentIssue.createdByUserId, priorIssueStatus: row.issueStatus as NativeAuthoritativeIssueStatus, agentId: row.agentId, }) @@ -211,7 +212,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 ? { contractId: contractRow.id, governanceGate } : null, + decisionId: row.decisionId, agentId: row.agentId, retiredAutomaticReview: retiredAutomaticReview ? { contractId: contractRow.id, governanceGate, hasUnresolvedIssueBlockers } : 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 98a14c49f23..f16e3b275c0 100644 --- a/server/src/services/native-runtime/status-decision-committer.ts +++ b/server/src/services/native-runtime/status-decision-committer.ts @@ -1606,7 +1606,7 @@ export async function commitNativeStatusDecision(input: { requireNoPendingChildCompletion?: NativeChildCompletionRecipient; requireReassessmentSource?: { phase: string; resultId: string; assessmentId: string; decisionId: string | null; agentId: string; - retiredAutomaticReview: { contractId: string; governanceGate: NativeGovernanceGate | null } | null; + retiredAutomaticReview: { contractId: string; governanceGate: NativeGovernanceGate | null; hasUnresolvedIssueBlockers: boolean } | null; }; requireModelRejectionOwner?: { agentId: string; reviewContext: NativeReviewAssignmentContext | null }; requireProviderFailureOwner?: { agentId: string; reviewContext: NativeReviewAssignmentContext | null }; @@ -1699,6 +1699,10 @@ export async function commitNativeStatusDecision(input: { issueId: input.issueId, runId: input.runId, executionState }); const expectedGate = expected.retiredAutomaticReview.governanceGate; if (gate?.kind !== expectedGate?.kind || gate?.id !== expectedGate?.id) throw new NativeStatusRaceError(); + const readiness = await issueService(tx as unknown as Db).getDependencyReadiness(input.issueId, tx, { lockSources: true }); + if ((readiness.unresolvedBlockerCount > 0) !== expected.retiredAutomaticReview.hasUnresolvedIssueBlockers) { + throw new NativeStatusRaceError(); + } } catch (error) { if (record(error).code === "55P03" || record(record(error).cause).code === "55P03") throw new NativeStatusRaceError(); throw error;