mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-31 06:04:11 +01:00
fix(server): fence reassessment against dependency changes
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
9e78f0edac
commit
a23679a5f0
7 files changed
+169
-11
No files matched your search
@@ -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.
|
||||
@@ -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.
|
||||
@@ -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<typeof createDb>;
|
||||
let svc!: ReturnType<typeof issueService>;
|
||||
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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();
|
||||
});
|
||||
|
||||
|
||||
@@ -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<void> | 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<void>(resolve => { entered = resolve; });
|
||||
const released = new Promise<void>(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 });
|
||||
|
||||
|
||||
@@ -2470,6 +2470,7 @@ async function listPendingFinalizeBlockerIssueIds(
|
||||
blockerIssueId: string;
|
||||
executionWorkspaceId: string;
|
||||
}>,
|
||||
lockSources = false,
|
||||
): Promise<Set<string>> {
|
||||
const pending = new Set<string>();
|
||||
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<Db, "select">,
|
||||
companyId: string,
|
||||
issueIds: string[],
|
||||
lockSources = false,
|
||||
) {
|
||||
const uniqueIssueIds = [...new Set(issueIds.filter(Boolean))];
|
||||
const readinessMap = new Map<string, IssueDependencyReadiness>();
|
||||
@@ -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);
|
||||
},
|
||||
|
||||
@@ -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"
|
||||
|
||||
@@ -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;
|
||||
|
||||
Reference in new issue
Block a user