fix(server): revalidate evidence before scheduled status commit

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-26 04:30:56 -05:00
1 parent a23679a5f0
commit 08091a8caf
6 files changed
+224 -17

No files matched your search

@@ -2546,3 +2546,13 @@ 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.
Slice 50 evidence checkpoint: seven races reproduced completion using revoked,
removed or concurrently edited evidence. The assessment owner now reclassifies
and locks referenced evidence inside scheduled status commit; changed facts and
contention defer through existing retry handling. All 177 status/lifecycle/
classifier consumer cases and server types pass. The audit preserves before-fix
failures and the remaining identity/producer and cost limits. Startup wiring,
periodic removal, remote gates, older stack findings and all remaining recurring
work are still incomplete. Nothing merged.
@@ -734,3 +734,50 @@ 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.
## Evidence freshness at status commit (2026-10-26)
All seven before-fix regressions committed stale completion when supporting
work-product review, approval, interaction, attachment, event, or case-handoff
state changed after classification and before commit. The work-product case also
holds a real uncommitted rejection until recovery returns, then rolls it back.
/tmp/sleep50-evidence-before.log/.status records seven failed assertions (exit 1),
not fixture/bootstrap failures.
The assessment owner now exposes lockCurrentNativeWorkAssessmentEvidence. It
loads the exact persisted reassessment with its bound contract/result, then calls
the existing classifier with opt-in source locking. Classification rules remain
in evidence-classifier.ts, including reference parsing, accepted/rejected policy,
question/approval outcomes and attention-target existence. The scheduled committer
only compares the resulting normalized assessment and handles contention through
its existing race/retry outcome. Both retired-review and other scheduled
reassessment paths use this guard. Ordinary classification retains its original
unlocked behavior. No extra transaction encloses the committer/publication path.
The classifier acquires SHARE NOWAIT on the referenced evidence rows and existing
attention target agents. The already-held task/run locks fence new task/run-bound
records through their existing FKs; the reference cache still deduplicates repeat
criterion/verification references. The assessment/contract/result records are
also share-locked. This protects mutable acceptance state and evidence deletion,
rather than assuming all durable evidence is immutable. Missing arbitrary agent
IDs have no row to lock; supported agent-creation identity/producer behavior and
other authority fields remain part of the outstanding producer audit. This is
not a blanket proof for independent SQL writers.
The added read cost is one exact assessment/contract/result query plus the normal
classifier's distinct-reference and attention-target queries. It runs only for
an actual scheduled decision commit, not on idle history. Quantitative history,
backlog and full transaction cost acceptance remains outstanding; no performance
claim is inferred from the narrow regression runtime.
After repair all 143 status-corpus, 30 worker-lifecycle and four classifier
consumer tests pass (177 total), as does server typecheck. The seven regressions
also restore/roll back evidence and prove subsequent completion. Evidence:
/tmp/sleep50-evidence-after.log/.status and /tmp/sleep50-evidence-types.log/.status,
both exit 0; all launched commands are terminal. Ordinary classifier consumer
cases still run without a locking API in their fake query builder.
Remaining: producer/authority coverage, query/history/backlog cost, startup
integration, removal of this periodic duty, PR publication and remote acceptance.
The full recurring-work stack and its older unresolved gates remain open.
@@ -17,6 +17,11 @@ import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi }
import { and, eq, sql } from "drizzle-orm";
import {
agents,
assets,
caseHandoffs,
cases,
heartbeatRunEvents,
issueAttachments,
approvals,
issueApprovals,
issueComments,
@@ -2754,6 +2759,101 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => {
}, 30_000,
);
it.each(["work product", "work product writer", "approval", "interaction", "attachment", "event", "handoff"] as const)(
"fences reassessment against changed evidence: %s", async kind => {
const seeded = await seedAutomaticReview();
await dismissAutomaticCompletionReviews(db, seeded.issueId);
const evidenceId = randomUUID(), assetId = randomUUID(), caseId = randomUUID();
let ref = `work_product:${seeded.workProductId}`;
const attachment = { id: evidenceId, companyId, issueId: seeded.issueId, assetId };
const event = { companyId, agentId, runId: seeded.runId, seq: 90001, eventType: "evidence.recorded",
payload: { prpEvent: { sourceKind: "control_plane", payload: { evidenceOutcome: "accepted" } } } };
if (kind === "approval") {
ref = `approval:${evidenceId}`;
await db.insert(approvals).values({ id: evidenceId, companyId, type: "hire_agent", status: "approved", payload: {} });
await db.insert(issueApprovals).values({ companyId, issueId: seeded.issueId, approvalId: evidenceId });
}
if (kind === "interaction") {
ref = `interaction:${evidenceId}`;
await db.insert(issueThreadInteractions).values({ id: evidenceId, companyId, issueId: seeded.issueId,
kind: "request_confirmation", status: "accepted", payload: { version: 1, prompt: "Evidence accepted" } });
}
if (kind === "attachment") {
ref = `attachment:${evidenceId}`;
await db.insert(assets).values({ id: assetId, companyId, provider: "local_disk", objectKey: assetId,
contentType: "text/plain", byteSize: 1, sha256: assetId });
await db.insert(issueAttachments).values(attachment);
}
if (kind === "event") { ref = "event:90001"; await db.insert(heartbeatRunEvents).values(event); }
if (kind === "handoff") {
ref = `case-handoff:${evidenceId}`;
const existing = await db.select({ id: cases.id }).from(cases).where(eq(cases.companyId, companyId));
await db.insert(cases).values({ id: caseId, companyId, caseNumber: existing.length + 1,
identifier: caseId, caseType: "evidence", title: "Handoff evidence" });
await db.insert(caseHandoffs).values({ id: evidenceId, companyId, caseId, issueId: seeded.issueId,
version: 1, destinationKind: "done", phase: "complete", instructions: "Complete", actorType: "system",
actorId: "test", sourceRunId: seeded.runId, wakeupRequestId: randomUUID(), idempotencyKey: evidenceId,
inputFingerprint: evidenceId, state: "completed" });
}
const [storedResult] = await db.select().from(nativeRunResults).where(eq(nativeRunResults.id, seeded.resultId!));
const result = storedResult.resultJson.result as Record<string, unknown>;
await db.update(nativeRunResults).set({ resultJson: { ...storedResult.resultJson, result: { ...result,
completionClaim: { ...(result.completionClaim as object), criteria: [{ criterionId: "objective", status: "satisfied", evidenceRefs: [ref] }] },
verification: [{ commandOrCheck: "evidence", status: "passed", artifactRef: ref }],
} } }).where(eq(nativeRunResults.id, seeded.resultId!));
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! };
async function change(database: typeof db, revoke: boolean) {
if (kind.startsWith("work product")) await database.update(issueWorkProducts)
.set({ reviewState: revoke ? "rejected" : "approved" }).where(eq(issueWorkProducts.id, seeded.workProductId));
if (kind === "approval") await database.update(approvals).set({ status: revoke ? "rejected" : "approved" }).where(eq(approvals.id, evidenceId));
if (kind === "interaction") await database.update(issueThreadInteractions).set({ status: revoke ? "rejected" : "accepted" }).where(eq(issueThreadInteractions.id, evidenceId));
if (kind === "handoff") await database.update(caseHandoffs).set({ state: revoke ? "cancelled" : "completed" }).where(eq(caseHandoffs.id, evidenceId));
if (kind === "attachment") {
if (revoke) await database.delete(issueAttachments).where(eq(issueAttachments.id, evidenceId));
else await database.insert(issueAttachments).values(attachment);
}
if (kind === "event") {
if (revoke) await database.delete(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, seeded.runId), eq(heartbeatRunEvents.seq, event.seq)));
else await database.insert(heartbeatRunEvents).values(event);
}
}
const transaction = db.transaction.bind(db);
let release!: () => void;
let writer: Promise<void> | undefined;
const intercept = vi.spyOn(db, "transaction").mockImplementationOnce((async (callback, options) => {
if (kind === "work product 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 evidence writer");
writer = transaction(async tx => { await change(tx as unknown as typeof db, true); entered(); await released; throw rollback; })
.catch(error => { if (error !== rollback) throw error; });
await Promise.race([ready, writer]);
} else await change(db, true);
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 (kind !== "work product writer") await change(db, false);
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 });
@@ -115,20 +115,22 @@ async function classifyEvidenceRef(input: {
issueId: string;
runId: string;
ref: string;
lockSources?: boolean;
}): Promise<NativeEvidenceRefAssessment> {
const handoffId = uuidRef(input.ref, ["case-handoff"]);
if (handoffId) {
const [row] = await input.db.select({ id: caseHandoffs.id, state: caseHandoffs.state }).from(caseHandoffs).where(and(
const query = input.db.select({ id: caseHandoffs.id, state: caseHandoffs.state }).from(caseHandoffs).where(and(
eq(caseHandoffs.id, handoffId), eq(caseHandoffs.companyId, input.companyId), eq(caseHandoffs.issueId, input.issueId),
eq(caseHandoffs.sourceRunId, input.runId),
)).limit(1);
const [row] = await (input.lockSources ? query.for("share", { noWait: true }) : query);
return { ref: input.ref, kind: "case_handoff", outcome: !row ? "missing" : row.state === "cancelled" ? "rejected" : "accepted",
reasonCode: !row ? "case_handoff_missing" : row.state === "cancelled" ? "case_handoff_cancelled" : "case_handoff_persisted", durableRecordId: row?.id ?? null };
}
const eventMatch = /^event:(\d+)$/.exec(input.ref);
if (eventMatch) {
const cursor = Number(eventMatch[1]);
const event = await input.db.select({
const query = input.db.select({
id: heartbeatRunEvents.id,
eventType: heartbeatRunEvents.eventType,
payload: heartbeatRunEvents.payload,
@@ -138,7 +140,8 @@ async function classifyEvidenceRef(input: {
eq(heartbeatRunEvents.companyId, input.companyId),
eq(heartbeatRunEvents.runId, input.runId),
or(eq(heartbeatRunEvents.seq, cursor), eq(heartbeatRunEvents.sourceSeq, cursor)),
)).limit(1).then((rows) => rows[0] ?? null);
)).limit(1);
const [event] = await (input.lockSources ? query.for("share", { noWait: true }) : query);
if (!event) {
return { ref: input.ref, kind: "event", outcome: "missing", reasonCode: "durable_run_event_missing", durableRecordId: null };
}
@@ -162,7 +165,7 @@ async function classifyEvidenceRef(input: {
const workProductId = uuidRef(input.ref, ["work_product", "work-product", "artifact"]);
if (workProductId) {
const row = await input.db.select({
const query = input.db.select({
id: issueWorkProducts.id,
status: issueWorkProducts.status,
reviewState: issueWorkProducts.reviewState,
@@ -170,7 +173,8 @@ async function classifyEvidenceRef(input: {
eq(issueWorkProducts.id, workProductId),
eq(issueWorkProducts.companyId, input.companyId),
eq(issueWorkProducts.issueId, input.issueId),
)).limit(1).then((rows) => rows[0] ?? null);
)).limit(1);
const [row] = await (input.lockSources ? query.for("share", { noWait: true }) : query);
if (!row) return { ref: input.ref, kind: "work_product", outcome: "missing", reasonCode: "work_product_missing", durableRecordId: null };
if (
["failed", "error", "cancelled", "rejected", "changes_requested"].includes(row.status)
@@ -189,7 +193,7 @@ async function classifyEvidenceRef(input: {
const approvalId = uuidRef(input.ref, ["approval"]);
if (approvalId) {
const row = await input.db.select({ id: approvals.id, status: approvals.status })
const query = input.db.select({ id: approvals.id, status: approvals.status })
.from(issueApprovals)
.innerJoin(approvals, eq(issueApprovals.approvalId, approvals.id))
.where(and(
@@ -197,7 +201,8 @@ async function classifyEvidenceRef(input: {
eq(issueApprovals.issueId, input.issueId),
eq(approvals.companyId, input.companyId),
eq(approvals.id, approvalId),
)).limit(1).then((rows) => rows[0] ?? null);
)).limit(1);
const [row] = await (input.lockSources ? query.for("share", { noWait: true }) : query);
if (!row) return { ref: input.ref, kind: "approval", outcome: "missing", reasonCode: "approval_missing", durableRecordId: null };
if (row.status === "approved") return { ref: input.ref, kind: "approval", outcome: "accepted", reasonCode: "approval_approved", durableRecordId: row.id };
if (["rejected", "cancelled"].includes(row.status)) return { ref: input.ref, kind: "approval", outcome: "rejected", reasonCode: "approval_rejected", durableRecordId: row.id };
@@ -206,7 +211,7 @@ async function classifyEvidenceRef(input: {
const interactionId = uuidRef(input.ref, ["interaction"]);
if (interactionId) {
const row = await input.db.select({
const query = input.db.select({
id: issueThreadInteractions.id,
status: issueThreadInteractions.status,
result: issueThreadInteractions.result,
@@ -214,7 +219,8 @@ async function classifyEvidenceRef(input: {
eq(issueThreadInteractions.id, interactionId),
eq(issueThreadInteractions.companyId, input.companyId),
eq(issueThreadInteractions.issueId, input.issueId),
)).limit(1).then((rows) => rows[0] ?? null);
)).limit(1);
const [row] = await (input.lockSources ? query.for("share", { noWait: true }) : query);
if (!row) return { ref: input.ref, kind: "interaction", outcome: "missing", reasonCode: "interaction_missing", durableRecordId: null };
const outcome = text(record(row.result).outcome);
if (["accepted", "answered", "resolved"].includes(outcome ?? "") || ["accepted", "answered"].includes(row.status)) {
@@ -228,11 +234,12 @@ async function classifyEvidenceRef(input: {
const attachmentId = uuidRef(input.ref, ["attachment"]);
if (attachmentId) {
const row = await input.db.select({ id: issueAttachments.id }).from(issueAttachments).where(and(
const query = input.db.select({ id: issueAttachments.id }).from(issueAttachments).where(and(
eq(issueAttachments.id, attachmentId),
eq(issueAttachments.companyId, input.companyId),
eq(issueAttachments.issueId, input.issueId),
)).limit(1).then((rows) => rows[0] ?? null);
)).limit(1);
const [row] = await (input.lockSources ? query.for("share", { noWait: true }) : query);
return row
? { ref: input.ref, kind: "attachment", outcome: "accepted", reasonCode: "attachment_persisted", durableRecordId: row.id }
: { ref: input.ref, kind: "attachment", outcome: "missing", reasonCode: "attachment_missing", durableRecordId: null };
@@ -248,7 +255,7 @@ async function classifyEvidenceRef(input: {
}
/**
* Classifies model claims against immutable server-owned records. A model's
* Classifies model claims against durable server-owned records. A model's
* `satisfied` or `passed` value is retained as a claim but never becomes proof.
*/
export async function classifyNativeEvidence(input: {
@@ -258,6 +265,8 @@ export async function classifyNativeEvidence(input: {
runId: string;
contract: Record<string, unknown>;
result: Record<string, unknown>;
/** Caller holds task/run FOR UPDATE in this transaction; evidence locks last through status commit. */
lockSources?: boolean;
}): Promise<NativeEvidenceAssessment> {
const claim = record(input.result.completionClaim);
const claimedCriteria = Array.isArray(claim.criteria) ? claim.criteria.map(record) : [];
@@ -361,11 +370,12 @@ export async function classifyNativeEvidence(input: {
const ignoredAttentionRequests = [...normalizedSignals.ignoredAttentionRequests];
for (const request of normalizedSignals.actionableAttentionRequests) {
if (request.ownerClass === "agent" && request.targetAgentId) {
const targetExists = await input.db.select({ id: agents.id }).from(agents).where(and(
const targetQuery = input.db.select({ id: agents.id }).from(agents).where(and(
eq(agents.id, request.targetAgentId),
eq(agents.companyId, input.companyId),
)).limit(1).then((rows) => rows.length > 0);
if (!targetExists) {
)).limit(1);
const targets = await (input.lockSources ? targetQuery.for("share", { noWait: true }) : targetQuery);
if (targets.length === 0) {
ignoredAttentionRequests.push({
sourceIndex: request.sourceIndex,
sourceKind: request.sourceKind,
@@ -1,3 +1,4 @@
import { lockCurrentNativeWorkAssessmentEvidence } from "./work-assessments.js";
import { hasNativeWorkspaceFinalizationOwner } from "./native-workspace-finalization-recovery-state.js";
import { tryLockNativeWorkspaceFinalization } from "./native-workspace-finalization-ownership.js";
import { lockCurrentNativeCompletionContract } from "./completion-contracts.js";
@@ -1715,6 +1716,15 @@ export async function commitNativeStatusDecision(input: {
inArray(statusDecisionEffects.deliveryState, ["pending", "failed"]),
)).limit(1).then(rows => rows[0]) : null;
if (pendingEffect) throw new NativeStatusRaceError();
try {
if (!await lockCurrentNativeWorkAssessmentEvidence(tx as unknown as Db, {
companyId: input.companyId, issueId: input.issueId, runId: input.runId,
resultId: expected.resultId, assessmentId: input.assessmentId,
})) throw new NativeStatusRaceError();
} catch (error) {
if (record(error).code === "55P03" || record(record(error).cause).code === "55P03") throw new NativeStatusRaceError();
throw error;
}
}
// A completed model turn cannot close a persistent conversation. Preserve
// the task here; the response finalizer records its durable waiting state.
@@ -1,7 +1,7 @@
import { and, eq } from "drizzle-orm";
import type { Db } from "@paperclipai/db";
import { workAssessments } from "@paperclipai/db";
import type { NativeEvidenceAssessment } from "./evidence-classifier.js";
import { completionContracts, nativeRunResults, workAssessments } from "@paperclipai/db";
import { classifyNativeEvidence, type NativeEvidenceAssessment } from "./evidence-classifier.js";
import { nativeSha256 } from "./canonical.js";
import type { NativeMonitorDelivery } from "./native-monitor-delivery.js";
@@ -142,3 +142,33 @@ export async function recordNativeAttentionAssessment(input: {
if (!row) throw new Error("native_attention_assessment_not_persisted");
return row;
}
/** Revalidate a scheduled reassessment inside the task/run status transaction. */
export async function lockCurrentNativeWorkAssessmentEvidence(db: Db, input: {
companyId: string;
issueId: string;
runId: string;
resultId: string;
assessmentId: string;
}): Promise<boolean> {
const [row] = await db.select({
assessment: workAssessments.assessmentJson,
contract: completionContracts.contractJson,
envelope: nativeRunResults.resultJson,
}).from(workAssessments)
.innerJoin(completionContracts, and(eq(completionContracts.id, workAssessments.contractId),
eq(completionContracts.companyId, input.companyId), eq(completionContracts.issueId, input.issueId)))
.innerJoin(nativeRunResults, and(eq(nativeRunResults.id, workAssessments.resultId),
eq(nativeRunResults.companyId, input.companyId), eq(nativeRunResults.issueId, input.issueId),
eq(nativeRunResults.runId, input.runId)))
.where(and(eq(workAssessments.id, input.assessmentId), eq(workAssessments.companyId, input.companyId),
eq(workAssessments.issueId, input.issueId), eq(workAssessments.runId, input.runId),
eq(workAssessments.resultId, input.resultId)))
.limit(1).for("share", { noWait: true });
if (!row) return false;
const current = await classifyNativeEvidence({ db, companyId: input.companyId,
issueId: input.issueId, runId: input.runId, contract: row.contract,
result: row.envelope.result as Record<string, unknown>, lockSources: true });
return nativeSha256(current) === nativeSha256(row.assessment);
}