diff --git a/doc/plans/2026-10-26-governed-review-response-sleep-audit.md b/doc/plans/2026-10-26-governed-review-response-sleep-audit.md index 22ff6b45453..283a737ffdd 100644 --- a/doc/plans/2026-10-26-governed-review-response-sleep-audit.md +++ b/doc/plans/2026-10-26-governed-review-response-sleep-audit.md @@ -1,7 +1,7 @@ # Governed-review response recovery: sleep migration audit -Status: owner and producer audit only; no worker or periodic removal implemented. -Inspected parent: `c4beddad1e` (slice 48, draft PR #17696). +Status: raw owner contract and initial worker verified; producer wiring and periodic removal remain incomplete. +Inspected parent: `fad35f2d7a` (slice 48, draft PR #17696). ## One reason for work @@ -136,3 +136,27 @@ files. Establish a baseline before editing the owner. Cover: No deployment, database suspension, Cloud alarm coverage, or cost-savings claim follows from this migration. The full original stack remains in scope. + +## Owner-contract checkpoint + +The scheduled entry point holds the issue transaction, checks the coordinator +claim and actual retry deadline under lock, restores the proof through its owner, +and publishes through addComment in that same transaction. Operational errors +propagate to the worker; existing foreground callers retain their logging contract. +The scoped worker reconstructs existing coordinators and excludes materialized +receipts, including deleted comments. No startup registration or periodic removal +has happened yet. + +The unchanged owner baseline passed 156 cases. The first expanded run passed +164 of 165; the new principal-revocation fixture incorrectly wrote a nonexistent +principal status field, also caught by typechecking. Using the existing company +membership suspension path repaired the fixture. The complete expanded suite now +passes all 165 cases and server typecheck passes. Evidence: +`/tmp/sleep49-baseline.log`, `/tmp/sleep49-contract-tests.log`, +`/tmp/sleep49-contract-tests-v2.log`, `/tmp/sleep49-contract-typecheck-v3.log`, +and `/tmp/sleep49-contract-v2.status`. The cases exercise concurrency, late claim/ +deadline races, real authorization-lock contention and rollback, revoked authority, +startup reconstruction, deleted receipts, and actual quiet SQL after settlement. + +The producer audit, query-cost measurements, startup/periodic wiring, remaining +lifecycle acceptance, full verification and remote gates are still outstanding. diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index a7eec6622b7..f62a357aa93 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db"; /** Topics for existing durable queues; register intent before their writes. */ export const DELIVERY_QUEUES = { + nativeReviewResponse: "native-review-response", nativeResponsePresentation: "native-response-presentation", nativePendingEffects: "native-pending-effects", unsafeWorkspaceExportRecovery: "unsafe-workspace-export-recovery", diff --git a/server/src/services/native-review-response-notifications.ts b/server/src/services/native-review-response-notifications.ts new file mode 100644 index 00000000000..39342be01c3 --- /dev/null +++ b/server/src/services/native-review-response-notifications.ts @@ -0,0 +1,8 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type NativeReviewResponseScope = { companyId: string; issueId?: string; runId?: string; agentId?: string }; + +export function notifyNativeReviewResponse(tx: object, scope: NativeReviewResponseScope) { + return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeReviewResponse, + JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, runId: scope.runId, agentId: scope.agentId })); +} diff --git a/server/src/services/native-runtime/external-chat-wait.integration.test.ts b/server/src/services/native-runtime/external-chat-wait.integration.test.ts index 1bbdd1380a3..c01eb0b30b2 100644 --- a/server/src/services/native-runtime/external-chat-wait.integration.test.ts +++ b/server/src/services/native-runtime/external-chat-wait.integration.test.ts @@ -47,7 +47,10 @@ import type { } from "../../vendor/paperclip-runner/index.js"; import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; -import { finalizeNativeRun } from "./native-run-finalizer.js"; +import { finalizeNativeRun, recoverCommittedNativeReviewResponse } from "./native-run-finalizer.js"; +import { nativeReviewResponseWork } from "./native-review-response-work.js"; +import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../task-admission.js"; import { nativeCompletionFeedback } from "./native-completion-feedback.js"; import { hasPublishedGitHubReview } from "../chat-channels/github/review-wait.js"; import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js"; @@ -2781,6 +2784,110 @@ describe("native external-chat response wait", () => { return run!.resultJson!; } + async function seedRecoverableReviewResponse() { + const fixture = await seedWaitTurn(); + const gate = await seedPriorCompletionReview(fixture); + await finalizeNativeRun({ db, runId: fixture.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: false }); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }) + .where(eq(heartbeatRuns.id, fixture.runId)); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId)); + expect(coordinator!.phase).toBe("committed"); + const input = { companyId: fixture.companyId, issueId: fixture.issueId, runId: fixture.runId, + decisionId: coordinator!.decisionId!, resultId: coordinator!.resultId!, assessmentId: coordinator!.assessmentId!, onlyWhenDue: true }; + return { ...fixture, gate, input }; + } + + it("recovers scheduled governed responses once under concurrent callers and preserves deleted receipts", async () => { + const fixture = await seedRecoverableReviewResponse(); + const work = nativeReviewResponseWork(db); + const [entry] = await work.load({ companyId: fixture.companyId }); + expect(entry?.id).toBe(fixture.runId); + expect(await Promise.all([recoverCommittedNativeReviewResponse(db, fixture.input), + recoverCommittedNativeReviewResponse(db, fixture.input)])).toEqual(expect.arrayContaining([true, false])); + const comments = await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId)); + expect(comments).toHaveLength(1); + expect(await db.select().from(chatPublications).where(eq(chatPublications.commentId, comments[0].id))).toHaveLength(1); + expect(await work.load({ companyId: fixture.companyId })).toEqual([]); + await db.update(issueComments).set({ deletedAt: new Date() }).where(eq(issueComments.id, comments[0].id)); + expect(await work.load({ companyId: fixture.companyId })).toEqual([]); + expect(await recoverCommittedNativeReviewResponse(db, fixture.input)).toBe(false); + expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(1); + }); + + it.each(["future", "claim"])("defers scheduled governed responses behind a new %s under the owner lock", async kind => { + const fixture = await seedRecoverableReviewResponse(); + const work = nativeReviewResponseWork(db); + const [entry] = await work.load({ companyId: fixture.companyId }); + const at = new Date(Date.now() + 60_000); + const transaction = db.transaction.bind(db); + const intercept = vi.spyOn(db, "transaction").mockImplementationOnce((async (callback, options) => { + await transaction(async tx => { + await tx.update(nativeRunFinalizations).set(kind === "future" ? { nextAttemptAt: at } + : { leaseOwner: "another-finalizer", leaseExpiresAt: at }).where(eq(nativeRunFinalizations.runId, fixture.runId)); + }); + return transaction(callback, options); + }) as typeof db.transaction); + try { + expect(await work.check(entry)).toEqual(kind === "future" ? { ...entry, dueAt: at.getTime() } : null); + } finally { intercept.mockRestore(); } + expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(0); + await db.update(nativeRunFinalizations).set({ nextAttemptAt: null, leaseOwner: null, leaseExpiresAt: null }) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + expect(await recoverCommittedNativeReviewResponse(db, fixture.input)).toBe(true); + }); + + it("propagates scheduled governed response contention and rolls back its proof restoration", async () => { + const fixture = await seedRecoverableReviewResponse(); + await db.update(heartbeatRuns).set({ resultJson: sql`${heartbeatRuns.resultJson} - 'externalChatReviewPresentation'` }) + .where(eq(heartbeatRuns.id, fixture.runId)); + let release!: () => void, observed!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const locked = new Promise(resolve => { observed = resolve; }); + const holder = db.transaction(async tx => { + await tx.select().from(chatEndpoints).where(eq(chatEndpoints.id, fixture.endpointId)).for("update"); + observed(); await hold; + }); + await locked; + try { + await expect(recoverCommittedNativeReviewResponse(db, fixture.input)).rejects.toBeDefined(); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + expect(run.resultJson?.externalChatReviewPresentation).toBeUndefined(); + expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(0); + } finally { release(); await holder; } + expect(await recoverCommittedNativeReviewResponse(db, fixture.input)).toBe(true); + }); + + it.each(["gate", "agent", "principal", "company"])("leaves scheduled governed response authority with its owner after %s changes", async kind => { + const fixture = await seedRecoverableReviewResponse(); + if (kind === "gate") await db.update(issueThreadInteractions).set({ status: "accepted" }).where(eq(issueThreadInteractions.id, fixture.gate.id)); + if (kind === "agent") await db.update(agents).set({ status: "paused" }).where(eq(agents.id, fixture.agentId)); + if (kind === "principal") await db.update(companyMemberships).set({ status: "suspended" }) + .where(and(eq(companyMemberships.companyId, fixture.companyId), eq(companyMemberships.principalId, fixture.userId))); + expect(await recoverCommittedNativeReviewResponse(db, { ...fixture.input, + ...(kind === "company" ? { companyId: randomUUID() } : {}) })).toBe(false); + expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(0); + }); + + it("reconstructs scheduled governed responses at startup and leaves their history quiet", async () => { + const fixture = await seedRecoverableReviewResponse(); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeReviewResponseWork(db).register(coordinator, () => true); + try { + await worker.ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 10_000 }).toEqual({ active: 0, next: null }); + expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(1); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const previousDebug = db.$client.options.debug; + const queries: string[] = []; + db.$client.options.debug = (_connection, query) => { queries.push(query); }; + try { await vi.advanceTimersByTimeAsync(120_000); expect(queries).toEqual([]); } + finally { db.$client.options.debug = previousDebug; } + expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + it.each(["telegram", "discord"] as const)( "presents the exact %s response without resolving a pre-existing native completion review", async (provider) => { diff --git a/server/src/services/native-runtime/native-chat-review-presentation.ts b/server/src/services/native-runtime/native-chat-review-presentation.ts index 366e564acb6..641c37c3b29 100644 --- a/server/src/services/native-runtime/native-chat-review-presentation.ts +++ b/server/src/services/native-runtime/native-chat-review-presentation.ts @@ -500,6 +500,7 @@ export async function restoreNativeChatReviewPresentationInTransaction( decisionId: string; resultId: string; assessmentId: string; + onlyWhenDue?: boolean; }, ): Promise { const [coordinator] = await db @@ -519,6 +520,8 @@ export async function restoreNativeChatReviewPresentationInTransaction( .for("update") .limit(1); if (!coordinator) return false; + if (input.onlyWhenDue && (coordinator.leaseOwner !== null + || coordinator.nextAttemptAt !== null && coordinator.nextAttemptAt.getTime() > Date.now())) return false; const [decision] = await db .select() .from(statusDecisions) diff --git a/server/src/services/native-runtime/native-review-response-work.ts b/server/src/services/native-runtime/native-review-response-work.ts new file mode 100644 index 00000000000..32df4ac1041 --- /dev/null +++ b/server/src/services/native-runtime/native-review-response-work.ts @@ -0,0 +1,83 @@ +import { and, asc, eq, gt, isNotNull, isNull, notExists, sql } from "drizzle-orm"; +import { heartbeatRuns, issueComments, nativeRunFinalizations, statusDecisions, type Db } from "@paperclipai/db"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import type { NativeReviewResponseScope } from "../native-review-response-notifications.js"; +import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; +import { recoverCommittedNativeReviewResponse } from "./native-run-finalizer.js"; + +type Entry = { id: string; companyId: string; issueId: string; agentId: string; dueAt: number }; + +export function nativeReviewResponseWork(db: Db) { + function candidates(scope?: NativeReviewResponseScope, after?: string, runId?: string) { + return db.select({ id: nativeRunFinalizations.runId, companyId: nativeRunFinalizations.companyId, + issueId: nativeRunFinalizations.issueId, agentId: heartbeatRuns.agentId, + decisionId: nativeRunFinalizations.decisionId, resultId: nativeRunFinalizations.resultId, + assessmentId: nativeRunFinalizations.assessmentId, nextAttemptAt: nativeRunFinalizations.nextAttemptAt, + }).from(nativeRunFinalizations) + .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId), + eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.nativeIssueId, nativeRunFinalizations.issueId))) + .innerJoin(statusDecisions, and(eq(statusDecisions.id, nativeRunFinalizations.decisionId), + eq(statusDecisions.companyId, nativeRunFinalizations.companyId), eq(statusDecisions.issueId, nativeRunFinalizations.issueId), + eq(statusDecisions.runId, nativeRunFinalizations.runId), eq(statusDecisions.assessmentId, nativeRunFinalizations.assessmentId))) + .where(and(eq(nativeRunFinalizations.phase, "committed"), isNull(nativeRunFinalizations.leaseOwner), + isNotNull(nativeRunFinalizations.resultId), isNotNull(nativeRunFinalizations.assessmentId), + eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.status, "succeeded"), + eq(statusDecisions.reasonCode, "governed_response_waiting"), eq(statusDecisions.applicationState, "applied"), + sql`${statusDecisions.decisionJson}->'externalChatReviewPresentation'->>'schema' = 'paperclip.chat_review_response_presentation.v1'`, + notExists(db.select({ id: issueComments.id }).from(issueComments).where(and( + eq(issueComments.companyId, nativeRunFinalizations.companyId), eq(issueComments.issueId, nativeRunFinalizations.issueId), + eq(issueComments.createdByRunId, nativeRunFinalizations.runId), eq(issueComments.authorAgentId, heartbeatRuns.agentId), + eq(issueComments.authorType, "agent"), sql`${issueComments.metadata}->>'authorizationReason' = 'allow_chat_run_presentation'`, + sql`${heartbeatRuns.resultJson}->'externalChatReviewPresentation'->>'decisionId' = ${nativeRunFinalizations.decisionId}::text`, + ))), + scope ? eq(nativeRunFinalizations.companyId, scope.companyId) : undefined, + scope?.issueId ? eq(nativeRunFinalizations.issueId, scope.issueId) : undefined, + scope?.runId ? eq(nativeRunFinalizations.runId, scope.runId) : undefined, + scope?.agentId ? eq(heartbeatRuns.agentId, scope.agentId) : undefined, + runId ? eq(nativeRunFinalizations.runId, runId) : undefined, + after ? gt(nativeRunFinalizations.runId, after) : undefined, + )).orderBy(asc(nativeRunFinalizations.runId)).limit(100); + } + async function load(scope?: NativeReviewResponseScope, signal?: AbortSignal, canRun = () => true): Promise { + const entries: Entry[] = []; + let after: string | undefined; + while (canRun() && !signal?.aborted) { + const rows = await candidates(scope, after); + for (const row of rows) entries.push({ id: row.id, companyId: row.companyId, + issueId: row.issueId, agentId: row.agentId, dueAt: row.nextAttemptAt?.getTime() ?? 0 }); + if (rows.length < 100) return entries; + after = rows[rows.length - 1].id; + } + throw new Error("Governed review response discovery stopped"); + } + async function check(entry: Entry, signal?: AbortSignal): Promise { + if (signal?.aborted) throw new Error("Governed review response recovery stopped"); + const scope = { companyId: entry.companyId, issueId: entry.issueId }; + const [current] = await candidates(scope, undefined, entry.id); + if (!current) return null; + if (current.nextAttemptAt && current.nextAttemptAt.getTime() > Date.now()) return { ...entry, dueAt: current.nextAttemptAt.getTime() }; + if (signal?.aborted) throw new Error("Governed review response recovery stopped"); + const recoveryStartedAt = Date.now(); + await recoverCommittedNativeReviewResponse(db, { ...scope, runId: entry.id, + decisionId: current.decisionId!, resultId: current.resultId!, assessmentId: current.assessmentId!, onlyWhenDue: true }); + const [remaining] = await candidates(scope, undefined, entry.id); + return remaining?.nextAttemptAt && remaining.nextAttemptAt.getTime() > recoveryStartedAt + ? { ...entry, dueAt: remaining.nextAttemptAt.getTime() } : null; + } + function register(coordinator: ReturnType, canRun: () => boolean) { + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.nativeReviewResponse, + decodeScope: (hint: string): NativeReviewResponseScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry: Entry, scope?: NativeReviewResponseScope) => !scope || entry.companyId === scope.companyId + && (!scope.issueId || entry.issueId === scope.issueId) && (!scope.runId || entry.id === scope.runId) + && (!scope.agentId || entry.agentId === scope.agentId), + load: (scope, signal) => load(scope, signal, canRun), + check: (entry, signal) => { + if (!canRun()) throw new Error("Governed review response recovery paused"); + return check(entry, signal); + }, + }, canRun); + } + return { load, check, register }; +} diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index 253d2499ecb..dc30b8112d5 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -721,64 +721,61 @@ async function projectCommittedRun(input: { } } -async function materializeCommittedReviewResponse(db: Db, runId: string) { - try { - const [run] = await db - .select() - .from(heartbeatRuns) - .where(eq(heartbeatRuns.id, runId)) - .limit(1); - if (!run?.nativeIssueId || !run.resultJson?.externalChatReviewPresentation) - return; - const decisionId = record( - run.resultJson.externalChatReviewPresentation, - ).decisionId; - if ( - typeof decisionId !== "string" || - (await hasMaterializedNativeReviewResponse(db, { - companyId: run.companyId, - issueId: run.nativeIssueId, - runId, - decisionId, - })) - ) - return; - const summary = record(run.resultJson.nativeResult).summary; - if ( - typeof summary !== "string" || - !(await authorizeNativeChatReviewPresentation(db, { - companyId: run.companyId, - issueId: run.nativeIssueId, - runId, - resultJson: run.resultJson, - })) - ) - return; - // The coordinator transaction is already committed. addComment owns a - // fresh issue transaction, rechecks the proof and carries selected files; - // its same-run/text and publication keys make recovery replay idempotent. - await issueService(db).addComment( - run.nativeIssueId, - summary, - { agentId: run.agentId, runId }, - { - authorizationReason: "allow_chat_run_presentation", - }, - ); - publishChatPublicationCommitSignal({ +async function materializeNativeReviewResponse(db: Db, runId: string, dbOrTx: Db = db) { + const [run] = await dbOrTx + .select() + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .limit(1); + if (!run?.nativeIssueId || !run.resultJson?.externalChatReviewPresentation) + return null; + const decisionId = record( + run.resultJson.externalChatReviewPresentation, + ).decisionId; + if ( + typeof decisionId !== "string" || + (await hasMaterializedNativeReviewResponse(dbOrTx, { companyId: run.companyId, issueId: run.nativeIssueId, - runId: run.id, - agentId: run.agentId, - eventType: "run.presentation.resolved", - }); + runId, + decisionId, + })) + ) + return null; + const summary = record(run.resultJson.nativeResult).summary; + if ( + typeof summary !== "string" || + !(await authorizeNativeChatReviewPresentation(dbOrTx, { + companyId: run.companyId, + issueId: run.nativeIssueId, + runId, + resultJson: run.resultJson, + })) + ) + return null; + await issueService(db).addComment( + run.nativeIssueId, + summary, + { agentId: run.agentId, runId }, + { + authorizationReason: "allow_chat_run_presentation", + }, + dbOrTx, + ); + return run; +} + +function publishNativeReviewResponse(run: typeof heartbeatRuns.$inferSelect) { + publishChatPublicationCommitSignal({ companyId: run.companyId, issueId: run.nativeIssueId!, + runId: run.id, agentId: run.agentId, eventType: "run.presentation.resolved" }); +} + +async function materializeCommittedReviewResponse(db: Db, runId: string) { + try { + const run = await materializeNativeReviewResponse(db, runId); + if (run) publishNativeReviewResponse(run); } catch (error) { - // The durable committed proof remains retryable by the next reconciler - // sweep. Presentation failure must not reclassify a successful native run. - logger.warn( - { runId, error }, - "Committed chat review response is awaiting presentation retry", - ); + logger.warn({ runId, error }, "Committed chat review response is awaiting presentation retry"); } } @@ -1153,6 +1150,22 @@ export async function repairCommittedNativeReviewResponse( ); if (restored) await materializeCommittedReviewResponse(db, input.runId); } +export async function recoverCommittedNativeReviewResponse(db: Db, input: { + companyId: string; issueId: string; runId: string; decisionId: string; + resultId: string; assessmentId: string; onlyWhenDue?: boolean; +}): Promise { + const materialized = await db.transaction(async transaction => { + const tx = transaction as unknown as Db; + const [issue] = await tx.select({ id: issues.id }).from(issues) + .where(and(eq(issues.companyId, input.companyId), eq(issues.id, input.issueId))).for("update"); + if (!issue || await hasMaterializedNativeReviewResponse(tx, input)) return null; + if (!await restoreNativeChatReviewPresentationInTransaction(tx, input)) return null; + return materializeNativeReviewResponse(db, input.runId, tx); + }); + if (materialized) publishNativeReviewResponse(materialized); + return materialized !== null; +} + export async function finalizeNativeRun(input: { db: Db; runId: string;