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 eb8c795e5fd..53535b60814 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2623,3 +2623,9 @@ Slice 43 final local acceptance: 120 owner/startup cases plus 209 affected consu 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. + +## Slice 44 accepted; slice 45 automatic completion reviews + +Obsolete policy-review cleanup is draft [#17679](https://github.com/paperclipai/paperclip/pull/17679) at `e671c66bba6e9a42aa570826e6f3420dba199399`. All 53 checks pass, Greptile reviewed that head at 5/5, and no unresolved threads remain. It is mergeable and unmerged. + +Slice 45 owns automatic completion-review retirement only. It preserves direct finalizer ordering and the durable withdrawal trigger for reassessment. The worker's raw lookup propagates failures so failed discovery cannot look empty. See [the audit](2026-10-25-automatic-completion-review-sleep-audit.md). All 370 local owner/startup/consumer tests pass, as do workspace types/build. Exact-head remote gates remain required. Older stack CI/review gaps, master reconciliation and the remaining original duties are still outstanding. diff --git a/doc/plans/2026-10-25-automatic-completion-review-sleep-audit.md b/doc/plans/2026-10-25-automatic-completion-review-sleep-audit.md new file mode 100644 index 00000000000..8ca7d84f992 --- /dev/null +++ b/doc/plans/2026-10-25-automatic-completion-review-sleep-audit.md @@ -0,0 +1,242 @@ +# Automatic completion-review retirement + +Date: 2026-10-25; current integration: 2026-10-29. Parent: obsolete policy +reviews #17679 at actual squash `dfd5b1ac6205b6ec8f56752dfc5d3f9c02c26b6d`. +The original draft parent was `e671c66bba6e9a42aa570826e6f3420dba199399`. + +## One reason and its owner + +The native finalization pass calls `dismissAutomaticCompletionReviews` globally, +or per issue when run-scoped. `finalizeNativeRun` also calls it for the active +issue before resolving status. This slice preserves that direct call and its +ordering and migrates only periodic discovery to a dedicated deadline worker. +The service in `native-runtime/automatic-completion-reviews.ts` retains the +historical predicate, withdrawal, chat publication, and audit ownership. + +This duty is separate from obsolete policy reviews and subsequent reassessment. +A retired card's existing cancelled result reason is the durable trigger that +`decisionHasRetiredAutomaticReview` checks. Retirement is not finalization +completion. Reassessment and workspace barriers remain with their owners. + +## Exact obligation and producers + +The existing candidate query identifies: + +- A pending system confirmation with no user or agent creator. +- An applied in-review decision with completion_claim_incomplete, + completion_claim_conflict, or external_verification_required. +- Matching effect target, company, issue, source run and assessment. +- A low-risk agent_claim_policy contract with no assessment attention requests. +- The exact native-review idempotency key, native_completion_review target, + decision revision, and historical prompt first line. + +Current arbitration no longer creates completion-claim reviews by default. +Explicit attention uses actionable_attention_pending and keyed requests, which +this query excludes. The typed committer still supports historical input; it +registers the new company/issue notification before its decision/effect writes +inside its existing transaction. The notification serializes only those IDs. +Outer commit, rollback, and uncertain writes use the existing shared protocol. + +The production inserter of decisions and status effects is +`commitNativeStatusDecision`. No production updates of completion contracts or +work assessments were found; these records are immutable inputs. Existing +decision updates adjust lineage or applicationState within the committer. +Interaction response, dismissal, expiry, summary, and connection-intent updates +do not create this historical decision/effect lineage or reopen answered native +reviews. A generic authored card also fails the system-card predicate. +Startup covers historical rows. Independent SQL writers or concurrent older +controllers require explicit integration; process-local signals do not cover them. + +## Worker and invariants + +The existing scoped deadline helper reconstructs 100-row keyset pages, stores +interaction/company/issue IDs and deadlines, and processes fair batches of +20. Normal commit hints inspect only their company/issue. Settled history leaves +no recurring SQL or deadline. No table, migration, separate timer, or provider +branch is added. + +Discovery now has a raw internal path that propagates SQL errors. The public +manual lookup retains its logging/empty-list compatibility. The worker must not +mistake a failed lookup for successful reconstruction and discard a dirty scope. +The shared helper retains failed discovery with backoff. + +Each attempt rereads the exact candidate. Withdrawal retains the existing +coordinator/issue locks and additionally matches the selected kind, source run, +idempotency key, system author fields and payload in its conditional update. +A changed or answered card is not withdrawn from stale discovery. The existing +durable withdrawal reason, terminal chat publication and activity remain in the +same transaction. A failed withdrawal leaves a matching card pending and retains +its 30-second retry; another candidate can still settle. A removed or completed +obligation disarms. No age or retry-count cutoff discards a card. + +The worker shares admission, drain and standby gates. Startup while paused and +writes during pause retain a request without passive database reads. Shutdown +joins admitted transactions and removes registration. Only periodic orphan +reaping passes the new skip flag through the finalization options. Startup, +manual reconciliation and the direct active finalizer call remain compatible. + +## Evidence and limits + +The previous slice's owner/consumer corpus is the pre-change baseline. The +first server typecheck and the typecheck with lifecycle additions passed. +The complete owner/startup/timing selection passed all 135 tests without skips: +72 status-authority corpus cases, 60 startup cases and three timing cases. +The corpus includes its existing 53-fixture native/legacy matrix test. + +Both review owners run the same lifecycle contract against real PostgreSQL: +genuine/settled quiet, real committer activation, outer commit and rollback, +restart, drain, standby, paused startup, isolated cleanup failure, retry after +pause, concurrent card edits, concurrent answers, shutdown join, and 103-card +paging. Automatic cleanup additionally uses the raw failure path. The injected +discovery failure leaves a pending deadline and repairs after resume. Existing +manual tests preserve explicit attention, stronger authority, later task edits, +newer contracts, workspace failure and unread child-completion obligations. + +A rolled-back fixture added 10,000 completed interactions on the same issue and +restored one pending automatic review. The scoped lookup used existing indexes, +returned one candidate, and took 0.080 ms locally. SQL and plan are retained in +`/tmp/sleep45-query-plan.sql` and `.log`. This supports the access path, not a +production latency promise. Global startup reconstruction still scales with +historical data and joins. + +Affected consumer suites, final workspace checks and exact-head remote gates +remain required until their results are recorded. The host uses fresh disposable +PostgreSQL 17 test databases because its embedded harness is unavailable. + +Other finalization duties, earlier stack CI/review findings and final master +reconciliation remain in scope. This slice makes no claim about Cloud alarm +coverage, host/database suspension, or measured fleet savings. + +Final affected-consumer verification passed all 235 tests without skips; together +with the owner/startup/timing selection, 370 local tests passed. Workspace +typecheck and build passed. The two initial typechecks and final checks all +passed; no failed selection was discarded. Remote gates are pending. + + +## Earlier parent and acceptance (2026-10-29) + +A normal merge adopts the verified obsolete-policy cleanup parent and current +recovery source. All substantive owned edits and unrelated upstream blobs are +preserved. The complete database source tree equals the parent: no migration, +table or executable history change is added, and the parent 0407 history with +five newest snapshots remains. No new code comments are added. + +Source `0d9087775b54dc7272bcae51345ff690e3407235` passes **462 cases across seven complete +files, no skips**, in 49.57 seconds. The status-authority corpus runs both review +workers against the lifecycle contract. Fixtures nested inside its matrix case +are not counted again. Exact file counts: + +| Complete file | Passing cases | +| --- | ---: | +| src/services/native-runtime/external-chat-wait.integration.test.ts | 234 | +| src/__tests__/native-status-arbiter-corpus.test.ts | 72 | +| 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 | 63 | +| src/services/native-runtime/native-finalization-reconciler-timing.test.ts | 3 | + +Five separate fresh PostgreSQL 17 databases are migrated before corpus, recovery, +finalizer, external-chat and governed-milestone fixture setup. All selected suites +use their supported external URL path; none is skipped because this host cannot +allocate another embedded cluster. Database build, server typecheck/build, module +boundaries, target migration order and all five fresh parent-0407 migrations pass. +The previous slice's circular-scope-payload failure remains recorded in its +owning audit and is not erased by these passes. + +These checks establish current cleanup, authority and affected-consumer behavior, +not deployed host/database suspension or external alarms. Actual landed parent +ancestry and fresh published-head review and CI remain required. + + +## Current migration409 parent and complete acceptance + +Normal integration `27378007cd64c308f41940097a999d51201b8ba8` preserves the +complete original runtime patch and all unrelated upstream blobs. The entire +database source tree equals the current parent, with 0409 history and five +newest snapshots unchanged. No migration or table is added. + +All **462 cases across seven complete files pass without skips**, in 49.75 +seconds. The selection and per-file counts match the preceding table. Five +separate fresh PostgreSQL 17 databases migrate through 0409 before setup. +Database build, server types/build, module boundaries and exact-head migration +order pass. All suites use their supported external-database fixture path; no +source adaptation or skipped suite is required. Earlier failures remain recorded. + +This proves current worker and affected consumer behavior, not deployed +host/database suspension or Cloud alarms. Publication stays held until the +actual prerequisite merge and squash ancestry reconciliation. Fresh current-head +CI and reviews remain required before merge. + + +## Published obsolete-review parent and actual preservation ancestry + +Normal merge `c0f3ada4c018943982731635c26d37c7bb4b9ad9` adopts published parent +`45cb0d9ccc9b0c37068f56a709f16572bd7b7ddc`. It carries the actual cancelled-preservation +squash ancestry and parent audit only. Complete server, package, UI and test +trees remain identical to accepted source `27378007cd64c308f41940097a999d51201b8ba8`, +including the whole 0409 database tree. The 462-case acceptance retains its +original scope. Publication remains held until the actual obsolete-review merge +and ancestry reconciliation. Fresh current-head CI and review remain required. + + +## Inherited corpus repeatability correction + +Published obsolete-review parent `9aa67199da64d700fd6230829d38f2ef522a07fe` +carries the verified two-value fixture prefix correction. Normal integration +`3f4d35ac0154aedc76cc39486d3758c73b1e6ff4` changes only those two corpus +fixture values and parent documentation/ancestry. All production source, other +test files and the complete 0409 database tree remain unchanged. The parent's +audit retains the collision reproduction and rejected foreign-key cleanup +attempt. No assertion, timing limit, teardown or worker behavior changes. + +All **462 distinct cases in seven complete files pass without skips**, in +49.31 seconds. The full 72-case corpus then passes again on the same database +in 12.95 seconds. Three preexisting PSC/OUT/sentinel rows remain unchanged in +the inspected ID/name/prefix fields after both runs. Each invocation appends +only its two fixture companies with distinct prefixes. The repeat is not added again +to the distinct case count. Five fresh PostgreSQL 17 fixtures migrate through +0409; database build, server types/build, boundaries and migration order pass. + +The actual prerequisite must still merge before publication. Fresh published-head +CI and reviews remain required. No deployed sleep or external-alarm claim is made. + + +## Actual obsolete-review squash ancestry + +Parent #17679 landed as `dfd5b1ac6205b6ec8f56752dfc5d3f9c02c26b6d`. Its whole tree +equals reviewed parent `9aa67199da64d700fd6230829d38f2ef522a07fe`, already present +in this child history. Normal merge `8c5c5da563e32b9a5520a4f8ddefc38c29b4755e` +resolves inspected duplicate squash-ancestry conflicts and preserves the entire +accepted child tree. Complete server, package, UI and test trees remain +identical to the 462-case source and same-database 72-case repeat. The complete +0409 database source remains exactly parent-identical. Fresh published-head +CI and reviews remain required. + + +## Current-head chat CI correction + +The published 40f head fails the initial passive native-presentation idle +observation in chat shard 1. The owner audit records the original failure, +unchanged local baseline, controlled red/green comparison and its limits: +[Native presentation startup observation](2026-10-25-github-terminal-response-sleep-audit.md#ci-native-presentation-startup-observation). +The permanent correction changes one observation from the default one-second +window to the existing five-second settlement bound. Production code and +business/worker deadlines do not change. All quietness, rollback and committed +settlement assertions remain. With the probe removed, all 274 selected GitHub +agent-review cases pass in 232.06 seconds. The earlier 462-case selection and +72-case repeat retain identical production and selected test sources. Fresh +CI and review are required on the updated published head. + + +## Independent master cleanup assertion preserved + +Master advanced from the actual obsolete-review squash to +`49fe8b87a49aa5a8985a08dd3d5ce57bdd24756c`. The sole change is an assertion +that company cleanup removes company skills. Normal merge +`4dbec260d36cae02bd73281bbdaac406ee989010` preserves both complete binary +patches: this PR's full fix and the independent upstream assertion. All production +source and the accepted GitHub integration file remain unchanged. The complete +cleanup-removal service suite passes all six cases in 8.16 seconds on a fresh +0409 PostgreSQL 17 fixture. The whole database source is unchanged. Current-base +migration order and fresh published-head CI/reviews remain required. diff --git a/doc/plans/2026-10-25-github-terminal-response-sleep-audit.md b/doc/plans/2026-10-25-github-terminal-response-sleep-audit.md index e908edceb58..9cc1a272828 100644 --- a/doc/plans/2026-10-25-github-terminal-response-sleep-audit.md +++ b/doc/plans/2026-10-25-github-terminal-response-sleep-audit.md @@ -377,3 +377,46 @@ Original producer, quietness, failure-isolation, skipped-suite and presentation results above retain their original scopes. Fresh exact-head CI and Greptile 5/5 with no actionable findings remain required. This does not establish a full local repository result, live GitHub delivery or physical host/database sleep. + + +## CI: native presentation startup observation + +PR #17680 head `40f84082526d6ba405f206475c919c3b4b67ec4c` fails chat shard +1 in run 40814818307, job 123253388098. The passive native-presentation case +expires its initial default one-second idle poll with no deadline and two active +tracked operations. The aggregate verify check fails because of that shard. +This is retained as a real CI failure; no unchanged-head retry was issued. + +The scoped worker's minimum pass interval is also one second. The initial poll +was added after the earlier coalescing correction documented above and still +used the default, while neighboring settlement observations use five seconds. +The unchanged local focused pair passes, so there is no claim of an unmodified +local reproduction of CI scheduling. + +A temporary controlled fixture commits a real milestone hint during the initial +registered run and holds the second registered run attempt for 600 ms after +delegating to the existing implementation. That attempt may defer at the +unchanged one-second gate, retaining a coalesced follow-up. Both old observations +fail at the original one-second bound. With the identical probe, both pass when +the initial observation uses the existing five-second bound. The controlled +failures have a pending deadline and one active token; CI saw no deadline and +two active tokens. This proves that the observation window cannot reliably fit +allowed startup follow-up, not the exact unrecorded CI token-owner timeline. + +The probe is removed byte for byte before final acceptance. The permanent +change is one argument on the initial poll. It still requires a null deadline +and zero active work, then proves two virtual minutes with zero SQL, transaction +or provider writes. Rollback, committed settlement and final drain assertions +are unchanged. Production code, scheduler cadence, retry/business deadlines and +the overall test budget are unchanged. Only this eventual-startup observation +uses the established five-second bound. + +The no-probe focused pair passes. The complete named GitHub agent-review +workflow then passes all **274 selected cases** in 232.06 seconds on fresh +PostgreSQL 17. The other 1507 cases in the large integration file are outside +that named selection, not counted as acceptance. Five dedicated fresh 0409 +fixtures isolate baseline, controlled red, controlled green, focused and workflow +runs. No embedded CI-environment equivalence is claimed. The source file hash +for permanent acceptance is +`d0dbef76741f03862dfef0a42338f97d5ea7a9f36d4b465c7538a5eaacd65b2c`. +Fresh published-head CI and review remain required. diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 1fba38d0cf6..3c2bc210f31 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -4114,7 +4114,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); try { await githubResponseCommentService(db, f.providerFetch).registerTerminalResponseWork(coordinator, () => true).ready; - await expect.poll(() => ({ next: coordinator.nextWakeAt(), active: idleWorkSnapshot().active })) + await expect.poll(() => ({ next: coordinator.nextWakeAt(), active: idleWorkSnapshot().active }), { timeout: 5000 }) .toEqual({ next: null, active: 0 }); expect(f.responseWrites).toHaveLength(baseline); vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); diff --git a/server/src/__tests__/native-status-arbiter-corpus.test.ts b/server/src/__tests__/native-status-arbiter-corpus.test.ts index da5da51bc23..8fea007c2e5 100644 --- a/server/src/__tests__/native-status-arbiter-corpus.test.ts +++ b/server/src/__tests__/native-status-arbiter-corpus.test.ts @@ -1,6 +1,6 @@ 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 { dismissAutomaticCompletionReviews, registerAutomaticCompletionReviewWork } 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"; @@ -2050,7 +2050,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { } - describe("obsolete policy review scheduling", () => { + describe.each(["obsolete policy", "automatic completion"] as const)("%s review scheduling", kind => { let coordinator: ReturnType | undefined; let admitted = true; const errors: unknown[] = []; @@ -2061,13 +2061,29 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { afterEach(async () => { await coordinator?.stop(); coordinator = undefined; vi.useRealTimers(); vi.restoreAllMocks(); - await dismissObsoleteNativePolicyReviews(db); + if (kind === "obsolete policy") await dismissObsoleteNativePolicyReviews(db); + else await dismissAutomaticCompletionReviews(db); expect(errors).toEqual([]); }); + async function seedReview(options: Parameters[0] = {}) { + if (kind === "obsolete policy" || options.genuine) return seedPolicyReview(options); + return seedPolicyReview({ ...options, commit: async input => { + const [assessment] = await db.select().from(workAssessments).where(eq(workAssessments.id, input.assessmentId)); + await db.update(completionContracts).set({ risk: "low", completionAuthority: "agent_claim_policy" }) + .where(eq(completionContracts.id, assessment!.contractId)); + await db.update(workAssessments).set({ assessmentJson: { ...assessment!.assessmentJson, attentionRequests: [] } }) + .where(eq(workAssessments.id, input.assessmentId)); + return (options.commit ?? commitNativeStatusDecision)({ ...input, decision: { ...input.decision, + reasonCode: "completion_claim_incomplete", effects: [{ kind: "bind_reviewer", ownerUserId: null, + prompt: "Review the persisted native-run evidence and confirm whether this issue may be completed." }], + } }); + } }); + } async function start() { coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => admitted, canReconcile: () => admitted, onError: error => errors.push(error) }); - await registerObsoletePolicyReviewWork(db, coordinator, () => admitted).ready; + const register = kind === "obsolete policy" ? registerObsoletePolicyReviewWork : registerAutomaticCompletionReviewWork; + await register(db, coordinator, () => admitted).ready; } async function flush() { await vi.waitFor(async () => { @@ -2089,9 +2105,10 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { } 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]); + const genuine = await seedReview({ genuine: true }); + const settled = await seedReview(); + if (kind === "obsolete policy") await dismissObsoleteNativePolicyReviews(db, [settled.runId]); + else await dismissAutomaticCompletionReviews(db, settled.issueId); await start(); await flush(); expect((await card(genuine.interaction.id)).status).toBe("pending"); expect((await card(settled.interaction.id)).status).toBe("cancelled"); @@ -2100,10 +2117,10 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { 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(); + const created = await seedReview(); await flush(); expect((await card(created.interaction.id)).status).toBe("cancelled"); - expect((await issueService(db).getById(created.issueId))!.status).toBe("in_progress"); + expect((await issueService(db).getById(created.issueId))!.status).toBe(kind === "obsolete policy" ? "in_progress" : "in_review"); expect((await db.select().from(statusDecisions).where(eq(statusDecisions.id, created.decision.id)))[0]).toEqual(created.decision); await quiet(); }); @@ -2112,7 +2129,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { await start(); await flush(); const entered = gate(), release = gate(); let issueId = ""; - const creating = seedPolicyReview({ commit: async input => db.transaction(async tx => { + const creating = seedReview({ 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; @@ -2137,7 +2154,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { 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(); + const created = await seedReview(); if (mode === "restart") await start(); else { await vi.advanceTimersByTimeAsync(1000); @@ -2149,14 +2166,14 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { }); it("retains startup discovery while admission is closed", async () => { - const created = await seedPolicyReview(); admitted = false; + const created = await seedReview(); 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 first = await seedReview(), second = await seedReview(); const write = vi.spyOn(db, "transaction").mockRejectedValueOnce(new Error("injected obsolete cleanup failure")); await start(); await vi.waitFor(async () => { @@ -2173,7 +2190,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { }); it("does not withdraw a review edited after discovery", async () => { - const created = await seedPolicyReview(); + const created = await seedReview(); 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." } }) @@ -2186,7 +2203,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { }); it("joins admitted cleanup on shutdown and reconstructs quiet state on restart", async () => { - const created = await seedPolicyReview(), entered = gate(), release = gate(); + const created = await seedReview(), 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); @@ -2199,9 +2216,34 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => { await start(); await flush(); await quiet(); }); + + it("retains discovery after a database error and retries on resume", async () => { + const created = await seedReview(); + const read = vi.spyOn(db, "select").mockImplementationOnce(() => { throw new Error("injected review discovery failure"); }); + await start(); + await vi.waitFor(() => expect(errors.length).toBeGreaterThan(0)); + read.mockRestore(); expect(coordinator!.nextWakeAt()).not.toBeNull(); + expect((await card(created.interaction.id)).status).toBe("pending"); + errors.length = 0; admitted = false; await quiet(); admitted = true; + await vi.advanceTimersByTimeAsync(30_000); await flush(); + expect((await card(created.interaction.id)).status).toBe("cancelled"); await quiet(); + }); + + it("preserves a newly answered card between discovery and withdrawal", async () => { + const created = await seedReview(); + const original = db.transaction.bind(db); + const write = vi.spyOn(db, "transaction").mockImplementationOnce(async (callback, config) => { + await db.update(issueThreadInteractions).set({ status: "accepted" }).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("accepted"); + expect((await issueService(db).getById(created.issueId))!.status).toBe("in_review"); 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()); + for (let index = 0; index < 103; index++) created.push(await seedReview()); await start(); await flush(); for (const value of created) expect((await card(value.interaction.id)).status).toBe("cancelled"); await quiet(); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index bfa9d07d7e1..3ed2a53a96c 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() })), + registerAutomaticCompletionReviewWork: 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() })), @@ -555,6 +556,13 @@ describe("startServer feedback export wiring", () => { process.env.BETTER_AUTH_SECRET = "test-secret"; }); + it.each([false, true])("registers automatic completion review cleanup only when maintenance is enabled (%s)", async enabled => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); + await startServer(); await settleStartupHousekeeping(); + expect(heartbeatServiceMock.registerAutomaticCompletionReviewWork).toHaveBeenCalledTimes(enabled ? 1 : 0); + if (enabled) expect(heartbeatServiceMock.registerAutomaticCompletionReviewWork.mock.calls[0][1]()).toBe(true); + }); + 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(); @@ -981,7 +989,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, skipObsoletePolicyReviews: true }); + expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true }); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(retiredDetector).not.toHaveBeenCalled(); } finally { diff --git a/server/src/index.ts b/server/src/index.ts index 162693b9cb1..788bf90b1f9 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.registerAutomaticCompletionReviewWork(deliveryWork, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; if (config.heartbeatSchedulerEnabled) await heartbeat.registerObsoletePolicyReviewWork(deliveryWork, () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; if (config.heartbeatSchedulerEnabled) await heartbeat.registerCancelledWorkspacePreservationWork(deliveryWork, @@ -1958,7 +1960,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, skipObsoletePolicyReviews: true }) + .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true }) .then(() => heartbeat.promoteDueScheduledRetries()) .then(async (promotion) => { await heartbeat.resumeQueuedRuns(); diff --git a/server/src/services/automatic-completion-review-notifications.ts b/server/src/services/automatic-completion-review-notifications.ts new file mode 100644 index 00000000000..dfe8ad92a0e --- /dev/null +++ b/server/src/services/automatic-completion-review-notifications.ts @@ -0,0 +1,8 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type AutomaticCompletionReviewScope = { companyId: string; issueId: string }; + +export async function notifyAutomaticCompletionReviews(transaction: object, scope: AutomaticCompletionReviewScope) { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.automaticCompletionReviews, + JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId })); +} diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index ec33ee21b29..383a917a571 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 = { + automaticCompletionReviews: "automatic-completion-reviews", obsoletePolicyReviews: "obsolete-policy-reviews", cancelledWorkspacePreservation: "cancelled-workspace-preservation", instructionCollection: "instruction-collection", diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index f42994f7aa8..c3fc549fef2 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +import { registerAutomaticCompletionReviewWork } from "./native-runtime/automatic-completion-reviews.js"; 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"; @@ -4086,6 +4087,8 @@ export function heartbeatService( triggerIssueMonitor, reportRunActivity: clearDetachedRunWarning, + registerAutomaticCompletionReviewWork: (coordinator: Parameters[1], canRun: () => boolean) => + registerAutomaticCompletionReviewWork(db, coordinator, () => canRun() && readTaskDrain(new Date()) === null), registerObsoletePolicyReviewWork: (coordinator: Parameters[1], canRun: () => boolean) => registerObsoletePolicyReviewWork(db, coordinator, () => canRun() && readTaskDrain(new Date()) === null), registerCancelledWorkspacePreservationWork: ( diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 4ae6637092e..bca38c1583e 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; skipObsoletePolicyReviews?: boolean }) { + async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean }) { if (!opts?.skipCancelledWorkspacePreservation) await preserveCancelledNativeWorkspaces(db); const staleThresholdMs = opts?.staleThresholdMs ?? 0; const now = new Date(); @@ -1966,6 +1966,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD // mode, never the current feature flag. await reconcileNativeFinalizations(db, undefined, { skipObsoletePolicyReviews: opts?.skipObsoletePolicyReviews, + skipAutomaticCompletionReviews: opts?.skipAutomaticCompletionReviews, environmentRuntime, pluginWorkerManager: dependencies.pluginWorkerManager, onWorkspaceSettled: settleRecoveredNativeWorkspace, diff --git a/server/src/services/native-runtime/automatic-completion-reviews.ts b/server/src/services/native-runtime/automatic-completion-reviews.ts index 6f12c75b7dc..9f7eea8c326 100644 --- a/server/src/services/native-runtime/automatic-completion-reviews.ts +++ b/server/src/services/native-runtime/automatic-completion-reviews.ts @@ -1,4 +1,8 @@ -import { and, eq, inArray, isNull, sql } from "drizzle-orm"; +import type { AutomaticCompletionReviewScope } from "../automatic-completion-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 { completionContracts, issueThreadInteractions, @@ -24,7 +28,7 @@ const automaticPrompt = "Review the persisted native-run evidence and confirm whether this issue may be completed."; /** Identify only proven system fallback cards; this lookup never changes state. */ -export async function findAutomaticCompletionReviews(db: Db, issueId?: string) { +function findAutomaticReviewCandidates(db: Db, options: { issueId?: string; companyId?: string; id?: string; after?: string } = {}) { return db .select({ interaction: issueThreadInteractions, decision: statusDecisions }) .from(issueThreadInteractions) @@ -83,17 +87,20 @@ export async function findAutomaticCompletionReviews(db: Db, issueId?: string) { sql`${issueThreadInteractions.payload}->'target'->>'key' = 'native_completion_review'`, sql`${issueThreadInteractions.payload}->'target'->>'revisionId' = ${statusDecisions.id}::text`, sql`split_part(${issueThreadInteractions.payload}->>'prompt', E'\n', 1) = ${automaticPrompt}`, - ...(issueId ? [eq(issueThreadInteractions.issueId, issueId)] : []), + options.issueId ? eq(issueThreadInteractions.issueId, options.issueId) : undefined, + options.companyId ? eq(issueThreadInteractions.companyId, options.companyId) : undefined, + options.id ? eq(issueThreadInteractions.id, options.id) : undefined, + options.after ? gt(issueThreadInteractions.id, options.after) : undefined, ), ) - .limit(100) - .catch((err) => { - logger.warn( - { err }, - "Automatic completion review lookup failed; will retry", - ); - return []; - }); + .orderBy(asc(issueThreadInteractions.id)).limit(100); +} + +export async function findAutomaticCompletionReviews(db: Db, issueId?: string) { + return findAutomaticReviewCandidates(db, { issueId }).catch((err) => { + logger.warn({ err }, "Automatic completion review lookup failed; will retry"); + return []; + }); } /** Narrow, replay-safe retirement. Explicit requests and answered cards are immutable here. */ @@ -101,7 +108,10 @@ export async function dismissAutomaticCompletionReviews( db: Db, issueId?: string, ) { - const candidates = await findAutomaticCompletionReviews(db, issueId); + await dismissCandidates(db, await findAutomaticCompletionReviews(db, issueId)); +} + +async function dismissCandidates(db: Db, candidates: Awaited>) { for (const { interaction, decision } of candidates) { const publications: ActivityPublication[] = []; try { @@ -147,6 +157,10 @@ export async function dismissAutomaticCompletionReviews( eq(issueThreadInteractions.companyId, issue.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(); @@ -213,3 +227,33 @@ export async function decisionHasRetiredAutomaticReview( .limit(1); return rows.length > 0; } + +export function registerAutomaticCompletionReviewWork(db: Db, coordinator: ReturnType, canRun: () => boolean) { + type Entry = { id: string; companyId: string; issueId: string; dueAt: number }; + async function load(scope?: AutomaticCompletionReviewScope, signal?: AbortSignal): Promise { + const entries = new Map(); + let after: string | undefined; + while (canRun() && !signal?.aborted) { + const rows = await findAutomaticReviewCandidates(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("Automatic completion review discovery stopped"); + } + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.automaticCompletionReviews, batchSize: 20, + decodeScope: (hint: string): AutomaticCompletionReviewScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry: Entry, scope?: AutomaticCompletionReviewScope) => !scope || entry.companyId === scope.companyId && entry.issueId === scope.issueId, + load, + check: async (entry: Entry, signal?: AbortSignal) => { + const candidates = await findAutomaticReviewCandidates(db, entry); + if (!canRun() || signal?.aborted) throw new Error("Automatic completion review cleanup stopped"); + await dismissCandidates(db, candidates); + const pending = await findAutomaticReviewCandidates(db, entry); + return pending.length ? { ...entry, dueAt: Date.now() + 30_000 } : null; + }, + }, canRun); +} 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 1de6bd22b01..255dad56f3d 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 @@ -32,6 +32,14 @@ describe("native finalization reconciliation timing", () => { timing.phase.mockClear(); timing.finish.mockClear(); }); + it("omits automatic review discovery when its worker owns that duty", async () => { + timing.phase.mockClear(); + const db = { select: vi.fn(() => { throw new Error("candidate scan reached"); }) }; + await expect(reconcileNativeFinalizations(db as never, ["run"], { skipAutomaticCompletionReviews: true })).rejects.toThrow("candidate scan reached"); + expect(timing.phase.mock.calls.map(([phase]) => phase)).toEqual(["legacy_workspace_exports", "obsolete_policy_reviews", "candidate_scan"]); + 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 e93e0355347..93959ef1e4a 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -638,6 +638,7 @@ export async function claimNativeSessionResumptions(input: { type NativeReconciliationOptions = { skipObsoletePolicyReviews?: boolean; + skipAutomaticCompletionReviews?: boolean; pluginWorkerManager?: WorkspaceWorkers; environmentRuntime?: EnvironmentRuntimeService; onWorkspaceSettled?: (input: { @@ -730,13 +731,15 @@ async function reconcileNativeFinalizationsPass( 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) - .where(inArray(nativeRunFinalizations.runId, runIds)); - for (const scope of scopes) await dismissAutomaticCompletionReviews(db, scope.issueId); - } else { - await dismissAutomaticCompletionReviews(db); + if (!options.skipAutomaticCompletionReviews) { + timing.phase("automatic_completion_reviews"); + if (runIds?.length) { + const scopes = await db.select({ issueId: nativeRunFinalizations.issueId }).from(nativeRunFinalizations) + .where(inArray(nativeRunFinalizations.runId, runIds)); + for (const scope of scopes) await dismissAutomaticCompletionReviews(db, scope.issueId); + } else { + await dismissAutomaticCompletionReviews(db); + } } timing.phase("candidate_scan"); const now = new Date(); diff --git a/server/src/services/native-runtime/status-decision-committer.ts b/server/src/services/native-runtime/status-decision-committer.ts index 1010b6b8b6a..f2c780837e1 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 { notifyAutomaticCompletionReviews } from "../automatic-completion-review-notifications.js"; import { notifyObsoletePolicyReviews } from "../obsolete-policy-review-notifications.js"; import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyInstructionCollection } from "../instruction-collection-notifications.js"; @@ -1983,6 +1984,7 @@ export async function commitNativeStatusDecision(input: { } await notifyChatRunMilestones(tx, input); await notifyObsoletePolicyReviews(tx, input); + await notifyAutomaticCompletionReviews(tx, input); if (!decisionRow) { const latestDecisionVersion = await tx .select({