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 ab2d7165048..eb8c795e5fd 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2617,3 +2617,9 @@ is still required. The full remaining stack and older unresolved gates remain in scope. Slice 43 final local acceptance: 120 owner/startup cases plus 209 affected consumer cases pass, without skips. Workspace typecheck/build, final server types, migration application/replay and index-plan checks passed. The negative fixture FK rejection and its correction remain in the owner audit. Remote CI and review are still required. + +## Slice 43 accepted; slice 44 obsolete policy reviews + +Cancelled workspace preservation is draft [#17677](https://github.com/paperclipai/paperclip/pull/17677) at `33567f91551d60053722c013f0d9b402c1db29e6`. All 53 checks passed, Greptile reviewed that head at 5/5, and there are no unresolved review threads. It remains unmerged. + +The native finalization audit found separate historical unsafe-export recovery, obsolete policy-review cleanup, automatic completion-review cleanup, effect delivery, committed response repair, evidence reassessment and workspace/result finalization. Slice 44 owns obsolete policy-review cleanup only. See [its audit](2026-10-25-obsolete-policy-review-sleep-audit.md). The owner/startup selection passes 116 tests after repairing a reproduced notification serialization bug. Consumer and final workspace verification remain pending. Older stack gates and final master reconciliation remain outstanding. diff --git a/doc/plans/2026-10-25-obsolete-policy-review-sleep-audit.md b/doc/plans/2026-10-25-obsolete-policy-review-sleep-audit.md new file mode 100644 index 00000000000..a8504b49656 --- /dev/null +++ b/doc/plans/2026-10-25-obsolete-policy-review-sleep-audit.md @@ -0,0 +1,222 @@ +# Obsolete native policy-review cleanup + +Date: 2026-10-25; current integration: 2026-10-29. Parent: cancelled workspace +preservation #17677 at `b34f0589847f33754ecde9413fd36bf007353d2c`. +The original draft parent was `33567f91551d60053722c013f0d9b402c1db29e6`. + +## Owner and scope + +The native finalization reconciler combines independent duties. Its initial +calls recover historical unsafe exports, dismiss obsolete policy reviews, and +retire automatic completion reviews. Its main loop also dispatches status +effects, repairs committed response presentation, reassesses changed evidence, +and resumes workspace/result finalization. Each duty needs its own migration; +removing the combined scan would lose work. + +This slice owns obsolete version-change policy reviews only. The existing +`obsolete-policy-reviews.ts` service identifies their full historical signature: +pending system confirmation, exact prompt, applied native review decision, +exact two-effect payload, and matching company/issue/run/effect/assessment. +Real completion reviews, authored requests and answered cards are excluded. +The current status arbiter does not generate this old prompt. The typed decision +committer still supports the old effect shape, so a startup-only scan is not +used as the complete writer contract. + +## Producers and durable truth + +The existing interaction, status decision, effect, and assessment rows are truth. +The only production inserter of status decisions and decision effects is +`commitNativeStatusDecision`. Its existing transaction registers an issue-scoped +notification before creating the decision/effects. The signal settles at outer +commit, including a caller-owned transaction. It adds no new transaction around +provider work and no table. A normal current decision can cause one scoped +lookup without leaving a recurring deadline. + +Other interaction writers cannot independently create the matching applied +native decision/effect lineage. Existing decision mutation paths update lineage +or delivery state, not the historical signature. Current-policy producers do +not create new obsolete reviews, but their common commit boundary remains covered. +Startup reconstructs old records. Independent SQL writers and concurrent older +controllers are not covered by process-local signals; they require an explicit +integration or restart reconstruction. + +## Scheduling and recovery + +The service registers with the existing scoped deadline helper. Startup and +committed issue hints reconstruct exact candidates in 100-row keyset pages. +The worker stores interaction/company/issue IDs and due times, not the decision +payload. Batches of twenty use the shared fairness mechanism. Each check rereads +the exact candidate and delegates to the existing transaction owner. + +The owner retains coordinator/issue/interaction locking, current status-version +and decision-pointer guards, other-review checks, chat terminal publication, +and activity persistence. The conditional withdrawal additionally matches the +selected payload, kind, source run, idempotency key, and system author fields. +A card changed after discovery is not withdrawn from a stale snapshot. + +After an attempt the worker rereads the exact pending predicate. Pending cards +retain a 30-second retry; successful, answered, changed, or deleted cards disarm. +An owner failure must not starve another candidate. Discovery failures use the +shared backoff and retain the dirty scope. No expiry or retry budget is invented +for historical records. No unresolved card is silently discarded for quiet. + +This preserves manual/scoped reconciliation and startup repair. Only periodic +orphan reaping passes the new skip flag, so that path stops repeating this +cleanup. The worker is registered when heartbeat maintenance is enabled and +shares the existing admission/drain/standby gates. Shutdown joins admitted +cleanup and removes the registration. Paused dispatch retains dirty requests. + +## Cost and boundaries + +The change is a worker migration, not proof of provider suspension. The combined +reaper and unrelated finalization duties still run until migrated separately. +Cloud alarms and sleep eligibility remain unchanged. + +Startup cost scales with historical pending candidates and their joins. Scoping +ordinary commit hints to company/issue uses the existing interaction indexes. +There is no new schema or unconditional hasPending query. Query-plan evidence +and final test results are recorded below when available. + +## Verification record + +Before production changes, all 44 existing status-authority corpus tests passed +on a fresh disposable PostgreSQL database. The host's embedded harness remains +unavailable; the corpus now accepts the repository's validated external-test DB +URL convention and closes its pool. No production connection behavior changed. + +The initial server typecheck and the typecheck including new worker tests both +passed. Full owner/startup, affected consumers, final workspace checks and exact +PR head CI/review remain required. No local or remote completion is claimed here. + +The first combined run had 67 passing tests and 49 failures. The producer helper +serialized the structurally assignable commit input rather than projecting its +company/issue fields. That included the database object and threw a circular +JSON error on real commits. The helper now serializes only its two domain IDs. +The failed run is retained at `/tmp/sleep44-owners-v1.log`. A fresh full selection +is required after this production repair. + +After the payload repair, all 116 tests in the owner/startup/timing selection pass +without skips (56 corpus tests, 58 startup tests and two timing tests). The corpus +includes the existing 53 native/legacy fixtures inside its matrix test. + +A rolled-back query fixture added 10,000 completed interaction rows on the same +issue and restored one matching pending historical review. The company/issue +lookup used existing decision, interaction-idempotency, effect-ordinal and +assessment indexes. It returned one candidate in 0.058 ms locally. This supports +the scoped-query access path, not a production latency claim. Startup remains +a global reconstruction whose cost scales with history; it does not recur when +settled. SQL and plan are in `/tmp/sleep44-query-plan.sql` and `.log`. + +Final affected consumer acceptance passed all 235 cases across heartbeat recovery, +native finalizer telemetry, external-chat response waits and governed chat +milestones. Combined with the 116 owner/startup cases, 351 tests passed without +skips. Workspace typecheck and build also passed. Remote CI and review remain +required for the published head. No workflow, lockfile or schema change is made. + + +## Earlier parent and acceptance (2026-10-29) + +Normal integration with the verified cancelled-preservation parent yields +`18809268f6292462bb8dc7aea4d081eca4851328`. Every substantive owned edit is preserved, +and unrelated upstream blobs are unchanged. The entire database source tree +is identical to the parent; this slice adds no migration or table and preserves +its 0407 history and five newest snapshots. No new code comments are added. + +All **443 cases across seven complete files pass with no skips**, in 47.08 +seconds. The corpus includes its existing native/legacy fixture matrix; fixtures +inside one matrix case are not added again to the test count. Exact counts: + +| Complete file | Passing cases | +| --- | ---: | +| src/services/native-runtime/external-chat-wait.integration.test.ts | 234 | +| src/__tests__/native-status-arbiter-corpus.test.ts | 56 | +| src/services/heartbeat/recovery.test.ts | 45 | +| src/__tests__/chat-run-milestone-governed-work.test.ts | 16 | +| src/services/native-runtime/native-run-finalizer-telemetry.test.ts | 29 | +| src/__tests__/server-startup-feedback-export.test.ts | 61 | +| src/services/native-runtime/native-finalization-reconciler-timing.test.ts | 2 | + +Five distinct fresh PostgreSQL 17 databases were created and migrated before +the corpus, recovery, finalizer, external-chat, and governed-milestone fixtures. +These external URLs use the supported test paths and avoid a new embedded +cluster while this host is near its semaphore limit; no selected suite is skipped. +Database build, server typecheck/build, module boundaries, target migration +order, and all five fresh parent-0407 migrations pass. The earlier circular +notification-payload failure remains recorded above; no passing run erases it. + +This verifies the combined cleanup owner, actual producer/transaction cases, +status authority and affected consumers. It does not establish deployed sleep +or Cloud alarms. Actual landed parent ancestry and fresh published-head review +and CI remain required. + + +## Current migration409 parent and complete acceptance + +Normal integration `4dff94ae9b261379b45c5b5e96a2bc7e8bbab772` preserves the +complete original runtime patch and every unrelated upstream blob. The entire +database source tree equals the current parent, including 0409 history and +the five newest snapshots. No migration or table is added. + +All **443 cases across seven complete files pass without skips**, in 47.38 +seconds. The selection and per-file counts are unchanged from the table above. +Five separate fresh PostgreSQL 17 databases migrate through 0409 before setup. +Database build, server types/build, module boundaries and exact-head migration +order pass. Existing supported external-database fixture paths are used; no +source fixture adaptation or skipped suite is required. Prior failures remain +recorded. These tests verify worker quiet and supported reactivation, not +deployed host/database suspension or external alarms. + +Adopting published parent `b34f0589847f33754ecde9413fd36bf007353d2c` through normal merge +`e57aa9ca1bbbac82ebb13e032487344f0ea4da1d` changes ancestry and parent documentation only. +Complete server, package, test and UI trees remain identical to the accepted +443-case source. The actual cancelled-preservation parent still must land. +Publication is held until its squash ancestry is reconciled. Fresh published-head +CI and reviews remain required. + + +## Actual cancelled-preservation squash ancestry + +Parent #17677 landed as `51c3f8924cb0bd5fe7cbda15ad9b8f06a6a11055`. Its +entire tree equals reviewed parent `b34f0589847f33754ecde9413fd36bf007353d2c`, +which is already an ancestor of this child. Normal merge +`3c36e163ae918833b15457a4da5f969b49fb2ded` resolves inspected duplicate +squash-ancestry conflicts. Its whole resulting tree equals the accepted child. +Complete server, package, UI and test trees remain identical to the 443-case +source. The database source remains exactly parent-identical, including all +0409 history and retained snapshots. Fresh published-head CI and reviews remain +required; no previous review score is used as current evidence. + + +## External corpus repeatability review correction + +Greptile reported that rerunning the supported external corpus fixture fails +because the prior companies retain the fixed PSC and OUT prefixes. The exact +reviewed head `45cb0d9ccc9b0c37068f56a709f16572bd7b7ddc` reproduces PostgreSQL +23505 on `companies_issue_prefix_idx` when rerun against its previously successful +fixture. Setup fails and all 56 cases are skipped; this is a failing reproduction, +not an assertion pass. + +An initial cleanup attempt at `669da4422df74e90eab324100ddab345f21df020` +passes all 443 assertions but fails the suite in afterAll. Activity-log records +hold non-cascading company foreign keys. That attempt is not accepted and no +passing retry erases it. The cleanup and import were fully removed. + +The final change derives each prefix from its own already-unique company UUID, +normalizes it to uppercase alphanumeric text, and keeps the PSC or OUT label. +It changes exactly two fixture values relative to the reviewed head. Existing +teardown, all assertions and timing limits, runtime code and database schema +remain unchanged. It performs no deletion and leaves retained fixture history +intact. The database prefix column accepts this form. + +Source `4cc0d03ae96bd92a4fdad44168c2a8647d42466c` passes all **443 distinct +cases across seven complete files**, without skips, in 46.46 seconds. The complete +56-case corpus immediately passes a second time against the same database in +10.15 seconds. The repeat is not added again to the distinct test count. Three +preexisting companies include retained PSC/OUT prefixes and an unrelated sentinel. +All three remain byte-identical in the inspected ID/name/prefix fields after +both runs. Only each invocation's two distinct companies are appended. + +Five new isolated PostgreSQL 17 fixtures migrate through 0409. Actual database +build, server typecheck/build, boundaries and exact-head migration order pass. +The complete database source stays unchanged. Fresh published-head CI and review +remain required; the old head's 5/5 score and green checks do not establish them. diff --git a/server/src/__tests__/native-status-arbiter-corpus.test.ts b/server/src/__tests__/native-status-arbiter-corpus.test.ts index 6f58845dfd1..da5da51bc23 100644 --- a/server/src/__tests__/native-status-arbiter-corpus.test.ts +++ b/server/src/__tests__/native-status-arbiter-corpus.test.ts @@ -1,10 +1,12 @@ +import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; import { dismissAutomaticCompletionReviews } from "../services/native-runtime/automatic-completion-reviews.js"; import * as childCompletionDelivery from "../services/native-runtime/native-child-completion-delivery.js"; import { nativeCompletionFeedback } from "../services/native-runtime/native-completion-feedback.js"; import { randomUUID } from "node:crypto"; import { readFileSync } from "node:fs"; import { fileURLToPath } from "node:url"; -import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { and, eq } from "drizzle-orm"; import { agents, @@ -67,10 +69,11 @@ import { resolveNativeReconciliationStatus, } from "../services/native-runtime/native-finalization-reconciler.js"; import * as activityLog from "../services/activity-log.js"; -import { dismissObsoleteNativePolicyReviews } from "../services/native-runtime/obsolete-policy-reviews.js"; +import { dismissObsoleteNativePolicyReviews, registerObsoletePolicyReviewWork } from "../services/native-runtime/obsolete-policy-reviews.js"; import { issueService } from "../services/issues.js"; import { issueThreadInteractionService } from "../services/issue-thread-interactions.js"; import { startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; type Fixture = { id: string; @@ -381,10 +384,11 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { const outsideAgentId = randomUUID(); beforeAll(async () => { - temporary = await startEmbeddedPostgresTestDatabase("paperclip-status-corpus-"); - db = createDb(temporary.connectionString); - await db.insert(companies).values({ id: companyId, name: "Status corpus", issuePrefix: "PSC" }); - await db.insert(companies).values({ id: outsideCompanyId, name: "Outside corpus", issuePrefix: "OUT" }); + const external = resolveExternalTestDatabaseUrl("PAPERCLIP_STATUS_CORPUS_TEST_DATABASE_URL"); + if (!external) temporary = await startEmbeddedPostgresTestDatabase("paperclip-status-corpus-"); + db = createDb(external ?? temporary!.connectionString); + await db.insert(companies).values({ id: companyId, name: "Status corpus", issuePrefix: `PSC${companyId.replaceAll("-", "").toUpperCase()}` }); + await db.insert(companies).values({ id: outsideCompanyId, name: "Outside corpus", issuePrefix: `OUT${outsideCompanyId.replaceAll("-", "").toUpperCase()}` }); await db.insert(agents).values([ { id: agentId, @@ -411,7 +415,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { ]); }, 30_000); - afterAll(async () => temporary?.cleanup()); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); async function seedFixture(fixture: Fixture) { const issueId = randomUUID(); @@ -2007,7 +2011,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { .rejects.toThrow("native_pending_effect_target_missing:enqueue_continuation"); }, 30_000); - async function seedPolicyReview(options: { genuine?: boolean; priorStatus?: "in_progress" | "blocked" | "in_review" } = {}) { + async function seedPolicyReview(options: { genuine?: boolean; priorStatus?: "in_progress" | "blocked" | "in_review"; commit?: typeof commitNativeStatusDecision } = {}) { const template = corpus.fixtures.find((candidate) => candidate.mode === "native")!; const priorStatus = options.priorStatus ?? "in_progress"; const seeded = await seedFixture({ @@ -2022,7 +2026,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { }); await db.update(workAssessments).set({ supersedesAssessmentId: previousAssessmentId }) .where(eq(workAssessments.id, seeded.assessmentId)); - const committed = await commitNativeStatusDecision({ + const committed = await (options.commit ?? commitNativeStatusDecision)({ db, companyId, issueId: seeded.issueId, runId: seeded.runId, assessmentId: seeded.assessmentId, priorStatus, priorStatusVersion: 0, priorDecisionId: null, decision: { @@ -2045,6 +2049,165 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { return { ...seeded, decision: decision!, interaction: interaction! }; } + + describe("obsolete policy review scheduling", () => { + let coordinator: ReturnType | undefined; + let admitted = true; + const errors: unknown[] = []; + beforeEach(() => { + admitted = true; errors.length = 0; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + }); + afterEach(async () => { + await coordinator?.stop(); coordinator = undefined; + vi.useRealTimers(); vi.restoreAllMocks(); + await dismissObsoleteNativePolicyReviews(db); + expect(errors).toEqual([]); + }); + async function start() { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => admitted, + onError: error => errors.push(error) }); + await registerObsoletePolicyReviewWork(db, coordinator, () => admitted).ready; + } + async function flush() { + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator!.nextWakeAt()).toBeNull(); + }, { timeout: 15_000 }); + } + async function quiet() { + const read = vi.spyOn(db, "select"), write = vi.spyOn(db, "transaction"); + try { await vi.advanceTimersByTimeAsync(120_000); expect(read).not.toHaveBeenCalled(); expect(write).not.toHaveBeenCalled(); } + finally { read.mockRestore(); write.mockRestore(); } + } + const card = async (id: string) => (await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, id)))[0]; + function gate() { + let release!: () => void; + const promise = new Promise(resolve => { release = resolve; }); + return { promise, release }; + } + + it("leaves genuine pending reviews and completed history quiet", async () => { + const genuine = await seedPolicyReview({ genuine: true }); + const settled = await seedPolicyReview(); + await dismissObsoleteNativePolicyReviews(db, [settled.runId]); + await start(); await flush(); + expect((await card(genuine.interaction.id)).status).toBe("pending"); + expect((await card(settled.interaction.id)).status).toBe("cancelled"); + await quiet(); + }); + + it("wakes from the real decision commit after becoming idle and preserves decision history", async () => { + await start(); await flush(); await quiet(); + const created = await seedPolicyReview(); + await flush(); + expect((await card(created.interaction.id)).status).toBe("cancelled"); + expect((await issueService(db).getById(created.issueId))!.status).toBe("in_progress"); + expect((await db.select().from(statusDecisions).where(eq(statusDecisions.id, created.decision.id)))[0]).toEqual(created.decision); + await quiet(); + }); + + it.each(["commit", "rollback"] as const)("waits for the outer transaction to %s", async outcome => { + await start(); await flush(); + const entered = gate(), release = gate(); + let issueId = ""; + const creating = seedPolicyReview({ commit: async input => db.transaction(async tx => { + issueId = input.issueId; + const result = await commitNativeStatusDecision({ ...input, db: tx as unknown as typeof db }); + entered.release(); await release.promise; + if (outcome === "rollback") throw new Error("rollback obsolete review"); + return result; + }) }); + const settled = creating.then(value => ({ value, error: null }), error => ({ value: null, error })); + await entered.promise; + await vi.advanceTimersByTimeAsync(2000); + expect(await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.issueId, issueId))).toEqual([]); + release.release(); const result = await settled; + await flush(); + if (outcome === "rollback") { + expect(result.error).toMatchObject({ message: "rollback obsolete review" }); + expect(await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.issueId, issueId))).toEqual([]); + } else { + expect(result.error).toBeNull(); + expect((await card(result.value!.interaction.id)).status).toBe("cancelled"); + } + await quiet(); + }); + + it.each(["restart", "drain", "standby"] as const)("recovers committed cards after %s", async mode => { + if (mode !== "restart") { await start(); await flush(); admitted = false; } + const created = await seedPolicyReview(); + if (mode === "restart") await start(); + else { + await vi.advanceTimersByTimeAsync(1000); + expect((await card(created.interaction.id)).status).toBe("pending"); + await quiet(); admitted = true; + await vi.advanceTimersByTimeAsync(30_000); + } + await flush(); expect((await card(created.interaction.id)).status).toBe("cancelled"); await quiet(); + }); + + it("retains startup discovery while admission is closed", async () => { + const created = await seedPolicyReview(); admitted = false; + await start(); await quiet(); expect((await card(created.interaction.id)).status).toBe("pending"); + admitted = true; await vi.advanceTimersByTimeAsync(30_000); await flush(); + expect((await card(created.interaction.id)).status).toBe("cancelled"); await quiet(); + }); + + it("retries only a failed cleanup while another candidate settles", async () => { + const first = await seedPolicyReview(), second = await seedPolicyReview(); + const write = vi.spyOn(db, "transaction").mockRejectedValueOnce(new Error("injected obsolete cleanup failure")); + await start(); + await vi.waitFor(async () => { + const states = [(await card(first.interaction.id)).status, (await card(second.interaction.id)).status]; + expect(states.sort()).toEqual(["cancelled", "pending"]); + }); + write.mockRestore(); expect(coordinator!.nextWakeAt()).not.toBeNull(); + await vi.advanceTimersByTimeAsync(5000); + expect([(await card(first.interaction.id)).status, (await card(second.interaction.id)).status].sort()).toEqual(["cancelled", "pending"]); + admitted = false; await quiet(); admitted = true; + await vi.advanceTimersByTimeAsync(30_000); await flush(); + expect((await card(first.interaction.id)).status).toBe("cancelled"); + expect((await card(second.interaction.id)).status).toBe("cancelled"); await quiet(); + }); + + it("does not withdraw a review edited after discovery", async () => { + const created = await seedPolicyReview(); + const original = db.transaction.bind(db); + const write = vi.spyOn(db, "transaction").mockImplementationOnce(async (callback, config) => { + await db.update(issueThreadInteractions).set({ payload: { ...created.interaction.payload, prompt: "Approve this deployment." } }) + .where(eq(issueThreadInteractions.id, created.interaction.id)); + return original(callback, config); + }); + await start(); await flush(); write.mockRestore(); + expect((await card(created.interaction.id)).status).toBe("pending"); + expect((await issueService(db).getById(created.issueId))!.status).toBe("in_review"); await quiet(); + }); + + it("joins admitted cleanup on shutdown and reconstructs quiet state on restart", async () => { + const created = await seedPolicyReview(), entered = gate(), release = gate(); + const original = db.transaction.bind(db); + const write = vi.spyOn(db, "transaction").mockImplementationOnce((callback, config) => original(async tx => { + entered.release(); await release.promise; return callback(tx); + }, config)); + const starting = start(); await entered.promise; + let stopped = false; const stopping = coordinator!.stop().then(() => { stopped = true; }); + await vi.advanceTimersByTimeAsync(1000); expect(stopped).toBe(false); + release.release(); await starting; await stopping; write.mockRestore(); + expect((await card(created.interaction.id)).status).toBe("cancelled"); + await start(); await flush(); await quiet(); + }); + + it("drains more than one reconstruction page without replaying settled reviews", async () => { + const created = []; + for (let index = 0; index < 103; index++) created.push(await seedPolicyReview()); + await start(); await flush(); + for (const value of created) expect((await card(value.interaction.id)).status).toBe("cancelled"); + await quiet(); + }, 60_000); + }); + async function seedAutomaticReview() { const seeded = await seedPolicyReview(); const prompt = "Review the persisted native-run evidence and confirm whether this issue may be completed."; diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index b06b38085a5..bfa9d07d7e1 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -116,6 +116,7 @@ const { reapOrphanedRuns: vi.fn(async () => ({ reaped: 0, runIds: [] })), registerScheduledRetryCancellationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerStoppedInstructionCollectionWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), + registerObsoletePolicyReviewWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerCancelledWorkspacePreservationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerInstructionDirectoryCleanupWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerCapturedInstructionCommitWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), @@ -554,6 +555,13 @@ describe("startServer feedback export wiring", () => { process.env.BETTER_AUTH_SECRET = "test-secret"; }); + it.each([false, true])("registers obsolete policy review cleanup only when maintenance is enabled (%s)", async enabled => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); + await startServer(); await settleStartupHousekeeping(); + expect(heartbeatServiceMock.registerObsoletePolicyReviewWork).toHaveBeenCalledTimes(enabled ? 1 : 0); + if (enabled) expect(heartbeatServiceMock.registerObsoletePolicyReviewWork.mock.calls[0][1]()).toBe(true); + }); + it.each([false, true])("registers cancelled workspace preservation only when maintenance is enabled (%s)", async enabled => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); await startServer(); await settleStartupHousekeeping(); @@ -973,7 +981,7 @@ describe("startServer feedback export wiring", () => { intervalCallback?.(); await new Promise((resolve) => setImmediate(resolve)); expect(heartbeatServiceMock.sweepStaleIssueLocks).toHaveBeenCalledTimes(2); - expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true }); + expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true }); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(retiredDetector).not.toHaveBeenCalled(); } finally { diff --git a/server/src/index.ts b/server/src/index.ts index 56b877c0a9a..162693b9cb1 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1824,6 +1824,8 @@ async function startServerWithDatabaseTeardown( () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; if (config.heartbeatSchedulerEnabled) await heartbeat.registerInstructionDirectoryCleanupWork(deliveryWork, () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; + if (config.heartbeatSchedulerEnabled) await heartbeat.registerObsoletePolicyReviewWork(deliveryWork, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; if (config.heartbeatSchedulerEnabled) await heartbeat.registerCancelledWorkspacePreservationWork(deliveryWork, () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; if (config.heartbeatSchedulerEnabled) await heartbeat.registerStoppedInstructionCollectionWork(deliveryWork, @@ -1956,7 +1958,7 @@ async function startServerWithDatabaseTeardown( // Periodically reap orphaned runs (5-min staleness threshold) and make sure // persisted queued work is still being driven forward. runPeriodicRecovery(() => heartbeat - .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true }) + .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true }) .then(() => heartbeat.promoteDueScheduledRetries()) .then(async (promotion) => { await heartbeat.resumeQueuedRuns(); diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index ff596f86a79..ec33ee21b29 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 = { + obsoletePolicyReviews: "obsolete-policy-reviews", cancelledWorkspacePreservation: "cancelled-workspace-preservation", instructionCollection: "instruction-collection", directoryCleanup: "directory-cleanup", diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 5df7f4e1408..f42994f7aa8 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +import { registerObsoletePolicyReviewWork } from "./native-runtime/obsolete-policy-reviews.js"; import { registerCancelledWorkspacePreservationWork } from "./native-runtime/native-failed-workspace-recovery.js"; import { readUpgradeReadiness } from "./heartbeat/upgrade-readiness.js"; import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; @@ -4085,6 +4086,8 @@ export function heartbeatService( triggerIssueMonitor, reportRunActivity: clearDetachedRunWarning, + registerObsoletePolicyReviewWork: (coordinator: Parameters[1], canRun: () => boolean) => + registerObsoletePolicyReviewWork(db, coordinator, () => canRun() && readTaskDrain(new Date()) === null), registerCancelledWorkspacePreservationWork: ( coordinator: Parameters[1], canRun: () => boolean, ) => registerCancelledWorkspacePreservationWork(db, coordinator, diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 350568dd3a5..4ae6637092e 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -1947,7 +1947,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD let settledProcessesDiscovered = false; - async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean }) { + async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean }) { if (!opts?.skipCancelledWorkspacePreservation) await preserveCancelledNativeWorkspaces(db); const staleThresholdMs = opts?.staleThresholdMs ?? 0; const now = new Date(); @@ -1965,6 +1965,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD // reconciler reads the durable workspace barrier and persisted runtime // mode, never the current feature flag. await reconcileNativeFinalizations(db, undefined, { + skipObsoletePolicyReviews: opts?.skipObsoletePolicyReviews, environmentRuntime, pluginWorkerManager: dependencies.pluginWorkerManager, onWorkspaceSettled: settleRecoveredNativeWorkspace, diff --git a/server/src/services/native-runtime/native-finalization-reconciler-timing.test.ts b/server/src/services/native-runtime/native-finalization-reconciler-timing.test.ts index b96aaa77c31..1de6bd22b01 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler-timing.test.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler-timing.test.ts @@ -25,6 +25,13 @@ vi.mock("./automatic-completion-reviews.js", async (importOriginal) => ({ const { reconcileNativeFinalizations } = await import("./native-finalization-reconciler.js"); describe("native finalization reconciliation timing", () => { + it("omits migrated policy cleanup from periodic reconciliation", async () => { + timing.phase.mockClear(); + await expect(reconcileNativeFinalizations({} as never, undefined, { skipObsoletePolicyReviews: true })).rejects.toThrow("review lookup failed"); + expect(timing.phase.mock.calls.map(([phase]) => phase)).toEqual(["legacy_workspace_exports", "automatic_completion_reviews"]); + timing.phase.mockClear(); timing.finish.mockClear(); + }); + it("records the phases and a failed outcome when a pass fails, and rethrows", async () => { await expect(reconcileNativeFinalizations({} as never)).rejects.toThrow("review lookup failed"); diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index df7d36fa7b0..e93e0355347 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -637,6 +637,7 @@ export async function claimNativeSessionResumptions(input: { } type NativeReconciliationOptions = { + skipObsoletePolicyReviews?: boolean; pluginWorkerManager?: WorkspaceWorkers; environmentRuntime?: EnvironmentRuntimeService; onWorkspaceSettled?: (input: { @@ -723,10 +724,12 @@ async function reconcileNativeFinalizationsPass( await recoverLegacyUnsafeWorkspaceExports(db, runIds).catch((err) => { logger.warn({ err }, "Historical unsafe export recovery remains pending"); }); - timing.phase("obsolete_policy_reviews"); - await dismissObsoleteNativePolicyReviews(db, runIds).catch((err) => { - logger.warn({ err }, "Obsolete native policy review lookup failed; continuing native reconciliation"); - }); + if (!options.skipObsoletePolicyReviews) { + timing.phase("obsolete_policy_reviews"); + await dismissObsoleteNativePolicyReviews(db, runIds).catch((err) => { + logger.warn({ err }, "Obsolete native policy review lookup failed; continuing native reconciliation"); + }); + } timing.phase("automatic_completion_reviews"); if (runIds?.length) { const scopes = await db.select({ issueId: nativeRunFinalizations.issueId }).from(nativeRunFinalizations) diff --git a/server/src/services/native-runtime/obsolete-policy-reviews.ts b/server/src/services/native-runtime/obsolete-policy-reviews.ts index 4330f3d8f0f..5233b37559c 100644 --- a/server/src/services/native-runtime/obsolete-policy-reviews.ts +++ b/server/src/services/native-runtime/obsolete-policy-reviews.ts @@ -1,4 +1,8 @@ -import { and, eq, inArray, isNull, sql } from "drizzle-orm"; +import type { ObsoletePolicyReviewScope } from "../obsolete-policy-review-notifications.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; +import { and, asc, eq, gt, inArray, isNull, sql } from "drizzle-orm"; import { approvals, issueApprovals, issueThreadInteractions, issues, nativeRunFinalizations, statusDecisionEffects, statusDecisions, workAssessments, @@ -14,8 +18,8 @@ import { persistActivity, publishActivity, type ActivityPublication } from "../a const obsoletePrompt = "Review the superseding native policy assessment."; /** Retire only the old version-change gate, never a real completion review. */ -export async function dismissObsoleteNativePolicyReviews(db: Db, runIds?: string[]) { - const candidates = await db.select({ +function findObsoletePolicyReviews(db: Db, options: { runIds?: string[]; scope?: ObsoletePolicyReviewScope; id?: string; after?: string } = {}) { + return db.select({ interaction: issueThreadInteractions, decision: statusDecisions, priorDecisionId: workAssessments.priorDecisionId, @@ -53,9 +57,18 @@ export async function dismissObsoleteNativePolicyReviews(db: Db, runIds?: string { kind: "bind_reviewer", prompt: obsoletePrompt, ownerUserId: null }, { kind: "append_superseding_assessment" }, ])}::jsonb`, - ...(runIds?.length ? [inArray(statusDecisions.runId, runIds)] : []), - )).limit(100); + ...(options.runIds?.length ? [inArray(statusDecisions.runId, options.runIds)] : []), + options.scope ? and(eq(issueThreadInteractions.companyId, options.scope.companyId), eq(issueThreadInteractions.issueId, options.scope.issueId)) : undefined, + options.id ? eq(issueThreadInteractions.id, options.id) : undefined, + options.after ? gt(issueThreadInteractions.id, options.after) : undefined, + )).orderBy(asc(issueThreadInteractions.id)).limit(100); +} +export async function dismissObsoleteNativePolicyReviews(db: Db, runIds?: string[]) { + await dismissCandidates(db, await findObsoletePolicyReviews(db, { runIds })); +} + +async function dismissCandidates(db: Db, candidates: Awaited>) { for (const { interaction, decision, priorDecisionId } of candidates) { const publications: ActivityPublication[] = []; try { @@ -80,6 +93,11 @@ export async function dismissObsoleteNativePolicyReviews(db: Db, runIds?: string eq(issueThreadInteractions.id, interaction.id), eq(issueThreadInteractions.companyId, decision.companyId), eq(issueThreadInteractions.status, "pending"), + eq(issueThreadInteractions.payload, interaction.payload), + eq(issueThreadInteractions.kind, interaction.kind), + eq(issueThreadInteractions.sourceRunId, decision.runId), + eq(issueThreadInteractions.idempotencyKey, interaction.idempotencyKey!), + isNull(issueThreadInteractions.createdByAgentId), isNull(issueThreadInteractions.createdByUserId), )).returning({ id: issueThreadInteractions.id }); if (!cancelled) return; const terminalInteraction = await issueThreadInteractionService(tx as unknown as Db).getById(cancelled.id); @@ -150,3 +168,33 @@ export async function dismissObsoleteNativePolicyReviews(db: Db, runIds?: string } } } + +export function registerObsoletePolicyReviewWork(db: Db, coordinator: ReturnType, canRun: () => boolean) { + type Entry = { id: string; companyId: string; issueId: string; dueAt: number }; + async function load(scope?: ObsoletePolicyReviewScope, signal?: AbortSignal): Promise { + const entries = new Map(); + let after: string | undefined; + while (canRun() && !signal?.aborted) { + const rows = await findObsoletePolicyReviews(db, { scope, after }); + for (const { interaction } of rows) entries.set(interaction.id, { + id: interaction.id, companyId: interaction.companyId, issueId: interaction.issueId, dueAt: 0, + }); + if (rows.length < 100) return [...entries.values()]; + after = rows[rows.length - 1].interaction.id; + } + throw new Error("Obsolete policy review discovery stopped"); + } + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.obsoletePolicyReviews, batchSize: 20, + decodeScope: (hint: string): ObsoletePolicyReviewScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry: Entry, scope?: ObsoletePolicyReviewScope) => !scope || entry.companyId === scope.companyId && entry.issueId === scope.issueId, + load, + check: async (entry: Entry, signal?: AbortSignal) => { + const candidates = await findObsoletePolicyReviews(db, { scope: entry, id: entry.id }); + if (!canRun() || signal?.aborted) throw new Error("Obsolete policy review cleanup stopped"); + await dismissCandidates(db, candidates); + const pending = await findObsoletePolicyReviews(db, { scope: entry, id: entry.id }); + return pending.length ? { ...entry, dueAt: Date.now() + 30_000 } : null; + }, + }, canRun); +} diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index 980b797152c..1010b6b8b6a 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 { notifyObsoletePolicyReviews } from "../obsolete-policy-review-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; @@ -1981,6 +1982,7 @@ export async function commitNativeStatusDecision(input: { return { decision: decisionRow, issue, replayed: true }; } await notifyChatRunMilestones(tx, input); + await notifyObsoletePolicyReviews(tx, input); if (!decisionRow) { const latestDecisionVersion = await tx .select({ diff --git a/server/src/services/obsolete-policy-review-notifications.ts b/server/src/services/obsolete-policy-review-notifications.ts new file mode 100644 index 00000000000..5424063cf31 --- /dev/null +++ b/server/src/services/obsolete-policy-review-notifications.ts @@ -0,0 +1,7 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type ObsoletePolicyReviewScope = { companyId: string; issueId: string }; + +export async function notifyObsoletePolicyReviews(transaction: object, scope: ObsoletePolicyReviewScope) { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.obsoletePolicyReviews, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId })); +}