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 f188627ed3d..e8f9aff047f 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -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. 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 2988fad88a3..fd252f21855 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 @@ -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. diff --git a/server/src/__tests__/native-status-arbiter-corpus.test.ts b/server/src/__tests__/native-status-arbiter-corpus.test.ts index b658e4f32c5..f2a79ad2284 100644 --- a/server/src/__tests__/native-status-arbiter-corpus.test.ts +++ b/server/src/__tests__/native-status-arbiter-corpus.test.ts @@ -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; + 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 | undefined; + const intercept = vi.spyOn(db, "transaction").mockImplementationOnce((async (callback, options) => { + if (kind === "work product writer") { + let entered!: () => void; + const ready = new Promise(resolve => { entered = resolve; }); + const released = new Promise(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 }); diff --git a/server/src/services/native-runtime/evidence-classifier.ts b/server/src/services/native-runtime/evidence-classifier.ts index d9736695f45..694d99c6ea5 100644 --- a/server/src/services/native-runtime/evidence-classifier.ts +++ b/server/src/services/native-runtime/evidence-classifier.ts @@ -115,20 +115,22 @@ async function classifyEvidenceRef(input: { issueId: string; runId: string; ref: string; + lockSources?: boolean; }): Promise { 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; result: Record; + /** Caller holds task/run FOR UPDATE in this transaction; evidence locks last through status commit. */ + lockSources?: boolean; }): Promise { 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, diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index f16e3b275c0..8914483393b 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 { 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. diff --git a/server/src/services/native-runtime/work-assessments.ts b/server/src/services/native-runtime/work-assessments.ts index 829b497040b..490fc0ce35d 100644 --- a/server/src/services/native-runtime/work-assessments.ts +++ b/server/src/services/native-runtime/work-assessments.ts @@ -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 { + 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, lockSources: true }); + return nativeSha256(current) === nativeSha256(row.assessment); +}