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 5f8d871973e..356ff0b6603 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 @@ -412,3 +412,47 @@ cost/backlog behavior, lifecycle wiring and remote PR gates. Current scheduled commit guards fence coordinator/run ownership and source selection, but the retired-review business-fact checks still run before the commit transaction. The old periodic reassessment remains enabled until this full slice is ready. + + +## Authoritative run-event producer checkpoint + +Event references may become accepted/rejected only when a persisted PRP event +has `sourceKind: control_plane` and the corresponding evidence verdict. The +classifier now exposes that existing pure verdict interpretation to the append +boundary, including its existing whitespace normalization. The append boundary +registers a run-scoped reassessment signal before the event write for native +runs. It does not look up a task context or scan evidence consumers to append +an event. Non-verdict events retain the ordinary atomic single-statement path; +runner-authored claims do not activate reassessment. + +The writer audit found three insertion paths. The native coordinator store accepts +only `sourceKind: runner` and rejects control-plane events, so its append path +cannot produce an authoritative verdict and needs no added signal. Scheduled +retry cancellation inserts ordinary lifecycle diagnostics without PRP evidence. +The shared append path owns the supported control-plane verdict writes. Event +rows are otherwise immutable in production; agent/company deletion owns removal +of their run history rather than producing new evidence for surviving runs. + +Four real-append regressions failed before the fix: ordinary append, native-source +append, outer transaction commit, and authoritative evidence after ordinary/runner +logs. The repaired tests observe task completion, outer rollback/commit fencing, +no reassessment signal for ordinary or runner-authored logs, replay idempotency, +and driver-observed SQL quiet. All 28 lifecycle cases and six existing atomic +append cases pass. The atomic tests still prove the one-statement ordinary-log +path and preserve binding/concurrency behavior. + +The initial combined run failed three sequencing cases during embedded PostgreSQL +bootstrap, before assertions. They now accept a validated dedicated external test +database URL, clean only their own company fixture between cases, and close both +clients. All three then pass, including concurrent retry receipt deduplication, +concurrent event allocation, and conflicting replay without cursor drift. Final +server typecheck passes. This is 37 distinct passing cases across the runs, not +a successful initial combined run. + +Evidence: `/tmp/sleep50-events-before.log` and `.status` (four failures), +`/tmp/sleep50-events-after.log` and `.status` (34 passes, three bootstrap failures), +`/tmp/sleep50-events-sequence.log` and `.status` (three passes), and +`/tmp/sleep50-events-final-types.log` and `.status` (pass). All test handles at +this checkpoint are terminal. Case-handoff creation/cancellation, remaining +contract/ownership/authority producers and the full acceptance tasks listed above +remain outstanding; the old periodic reassessment path is still enabled. diff --git a/server/src/__tests__/heartbeat-run-event-sequencing.test.ts b/server/src/__tests__/heartbeat-run-event-sequencing.test.ts index 28c2dd00d24..d5ecb303484 100644 --- a/server/src/__tests__/heartbeat-run-event-sequencing.test.ts +++ b/server/src/__tests__/heartbeat-run-event-sequencing.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { describe, expect, it } from "vitest"; import { eq } from "drizzle-orm"; import { @@ -14,11 +15,20 @@ import { } from "../services/heartbeat-run-events.js"; import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_EVENT_SEQUENCING_TEST_DATABASE_URL"); +async function clearExternalFixture(db: ReturnType, companyId: string) { + if (!externalUrl) return; + await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.companyId, companyId)); + await db.delete(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId)); + await db.delete(agents).where(eq(agents.companyId, companyId)); + await db.delete(companies).where(eq(companies.id, companyId)); +} + describe("P6-11..13 / P6-17 canonical event allocator", () => { it.each([false, true])("deduplicates concurrent retry exhaustion (historical receipt: %s)", async (historical) => { - const temporary = await startEmbeddedPostgresTestDatabase("paperclip-exhaustion-events-"); - const db = createDb(temporary.connectionString); - const otherDb = createDb(temporary.connectionString); + const temporary = externalUrl ? undefined : await startEmbeddedPostgresTestDatabase("paperclip-exhaustion-events-"); + const db = createDb(externalUrl ?? temporary!.connectionString); + const otherDb = createDb(externalUrl ?? temporary!.connectionString); const companyId = "20000000-0000-4000-8000-000000000001"; const agentId = "20000000-0000-4000-8000-000000000002"; const runId = "20000000-0000-4000-8000-000000000003"; @@ -72,13 +82,16 @@ describe("P6-11..13 / P6-17 canonical event allocator", () => { ...event, companyId: "20000000-0000-4000-8000-000000000099", retryExhaustion: exhaustion, })).rejects.toThrow("heartbeat_run_event_binding_mismatch"); } finally { - await temporary.cleanup(); + await clearExternalFixture(db, companyId); + await otherDb.$client.end({ timeout: 1 }); + await db.$client.end({ timeout: 1 }); + await temporary?.cleanup(); } }, 60_000); it("serializes concurrent writers and rejects conflicting replay without cursor drift", async () => { - const temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-events-"); - const db = createDb(temporary.connectionString); + const temporary = externalUrl ? undefined : await startEmbeddedPostgresTestDatabase("paperclip-native-events-"); + const db = createDb(externalUrl ?? temporary!.connectionString); const companyId = "20000000-0000-4000-8000-000000000001"; const agentId = "20000000-0000-4000-8000-000000000002"; const runId = "20000000-0000-4000-8000-000000000003"; @@ -162,7 +175,9 @@ describe("P6-11..13 / P6-17 canonical event allocator", () => { })).rejects.toBeInstanceOf(HeartbeatRunEventConflictError); expect(await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, runId))).toHaveLength(48); } finally { - await temporary.cleanup(); + await clearExternalFixture(db, companyId); + await db.$client.end({ timeout: 1 }); + await temporary?.cleanup(); } }, 60_000); }); diff --git a/server/src/__tests__/native-status-reassessment-work.test.ts b/server/src/__tests__/native-status-reassessment-work.test.ts index 4ba8b7e1b47..e869ab3efb8 100644 --- a/server/src/__tests__/native-status-reassessment-work.test.ts +++ b/server/src/__tests__/native-status-reassessment-work.test.ts @@ -1,4 +1,6 @@ import express from "express"; +import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js"; +import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js"; import { mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -122,6 +124,67 @@ describe("native reassessment worker lifecycle", () => { return { issueId, runId, workProductId, decisionId: committed.decision.id }; } + it.each(["ordinary", "native source"] as const)("wakes from a committed control-plane evidence event through the %s append path", async path => { + const seeded = await seedReview("event:1"); + await vi.advanceTimersByTimeAsync(1000); + await db.update(issueWorkProducts).set({ updatedAt: new Date() }).where(eq(issueWorkProducts.id, seeded.workProductId)); + await start(); await flush(); await quiet(); + const prpEvent = { sourceKind: "control_plane", payload: { evidenceOutcome: " accepted " } }; + const input = { companyId, agentId, runId: seeded.runId, eventType: "evidence.recorded", payload: { prpEvent }, + ...(path === "native source" ? { nativeSource: { sourceInstanceId: randomUUID(), sourceEventId: randomUUID(), + sourceSeq: 1, protocolSchemaVersion: 1, canonicalPayload: prpEvent } } : {}) }; + expect((await appendHeartbeatRunEvent(db, input)).disposition).toBe("committed"); + await flush(); + expect(await status(seeded.issueId)).toBe("done"); + if (path === "native source") expect((await appendHeartbeatRunEvent(db, input)).disposition).toBe("duplicate"); + expect(await db.select().from(statusDecisions).where(eq(statusDecisions.issueId, seeded.issueId))).toHaveLength(2); + await quiet(); + }); + + it("preserves event evidence until outer commit and ignores rollback", async () => { + const seeded = await seedReview("event:1"); + await vi.advanceTimersByTimeAsync(1000); + await db.update(issueWorkProducts).set({ updatedAt: new Date() }).where(eq(issueWorkProducts.id, seeded.workProductId)); + await start(); await flush(); await quiet(); + const write = async (tx: typeof db) => { + await appendHeartbeatRunEvent(tx, { companyId, agentId, runId: seeded.runId, eventType: "evidence.recorded", + payload: { prpEvent: { sourceKind: "control_plane", payload: { evidenceOutcome: "accepted" } } } }); + await vi.advanceTimersByTimeAsync(1000); + expect(await status(seeded.issueId)).toBe("in_review"); + }; + await expect(db.transaction(async tx => { + await write(tx as unknown as typeof db); throw new Error("rollback event evidence"); + })).rejects.toThrow("rollback event evidence"); + await flush(); await quiet(); + expect(await status(seeded.issueId)).toBe("in_review"); + await db.transaction(tx => write(tx as unknown as typeof db)); + await flush(); + expect(await status(seeded.issueId)).toBe("done"); + await quiet(); + }); + + it("keeps ordinary and runner-authored logs from waking reassessment", async () => { + const seeded = await seedReview("event:3"); + await vi.advanceTimersByTimeAsync(1000); + await db.update(issueWorkProducts).set({ updatedAt: new Date() }).where(eq(issueWorkProducts.id, seeded.workProductId)); + await start(); await flush(); await quiet(); + const wakes = vi.fn(); + const stop = subscribeDeliveryWork(db, DELIVERY_QUEUES.nativeStatusReassessment, wakes); + try { + await appendHeartbeatRunEvent(db, { companyId, agentId, runId: seeded.runId, eventType: "log", message: "Still working" }); + await appendHeartbeatRunEvent(db, { companyId, agentId, runId: seeded.runId, eventType: "evidence.recorded", + payload: { prpEvent: { sourceKind: "runner", payload: { evidenceOutcome: "accepted" } } } }); + expect(wakes).not.toHaveBeenCalled(); + await quiet(); + expect(await status(seeded.issueId)).toBe("in_review"); + await appendHeartbeatRunEvent(db, { companyId, agentId, runId: seeded.runId, eventType: "evidence.recorded", + payload: { prpEvent: { sourceKind: "control_plane", payload: { evidenceOutcome: "accepted" } } } }); + await flush(); + expect(await status(seeded.issueId)).toBe("done"); + await quiet(); + } finally { stop(); } + }); + it("wakes when a real workspace finalization settles after a quiet failure", async () => { const seeded = await seedReview(); await db.update(workspaceOperations).set({ status: "running" }).where(eq(workspaceOperations.heartbeatRunId, seeded.runId)); diff --git a/server/src/services/heartbeat-run-events.ts b/server/src/services/heartbeat-run-events.ts index a00a195c39b..a89934dab72 100644 --- a/server/src/services/heartbeat-run-events.ts +++ b/server/src/services/heartbeat-run-events.ts @@ -1,3 +1,5 @@ +import { readNativeEventEvidenceOutcome } from "./native-runtime/evidence-classifier.js"; +import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js"; import { isNativeResponseEvidence, notifyNativeResponsePresentation } from "./native-response-presentation-notifications.js"; import { isSafeNativeChatProgressEventType } from "./safe-native-chat-progress.js"; import { notifyChatNativeProgress } from "./chat-native-progress-notifications.js"; @@ -72,9 +74,10 @@ export async function appendHeartbeatRunEvent( db: Db, input: AppendHeartbeatRunEventInput, ): Promise { + const projectsReassessment = readNativeEventEvidenceOutcome(input.payload?.prpEvent) !== null; const projectsProgress = isSafeNativeChatProgressEventType(input.eventType); const projectsResponse = isNativeResponseEvidence(input.eventType, input.payload?.prpEvent); - if (!input.nativeSource && !input.retryExhaustion && !projectsProgress && !projectsResponse) { + if (!input.nativeSource && !input.retryExhaustion && !projectsProgress && !projectsResponse && !projectsReassessment) { const allocated = db.$with("allocated_run_event", {}).as(sql` update ${heartbeatRuns} set next_event_seq = ${heartbeatRuns.nextEventSeq} + 1, @@ -103,7 +106,7 @@ export async function appendHeartbeatRunEvent( .select({ companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, // Only progress needs the task binding. Other native/recovery events // must not read a potentially large context snapshot to append a log. - ...(projectsProgress || projectsResponse ? { runtimeMode: heartbeatRuns.runtimeMode } : {}), + ...(projectsProgress || projectsResponse || projectsReassessment ? { runtimeMode: heartbeatRuns.runtimeMode } : {}), ...(projectsProgress ? { issueId: boundRunIssueIdSql() } : {}) }) .from(heartbeatRuns) .where(eq(heartbeatRuns.id, input.runId)) @@ -175,6 +178,9 @@ export async function appendHeartbeatRunEvent( } } + if (projectsReassessment && run.runtimeMode === "native") { + await notifyNativeStatusReassessment(tx, { companyId: run.companyId, runId: input.runId }); + } if (projectsProgress && run.runtimeMode === "native" && run.issueId) { await notifyChatNativeProgress(tx, { companyId: run.companyId, issueId: run.issueId }); } diff --git a/server/src/services/native-runtime/evidence-classifier.ts b/server/src/services/native-runtime/evidence-classifier.ts index 835dacf57dd..d9736695f45 100644 --- a/server/src/services/native-runtime/evidence-classifier.ts +++ b/server/src/services/native-runtime/evidence-classifier.ts @@ -102,6 +102,13 @@ function uuidRef(ref: string, prefixes: string[]): string | null { return null; } +export function readNativeEventEvidenceOutcome(value: unknown): "accepted" | "rejected" | null { + const event = record(value); + if (event.sourceKind !== "control_plane") return null; + const outcome = text(record(event.payload).evidenceOutcome); + return outcome === "accepted" || outcome === "rejected" ? outcome : null; +} + async function classifyEvidenceRef(input: { db: Db; companyId: string; @@ -135,12 +142,11 @@ async function classifyEvidenceRef(input: { if (!event) { return { ref: input.ref, kind: "event", outcome: "missing", reasonCode: "durable_run_event_missing", durableRecordId: null }; } - const prpEvent = record(record(event.payload).prpEvent); - const evidenceOutcome = text(record(prpEvent.payload).evidenceOutcome); - if (prpEvent.sourceKind === "control_plane" && evidenceOutcome === "accepted") { + const evidenceOutcome = readNativeEventEvidenceOutcome(record(event.payload).prpEvent); + if (evidenceOutcome === "accepted") { return { ref: input.ref, kind: "event", outcome: "accepted", reasonCode: "control_plane_evidence_accepted", durableRecordId: String(event.id) }; } - if (prpEvent.sourceKind === "control_plane" && evidenceOutcome === "rejected") { + if (evidenceOutcome === "rejected") { return { ref: input.ref, kind: "event", outcome: "rejected", reasonCode: "control_plane_evidence_rejected", durableRecordId: String(event.id) }; } return {