diff --git a/cli/src/__tests__/company-import-export-e2e.test.ts b/cli/src/__tests__/company-import-export-e2e.test.ts index bf327b0915b..5d49d4a8b84 100644 --- a/cli/src/__tests__/company-import-export-e2e.test.ts +++ b/cli/src/__tests__/company-import-export-e2e.test.ts @@ -13,13 +13,14 @@ import { } from "./helpers/embedded-postgres.js"; import { createStoredZipArchive } from "./helpers/zip.js"; import { removeTestHome } from "../../../tests/helpers/remove-test-home.js"; +import { PAPERCLIP_RUNTIME_INFO_FILENAME, readRuntimeInfo } from "../runtime-info.js"; const execFileAsync = promisify(execFile); type ServerProcess = ReturnType; -async function getAvailablePort(): Promise { +async function reserveRequestedPort(): Promise<{ port: number; release: () => Promise }> { return await new Promise((resolve, reject) => { - const server = net.createServer(); + const server = net.createServer((socket) => socket.destroy()); server.unref(); server.on("error", reject); server.listen(0, "127.0.0.1", () => { @@ -29,9 +30,11 @@ async function getAvailablePort(): Promise { return; } const { port } = address; - server.close((error) => { - if (error) reject(error); - else resolve(port); + resolve({ + port, + release: () => new Promise((resolveClose, rejectClose) => { + server.close((error) => error ? rejectClose(error) : resolveClose()); + }), }); }); }); @@ -247,7 +250,8 @@ async function runCliJson( } async function waitForServer( - apiBase: string, + instanceId: string, + runtimeInfoPath: string, child: ServerProcess, output: { stdout: string[]; stderr: string[] }, ) { @@ -259,18 +263,22 @@ async function waitForServer( ); } - try { - const res = await fetch(`${apiBase}/api/health`); - if (res.ok) return; - } catch { - // Server is still starting. + const runtime = readRuntimeInfo(instanceId, runtimeInfoPath); + if (runtime && runtime.pid === child.pid) { + const apiBase = `http://127.0.0.1:${runtime.port}`; + try { + const res = await fetch(`${apiBase}/api/health`, { signal: AbortSignal.timeout(1_000) }); + if (res.ok) return apiBase; + } catch { + + } } await new Promise((resolve) => setTimeout(resolve, 250)); } throw new Error( - `Timed out waiting for ${apiBase}/api/health.\nstdout:\n${output.stdout.join("")}\nstderr:\n${output.stderr.join("")}`, + `Timed out waiting for the child server recorded in ${runtimeInfoPath}.\nstdout:\n${output.stdout.join("")}\nstderr:\n${output.stderr.join("")}`, ); } @@ -283,6 +291,7 @@ describeEmbeddedPostgres("paperclipai company import/export e2e", () => { let cliShellHome = ""; let paperclipInstanceId = ""; let serverProcess: ServerProcess | null = null; + let requestedPortReservation: Awaited> | null = null; let tempDb: Awaited> | null = null; beforeAll(async () => { @@ -297,9 +306,11 @@ describeEmbeddedPostgres("paperclipai company import/export e2e", () => { tempDb = await startEmbeddedPostgresTestDatabase("paperclip-company-cli-db-"); - const port = await getAvailablePort(); + + + requestedPortReservation = await reserveRequestedPort(); + const { port } = requestedPortReservation; writeTestConfig(configPath, tempRoot, port, tempDb.connectionString); - apiBase = `http://127.0.0.1:${port}`; const repoRoot = path.resolve(path.dirname(fileURLToPath(import.meta.url)), "../../.."); const output = { stdout: [] as string[], stderr: [] as string[] }; @@ -324,11 +335,18 @@ describeEmbeddedPostgres("paperclipai company import/export e2e", () => { output.stderr.push(String(chunk)); }); - await waitForServer(apiBase, child, output); + apiBase = await waitForServer( + paperclipInstanceId, + path.join(paperclipHome, "instances", paperclipInstanceId, PAPERCLIP_RUNTIME_INFO_FILENAME), + child, + output, + ); + expect(new URL(apiBase).port).not.toBe(String(port)); }, 60_000); afterAll(async () => { await stopServerProcess(serverProcess); + await requestedPortReservation?.release(); await tempDb?.cleanup(); if (tempRoot) { removeTestHome(tempRoot); 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 d800eeea39f..3b5b3b9f45b 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -2826,3 +2826,53 @@ null deadline or unfinished startup batch. Their settlement predicate now includ zero active work. All 18 affected cases pass against the full suite's retained history. Fresh remote acceptance remains pending; these are recorded test repairs, not a waiver of failed gates or a change to production batch behavior. + +## Slice 48 accepted; slice 49 approval producer checkpoint + +#17696 at `fad35f2d7ae6bdbb85e709f1bab4633d197fd108` has all 53 checks +passing, Greptile 5/5 at that commit, and zero unresolved review threads. Its PR +body now records those results and preserves the earlier local/remote failures. +It remains draft and unmerged. Evidence is in /tmp/sleep48-final-gates.json, +/tmp/sleep48-final-comments.json and /tmp/sleep48-final-review-threads.json. + +Slice 49 has a guarded recovery owner and scoped worker, plus verified approval +producer wakes. The 171-case owner suite and 70 existing approval/budget/queue/ +retention consumer cases pass, as does server typecheck. See its audit for the +reproduced missed-wake failures and repaired test stub. Interaction producer +coverage, discovery cost, startup wiring, periodic removal and full acceptance +remain outstanding. No slice 49 PR or whole-instance sleep is claimed. Earlier +stack gates, master reconciliation and every remaining original duty stay in scope. + +Slice 49 lifecycle checkpoint: the real agent-resume missed wake was reproduced +and repaired through the existing lifecycle hook. Milestone/finalization scopes +now feed review recovery, and pending-effect settlement supplies its task identity. +The 172-case owner suite and 162 lifecycle/milestone/finalization consumer cases +pass, with server types passing. The owner audit lists remaining interaction, task +authority and coordinator producer evidence. Registration, periodic removal, query +cost, full verification and publication remain outstanding; the full stack scope +and older unresolved gates remain unchanged. + +Slice 49 interaction checkpoint: five interaction misses plus task-status and +OAuth-denial misses are reproduced and fixed. Task-scoped notifications cover +existing resolution transactions and direct OAuth/connection cleanup writers. +The complete owner suite passes 180 cases; interaction/telemetry/satisfied-connection/ +generic-OAuth consumers pass 249 cases, and server types pass. Skipped telemetry, +external pool-fixture and callback-narrowing failures and repairs are preserved +in the owner audit. Execution-state-only restoration, real coordinator release, +query cost, registration/periodic removal and final acceptance remain. No PR yet, +no merge, and the full original stack scope remains open. + + +## Slice 49 startup and cost checkpoint + +Governed-review response recovery now owns both periodic repair entrances and +registers through heartbeat maintenance. All 255 owner/startup cases, workspace +typecheck and build pass. Pending-policy release, real coordinator retry, SQL-quiet +drain, contention/resume and shutdown/restart are covered. A run-first concurrent +receipt index removes the measured quadratic history lookup (about 3,000 ms to +45 ms on 10,000 selected historical responses). The 10,000 outstanding-item +fixture remains proportional: 101 queries, 2.25 seconds, about 2.05 MB; it stops +between pages on drain. Fresh migration and interrupted/concurrent replay pass. +The audit preserves the initial query, migration and test-clock failures. Full +local tests are running; remote acceptance, older stack findings, master +reconciliation and every remaining original duty are still open. Nothing merged. diff --git a/doc/plans/2026-10-26-governed-review-response-sleep-audit.md b/doc/plans/2026-10-26-governed-review-response-sleep-audit.md new file mode 100644 index 00000000000..0fd24112942 --- /dev/null +++ b/doc/plans/2026-10-26-governed-review-response-sleep-audit.md @@ -0,0 +1,623 @@ +# Governed-review response recovery: sleep migration audit + +Status: raw owner contract and initial worker verified; producer wiring and periodic removal remain incomplete. +Inspected parent: `fad35f2d7a` (slice 48, draft PR #17696). + +## One reason for work + +A native run can commit a successful response while the task remains in review +under an earlier server-created completion-review gate. Its saved response and +selected files still need one authorized comment/publication. Current periodic +native finalization reconciliation repairs that presentation even when no source +changes. This slice owns only that owed presentation, not review resolution, +status reassessment, provider execution, or pending status-effect delivery. + +The durable sources already exist: native_run_finalizations, native_run_results, +work_assessments, status_decisions, heartbeat_runs, and the original +issue_thread_interactions review gate. No new queue table is needed. + +## Existing owner and contracts + +- `native-chat-review-presentation.ts` owns proof preparation, restoration, + current authorization, contention handling, and the materialized receipt. +- `native-run-finalizer.ts` owns `materializeCommittedReviewResponse` and + `repairCommittedNativeReviewResponse`. The former currently catches operational + failures and returns no outcome. A scheduled consumer needs a narrow raw + outcome/error contract; retain current behavior for existing foreground callers. +- `issues.ts` addComment owns the issue transaction, nonblocking reauthorization, + selected-file attachment, comment deduplication, and publication intent. + Recovery must continue through it, never insert a parallel comment/outbox path. +- `native-finalization-reconciler.ts` repairs governed presentation before the + superseded-status early return. A later task decision must not erase an older, + still-authorized response. The already-committed branch of `finalizeNativeRun` + also restores/materializes it; periodic migration must address both entrances. +- Fresh finalization still presents immediately. Explicit/startup recovery retains + its behavior unless the scheduler owns that call explicitly. + +Receipt semantics: `hasMaterializedNativeReviewResponse` recognizes an agent +comment for the exact company/task/run with the presentation authorization reason +and matching current decision proof. Soft-deleted comments remain receipts. +Restoration already avoids rewriting identical metadata or signaling another +wake. Preserve those properties. + +## Current authority that must remain in the owner + +Authorization checks committed coordinator/result/assessment/decision bindings; +applied governed_response_waiting decision; successful run without held native +ownership; in_review task assigned to the same agent; unchanged response context; +agent not paused/terminated/pending approval; and current external destination +and principal authority. Runtime health status error alone does not revoke an +already completed response. + +The proof requires an accepted yielded response_wake result and matching server +assessment, successful terminal evidence, no attention requests, and a nonempty +summary/idempotency key. Its prior pending review gate must be server-created, +older than this run, from another source run, correctly bound to the native +completion-review revision and delivered bind_reviewer effect. The hashes bind +semantic result, gate policy, context, and summary. Competing pending interactions, +pending/revision_requested linked approvals, or a run-owned provider prompt can +prevent presentation. These checks are publication authority, not scheduler policy. + +## Planned worker boundary + +Use the existing scoped deadline helper and delivery coordinator. Discover only +structurally eligible committed governed responses without a materialized receipt; +revalidate exact durable state and current authorization through the owner before +presentation. Failed I/O/contention retains retry ownership. Settled or currently +unauthorized rows become dormant until a relevant source change. Preserve saved +coordinator retry deadlines, active claims, drain/resume, and shutdown. + +Before implementation, establish the scheduled owner guard under its existing +lock: a previously discovered row must not bypass a newly acquired finalization +claim or a moved-later retry deadline. Keep issue/coordinator/run lock ordering +compatible with addComment's nonblocking recheck. Do not conflate authorization +denial with operational errors. Preserve deleted receipts and idempotent replay. + +Measure discovery with selected history and a large unpresented backlog. Slice +48 showed why per-page limits do not make a company-wide reconstruction cheap. +Normal run writes must retain task/run scope; broad reloads are reserved for +changes whose authority can actually affect the broader scope. + +## Producer audit still required before retiring the backstop + +Already identified reusable boundaries: + +- Finalization/result/decision/effect settlement: native-finalization-notifications + and the existing milestone boundary. Existing review proof restoration signals + only a real metadata change. +- Run status, ownership release, task status/assignment, conversation lineage, + endpoint/principal configuration, publication outcomes, and agent resume: + inspect existing milestone/finalization producers and extend one owning boundary + per dependency. Do not scatter caller recovery wrappers. +- Publication changes already use notifyChatPublicationWork with task scope only + when prompt authority or delivery outcome changes; transport lease bookkeeping + intentionally does not invalidate presentation. + +Concrete gaps found in this audit: + +- `writeApprovalChange` currently signals approval retention and queue seeding, + not governed-response recovery. Resolving or unlinking a competing linked + approval can make an existing proof eligible. Audit approve/reject/revise, + resubmit, link/unlink, bulk linking, retention, and alternative writers. Prefer + an approval identity/task scope over company-wide reconstruction. +- The proof rejects *any* competing pending interaction, while current milestone + notifications often apply only to questions and confirmations. Resolving a + connection intent or another kind may restore eligibility. Inventory common + interaction mutation owners and direct cleanup/recovery writers before choosing + a notification hook; do not assume confirmation coverage is sufficient. +- Run-owned prompt publication states can retain or release the response slot. + Confirm every relevant publication transition reaches the existing task-scoped + publication boundary. + +This is not a completed producer audit. Other supported writers and late authority +restoration must be traced before removing periodic governed-review repair. + +## Acceptance requirements + +Reuse the genuine native review fixture in chat-channels.integration.test.ts, +which creates the server review through native finalization and records selected +files. Establish a baseline before editing the owner. Cover: + +- Startup reconstructs a missing response and creates one comment/publication with + the selected files; duplicate/concurrent recovery is idempotent. +- Existing and soft-deleted receipts produce no recurring SQL; unchanged proof + restoration does not repeatedly rewrite or wake itself. +- Task-wide lineage, late producer commits, and outer transaction rollback/commit + retain their existing authority boundaries. +- Actual approval resolution/unlink and competing-interaction settlement reactivate + dormant work without restart. Unrelated run/task mutations do not reload it. +- Agent/principal/endpoint revocation or changed gate blocks publication; supported + restoration reactivates it. Transport reauthorization still blocks stale files. +- Saved future deadlines, moved deadlines, claims, real lock contention, operational + retry, first wake during drain, failure before drain, resume, and shutdown retain + concrete recovery obligations without passive SQL. +- Both periodic entrances defer only this duty; fresh finalization and other + recovery reasons retain their behavior. + +No deployment, database suspension, Cloud alarm coverage, or cost-savings claim +follows from this migration. The full original stack remains in scope. + +## Owner-contract checkpoint + +The scheduled entry point holds the issue transaction, checks the coordinator +claim and actual retry deadline under lock, restores the proof through its owner, +and publishes through addComment in that same transaction. Operational errors +propagate to the worker; existing foreground callers retain their logging contract. +The scoped worker reconstructs existing coordinators and excludes materialized +receipts, including deleted comments. No startup registration or periodic removal +has happened yet. + +The unchanged owner baseline passed 156 cases. The first expanded run passed +164 of 165; the new principal-revocation fixture incorrectly wrote a nonexistent +principal status field, also caught by typechecking. Using the existing company +membership suspension path repaired the fixture. The complete expanded suite now +passes all 165 cases and server typecheck passes. Evidence: +`/tmp/sleep49-baseline.log`, `/tmp/sleep49-contract-tests.log`, +`/tmp/sleep49-contract-tests-v2.log`, `/tmp/sleep49-contract-typecheck-v3.log`, +and `/tmp/sleep49-contract-v2.status`. The cases exercise concurrency, late claim/ +deadline races, real authorization-lock contention and rollback, revoked authority, +startup reconstruction, deleted receipts, and actual quiet SQL after settlement. + +The producer audit, query-cost measurements, startup/periodic wiring, remaining +lifecycle acceptance, full verification and remote gates are still outstanding. + +## Approval producer checkpoint + +Approval resolution, cancellation, revision/resubmission, link/unlink and bulk +link now pass their existing approval/task identity through writeApprovalChange. +The notification owner finds linked tasks before the mutation and registers only +those task scopes. Budget resolution supplies its approval identity inside the +existing accounting transaction. Creation without a task link and company-wide +retention notifications retain their previous behavior and do not reload response +history. No additional transaction wrapper or approval execution policy was added. +All production approval updates/deletes were inventoried; company deletion leaves +no surviving response obligation. Tool-gateway's separate link insertion still +needs its explicit task scope during the remaining producer pass. + +Four real-producer regressions (approve, reject, cancel, unlink) failed before +wiring and passed afterward. Two further cases exercise nested cancel/unlink +with outer rollback/commit and measure no governed-response candidate SQL for an +unrelated approval. All 171 owner cases pass. In the combined run two older +approval unit cases failed because their ordered stub consumed an approval row +for the new issue_approvals read. The stub now distinguishes that table and returns +no links. Existing consumers pass 19 approval service/route cases, five budget +cases, 25 queue-seeding cases, and 21 retention cases. Server typecheck passes. +Logs: /tmp/sleep49-approval-before.log, /tmp/sleep49-approval-after.log, +/tmp/sleep49-approval-full.log, /tmp/sleep49-approval-unit.log, +/tmp/sleep49-budgets.log, /tmp/sleep49-approval-queue.log, +/tmp/sleep49-approval-retention.log, /tmp/sleep49-approval-final-types.log. + +Remaining interaction gaps include connection-intent resolution/expiry, generic +interaction administrative settlement, satisfied-connection cleanup, and OAuth +confirmation callbacks/denials. Some existing milestone hooks cover only question +and confirmation kinds. Payload-only updates that leave a competing card pending +do not reactivate a response. Task ownership changes already carry task invalidation, +but direct cleanup writers must be individually accounted for. The polling backstop +is retained until this audit and the remaining acceptance requirements are complete. + +## Lifecycle and finalization notification checkpoint + +A real agent-resume regression failed before wiring and passes afterward. The +existing lifecycle pre-status-write hook now signals this worker when an agent +becomes invokable. Existing task/run milestones and finalization writes also +fan out to the new topic. The finalization scope accepts an optional issueId so +issue-wide pending-effect settlement can target its affected task instead of +reloading all governed responses in the company. Existing pending-effect delivery +retains its previous topic semantics. The tool gateway's separate approval link +insertion now supplies its known issue identity. + +All 172 owner cases pass, plus 58 agent lifecycle, 72 milestone and 32 native +finalization consumer cases. Server typecheck passes. Evidence is in +/tmp/sleep49-agent-before.log (reproduced failure), +/tmp/sleep49-lifecycle-full.log, /tmp/sleep49-lifecycle-types.log, +/tmp/sleep49-agent-consumers.log, /tmp/sleep49-milestone-consumers.log, +/tmp/sleep49-finalization-consumers.log and their status files. + +Further producer audit findings, not completed migration claims: + +- Main interaction service: suggest-task accept/reject, item-verdict completion, + connection-intent resolve, administrative cancel/withdraw/skip and comment + supersession can release a competing pending interaction. Several existing + milestone hooks cover only ask_user_questions/request_confirmation. Checkbox + confirmation resolution and other kinds need their own governed-response wake + without unnecessarily broadening unrelated workers. Some item-verdict comment + expiry writes do not currently own a transaction. +- Direct interaction writers: satisfied-connection cleanup and OAuth confirmation + acceptance/denial need coverage. OAuth connection-intent retry-phase changes + leave the interaction pending and cannot by themselves restore presentation. + Dismissal also leaves status pending; confirm the owning predicate before + introducing a wake for it. +- Existing agent-conversation reset, tool-action review, obsolete policy review, + automatic completion review and native decision commits already emit task + milestones/finalization signals. Verify every alternate branch before marking + those writer groups covered. +- Issue service currently signals response availability on task status/assignment + changes, but that boundary only targets GitHub response queues. Governed-response + task status/assignment/execution-state restoration still needs explicit coverage. +- Coordinator claim/retry notifications now reach the worker, but add a real-store + release test; synthetic late claim/deadline owner tests alone do not prove that + producer path. Current authority registration and all publication transitions + also remain part of the producer audit. + +The worker remains unregistered and periodic repair is unchanged. These commits +are checkpoints, not PR readiness or whole-instance sleep acceptance. + +## Interaction and task-authority checkpoint + +Five real interaction producer cases failed before wiring: skip, withdrawal, +rejecting suggestions, completing item verdicts and declining a connection intent. +They now pass through task-scoped notifications before the existing transaction's +write. Checkbox and administrative settlement use the same narrow topic when +the existing question/confirmation milestone fanout does not apply. Comment +supersession of item verdicts now owns its notification and row mutation together; +connection-intent ownership expiry receives its caller's issue transaction. + +Task status restoration and OAuth denial also reproduced missed wakes. Task-scoped +response-availability changes now reach governed-response recovery. Direct and +Cloud OAuth confirmation acceptance and automatic satisfied-connection retirement +notify the affected task before their writes. OAuth terminal denial owns the +signal and card mutation in one transaction; connection-intent retry phase changes +remain pending and do not wake this worker. Existing policy remains authoritative: +the attempted task-assignment restoration was rejected because externally bound +tasks cannot be reassigned through issueService. Its test now verifies that +restriction rather than bypassing it. Endpoint/identity configuration already +uses the company milestone boundary for supported authority restoration. + +Current evidence: all 180 owner cases pass, including OAuth outer rollback/commit, +plus 106 interaction service, nine telemetry, 50 satisfied-connection, and 84 +generic MCP/OAuth cases. Server typecheck passes. The first interaction consumer +run skipped the nine telemetry cases because embedded PostgreSQL was unavailable; +they now run through the standard validated external-database test option and pass. +The satisfied-connection suite initially passed 49/50: its one-connection-pool +fixture still used temp.connectionString after enabling external DB mode. That +fixture now uses the selected URL and all 50 pass. The first OAuth typecheck caught +nullable interactionId captured by a transaction callback; binding the checked ID +locally repaired it without changing policy. Preserved evidence: +/tmp/sleep49-interactions-before.log, /tmp/sleep49-interactions-after.log, +/tmp/sleep49-interactions-consumers.log, /tmp/sleep49-authority-before.log, +/tmp/sleep49-authority-after.log, /tmp/sleep49-authority-types.log, +/tmp/sleep49-telemetry-consumers.log, /tmp/sleep49-satisfied-consumers.log, +/tmp/sleep49-satisfied-consumers-v2.log, /tmp/sleep49-oauth.log, +/tmp/sleep49-authority-final.log and /tmp/sleep49-authority-final-types.log. + +Further audit: change-consent consumes already accepted metadata and cannot clear +a competing pending card. Confirmation-from-comment delegates resolution to the +covered interaction service. Native interaction audit records change summaries +only. Superseded native review retirement commits with a task-scoped native status +decision. Tool-review projection and secret-proposal settlement already signal +task milestones. Board claim and trusted-header membership restoration also +signal existing company milestones; compatibility membership backfill is agent +only. No new global live-authority registration dependency was found in governed +review authorization, which reads its durable binding/principal/destination proof. + +Still inspect execution-state-only transitions (pending policy state can block a +response without changing task status), prove real coordinator lease/retry release, +and finish candidate query-cost/lifecycle/wiring acceptance. The worker remains +unregistered and both periodic entrances remain intact. + + +## Policy release and discovery-cost checkpoint + +A pending execution policy can block presentation without changing task status. +The regression failed before the issue mutation boundary signaled pending-to-clear +transitions. All 182 owner cases now pass, including a real coordinator claim and +retry release; server types pass. Evidence: /tmp/sleep49-policy-before.log, +/tmp/sleep49-policy-after.log and /tmp/sleep49-policy-types.log. + +Discovery was measured against 10,000 cloned, already-presented governed response +rows plus retained test history. The clones preserve relational selection shape; +they are not valid authorization proofs and are never used to test publication. +Initial startup/company reconstruction took about 3,000 ms: receipt lookup scanned +roughly 4,795 unrelated comments per candidate. A company-first receipt index and +expression statistics did not reliably fix startup's plan. A run-first partial +receipt index alone reduced measured startup to 45 ms, company/task discovery to +33–35 ms and exact-run discovery to 1.5 ms. No expression statistics are retained. +The generated migration builds that index concurrently and preserves deleted +comments as receipts. These are local query measurements, not host sleep evidence. +Logs: /tmp/sleep49-query-plan.log, /tmp/sleep49-query-index-plan.log, +/tmp/sleep49-query-stats-plan.log, /tmp/sleep49-query-run-index-plan.log, +/tmp/sleep49-query-run-only-plan.log. The first root-level tsx invocation failed +because tsx is a server-package dependency; the corrected command uses its owner. + +Full outstanding-backlog cost, lifecycle coverage, startup registration and both +periodic entrance removals remain required before PR acceptance. + + +## Startup integration and lifecycle acceptance + +The worker now registers with enabled heartbeat maintenance and uses its existing +instance/task-drain gates. The shared coordinator owns notification teardown and +shutdown. Both periodic repair entrances defer to this worker: direct reconciler +repair and already-committed finalizer replay. Fresh finalization and explicit +recovery keep their behavior. Three regression cases failed before this wiring +(two periodic entrances and enabled startup registration); all now pass. + +The combined owner/startup selection passes all 255 cases. Actual SQL capture +shows no passive SQL during idle drain or after shutdown. Real approval cancellation +reactivates dormant work through drain; endpoint-lock contention retains a retry +that succeeds after resume; a real coordinator retry survives shutdown and is +reconstructed without early publication. One initial drain test mixed a real timer +with a subsequently installed fake clock. Installing the clock before coordinator +creation repaired the fixture; all three new lifecycle cases then passed. +Evidence: /tmp/sleep49-drain.log, /tmp/sleep49-drain-v2.log, +/tmp/sleep49-wiring-before.log, /tmp/sleep49-wiring-after.log. + +Workspace typecheck and build pass. Migration generation and safety checks pass. +The first migration application exposed the missing top-level migration allowlist +entry; after registration, fresh application, invalid concurrent-index repair, +concurrent migrator replay and no-op replay all pass. Evidence: +/tmp/sleep49-migration-apply.log, /tmp/sleep49-migration-apply-v2.log, +/tmp/sleep49-migration-replay.log, /tmp/sleep49-workspace-types.log, +/tmp/sleep49-build.log. The five newest schema snapshots are retained. + +The synthetic 10,000-item unpresented cohort requires 101 candidate queries, +about 2.25 seconds and 2.05 MB of serialized entries. Discovery checks admission +and cancellation between pages: the drain experiment stops after two queries; +an already aborted signal executes none. This remains proportional startup or +legitimate broad-authority reconstruction, not a bounded total-memory claim. +Normal writes retain task/run/agent scope. Synthetic proof copies establish +selection cost only; genuine owner fixtures establish publication authority. +Evidence: /tmp/sleep49-backlog.log. + +Full local tests are running. Remote CI/review, earlier stack findings and master +reconciliation remain open. No merge or whole-instance sleep is claimed. + +## Concurrent approval links and resolution + +Review finding `PRRT_kwDORclPyM6wgKOl` identified a missed-wake race in the +approval producer. Resolution could discover no links, a concurrent link could +commit and wake recovery while the approval was still pending, and resolution +could then commit without any task-scoped notification. Parking pending responses +without polling makes that ordering observable as an indefinitely missing reply. + +The existing approval notification owner now locks the company-scoped approval +row `FOR UPDATE` before discovering linked tasks. Existing `issue_approvals` +foreign-key checks hold `KEY SHARE` on that same row until their outer transaction +settles. If linking wins, resolution waits and then discovers the committed link. +If resolution wins, linking waits and its own task wake observes the committed +decision. Both cases use the existing transaction signal and durable records. +There is no new table, timer, wrapper at each caller, or company-wide discovery. +This adds one primary-key lookup/row lock to approval-identity mutations; task-only +link notifications and unlinked approval creation keep their existing cost. + +The writer audit covers single/bulk issue linking and the gateway's separate +formal-approval link insertion, all using the existing immediate foreign key. +Approval decisions, cancellation, revision/resubmission, and budget resolution +already supply the approval identity through this owner. Hire decisions retain +company-before-approval ordering; budget decisions retain their accounting +transaction. Unlink carries the removed task identity explicitly. The lock stays +owned by the outer transaction, and errors retain rollback/normal error handling. +This contract uses the application's read-committed write transactions and +integrated writer notifications; independent SQL writers remain unsupported. +Changing or deferring the link foreign key requires revisiting this contract. + +Four real PostgreSQL regressions force both transaction orderings for single and +bulk link APIs. They observe task-scoped settled notifications by reading the +committed approval/link state. A controlled table lock pauses resolution after +its link discovery without replacing either service or database. Before the fix, +both resolver-first cases observed only `pending`; both linker-first cases passed. +After the fix all four see a wake with `approved`. Six existing cases additionally +prove actual worker publication after approve/reject/cancel/unlink and nested +rollback/outer-commit behavior. All 286 tests across six complete affected suites +passed, including budgets, queue seeding, retention and approval API consumers; +server typechecking passed. Evidence: `/tmp/sleep-approval-links-{red,green}.log`, +`/tmp/sleep-approval-links-owning-tests.log`, and +`/tmp/sleep-approval-links-root-types.log` with matching status files. Integrated +stack verification and fresh remote acceptance are still required. This is a +producer-correctness fix, not deployed host-sleep qualification. + +## Status replay and response recovery lock order (2026-10-27) + +Review found that scheduled response recovery locked the issue before restoring +presentation under the finalization coordinator lock. Status commitment and +replay acquire those locks in the opposite order. Concurrent operations could +therefore deadlock even though each isolated path passed its tests. + +Recovery now acquires the scoped coordinator row before the issue. Its existing +presentation owner still rechecks the accepted decision, result, assessment, +claim, retry deadline and current authority. This changes no durable schema or +polling behavior. The extra primary-key lock query occurs only for actual work. + +The regression holds a real coordinator lock, starts response recovery, observes +PostgreSQL reporting that recovery is blocked, then calls the actual status +committer inside the owning transaction. Before the fix that replay fails with +55P03 while locking the issue. After the fix replay completes, recovery publishes +one response after the transaction ends, and repeated recovery produces no second +response. The nine targeted authority/concurrency/deadline cases and all 270 +cases in the complete external-chat wait suite pass. Server typechecking passes; +an initial test-only persisted-JSON type assertion error was corrected separately. +Latest-head CI and integration with downstream changes remain required. + + +## Current 0413 integration and governed-review acceptance + +The current accepted-response parent is integrated through a normal merge. +Nineteen owned source files preserve their original bytes; ten files combine +original and upstream edits automatically. All 10,580 unrelated upstream source +files remain unchanged. Four reviewed conflicts combine the owned approval +scope with the current async transaction writer, retain explicit tool-review +producer notifications instead of restoring removed shared fanout, and preserve +the source conversation/session-generation authority locks before the new +review hint and row mutation. The real approval-link concurrency and status +replay paths are exercised below. Newly added code comments were removed with +nontrivia token identity proof; no behavior was removed. + +Only the owned unlanded migration artifacts were deferred. `pnpm db:generate` +produced `0413_tired_kronos.sql`; its original concurrent index statement remains +byte-identical. Prior executable SQL, journal entries and retained snapshot +bytes are unchanged. Five snapshots 0409 through 0413 remain. Actual database +build, server types/build, module boundaries, exact-head migration order and +no-new-code-comments checks pass. + +Tested source `320146c29489bb1f7308422981fa946b6a3d3368` passes **2828 distinct +cases across 37 complete files and one focused chat file**. The main selection +passes 2399 cases in 272.11 seconds; safe replacement passes 59 in 26.04 +seconds; all 270 governed external-chat wait cases pass in 53.07 seconds; the +complete CLI company import/export E2E passes its real child-server journey in +13.31 seconds. The focused chat selection passes 39 in 40.41 seconds; 1760 +other chat cases are outside its name pattern and are not claimed. The main +invocation initially skips two approval files (60 cases) because their external +database variables were omitted. A targeted add-on supplies their own fresh +databases and passes all 60 cases in 26.20 seconds, without repeating the +already-passing suites. The original skipped output remains retained. Counts: + +| File | Passing cases | +| --- | ---: | +| cli/src/__tests__/company-import-export-e2e.test.ts | 1 | +| server/src/__tests__/approval-queue-seeding-work.test.ts | 28 | +| server/src/__tests__/approval-retention-work.test.ts | 32 | +| server/src/__tests__/approvals-service.test.ts | 8 | +| server/src/__tests__/chat-channels.integration.test.ts (focused) | 39 | +| server/src/__tests__/committed-chat-response-authority.test.ts | 7 | +| server/src/__tests__/issue-recovery-actions.test.ts | 89 | +| server/src/__tests__/issue-thread-interactions-telemetry.test.ts | 9 | +| server/src/__tests__/native-finalization-recovery.test.ts | 32 | +| server/src/__tests__/native-interaction-bridge.test.ts | 7 | +| server/src/__tests__/native-status-arbiter-corpus.test.ts | 94 | +| server/src/__tests__/native-workspace-sync.test.ts | 10 | +| server/src/__tests__/recovery-stale-issue-lock-sweep.test.ts | 40 | +| server/src/__tests__/satisfied-connection-intents.test.ts | 52 | +| server/src/__tests__/server-startup-feedback-export.test.ts | 71 | +| server/src/__tests__/slack-conversation-lifecycle.test.ts | 41 | +| server/src/__tests__/tool-access-oauth-protocol.integration.test.ts | 2 | +| server/src/modules/run-dispatch/adapters/postgres.test.ts | 69 | +| server/src/modules/wake-queue/adapters/postgres.test.ts | 93 | +| server/src/modules/wake-queue/adapters/queued-comment-postgres.test.ts | 5 | +| server/src/services/chat-monitor-continuation.test.ts | 21 | +| server/src/services/execution-control-reconciliation.test.ts | 10 | +| server/src/services/explicit-native-continuation.test.ts | 335 | +| server/src/services/heartbeat/queue.test.ts | 32 | +| server/src/services/heartbeat/recovery.test.ts | 45 | +| server/src/services/heartbeat/run-control.test.ts | 51 | +| server/src/services/heartbeat/run-lifecycle.test.ts | 34 | +| server/src/services/heartbeat/upgrade-readiness.test.ts | 7 | +| server/src/services/issue-thread-interactions.test.ts | 13 | +| server/src/services/legacy-execution-recovery.test.ts | 15 | +| server/src/services/native-runtime/external-chat-wait.integration.test.ts | 270 | +| server/src/services/native-runtime/native-restart-recovery.test.ts | 25 | +| server/src/services/native-runtime/native-run-finalizer-telemetry.test.ts | 29 | +| server/src/services/native-runtime/native-safe-replacement.test.ts | 59 | +| server/src/services/native-runtime/native-session-executor.test.ts | 968 | +| server/src/services/native-runtime/native-workspace-export-retry.test.ts | 80 | +| server/src/services/native-runtime/native-workspace-reconstruction.test.ts | 50 | +| server/src/services/native-runtime/paperclip-control-plane-port.test.ts | 55 | + +The complete 94-case status-authority corpus also passes again on the same +database. The repeat is not double-counted. Retained PSC/OUT/sentinel rows +remain unchanged and each run uses distinct fixture prefixes. Thirty-one +separate fresh PostgreSQL 17 databases migrate through 0413. Five main +embedded-only fixture imports and the CLI fixture import temporarily use +their own external databases on this IPC-exhausted host, then restore byte +for byte. Assertions and timing limits remain unchanged. This qualifies +external-database behavior; it does not reproduce embedded CI locally. + +Validation started at `2b7f491641f22fb7577859681f8b632117113c8d`. The CLI +comment cleanup was already present when pre-run source hashes were captured +but was omitted from that commit's server/database-only staging. Committing +those exact bytes as `320146c29489bb1f7308422981fa946b6a3d3368` changed no working +source during validation. All 38 selected committed files match the pre-run +hashes after fixture restoration. The comment cleanup has token identity proof. + +A populated 0412-to0413 upgrade preserves all 411 prior receipts and all +38 companies, appends exactly one receipt, and adds none on replay. The +receipt index is valid and ready on fresh and upgraded databases. The +preexisting response-selection statistic is retained. Earlier failed checks +and restricted retries above keep their original scope. The first local +composition bookkeeping check omitted the unstaged CLI comment cleanup; the +aligned source proof now passes. No functional failure is reclassified by it. + +The published response-presentation prerequisite is adopted without changing +any accepted complete core tree. Publication stays held until #17696 lands +and actual squash ancestry is reconciled. Fresh published-head CI and +exact-head Greptile 5/5 remain required. These checks do not establish deployed +suspension, external Cloud alarm delivery or fleet savings. + + +### Scoped wake repair after the published parent review + +Source `2341714720b5c273f0ce011b48345af0ef39227f` inherits the published +parent `ef6cb73777f0c31e3622d23d039d702b2fc08263` through a normal merge. +Both response queues now require an issue or run on ordinary milestone and +finalization notifications. The explicit company authority boundary invalidates +both workers globally, preserving actual membership/configuration restoration. +The real coalesced-input producer test covers both queues after outer commit. +Existing pending-effects and older chat queue payloads remain unchanged. + +The exact three new governed-response regressions first reproduce two failures +for repeated coarse hints while actual membership restoration passes. The same +source and assertion limits pass after repair: repeated coarse hints do not +reconstruct dormant governed response history, and actual membership restoration +still publishes once. Source qualifies **2053 distinct passing cases**: 1703 +in 15 complete files and 350 selected chat cases. The complete files include +all 273 governed response cases, 95 wake-queue adapter cases, 968 native-session +cases, authority/lifecycle/approval consumers, and inherited monitor/finalizer +coverage. Another 1463 chat cases are outside the selection; the complete +1813-case chat file is not claimed for this repair. The three regressions and +two coalescing cases are included, not added again. + +Fourteen complete consumer files pass 1430 cases in 123.59 seconds; all 273 +governed-response cases pass in 53.97 seconds; focused chat passes 350 in +301.76 seconds. Thirteen dedicated PostgreSQL 17 fixtures migrate through +0413. Only the embedded-only wake-queue import temporarily selects its own +external fixture and restores byte for byte. All 16 captured test hashes match. +Database build, server types/build, module boundaries and full-SHA migration +order pass. No assertion or time-limit changes qualify the repair. + +The composition proof retains 12646 parent/unrelated tree entries exactly, +36 owned entries exactly and one identical shared change. Only two notification +files need manual scope resolutions; the third overlap, the native finalizer, +preserves the exact parent hunk bodies automatically. The database tree is +identical to the previously qualified 0413 source; no SQL, executable history, +journal or snapshot is regenerated. Earlier 2828-case acceptance and populated +upgrade/replay evidence retain their earlier source scope. The remote PR remains +held until its parent merges and fresh current-head CI and reviews pass. + + +### Landed parent, Photon boundary and independent master contract + +Parent #17696 merged as `ec3b558277be5488c92e9b388e25300826caee89` +from exact reviewed head `ef6cb73777f0c31e3622d23d039d702b2fc08263`, +with 58 clean checks, Greptile 5/5 and unconditional Ponytail 5/5. The normal +ancestry reconciliation proves whole parent/squash tree equality and preserves +the complete held child tree, including the owned scope repair. + +A later security warning claimed Photon ownership assertions caused a global +response reconstruction on every event. The existing callback already reads the +endpoint under a row lock, requires attention status and transitions it to active +or verifying before commit. The real published callback test disproved the claim: +ten active assertions cause zero response hints and zero candidate SELECTs; +three concurrent attention assertions cause exactly one hint and one SELECT, +then return to quiet. The finding was resolved with that evidence and its +current-head security check became successful. No production change was needed. +The regression is retained here and covers both response workers with real +Paperclip callbacks and PostgreSQL; the external Photon transport is mocked. +The complete 22-case Photon file passes on this combined source. An earlier +single-worker 21-case acceptance retains its own parent source scope. + +Master independently added the runtime activity contract in #18129 at +`80a2b9284bf189e93ca4313085814548f48b8dc8`. Its 32 changed paths have no +intersection with the owned sleep patch. A normal merge preserves both complete +binary patches. The inherited contract changes event validation, redaction and +UI event classification; it does not change this worker or its durable predicate. + +Combined source `586f5674cd9798b33cd22437255e0480b5726d20` passes +**1297 distinct cases**: 1093 across five complete server files in 96.03 seconds +(968 native-session, 22 Photon and 103 inherited contract cases), 185 in five +complete runner files in 1.55 seconds, 14 Node protocol-contract cases in +1614.48 milliseconds and five UI registry cases. These results are separate +from the earlier 2053-case source; overlapping cases are not cumulatively counted. +The three dedicated fresh 0413 fixtures and the live-event helper's own fresh +database use PostgreSQL 17, with no source adaptations. All twelve new-context +source hashes and all sixteen earlier scoped-qualification hashes match. +Database build, server types/build, module boundaries, full-SHA migration order +and generated TypeScript/Rust profile checks pass. The Runner vendor rebuild +completes successfully for the new inherited inputs. No deployment or live +external-provider qualification is claimed. + +The database tree remains identical to the qualified 0413 source. The populated +upgrade evidence therefore retains its unchanged executable migration bytes; +no new migration or snapshot generation was necessary. Before publication, +the original owned patch and independent master patch are checked again as +complete binary diffs, including the retained Photon test and this audit. +Fresh exact-head CI, Greptile 5/5 and actionable review resolution remain merge +gates for this PR. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 55d1a1ca9a1..af421ce9779 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -562,6 +562,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0410_freezing_kinsey_walden.sql", "0411_real_scrambler.sql", "0412_tearful_squadron_supreme.sql", + "0413_tired_kronos.sql", "0384_funny_adam_warlock.sql", diff --git a/packages/db/src/migrations/0413_tired_kronos.sql b/packages/db/src/migrations/0413_tired_kronos.sql new file mode 100644 index 00000000000..1a3a6438728 --- /dev/null +++ b/packages/db/src/migrations/0413_tired_kronos.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "issue_comments_review_response_receipt_idx" ON "issue_comments" USING btree ("created_by_run_id","company_id","issue_id","author_agent_id") WHERE "issue_comments"."author_type" = 'agent' AND "issue_comments"."metadata"->>'authorizationReason' = 'allow_chat_run_presentation'; diff --git a/packages/db/src/migrations/meta/0408_snapshot.json b/packages/db/src/migrations/meta/0413_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0408_snapshot.json rename to packages/db/src/migrations/meta/0413_snapshot.json index 195b96fbb8e..c27c8a4f7b2 100644 --- a/packages/db/src/migrations/meta/0408_snapshot.json +++ b/packages/db/src/migrations/meta/0413_snapshot.json @@ -1,6 +1,6 @@ { - "id": "6bd67ad8-0d80-4163-873b-8b8eacf32bb2", - "prevId": "856a147c-73dc-4908-a8d9-4f19cab7dbf6", + "id": "5690a14a-f9cc-41a6-af51-c27f688f4ef6", + "prevId": "3b8bb508-028d-4154-abdd-af870f1f3ca6", "version": "7", "dialect": "postgresql", "tables": { @@ -33771,6 +33771,56 @@ } }, "indexes": { + "heartbeat_runs_native_unpresented_response_idx": { + "name": "heartbeat_runs_native_unpresented_response_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "native_issue_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_runs\".\"runtime_mode\" = 'native' AND \"heartbeat_runs\".\"status\" in ('succeeded', 'failed', 'timed_out') AND \"heartbeat_runs\".\"finished_at\" is not null AND jsonb_typeof(\"heartbeat_runs\".\"result_json\" #> '{presentationDecision,commentId}') is distinct from 'string'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "heartbeat_runs_cancelled_workspace_preservation_idx": { + "name": "heartbeat_runs_cancelled_workspace_preservation_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"heartbeat_runs\".\"runtime_mode\" = 'native' AND \"heartbeat_runs\".\"status\" = 'cancelled'\n AND \"heartbeat_runs\".\"runner_profile_json\"->'nativeWorkspaceSync'->>'state' = 'prepared'\n AND \"heartbeat_runs\".\"result_json\"->'nativeCancellation'->>'dispatchState' = 'acknowledged'\n AND (\"heartbeat_runs\".\"result_json\"->'workspaceRestoreRecovery'->>'schema' = 'paperclip.workspace-restore-recovery.v1'\n AND \"heartbeat_runs\".\"result_json\"->>'workspaceRestoreFailure' IN ('restore_permission_denied', 'restore_lock_timeout', 'restore_unsafe_archive', 'restore_failed')) IS NOT TRUE", + "concurrently": false, + "method": "btree", + "with": {} + }, "heartbeat_runs_retained_recovery_adoption_idx": { "name": "heartbeat_runs_retained_recovery_adoption_idx", "columns": [ @@ -36453,6 +36503,40 @@ "method": "btree", "with": {} }, + "issue_comments_review_response_receipt_idx": { + "name": "issue_comments_review_response_receipt_idx", + "columns": [ + { + "expression": "created_by_run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "issue_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "author_agent_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"issue_comments\".\"author_type\" = 'agent' AND \"issue_comments\".\"metadata\"->>'authorizationReason' = 'allow_chat_run_presentation'", + "concurrently": false, + "method": "btree", + "with": {} + }, "issue_comments_company_idx": { "name": "issue_comments_company_idx", "columns": [ @@ -43086,6 +43170,62 @@ } }, "indexes": { + "native_run_finalizations_pending_effect_owner_idx": { + "name": "native_run_finalizations_pending_effect_owner_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "issue_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "decision_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"native_run_finalizations\".\"decision_id\" is not null AND \"native_run_finalizations\".\"result_id\" is not null AND \"native_run_finalizations\".\"phase\" <> 'terminal_failure'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "native_run_finalizations_unsafe_export_recovery_idx": { + "name": "native_run_finalizations_unsafe_export_recovery_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "issue_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "run_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"native_run_finalizations\".\"phase\" = 'terminal_failure' AND \"native_run_finalizations\".\"failure_code\" = 'native_workspace_sync_out_unsafe_archive'", + "concurrently": false, + "method": "btree", + "with": {} + }, "native_run_finalizations_control_deadline_idx": { "name": "native_run_finalizations_control_deadline_idx", "columns": [ @@ -52390,6 +52530,34 @@ } }, "indexes": { + "status_decision_effects_pending_delivery_idx": { + "name": "status_decision_effects_pending_delivery_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "issue_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "decision_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"status_decision_effects\".\"delivery_state\" in ('pending', 'failed')", + "concurrently": false, + "method": "btree", + "with": {} + }, "status_decision_effects_decision_ordinal_uq": { "name": "status_decision_effects_decision_ordinal_uq", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 76a38c88a5b..16ac6fbe13b 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2878,6 +2878,13 @@ "when": 1793249495056, "tag": "0412_tearful_squadron_supreme", "breakpoints": true + }, + { + "idx": 413, + "version": "7", + "when": 1793251083562, + "tag": "0413_tired_kronos", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema/issue_comments.ts b/packages/db/src/schema/issue_comments.ts index 8d17c216628..ba5d74007e4 100644 --- a/packages/db/src/schema/issue_comments.ts +++ b/packages/db/src/schema/issue_comments.ts @@ -5,6 +5,7 @@ import type { IssueCommentPresentation, SourceTrustMetadata, } from "@paperclipai/shared"; +import { sql } from "drizzle-orm"; import { pgTable, uuid, text, timestamp, index, jsonb, unique, integer } from "drizzle-orm/pg-core"; import { companies } from "./companies.js"; import { issues } from "./issues.js"; @@ -50,6 +51,9 @@ export const issueComments = pgTable( clientRequestUq: unique("issue_comments_client_request_uq").on(table.issueId, table.authorUserId, table.clientRequestId), companyIdUq: unique("issue_comments_company_id_uq").on(table.companyId, table.id), issueIdx: index("issue_comments_issue_idx").on(table.issueId), + reviewResponseReceiptIdx: index("issue_comments_review_response_receipt_idx").on( + table.createdByRunId, table.companyId, table.issueId, table.authorAgentId, + ).where(sql`${table.authorType} = 'agent' AND ${table.metadata}->>'authorizationReason' = 'allow_chat_run_presentation'`), companyIdx: index("issue_comments_company_idx").on(table.companyId), companyIssueCreatedAtIdx: index("issue_comments_company_issue_created_at_idx").on( table.companyId, diff --git a/server/src/__tests__/approvals-service.test.ts b/server/src/__tests__/approvals-service.test.ts index 4049b3b0fe0..128733d8bdd 100644 --- a/server/src/__tests__/approvals-service.test.ts +++ b/server/src/__tests__/approvals-service.test.ts @@ -1,6 +1,6 @@ import { beforeEach, describe, expect, it, vi } from "vitest"; import { approvalService } from "../services/approvals.ts"; -import { companies } from "@paperclipai/db"; +import { approvals, companies, issueApprovals } from "@paperclipai/db"; vi.mock("../services/budgets.js", async importOriginal => ({ ...await importOriginal(), @@ -50,10 +50,14 @@ function createDbStub(selectResults: ApprovalRecord[][], updateResults: Approval const pendingSelectResults = [...selectResults]; let inTransaction = false; const selectWhere = vi.fn(async () => (inTransaction ? pendingSelectResults.shift() : pendingSelectResults[0]) ?? []); - const from = vi.fn((table) => table === companies + const from = vi.fn((table) => table === issueApprovals + ? { where: async () => [] } + : table === companies ? { where: () => ({ for: async () => [{ id: "company-1" }] }) } : { where: selectWhere }); - const select = vi.fn(() => ({ from })); + const select = vi.fn((fields?: { id?: unknown }) => fields?.id === approvals.id + ? { from: () => ({ where: () => ({ for: async () => [{ id: "approval-1" }] }) }) } + : { from }); const returning = vi.fn(async () => updateResults); const updateWhere = vi.fn(() => ({ returning })); diff --git a/server/src/__tests__/issue-thread-interactions-telemetry.test.ts b/server/src/__tests__/issue-thread-interactions-telemetry.test.ts index d3db5df11d3..78bd4a1ca70 100644 --- a/server/src/__tests__/issue-thread-interactions-telemetry.test.ts +++ b/server/src/__tests__/issue-thread-interactions-telemetry.test.ts @@ -1,3 +1,4 @@ +import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { randomUUID } from "node:crypto"; import { eq } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; @@ -29,7 +30,8 @@ vi.mock("../telemetry.js", () => ({ getTelemetryClient: telemetryMocks.getTelemetryClient, })); -const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_INTERACTION_TELEMETRY_TEST_DATABASE_URL"); +const embeddedPostgresSupport = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; describeEmbeddedPostgres("issueThreadInteractionService telemetry", () => { @@ -38,8 +40,8 @@ describeEmbeddedPostgres("issueThreadInteractionService telemetry", () => { let tempDb: Awaited> | null = null; beforeAll(async () => { - tempDb = await startEmbeddedPostgresTestDatabase("paperclip-issue-interaction-telemetry-"); - db = createDb(tempDb.connectionString); + if (!externalUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-issue-interaction-telemetry-"); + db = createDb(externalUrl ?? tempDb!.connectionString); interactionsSvc = issueThreadInteractionService(db); }, 20_000); @@ -65,6 +67,7 @@ describeEmbeddedPostgres("issueThreadInteractionService telemetry", () => { }); afterAll(async () => { + await db?.$client.end({ timeout: 1 }); await tempDb?.cleanup(); }); diff --git a/server/src/__tests__/photon/channel.integration.test.ts b/server/src/__tests__/photon/channel.integration.test.ts index 812ce3740e9..84985e0face 100644 --- a/server/src/__tests__/photon/channel.integration.test.ts +++ b/server/src/__tests__/photon/channel.integration.test.ts @@ -1,3 +1,7 @@ +import { nativeReviewResponseWork } from "../../services/native-runtime/native-review-response-work.js"; +import { nativeResponsePresentationWork } from "../../services/native-runtime/native-response-presentation-work.js"; +import { DELIVERY_QUEUES } from "../../services/delivery-work-notifications.js"; +import { idleWorkSnapshot } from "../../services/task-admission.js"; import { createDeliveryWorkCoordinator } from "../../services/delivery-work-coordinator.js"; import { resolveExternalTestDatabaseUrl } from "../helpers/external-test-database.js"; import { Readable } from "node:stream"; @@ -24,6 +28,7 @@ import { import { and, eq, sql } from "drizzle-orm"; import { applyPendingMigrations, + subscribeDatabaseWork, createDb, companies, agents, @@ -372,6 +377,53 @@ describe("iMessage Photon channel control plane", () => { }, }; } + it.each([DELIVERY_QUEUES.nativeResponsePresentation, DELIVERY_QUEUES.nativeReviewResponse])( + "keeps repeated active Photon ownership assertions out of %s reconstruction and wakes once on recovery", async queue => { + const t = await setup(); + await t.start(); + await t.qualify(); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + coordinators.push(coordinator); + const work = queue === DELIVERY_QUEUES.nativeResponsePresentation ? nativeResponsePresentationWork(db) : nativeReviewResponseWork(db); + const worker = work.register(coordinator, () => true); + const debug = db.$client.options.debug; + const hints: unknown[] = []; + const queries: string[] = []; + const unsubscribe = subscribeDatabaseWork(db, queue, (event, values) => { + if (event === "settled") for (const value of values ?? []) { + const scope = JSON.parse(value); + if (scope.companyId === t.companyId) hints.push(scope); + } + }); + try { + await worker.ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), { timeout: 5000 }) + .toEqual({ active: 0, next: null }); + db.$client.options.debug = (_connection, query) => { + if (query.startsWith("select") && query.includes("native_run_finalizations")) queries.push(query); + }; + const assertOwned = t.callbacks().onPhotonAssertOwned!; + for (let i = 0; i < 5; i++) await assertOwned(); + expect(hints).toEqual([]); + expect(queries).toEqual([]); + await db.update(chatEndpoints).set({ status: "attention", healthMessage: "Recovering transport" }) + .where(eq(chatEndpoints.id, t.endpoint.id)); + await Promise.all([assertOwned(), assertOwned(), assertOwned()]); + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), { timeout: 5000 }) + .toEqual({ active: 0, next: null }); + expect(hints).toEqual([{ companyId: t.companyId }]); + expect(queries).toHaveLength(1); + expect((await t.service.get(t.endpoint.id)).status).toBe("active"); + hints.length = 0; + queries.length = 0; + for (let i = 0; i < 5; i++) await assertOwned(); + expect(hints).toEqual([]); + expect(queries).toEqual([]); + expect(errors).toEqual([]); + } finally { unsubscribe(); db.$client.options.debug = debug; if ("stop" in worker) worker.stop(); } + }); + it("recovers a Photon transport loss through the receiver worker and cancels fatal recovery", async () => { const t = await setup(); const onError = vi.fn(); diff --git a/server/src/__tests__/satisfied-connection-intents.test.ts b/server/src/__tests__/satisfied-connection-intents.test.ts index 54e4d934aae..4809931df08 100644 --- a/server/src/__tests__/satisfied-connection-intents.test.ts +++ b/server/src/__tests__/satisfied-connection-intents.test.ts @@ -21,19 +21,16 @@ import { idleWorkSnapshot } from "../services/task-admission.js"; import { readIdleSleepSafety, type IdleSleepDrainStatus } from "../services/idle-sleep-safety.js"; import { reserveRunBudget } from "../services/budget-reservations.js"; -const externalTestDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_SATISFIED_CONNECTION_TEST_DATABASE_URL"); -const support = externalTestDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); +const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_SATISFIED_CONNECTION_TEST_DATABASE_URL"); +const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport(); const describePostgres = support.supported ? describe : describe.skip; describePostgres("already available connection requests", () => { let db!: ReturnType; let temp!: Awaited>; beforeAll(async () => { - if (externalTestDatabaseUrl) db = createDb(externalTestDatabaseUrl); - else { - temp = await startEmbeddedPostgresTestDatabase("paperclip-satisfied-connections-"); - db = createDb(temp.connectionString); - } + if (!externalUrl) temp = await startEmbeddedPostgresTestDatabase("paperclip-satisfied-connections-"); + db = createDb(externalUrl ?? temp.connectionString); await db.insert(authUsers).values({ id: "connection-owner", name: "Owner", email: "owner@example.test", createdAt: new Date(), updatedAt: new Date() }).onConflictDoNothing(); }, 20_000); beforeEach(async () => { await db.execute(sql`truncate table companies cascade`); }); @@ -65,6 +62,19 @@ describePostgres("already available connection requests", () => { return run!; } + it("wakes governed response recovery only when automatic cleanup retires a connection card", async () => { + const f = await seed(); + const notified = vi.fn(); + const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.nativeReviewResponse, notified); + try { + expect(await satisfiedConnectionIntentService(db).sweepPending()).toMatchObject({ satisfied: 1 }); + expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, f.intent.id)))[0].status).toBe("expired"); + expect(notified).toHaveBeenCalledOnce(); + expect(await satisfiedConnectionIntentService(db).sweepPending()).toMatchObject({ satisfied: 0 }); + expect(notified).toHaveBeenCalledOnce(); + } finally { unsubscribe(); } + }); + it("notifies terminal confirmation recovery after existing-access retirement commits", async () => { const f = await seed(); const notified = vi.fn(); @@ -398,7 +408,7 @@ describePostgres("already available connection requests", () => { it("records a system completion and one durable continuation with a one-connection pool", async () => { const f = await seed(); - const single = createDb(externalTestDatabaseUrl ?? temp.connectionString, { maxConnections: 1 }); + const single = createDb(externalUrl ?? temp.connectionString, { maxConnections: 1 }); try { const outcomes = await Promise.all([satisfiedConnectionIntentService(single).sweepPending(), satisfiedConnectionIntentService(db).sweepPending()]); expect(outcomes.reduce((sum, outcome) => sum + outcome.satisfied, 0)).toBe(1); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 8777ebc7858..feb240892e4 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() })), + registerNativeReviewResponseWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerNativeResponsePresentationWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve(), stop: vi.fn() })), registerNativePendingEffectWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), registerUnsafeWorkspaceExportWork: vi.fn((_coordinator: unknown, _canRun: () => boolean) => ({ ready: Promise.resolve() })), @@ -570,6 +571,13 @@ describe("startServer feedback export wiring", () => { } }); + it.each([false, true])("registers governed-response recovery only when maintenance is enabled (%s)", async enabled => { + loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); + await startServer(); await settleStartupHousekeeping(); + expect(heartbeatServiceMock.registerNativeReviewResponseWork).toHaveBeenCalledTimes(enabled ? 1 : 0); + if (enabled) expect(heartbeatServiceMock.registerNativeReviewResponseWork.mock.calls[0][1]()).toBe(true); + }); + it.each([false, true])("registers native pending effects only when maintenance is enabled (%s)", async enabled => { loadConfigMock.mockReturnValue(buildTestConfig({ heartbeatSchedulerEnabled: enabled })); await startServer(); await settleStartupHousekeeping(); @@ -1017,7 +1025,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, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true }); + expect(heartbeatServiceMock.reapOrphanedRuns).toHaveBeenLastCalledWith({ staleThresholdMs: 300_000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true }); expect(heartbeatServiceMock.reconcileCostAccounting).toHaveBeenCalledTimes(1); expect(retiredDetector).not.toHaveBeenCalled(); } finally { diff --git a/server/src/index.ts b/server/src/index.ts index 86a7b9da72c..9e25bd0a284 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1825,6 +1825,8 @@ async function startServerWithDatabaseTeardown( () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; if (config.heartbeatSchedulerEnabled) await heartbeat.registerInstructionDirectoryCleanupWork(deliveryWork, () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; + if (config.heartbeatSchedulerEnabled) await heartbeat.registerNativeReviewResponseWork(deliveryWork, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; if (config.heartbeatSchedulerEnabled) { const worker = heartbeat.registerNativeResponsePresentationWork(deliveryWork, () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()); @@ -1972,7 +1974,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, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true }) + .reapOrphanedRuns({ staleThresholdMs: 5 * 60 * 1000, skipScheduledRetryCancellation: true, skipCapturedInstructionCommit: true, skipInstructionDirectoryCleanup: true, skipStoppedInstructionCollection: true, skipCancelledWorkspacePreservation: true, skipObsoletePolicyReviews: true, skipAutomaticCompletionReviews: true, skipUnsafeWorkspaceExports: true, skipNativePendingEffects: true, skipNativeResponsePresentation: true, skipNativeReviewResponse: true }) .then(() => heartbeat.promoteDueScheduledRetries()) .then(async (promotion) => { await heartbeat.resumeQueuedRuns(); diff --git a/server/src/lib/approval-records.ts b/server/src/lib/approval-records.ts index bfa7a38b3dc..a4027fc4742 100644 --- a/server/src/lib/approval-records.ts +++ b/server/src/lib/approval-records.ts @@ -52,7 +52,7 @@ export function approvalResolutionRecords(db: Db) { .where(and(eq(approvals.id, id), inArray(approvals.status, resolvableStatuses))) .returning() .then((rows) => rows[0] ?? null); - }); + }, { approvalId: id }); if (updated) { return { approval: updated, applied: true }; diff --git a/server/src/modules/wake-queue/adapters/postgres.test.ts b/server/src/modules/wake-queue/adapters/postgres.test.ts index 240776e2278..79f5ea78968 100644 --- a/server/src/modules/wake-queue/adapters/postgres.test.ts +++ b/server/src/modules/wake-queue/adapters/postgres.test.ts @@ -1224,12 +1224,13 @@ describeEmbeddedPostgres("wake-queue postgres adapter", () => { expect(wakeRow?.payload).not.toHaveProperty("foo"); }); - it("scopes a real coalesced-input response wake to its active run after commit", async () => { + it.each([DELIVERY_QUEUES.nativeResponsePresentation, DELIVERY_QUEUES.nativeReviewResponse])( + "scopes a real coalesced-input %s response wake to its active run after commit", async queue => { const companyId = await seedCompany(); const agentId = await seedAgent({ companyId }); const runId = await seedRun({ companyId, agentId, status: "running" }); const hints: Array> = []; - const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.nativeResponsePresentation, (event, values) => { + const unsubscribe = subscribeDatabaseWork(db, queue, (event, values) => { if (event === "settled") for (const value of values ?? []) hints.push(JSON.parse(value)); }); try { diff --git a/server/src/services/agent-lifecycle.ts b/server/src/services/agent-lifecycle.ts index 041ae077db4..80d8a816fec 100644 --- a/server/src/services/agent-lifecycle.ts +++ b/server/src/services/agent-lifecycle.ts @@ -1,3 +1,4 @@ +import { notifyNativeReviewResponse } from "./native-review-response-notifications.js"; import { isAgentStatusInvokable } from "@paperclipai/shared"; import { notifyNativeResponsePresentation } from "./native-response-presentation-notifications.js"; import { notifyChatInboundWakeNotices } from "./chat-inbound-wake-notice-notifications.js"; @@ -39,6 +40,7 @@ export function createAgentLifecycleEffects(_hooks: BudgetServiceHooks = {}): li async beforeStatusChange(tx, companyId, agentId, before, after) { if (!isAgentStatusInvokable(before) && isAgentStatusInvokable(after)) { await notifyNativeResponsePresentation(tx, { companyId, agentId }); + await notifyNativeReviewResponse(tx, { companyId, agentId }); await notifyChatInboundWakeNotices(tx, { companyId }); } }, diff --git a/server/src/services/approval-work-notifications.ts b/server/src/services/approval-work-notifications.ts index 5c914477d15..6b74cb6c665 100644 --- a/server/src/services/approval-work-notifications.ts +++ b/server/src/services/approval-work-notifications.ts @@ -1,16 +1,32 @@ -import type { Db } from "@paperclipai/db"; +import { and, eq } from "drizzle-orm"; +import { approvals, issueApprovals, type Db } from "@paperclipai/db"; +import { notifyNativeReviewResponse } from "./native-review-response-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; export const notifyApprovalQueueWork = (transaction: object, companyId: string) => notifyDeliveryWork(transaction, DELIVERY_QUEUES.approvalQueueSeeding, companyId); -export async function notifyApprovalWork(transaction: object, companyId: string) { +type ApprovalChangeScope = { approvalId?: string; issueIds?: string[] }; + +export async function notifyApprovalWork(transaction: object, companyId: string, scope?: ApprovalChangeScope) { await notifyDeliveryWork(transaction, DELIVERY_QUEUES.approvalRetention); await notifyApprovalQueueWork(transaction, companyId); + const issueIds = new Set(scope?.issueIds); + if (scope?.approvalId) { + + + + await (transaction as Db).select({ id: approvals.id }).from(approvals) + .where(and(eq(approvals.companyId, companyId), eq(approvals.id, scope.approvalId))).for("update"); + const links = await (transaction as Db).select({ issueId: issueApprovals.issueId }).from(issueApprovals) + .where(and(eq(issueApprovals.companyId, companyId), eq(issueApprovals.approvalId, scope.approvalId))); + for (const link of links) issueIds.add(link.issueId); + } + for (const issueId of issueIds) await notifyNativeReviewResponse(transaction, { companyId, issueId }); } -export function writeApprovalChange(db: Db, companyId: string, write: (transaction: Db) => Promise): Promise { +export function writeApprovalChange(db: Db, companyId: string, write: (transaction: Db) => Promise, scope?: ApprovalChangeScope): Promise { return db.transaction(async tx => { - await notifyApprovalWork(tx, companyId); + await notifyApprovalWork(tx, companyId, scope); return write(tx as unknown as Db); }); } diff --git a/server/src/services/approvals.ts b/server/src/services/approvals.ts index cfa7624eee8..301a12abb3e 100644 --- a/server/src/services/approvals.ts +++ b/server/src/services/approvals.ts @@ -80,7 +80,7 @@ export function approvalRecords(db: Db) { .where(and(eq(approvals.id, id), inArray(approvals.status, resolvableStatuses))) .returning() .then((rows) => rows[0] ?? null); - }); + }, { approvalId: id }); return updated; }, @@ -102,7 +102,7 @@ export function approvalRecords(db: Db) { }) .where(eq(approvals.id, id)) .returning() - .then((rows) => rows[0])); + .then((rows) => rows[0]), { approvalId: id }); }, resubmit: async (id: string, payload?: Record) => { @@ -124,7 +124,7 @@ export function approvalRecords(db: Db) { }) .where(eq(approvals.id, id)) .returning() - .then((rows) => rows[0])); + .then((rows) => rows[0]), { approvalId: id }); }, listComments: async (approvalId: string) => { diff --git a/server/src/services/budgets.ts b/server/src/services/budgets.ts index ccff957a9ed..9a84916484b 100644 --- a/server/src/services/budgets.ts +++ b/server/src/services/budgets.ts @@ -262,7 +262,7 @@ async function markApprovalStatus( if (!approvalId) return; const [approval] = await db.select({ companyId: approvals.companyId }).from(approvals).where(eq(approvals.id, approvalId)); if (!approval) return; - await notifyApprovalWork(db, approval.companyId); + await notifyApprovalWork(db, approval.companyId, { approvalId }); await db .update(approvals) .set({ diff --git a/server/src/services/chat-response-availability.ts b/server/src/services/chat-response-availability.ts index d1459d65d9d..8779eab4a27 100644 --- a/server/src/services/chat-response-availability.ts +++ b/server/src/services/chat-response-availability.ts @@ -1,8 +1,10 @@ +import { notifyNativeReviewResponse } from "./native-review-response-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; export type ChatResponseScope = { companyId: string; issueId?: string; deliveryId?: string; conversationId?: string }; export async function notifyChatResponseAvailability(tx: object, scope: ChatResponseScope) { + if (scope.issueId) await notifyNativeReviewResponse(tx, { companyId: scope.companyId, issueId: scope.issueId }); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubTerminalResponses, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, deliveryId: scope.deliveryId, conversationId: scope.conversationId })); await notifyDeliveryWork(tx, DELIVERY_QUEUES.chatGitHubAcknowledgements, JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, deliveryId: scope.deliveryId, conversationId: scope.conversationId })); } diff --git a/server/src/services/chat-run-milestone-notifications.ts b/server/src/services/chat-run-milestone-notifications.ts index 17316dbc678..6736b0e6ffc 100644 --- a/server/src/services/chat-run-milestone-notifications.ts +++ b/server/src/services/chat-run-milestone-notifications.ts @@ -1,3 +1,4 @@ +import { notifyNativeReviewResponse } from "./native-review-response-notifications.js"; import { notifyNativeResponsePresentation } from "./native-response-presentation-notifications.js"; import { notifyChatResponseAvailability } from "./chat-response-availability.js"; import { notifyChatConfirmationWork } from "./chat-confirmation-notifications.js"; @@ -11,6 +12,8 @@ export type ChatRunMilestoneScope = { companyId: string; issueId?: string; runId export async function notifyChatRunMilestones(tx: object, scope: ChatRunMilestoneScope) { if (scope.issueId || scope.runId) { + await notifyNativeReviewResponse(tx, { companyId: scope.companyId, issueId: scope.issueId, + runId: scope.issueId ? undefined : scope.runId }); await notifyNativeResponsePresentation(tx, { companyId: scope.companyId, issueId: scope.issueId, runId: scope.issueId ? undefined : scope.runId }); } @@ -30,4 +33,5 @@ export async function notifyChatRunMilestones(tx: object, scope: ChatRunMileston export async function notifyChatCompanyAuthorityChange(tx: object, scope: { companyId: string }) { await notifyChatRunMilestones(tx, scope); await notifyNativeResponsePresentation(tx, scope); + await notifyNativeReviewResponse(tx, scope); } diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 413c91f3b76..b1b72704e53 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db"; /** Topics for existing durable queues; register intent before their writes. */ export const DELIVERY_QUEUES = { + nativeReviewResponse: "native-review-response", nativeResponsePresentation: "native-response-presentation", nativePendingEffects: "native-pending-effects", unsafeWorkspaceExportRecovery: "unsafe-workspace-export-recovery", diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 141c1d6d7d9..2aab7492427 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +import { registerNativeReviewResponseWork } from "./native-runtime/native-review-response-work.js"; import { registerNativeResponsePresentationWork } from "./native-runtime/native-response-presentation-work.js"; import { registerNativePendingEffectWork } from "./native-runtime/native-pending-effect-work.js"; import { registerUnsafeWorkspaceExportWork } from "./native-runtime/native-workspace-export-recovery.js"; @@ -4090,6 +4091,8 @@ export function heartbeatService( triggerIssueMonitor, reportRunActivity: clearDetachedRunWarning, + registerNativeReviewResponseWork: (coordinator: Parameters[1], canRun: () => boolean) => + registerNativeReviewResponseWork(db, coordinator, () => canRun() && readTaskDrain(new Date()) === null), registerNativeResponsePresentationWork: (coordinator: Parameters[1], canRun: () => boolean) => registerNativeResponsePresentationWork(db, coordinator, () => canRun() && readTaskDrain(new Date()) === null), registerNativePendingEffectWork: (coordinator: Parameters[1], canRun: () => boolean) => diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index a6fbf3432d7..7225dba15b5 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; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean; skipNativePendingEffects?: boolean; skipNativeResponsePresentation?: boolean }) { + async function reapOrphanedRuns(opts?: { staleThresholdMs?: number; skipScheduledRetryCancellation?: boolean; skipCapturedInstructionCommit?: boolean; skipInstructionDirectoryCleanup?: boolean; skipStoppedInstructionCollection?: boolean; skipCancelledWorkspacePreservation?: boolean; skipObsoletePolicyReviews?: boolean; skipAutomaticCompletionReviews?: boolean; skipUnsafeWorkspaceExports?: boolean; skipNativePendingEffects?: boolean; skipNativeResponsePresentation?: boolean; skipNativeReviewResponse?: boolean }) { if (!opts?.skipCancelledWorkspacePreservation) await preserveCancelledNativeWorkspaces(db); const staleThresholdMs = opts?.staleThresholdMs ?? 0; const now = new Date(); @@ -1970,6 +1970,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD skipUnsafeWorkspaceExports: opts?.skipUnsafeWorkspaceExports, skipNativePendingEffects: opts?.skipNativePendingEffects, skipNativeResponsePresentation: opts?.skipNativeResponsePresentation, + skipNativeReviewResponse: opts?.skipNativeReviewResponse, environmentRuntime, pluginWorkerManager: dependencies.pluginWorkerManager, onWorkspaceSettled: settleRecoveredNativeWorkspace, diff --git a/server/src/services/issue-approvals.ts b/server/src/services/issue-approvals.ts index 682262ca899..62b36fec675 100644 --- a/server/src/services/issue-approvals.ts +++ b/server/src/services/issue-approvals.ts @@ -117,7 +117,7 @@ export function issueApprovalService(db: Db) { linkedByAgentId: actor?.agentId ?? null, linkedByUserId: actor?.userId ?? null, }) - .onConflictDoNothing()); + .onConflictDoNothing(), { issueIds: [issueId] }); return db .select() @@ -130,7 +130,7 @@ export function issueApprovalService(db: Db) { const { issue } = await assertIssueAndApprovalSameCompany(issueId, approvalId); await writeApprovalChange(db, issue.companyId, tx => tx .delete(issueApprovals) - .where(and(eq(issueApprovals.issueId, issueId), eq(issueApprovals.approvalId, approvalId)))); + .where(and(eq(issueApprovals.issueId, issueId), eq(issueApprovals.approvalId, approvalId))), { issueIds: [issueId] }); }, linkManyForApproval: async (approvalId: string, issueIds: string[], actor?: LinkActor) => { @@ -169,7 +169,7 @@ export function issueApprovalService(db: Db) { linkedByUserId: actor?.userId ?? null, })), ) - .onConflictDoNothing()); + .onConflictDoNothing(), { issueIds: uniqueIssueIds }); }, }; } diff --git a/server/src/services/issue-thread-interactions.ts b/server/src/services/issue-thread-interactions.ts index 081bf96d80c..295cf894a36 100644 --- a/server/src/services/issue-thread-interactions.ts +++ b/server/src/services/issue-thread-interactions.ts @@ -1,3 +1,4 @@ +import { notifyNativeReviewResponse } from "./native-review-response-notifications.js"; import { notifyChatConfirmationResolution } from "./chat-confirmation-notifications.js"; import { notifyChatRunMilestones } from "./chat-run-milestone-notifications.js"; import { notifyToolReviewChange } from "./tool-review-work-notifications.js"; @@ -1944,6 +1945,7 @@ async function expireStaleRequestConfirmationTarget( const expireInTransaction = async (tx: Db | any) => { await notifyChatConfirmationResolution(tx, args.row); if (args.row.kind === "request_confirmation") await notifyChatRunMilestones(tx, args.row); + else await notifyNativeReviewResponse(tx, args.row); const [updated] = await tx .update(issueThreadInteractions) .set({ @@ -2223,6 +2225,7 @@ export function issueThreadInteractionService( await assertInteractionRunWriteAllowed(tx as unknown as Db, args.issue, args.actor); await notifyChatConfirmationResolution(tx, args.current); if (args.current.kind === "request_confirmation") await notifyChatRunMilestones(tx, args.current); + else await notifyNativeReviewResponse(tx, args.current); await args.mutationOptions?.beforeResolveInTransaction?.(tx); // Policy mutations and review transitions use the same issue-row lock, // so the authoritative review policy and requester are stable through @@ -2513,6 +2516,7 @@ export function issueThreadInteractionService( await assertInteractionRunWriteAllowed(tx as unknown as Db, args.issue, args.actor); await notifyChatConfirmationResolution(tx, args.current); if (args.current.kind === "request_confirmation") await notifyChatRunMilestones(tx, args.current); + else await notifyNativeReviewResponse(tx, args.current); await args.mutationOptions?.beforeResolveInTransaction?.(tx); const issueContext = await tx .select({ @@ -2905,6 +2909,7 @@ export function issueThreadInteractionService( const updated = await db.transaction(async (tx) => { await notifyChatConfirmationResolution(tx, current); await mutationOptions.beforeResolveInTransaction?.(tx); + await notifyNativeReviewResponse(tx, current); const [row] = await tx .update(issueThreadInteractions) .set({ @@ -3340,6 +3345,7 @@ export function issueThreadInteractionService( const updated = await db.transaction(async (tx) => { await notifyChatConfirmationResolution(tx, row); if (["ask_user_questions", "request_confirmation"].includes(row.kind)) await notifyChatRunMilestones(tx, row); + else await notifyNativeReviewResponse(tx, row); if (row.kind === "request_confirmation") await notifyToolProjectionForInteractions(tx, row.companyId, [row.id]); const [resolved] = await tx .update(issueThreadInteractions) @@ -3427,6 +3433,7 @@ export function issueThreadInteractionService( const updated = await db.transaction(async (tx) => { await notifyChatConfirmationResolution(tx, row); if (["ask_user_questions", "request_confirmation"].includes(row.kind)) await notifyChatRunMilestones(tx, row); + else await notifyNativeReviewResponse(tx, row); if (row.kind === "request_confirmation") await notifyToolProjectionForInteractions(tx, row.companyId, [row.id]); const [updatedRow] = await tx .update(issueThreadInteractions) @@ -4208,6 +4215,7 @@ export function issueThreadInteractionService( } } } + await notifyNativeReviewResponse(tx, { companyId: issue.companyId, issueId: issue.id }); const resolvedAt = new Date(); const [claimed] = await tx .update(issueThreadInteractions) @@ -4471,6 +4479,7 @@ export function issueThreadInteractionService( complete, items, } satisfies RequestItemVerdictsResult; + if (complete) await notifyNativeReviewResponse(tx, { companyId: issue.companyId, issueId: issue.id }); const [updated] = await tx .update(issueThreadInteractions) .set({ @@ -4537,6 +4546,7 @@ export function issueThreadInteractionService( await notifyChatConfirmationResolution(tx, current); await mutationOptions.beforeResolveInTransaction?.(tx); await assertInteractionRunWriteAllowed(tx as unknown as Db, issue, actor); + await notifyNativeReviewResponse(tx, { companyId: issue.companyId, issueId: issue.id }); const rows = await tx.update(issueThreadInteractions) .set({ status: "rejected", @@ -4628,6 +4638,7 @@ export function issueThreadInteractionService( const updated = await db.transaction(async (tx) => { await notifyChatConfirmationResolution(tx, row); if (["ask_user_questions", "request_confirmation"].includes(row.kind)) await notifyChatRunMilestones(tx, row); + else await notifyNativeReviewResponse(tx, row); if (row.kind === "connection_intent") { await tx .delete(toolOauthStates) @@ -4805,6 +4816,7 @@ export function issueThreadInteractionService( const updatedInteractions = await db.transaction(async (tx) => { await notifyChatConfirmationResolution(tx, sampleConfirmationRow); if (commentRows.some(row => row.kind === "request_confirmation")) await notifyChatRunMilestones(tx, sampleConfirmationRow); + else await notifyNativeReviewResponse(tx, sampleConfirmationRow); await notifyToolProjectionForInteractions(tx, issue.companyId, confirmationRowIds); const updatedRows = await tx .update(issueThreadInteractions) @@ -4840,6 +4852,7 @@ export function issueThreadInteractionService( for (const row of itemVerdictRows) { const updated = await db.transaction(async tx => { + await notifyNativeReviewResponse(tx, row); const [retired] = await tx .update(issueThreadInteractions) .set({ @@ -4865,6 +4878,7 @@ export function issueThreadInteractionService( for (const row of connectionIntentRows) { const [updated] = await db.transaction(async (tx) => { + await notifyNativeReviewResponse(tx, row); await notifyChatConfirmationResolution(tx, row); await tx .delete(toolOauthStates) @@ -4953,6 +4967,7 @@ export function issueThreadInteractionService( const updated = await db.transaction(async (tx) => { await notifyChatConfirmationResolution(tx, row); if (row.kind === "request_confirmation") await notifyChatRunMilestones(tx, row); + else await notifyNativeReviewResponse(tx, row); const [resolved] = await tx .update(issueThreadInteractions) .set({ @@ -5016,6 +5031,7 @@ export function issueThreadInteractionService( ? "the AI account" : `the ${payload?.serviceName?.trim() || "requested"} connection`; const updated = await db.transaction(async tx => { + await notifyNativeReviewResponse(tx, { companyId: issue.companyId, issueId: issue.id }); await notifyChatConfirmationResolution(tx, { companyId: issue.companyId, issueId: issue.id, kind: "connection_intent" }); const [resolved] = await tx .update(issueThreadInteractions) @@ -5077,6 +5093,7 @@ export function issueThreadInteractionService( .transaction(async (tx) => { await notifyChatConfirmationResolution(tx, row); if (["ask_user_questions", "request_confirmation"].includes(row.kind)) await notifyChatRunMilestones(tx, row); + else await notifyNativeReviewResponse(tx, row); if (row.kind === "connection_intent") { await tx .delete(toolOauthStates) @@ -5169,6 +5186,7 @@ export function issueThreadInteractionService( await assertInteractionRunWriteAllowed(tx as unknown as Db, issue, actor); await notifyChatConfirmationResolution(tx, current); if (["ask_user_questions", "request_confirmation"].includes(current.kind)) await notifyChatRunMilestones(tx, current); + else await notifyNativeReviewResponse(tx, current); await resolveLinkedToolActionRequests(tx, current, { status: "cancelled", fromStatuses: ["pending", "approved"], @@ -5431,6 +5449,7 @@ export function issueThreadInteractionService( await assertInteractionRunWriteAllowed(tx as unknown as Db, issue, actor); await notifyChatConfirmationResolution(tx, current); if (["ask_user_questions", "request_confirmation"].includes(current.kind)) await notifyChatRunMilestones(tx, current); + else await notifyNativeReviewResponse(tx, current); await resolveLinkedToolActionRequests(tx, current, { status: "cancelled", fromStatuses: ["pending", "approved"], diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index 04387d35298..23a0f9c1c99 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -11453,7 +11453,9 @@ export function issueService(db: Db) { await notifyDecisionTargetWork(tx, `issue:${id}`); } if ((patch.status !== undefined && patch.status !== receiptExisting.status) || - (patch.assigneeAgentId !== undefined && patch.assigneeAgentId !== receiptExisting.assigneeAgentId)) { + (patch.assigneeAgentId !== undefined && patch.assigneeAgentId !== receiptExisting.assigneeAgentId) || + (patch.executionState !== undefined && parseObject(receiptExisting.executionState).status === "pending" && + parseObject(patch.executionState).status !== "pending")) { await notifyChatResponseAvailability(tx, { companyId: receiptExisting.companyId, issueId: id }); } if (patch.status !== undefined && patch.status !== receiptExisting.status) { diff --git a/server/src/services/native-finalization-notifications.ts b/server/src/services/native-finalization-notifications.ts index f14f1e79db8..91ccfc2023d 100644 --- a/server/src/services/native-finalization-notifications.ts +++ b/server/src/services/native-finalization-notifications.ts @@ -1,10 +1,14 @@ +import { notifyNativeReviewResponse } from "./native-review-response-notifications.js"; import { notifyNativeResponsePresentation } from "./native-response-presentation-notifications.js"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; export type NativeFinalizationWorkScope = { companyId: string; runId?: string; issueId?: string }; export async function notifyNativeFinalizationWork(transaction: object, scope: NativeFinalizationWorkScope) { - if (scope.issueId || scope.runId) await notifyNativeResponsePresentation(transaction, scope); + if (scope.issueId || scope.runId) { + await notifyNativeResponsePresentation(transaction, scope); + await notifyNativeReviewResponse(transaction, scope); + } await notifyDeliveryWork(transaction, DELIVERY_QUEUES.nativePendingEffects, JSON.stringify({ companyId: scope.companyId, runId: scope.runId })); } diff --git a/server/src/services/native-review-response-notifications.ts b/server/src/services/native-review-response-notifications.ts new file mode 100644 index 00000000000..39342be01c3 --- /dev/null +++ b/server/src/services/native-review-response-notifications.ts @@ -0,0 +1,8 @@ +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; + +export type NativeReviewResponseScope = { companyId: string; issueId?: string; runId?: string; agentId?: string }; + +export function notifyNativeReviewResponse(tx: object, scope: NativeReviewResponseScope) { + return notifyDeliveryWork(tx, DELIVERY_QUEUES.nativeReviewResponse, + JSON.stringify({ companyId: scope.companyId, issueId: scope.issueId, runId: scope.runId, agentId: scope.agentId })); +} diff --git a/server/src/services/native-runtime/external-chat-wait.integration.test.ts b/server/src/services/native-runtime/external-chat-wait.integration.test.ts index 50d7910bb96..f2951d3b8c0 100644 --- a/server/src/services/native-runtime/external-chat-wait.integration.test.ts +++ b/server/src/services/native-runtime/external-chat-wait.integration.test.ts @@ -1,3 +1,6 @@ +import { notifyChatRunMilestones } from "../chat-run-milestone-notifications.js"; +import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; +import { accessService } from "../access.js"; import { initializeRunIdentity } from "../run-identity.js"; import { createHash, randomUUID } from "node:crypto"; import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; @@ -28,6 +31,7 @@ import { completionContracts, closeRegisteredClients, createDb, + subscribeDatabaseWork, heartbeatRuns, issueComments, issueAttachments, @@ -41,6 +45,7 @@ import { statusDecisions, toolApplications, toolConnections, + toolOauthStates, chatVoiceSessions, chatVoiceReplies, chatVoiceToolCalls, workspaceOperations, } from "@paperclipai/db"; @@ -51,7 +56,17 @@ import type { } from "../../vendor/paperclip-runner/index.js"; import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; -import { finalizeNativeRun } from "./native-run-finalizer.js"; +import { finalizeNativeRun, recoverCommittedNativeReviewResponse } from "./native-run-finalizer.js"; +import { commitNativeStatusDecision } from "./status-decision-committer.js"; +import { configureAgentLifecycle, createAgentLifecycle, processAgentLifecycle } from "../agent-lifecycle.js"; +import { createOAuthStateStore } from "../tool-access/oauth/state.js"; +import { approvalRecords } from "../approvals.js"; +import { issueApprovalService } from "../issue-approvals.js"; +import { NativeRunCoordinatorStore } from "./native-run-coordinator-store.js"; +import { nativeReviewResponseWork } from "./native-review-response-work.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { idleWorkSnapshot } from "../task-admission.js"; import { nativeCompletionFeedback } from "./native-completion-feedback.js"; import { hasPublishedGitHubReview, readInheritedGitHubReviewWait } from "../chat-channels/github/review-wait.js"; import * as chatWaitAuthority from "./chat-attachment-reuse.js"; @@ -422,6 +437,10 @@ describe("native external-chat response wait", () => { userId, commentId, principalId, + storeBinding: { companyId, issueId, runId, agentId, normalizedSessionId: sessionId, + runnerSourceInstanceId: runnerInstanceId, completionContractId: contractId, + completionContractSha256: contractSha256, completionContractRevision: "external-chat-wait-v1", + completionContractCriterionIds: ["response"] }, }; } @@ -2869,6 +2888,653 @@ describe("native external-chat response wait", () => { return run!.resultJson!; } + async function seedRecoverableReviewResponse() { + const fixture = await seedWaitTurn(); + const gate = await seedPriorCompletionReview(fixture); + await finalizeNativeRun({ db, runId: fixture.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: false }); + await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() }) + .where(eq(heartbeatRuns.id, fixture.runId)); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId)); + expect(coordinator!.phase).toBe("committed"); + const input = { companyId: fixture.companyId, issueId: fixture.issueId, runId: fixture.runId, + decisionId: coordinator!.decisionId!, resultId: coordinator!.resultId!, assessmentId: coordinator!.assessmentId!, onlyWhenDue: true }; + return { ...fixture, gate, input }; + } + + it.each(["reconciler", "finalizer"] as const)("defers periodic governed-response %s repair to its registered worker", async path => { + const fixture = await seedRecoverableReviewResponse(); + await db.update(heartbeatRuns).set({ resultJson: sql`${heartbeatRuns.resultJson} - 'externalChatReviewPresentation'` }) + .where(eq(heartbeatRuns.id, fixture.runId)); + if (path === "reconciler") await reconcileNativeFinalizations(db, [fixture.runId], { skipNativeReviewResponse: true }); + else await finalizeNativeRun({ db, runId: fixture.runId, workspaceFinalizeStatus: "succeeded", + projectRunStatus: true, skipNativeReviewResponse: true }); + expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(0); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + expect(run.resultJson?.externalChatReviewPresentation).toBeUndefined(); + const work = nativeReviewResponseWork(db); + const [entry] = await work.load({ companyId: fixture.companyId, runId: fixture.runId }); + expect(entry?.id).toBe(fixture.runId); + expect(await work.check(entry)).toBeNull(); + expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(1); + }); + + it("recovers scheduled governed responses once under concurrent callers and preserves deleted receipts", async () => { + const fixture = await seedRecoverableReviewResponse(); + const work = nativeReviewResponseWork(db); + const [entry] = await work.load({ companyId: fixture.companyId }); + expect(entry?.id).toBe(fixture.runId); + expect(await Promise.all([recoverCommittedNativeReviewResponse(db, fixture.input), + recoverCommittedNativeReviewResponse(db, fixture.input)])).toEqual(expect.arrayContaining([true, false])); + const comments = await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId)); + expect(comments).toHaveLength(1); + expect(await db.select().from(chatPublications).where(eq(chatPublications.commentId, comments[0].id))).toHaveLength(1); + expect(await work.load({ companyId: fixture.companyId })).toEqual([]); + await db.update(issueComments).set({ deletedAt: new Date() }).where(eq(issueComments.id, comments[0].id)); + expect(await work.load({ companyId: fixture.companyId })).toEqual([]); + expect(await recoverCommittedNativeReviewResponse(db, fixture.input)).toBe(false); + expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(1); + }); + + it("lets status replay finish while governed response recovery waits for its coordinator", async () => { + const fixture = await seedRecoverableReviewResponse(); + const [issue] = await db.select().from(issues).where(eq(issues.id, fixture.issueId)); + const [decision] = await db.select().from(statusDecisions).where(eq(statusDecisions.id, fixture.input.decisionId)); + let recovery: Promise<{ value?: boolean; error?: unknown }> | undefined; + try { + await db.transaction(async tx => { + await tx.select().from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, fixture.runId)).for("update"); + const [backend] = await tx.execute<{ pid: number }>(sql`select pg_backend_pid() as pid`); + recovery = recoverCommittedNativeReviewResponse(db, fixture.input) + .then(value => ({ value }), error => ({ error })); + await vi.waitFor(async () => { + const [waiting] = await db.execute<{ count: number }>(sql` + select count(*)::int as count from pg_stat_activity + where datname = current_database() and wait_event_type = 'Lock' + and ${backend.pid} = any(pg_blocking_pids(pid))`); + expect(waiting.count).toBe(1); + }); + + + await tx.execute(sql`set local lock_timeout = '500ms'`); + const replay = await commitNativeStatusDecision({ + db: tx as unknown as typeof db, companyId: fixture.companyId, + issueId: fixture.issueId, runId: fixture.runId, + assessmentId: fixture.input.assessmentId, + priorStatus: issue.status, priorStatusVersion: issue.statusVersion, + priorDecisionId: issue.lastStatusDecisionId, + decision: decision.decisionJson as unknown as Parameters[0]["decision"], + }); + expect(replay.replayed).toBe(true); + }); + } finally { + if (recovery) { + const result = await recovery; + expect(result.error).toBeUndefined(); + expect(result.value).toBe(true); + } + } + expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(1); + expect(await recoverCommittedNativeReviewResponse(db, fixture.input)).toBe(false); + }); + + it.each(["future", "claim"])("defers scheduled governed responses behind a new %s under the owner lock", async kind => { + const fixture = await seedRecoverableReviewResponse(); + const work = nativeReviewResponseWork(db); + const [entry] = await work.load({ companyId: fixture.companyId }); + const at = new Date(Date.now() + 60_000); + const transaction = db.transaction.bind(db); + const intercept = vi.spyOn(db, "transaction").mockImplementationOnce((async (callback, options) => { + await transaction(async tx => { + await tx.update(nativeRunFinalizations).set(kind === "future" ? { nextAttemptAt: at } + : { leaseOwner: "another-finalizer", leaseExpiresAt: at }).where(eq(nativeRunFinalizations.runId, fixture.runId)); + }); + return transaction(callback, options); + }) as typeof db.transaction); + try { + expect(await work.check(entry)).toEqual(kind === "future" ? { ...entry, dueAt: at.getTime() } : null); + } finally { intercept.mockRestore(); } + expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(0); + await db.update(nativeRunFinalizations).set({ nextAttemptAt: null, leaseOwner: null, leaseExpiresAt: null }) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + expect(await recoverCommittedNativeReviewResponse(db, fixture.input)).toBe(true); + }); + + it("propagates scheduled governed response contention and rolls back its proof restoration", async () => { + const fixture = await seedRecoverableReviewResponse(); + await db.update(heartbeatRuns).set({ resultJson: sql`${heartbeatRuns.resultJson} - 'externalChatReviewPresentation'` }) + .where(eq(heartbeatRuns.id, fixture.runId)); + let release!: () => void, observed!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const locked = new Promise(resolve => { observed = resolve; }); + const holder = db.transaction(async tx => { + await tx.select().from(chatEndpoints).where(eq(chatEndpoints.id, fixture.endpointId)).for("update"); + observed(); await hold; + }); + await locked; + try { + await expect(recoverCommittedNativeReviewResponse(db, fixture.input)).rejects.toBeDefined(); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + expect(run.resultJson?.externalChatReviewPresentation).toBeUndefined(); + expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(0); + } finally { release(); await holder; } + expect(await recoverCommittedNativeReviewResponse(db, fixture.input)).toBe(true); + }); + + it.each(["gate", "agent", "principal", "company"])("leaves scheduled governed response authority with its owner after %s changes", async kind => { + const fixture = await seedRecoverableReviewResponse(); + if (kind === "gate") await db.update(issueThreadInteractions).set({ status: "accepted" }).where(eq(issueThreadInteractions.id, fixture.gate.id)); + if (kind === "agent") await db.update(agents).set({ status: "paused" }).where(eq(agents.id, fixture.agentId)); + if (kind === "principal") await db.update(companyMemberships).set({ status: "suspended" }) + .where(and(eq(companyMemberships.companyId, fixture.companyId), eq(companyMemberships.principalId, fixture.userId))); + expect(await recoverCommittedNativeReviewResponse(db, { ...fixture.input, + ...(kind === "company" ? { companyId: randomUUID() } : {}) })).toBe(false); + expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(0); + }); + + it.each(["milestone", "finalization"] as const)("ignores repeated unscoped routine %s hints without reconstructing dormant governed responses", async source => { + const fixture = await seedRecoverableReviewResponse(); + await db.update(companyMemberships).set({ status: "suspended" }).where(and( + eq(companyMemberships.companyId, fixture.companyId), eq(companyMemberships.principalId, fixture.userId))); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeReviewResponseWork(db).register(coordinator, () => true); + try { + await worker.ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), { timeout: 10_000 }) + .toEqual({ active: 0, next: null }); + const debug = db.$client.options.debug; + const queries: string[] = []; + db.$client.options.debug = (_connection, query) => { + if (query.startsWith("select") && query.includes("native_run_finalizations")) queries.push(query); + }; + try { + for (let i = 0; i < 3; i++) { + await db.transaction(tx => source === "milestone" ? notifyChatRunMilestones(tx, { companyId: fixture.companyId }) + : notifyNativeFinalizationWork(tx, { companyId: fixture.companyId })); + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), { timeout: 5000 }) + .toEqual({ active: 0, next: null }); + } + expect(queries).toEqual([]); + } finally { db.$client.options.debug = debug; } + expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(0); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + + it("wakes dormant governed responses through the real company membership restoration owner", async () => { + const fixture = await seedRecoverableReviewResponse(); + const [member] = await db.select().from(companyMemberships).where(and( + eq(companyMemberships.companyId, fixture.companyId), eq(companyMemberships.principalId, fixture.userId))); + await db.update(companyMemberships).set({ status: "suspended" }).where(eq(companyMemberships.id, member.id)); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeReviewResponseWork(db).register(coordinator, () => true); + try { + await worker.ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), { timeout: 10_000 }) + .toEqual({ active: 0, next: null }); + expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(0); + await accessService(db).updateMember(fixture.companyId, member.id, { status: "active" }); + await expect.poll(async () => ({ count: (await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).length, + active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), { timeout: 5000 }) + .toEqual({ count: 1, active: 0, next: null }); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + + it.each(["approve", "reject", "cancel", "unlink"] as const)("reactivates dormant governed responses after approval %s", async action => { + const fixture = await seedRecoverableReviewResponse(); + const [approval] = await db.insert(approvals).values({ companyId: fixture.companyId, + type: "hire_agent", status: "pending", payload: {} }).returning(); + await issueApprovalService(db).link(fixture.issueId, approval.id); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeReviewResponseWork(db).register(coordinator, () => true); + const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId)); + try { + await worker.ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 10_000 }).toEqual({ active: 0, next: null }); + expect(await comments()).toHaveLength(0); + if (action === "unlink") await issueApprovalService(db).unlink(fixture.issueId, approval.id); + else if (action === "cancel") await approvalRecords(db).cancel(approval.id); + else await approvalRecords(db).resolveApproval(approval.id, action === "approve" ? "approved" : "rejected", fixture.userId, null); + await expect.poll(async () => (await comments()).length, { timeout: 5_000 }).toBe(1); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + + it.each([ + ["resolver", "single"], ["resolver", "bulk"], + ["linker", "single"], ["linker", "bulk"], + ] as const)("retains a settled task wake when %s starts first during concurrent %s approval linking", async (first, kind) => { + const fixture = await seedWaitTurn(); + const [approval] = await db.insert(approvals).values({ companyId: fixture.companyId, + type: "request_board_approval", status: "pending", payload: {} }).returning(); + const observed: string[] = []; + const reads: Promise[] = []; + const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.nativeReviewResponse, (event, hints) => { + if (event !== "settled" || ![...(hints ?? [])].some(hint => { + const scope = JSON.parse(hint); + return scope.companyId === fixture.companyId && scope.issueId === fixture.issueId; + })) return; + reads.push(db.select({ status: approvals.status }).from(issueApprovals) + .innerJoin(approvals, eq(issueApprovals.approvalId, approvals.id)) + .where(and(eq(issueApprovals.issueId, fixture.issueId), eq(approvals.id, approval.id))) + .then(rows => { for (const row of rows) observed.push(row.status); })); + }); + let release!: () => void, ready!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const held = new Promise(resolve => { ready = resolve; }); + let holderPid = 0, resolverPid = 0, linkerPid = 0; + let linked = false, resolved = false; + const link = async (tx: typeof db) => { + if (kind === "single") await issueApprovalService(tx).link(fixture.issueId, approval.id); + else await issueApprovalService(tx).linkManyForApproval(approval.id, [fixture.issueId]); + }; + const holder = db.transaction(async tx => { + const [backend] = await tx.execute(sql`select pg_backend_pid() as pid`); + holderPid = Number(backend.pid); + if (first === "resolver") { + + + await tx.execute(sql`lock table approvals in share mode`); + } else { + await link(tx as unknown as typeof db); + } + ready(); + await hold; + }); + let resolver: Promise | undefined, linker: Promise | undefined; + const blocked = async (pid: number, by: number) => { + if (!pid || !by) return false; + const [row] = await db.execute(sql`select ${by} = any(pg_blocking_pids(${pid})) as blocked`); + return row.blocked === true; + }; + try { + await held; + resolver = db.transaction(async tx => { + const [backend] = await tx.execute(sql`select pg_backend_pid() as pid`); + resolverPid = Number(backend.pid); + await approvalRecords(tx as unknown as typeof db).resolveApproval(approval.id, "approved", fixture.userId, null); + }).then(() => { resolved = true; }); + await expect.poll(async () => resolved || await blocked(resolverPid, holderPid), { timeout: 5000 }).toBe(true); + if (first === "resolver") { + linker = db.transaction(async tx => { + const [backend] = await tx.execute(sql`select pg_backend_pid() as pid`); + linkerPid = Number(backend.pid); + await link(tx as unknown as typeof db); + }).then(() => { linked = true; }); + await expect.poll(async () => linked || await blocked(linkerPid, resolverPid), { timeout: 5000 }).toBe(true); + } + + + await Promise.all(reads); + release(); + await Promise.all([holder, resolver, linker]); + await Promise.all(reads); + expect(observed).toContain("approved"); + } finally { + release(); + await Promise.allSettled([holder, resolver, linker]); + await Promise.all(reads); + unsubscribe(); + } + }); + + it.each(["cancel", "unlink"] as const)("keeps governed response approval %s behind outer commit and scoped to linked tasks", async action => { + const fixture = await seedRecoverableReviewResponse(); + const [approval, unrelated] = await db.insert(approvals).values([ + { companyId: fixture.companyId, type: "hire_agent", status: "pending", payload: {} }, + { companyId: fixture.companyId, type: "hire_agent", status: "pending", payload: {} }, + ]).returning(); + await issueApprovalService(db).link(fixture.issueId, approval.id); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeReviewResponseWork(db).register(coordinator, () => true); + const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId)); + const mutate = async (tx: typeof db) => { + if (action === "cancel") await approvalRecords(tx).cancel(approval.id); + else await issueApprovalService(tx).unlink(fixture.issueId, approval.id); + }; + try { + await worker.ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 10_000 }).toEqual({ active: 0, next: null }); + const previousDebug = db.$client.options.debug; + const queries: string[] = []; + db.$client.options.debug = (_connection, query) => { if (query.includes('"native_run_finalizations"')) queries.push(query); }; + try { + await approvalRecords(db).cancel(unrelated.id); + await expect.poll(() => idleWorkSnapshot().active).toBe(0); + expect(queries).toEqual([]); + } finally { db.$client.options.debug = previousDebug; } + const rollback = new Error("rollback governed approval change"); + await expect(db.transaction(async tx => { + await mutate(tx as unknown as typeof db); + expect(await comments()).toHaveLength(0); + throw rollback; + })).rejects.toBe(rollback); + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 10_000 }).toEqual({ active: 0, next: null }); + expect(await comments()).toHaveLength(0); + await db.transaction(async tx => { + await mutate(tx as unknown as typeof db); + expect(await comments()).toHaveLength(0); + }); + await expect.poll(async () => (await comments()).length, { timeout: 5_000 }).toBe(1); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + + it("wakes dormant governed response recovery after real agent resume", async () => { + const fixture = await seedRecoverableReviewResponse(); + await db.update(agents).set({ status: "paused", lifecycleState: "paused", lifecycleHolds: ["manual"], lifecycleOperation: null }) + .where(eq(agents.id, fixture.agentId)); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeReviewResponseWork(db).register(coordinator, () => true); + const lifecycle = configureAgentLifecycle(db, { + requiredPluginIds: async () => [], runHost: async () => "complete", runPlugin: async () => "complete", + }); + try { + await worker.ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 10_000 }).toEqual({ active: 0, next: null }); + expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(0); + await createAgentLifecycle(db).resumeAgent(fixture.agentId); + await processAgentLifecycle(db, fixture.agentId); + await expect.poll(async () => (await db.select().from(agents).where(eq(agents.id, fixture.agentId)))[0].status).toBe("idle"); + await expect.poll(async () => (await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).length, + { timeout: 5_000 }).toBe(1); + expect(errors).toEqual([]); + } finally { await lifecycle.stop(); await coordinator.stop(); } + }); + + it.each(["skip", "withdraw", "reject_suggestions", "complete_verdicts", "decline_connection"] as const)( + "reactivates governed responses after competing interaction %s", async action => { + const fixture = await seedRecoverableReviewResponse(); + const service = issueThreadInteractionService(db); + const issue = { id: fixture.issueId, companyId: fixture.companyId }; + const actor = { userId: fixture.userId }; + const competing = action === "decline_connection" + ? await service.createConnectionIntent(issue, { sourceRunId: fixture.runId, + addresseeUserId: fixture.userId, idempotencyKey: `review-competing:${randomUUID()}`, + payload: { version: 1, serviceSlug: "notion", serviceName: "Notion", serviceLogoUrl: null, + requestingAgentId: fixture.agentId, requestingAgentName: "Agent", phase: "requested" } }) + : action === "complete_verdicts" + ? await service.create(issue, { kind: "request_item_verdicts", payload: { version: 1, + prompt: "Review this item", items: [{ id: "one", label: "One" }] } }, actor) + : await service.create(issue, { kind: "suggest_tasks", payload: { version: 1, + tasks: [{ clientKey: "child", title: "A suggested task" }] } }, actor); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeReviewResponseWork(db).register(coordinator, () => true); + const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId)); + try { + await worker.ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 10_000 }).toEqual({ active: 0, next: null }); + expect(await comments()).toHaveLength(0); + if (action === "skip") await service.skipInteraction(issue, competing.id, {}, actor); + else if (action === "withdraw") await service.withdrawInteraction(issue, competing.id, {}, actor); + else if (action === "reject_suggestions") await service.rejectInteraction(issue, competing.id, {}, actor); + else if (action === "complete_verdicts") await service.submitItemVerdicts(issue, competing.id, + { verdicts: [{ id: "one", verdict: "approve" }] }, actor); + else await service.resolveConnectionIntent(issue, competing.id, { version: 1, outcome: "declined" }, actor); + await expect.poll(async () => (await comments()).length, { timeout: 5_000 }).toBe(1); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + + it.each(["status", "assignment"] as const)("preserves governed response policy through task %s changes", async change => { + const fixture = await seedRecoverableReviewResponse(); + await db.update(issues).set(change === "status" ? { status: "blocked" } : { assigneeAgentId: null }) + .where(eq(issues.id, fixture.issueId)); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeReviewResponseWork(db).register(coordinator, () => true); + const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId)); + try { + await worker.ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 10_000 }).toEqual({ active: 0, next: null }); + expect(await comments()).toHaveLength(0); + if (change === "assignment") { + await expect(issueService(db).update(fixture.issueId, { assigneeAgentId: fixture.agentId })) + .rejects.toThrow("Agent assignment cannot change while this task is bound to an external channel"); + expect(await comments()).toHaveLength(0); + } else { + await issueService(db).update(fixture.issueId, { status: "in_review" }); + await expect.poll(async () => (await comments()).length, { timeout: 5_000 }).toBe(1); + } + expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + + it("reactivates governed responses after OAuth confirmation denial", async () => { + const fixture = await seedRecoverableReviewResponse(); + const [connection] = await db.select().from(toolConnections).where(eq(toolConnections.companyId, fixture.companyId)); + const competing = await issueThreadInteractionService(db).create({ id: fixture.issueId, companyId: fixture.companyId }, + { kind: "request_confirmation", payload: { version: 1, prompt: "Connect your account", target: { + type: "custom", key: `connection:${connection.id}`, revisionId: randomUUID(), label: "Connect account" } } }, + { userId: fixture.userId }); + const [state] = await db.insert(toolOauthStates).values({ state: randomUUID(), companyId: fixture.companyId, + connectionId: connection.id, codeVerifier: "test-verifier", issueId: fixture.issueId, interactionId: competing.id, + subjectUserId: fixture.userId, expiresAt: new Date(Date.now() + 60_000) }).returning(); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeReviewResponseWork(db).register(coordinator, () => true); + const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId)); + try { + await worker.ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 10_000 }).toEqual({ active: 0, next: null }); + expect(await comments()).toHaveLength(0); + const rollback = new Error("rollback OAuth denial"); + await expect(db.transaction(async tx => { + await createOAuthStateStore(tx as unknown as typeof db).rejectPendingOAuthInteraction(state, + { actorType: "user", actorId: fixture.userId }); + expect(await comments()).toHaveLength(0); + throw rollback; + })).rejects.toBe(rollback); + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 10_000 }).toEqual({ active: 0, next: null }); + expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, competing.id)))[0].status).toBe("pending"); + await db.transaction(async tx => { + await createOAuthStateStore(tx as unknown as typeof db).rejectPendingOAuthInteraction(state, + { actorType: "user", actorId: fixture.userId }); + expect(await comments()).toHaveLength(0); + }); + await expect.poll(async () => (await comments()).length, { timeout: 5_000 }).toBe(1); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + + it("reactivates governed responses when pending execution policy is cleared without changing task status", async () => { + const fixture = await seedRecoverableReviewResponse(); + await db.update(issues).set({ executionState: { status: "pending", currentStageId: randomUUID(), + currentStageIndex: 0, currentStageType: "review", currentParticipant: { type: "user", userId: fixture.userId, agentId: null }, + returnAssignee: { type: "agent", agentId: fixture.agentId, userId: null }, reviewRequest: null, + completedStageIds: [], lastDecisionId: null, lastDecisionOutcome: null, monitor: null, changesRequestedCount: 0 } }) + .where(eq(issues.id, fixture.issueId)); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeReviewResponseWork(db).register(coordinator, () => true); + const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId)); + try { + await worker.ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 10_000 }).toEqual({ active: 0, next: null }); + expect(await comments()).toHaveLength(0); + await issueService(db).updateForCompany(fixture.issueId, fixture.companyId, { executionState: null }); + expect((await db.select().from(issues).where(eq(issues.id, fixture.issueId)))[0].status).toBe("in_review"); + await expect.poll(async () => (await comments()).length, { timeout: 5_000 }).toBe(1); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + + it("wakes governed responses from the real coordinator retry release", async () => { + const fixture = await seedRecoverableReviewResponse(); + const store = new NativeRunCoordinatorStore(db, fixture.storeBinding); + await store.claimFinalization({ leaseOwner: "review-response-fixture", leaseTtlMs: 30_000 }); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeReviewResponseWork(db).register(coordinator, () => true); + const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId)); + try { + await worker.ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 10_000 }).toEqual({ active: 0, next: null }); + expect(await comments()).toHaveLength(0); + await store.markFinalizationRetry({ leaseOwner: "review-response-fixture", failureCode: "review_response_retry", retryAfterMs: 1000 }); + expect(await comments()).toHaveLength(0); + await expect.poll(async () => (await comments()).length, { timeout: 5_000 }).toBe(1); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); } + }); + + it("retains the first governed-response wake through idle drain without passive SQL", async () => { + const fixture = await seedRecoverableReviewResponse(); + const [approval] = await db.insert(approvals).values({ companyId: fixture.companyId, + type: "hire_agent", status: "pending", payload: {} }).returning(); + await issueApprovalService(db).link(fixture.issueId, approval.id); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + let canRun = true; + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => true, onError: error => errors.push(error) }); + const worker = nativeReviewResponseWork(db).register(coordinator, () => canRun); + const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId)); + try { + await worker.ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 10_000 }).toEqual({ active: 0, next: null }); + canRun = false; + await approvalRecords(db).cancel(approval.id); + await expect.poll(() => idleWorkSnapshot().active).toBe(0); + expect(await comments()).toHaveLength(0); + expect(coordinator.nextWakeAt()).not.toBeNull(); + const previousDebug = db.$client.options.debug; + const queries: string[] = []; + db.$client.options.debug = (_connection, query) => { queries.push(query); }; + try { await vi.advanceTimersByTimeAsync(120_000); expect(queries).toEqual([]); } + finally { db.$client.options.debug = previousDebug; } + canRun = true; worker.wake(); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await comments()).toHaveLength(1); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + + it("retains failed governed-response recovery across drain and retries after resume", async () => { + const fixture = await seedRecoverableReviewResponse(); + let release!: () => void, observed!: () => void; + const hold = new Promise(resolve => { release = resolve; }); + const locked = new Promise(resolve => { observed = resolve; }); + const holder = db.transaction(async tx => { + await tx.select().from(chatEndpoints).where(eq(chatEndpoints.id, fixture.endpointId)).for("update"); + observed(); await hold; + }); + await locked; + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + let canRun = true; + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => canRun, + canReconcile: () => true, onError: error => errors.push(error) }); + const worker = nativeReviewResponseWork(db).register(coordinator, () => canRun); + const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId)); + try { + await worker.ready; + await expect.poll(() => ({ errors: errors.length, active: idleWorkSnapshot().active }), + { timeout: 10_000 }).toEqual({ errors: 1, active: 0 }); + expect(await comments()).toHaveLength(0); + expect(coordinator.nextWakeAt()).toBeGreaterThan(Date.now()); + canRun = false; + worker.wake(); + await vi.waitFor(() => expect(idleWorkSnapshot().active).toBe(0)); + const previousDebug = db.$client.options.debug; + const queries: string[] = []; + db.$client.options.debug = (_connection, query) => { queries.push(query); }; + try { await vi.advanceTimersByTimeAsync(120_000); expect(queries).toEqual([]); } + finally { db.$client.options.debug = previousDebug; } + release(); await holder; + canRun = true; worker.wake(); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await comments()).toHaveLength(1); + expect(idleWorkSnapshot().active).toBe(0); + expect(coordinator.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + expect(errors).toHaveLength(1); + } finally { release(); await holder; await coordinator.stop(); vi.useRealTimers(); } + }); + + it("reconstructs the governed-response deadline after shutdown without early presentation", async () => { + const fixture = await seedRecoverableReviewResponse(); + const store = new NativeRunCoordinatorStore(db, fixture.storeBinding); + await store.claimFinalization({ leaseOwner: "review-future", leaseTtlMs: 30_000 }); + await store.markFinalizationRetry({ leaseOwner: "review-future", failureCode: "review_fixture", retryAfterMs: 60_000 }); + const [state] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId)); + const at = state.nextAttemptAt!.getTime(); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeReviewResponseWork(db).register(coordinator, () => true); + const comments = () => db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId)); + await worker.ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 10_000 }).toEqual({ active: 0, next: at }); + await coordinator.stop(); + expect(coordinator.nextWakeAt()).toBeNull(); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const previousDebug = db.$client.options.debug; + const queries: string[] = []; + db.$client.options.debug = (_connection, query) => { queries.push(query); }; + try { await vi.advanceTimersByTimeAsync(30_000); expect(queries).toEqual([]); } + finally { db.$client.options.debug = previousDebug; } + const resumed = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + try { + const restarted = nativeReviewResponseWork(db).register(resumed, () => true); + await restarted.ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: resumed.nextWakeAt() }), + { timeout: 10_000 }).toEqual({ active: 0, next: at }); + expect(await comments()).toHaveLength(0); + await vi.advanceTimersByTimeAsync(Math.max(0, at - Date.now())); + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(1000); + expect(await comments()).toHaveLength(1); + expect(idleWorkSnapshot().active).toBe(0); + expect(resumed.nextWakeAt()).toBeNull(); + }, { timeout: 10_000 }); + expect(errors).toEqual([]); + } finally { await resumed.stop(); vi.useRealTimers(); } + }); + + it("reconstructs scheduled governed responses at startup and leaves their history quiet", async () => { + const fixture = await seedRecoverableReviewResponse(); + const errors: unknown[] = []; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => errors.push(error) }); + const worker = nativeReviewResponseWork(db).register(coordinator, () => true); + try { + await worker.ready; + await expect.poll(() => ({ active: idleWorkSnapshot().active, next: coordinator.nextWakeAt() }), + { timeout: 10_000 }).toEqual({ active: 0, next: null }); + expect(await db.select().from(issueComments).where(eq(issueComments.createdByRunId, fixture.runId))).toHaveLength(1); + vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); + const previousDebug = db.$client.options.debug; + const queries: string[] = []; + db.$client.options.debug = (_connection, query) => { queries.push(query); }; + try { await vi.advanceTimersByTimeAsync(120_000); expect(queries).toEqual([]); } + finally { db.$client.options.debug = previousDebug; } + expect(errors).toEqual([]); + } finally { await coordinator.stop(); vi.useRealTimers(); } + }); + it.each(["telegram", "discord"] as const)( "presents the exact %s response without resolving a pre-existing native completion review", async (provider) => { diff --git a/server/src/services/native-runtime/native-chat-review-presentation.ts b/server/src/services/native-runtime/native-chat-review-presentation.ts index 366e564acb6..641c37c3b29 100644 --- a/server/src/services/native-runtime/native-chat-review-presentation.ts +++ b/server/src/services/native-runtime/native-chat-review-presentation.ts @@ -500,6 +500,7 @@ export async function restoreNativeChatReviewPresentationInTransaction( decisionId: string; resultId: string; assessmentId: string; + onlyWhenDue?: boolean; }, ): Promise { const [coordinator] = await db @@ -519,6 +520,8 @@ export async function restoreNativeChatReviewPresentationInTransaction( .for("update") .limit(1); if (!coordinator) return false; + if (input.onlyWhenDue && (coordinator.leaseOwner !== null + || coordinator.nextAttemptAt !== null && coordinator.nextAttemptAt.getTime() > Date.now())) return false; const [decision] = await db .select() .from(statusDecisions) diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index 5fc5567ea2e..cc97199948b 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/server/src/services/native-runtime/native-finalization-reconciler.ts @@ -645,6 +645,7 @@ type NativeReconciliationOptions = { skipUnsafeWorkspaceExports?: boolean; skipNativePendingEffects?: boolean; skipNativeResponsePresentation?: boolean; + skipNativeReviewResponse?: boolean; pluginWorkerManager?: WorkspaceWorkers; environmentRuntime?: EnvironmentRuntimeService; onWorkspaceSettled?: (input: { @@ -875,7 +876,8 @@ async function reconcileNativeFinalizationsPass( row.decisionId && row.resultId && row.assessmentId && - currentDecisionJson.externalChatReviewPresentation + currentDecisionJson.externalChatReviewPresentation && + !options.skipNativeReviewResponse ) { // A later chat turn may retain the same pending review. Recover the // earlier response independently before status reconciliation skips @@ -1125,6 +1127,7 @@ async function reconcileNativeFinalizationsPass( workspaceFinalizeStatus, projectRunStatus: true, skipNativeResponsePresentation: options.skipNativeResponsePresentation, + skipNativeReviewResponse: options.skipNativeReviewResponse, preserveProviderAttempt: Boolean( readNativeWorkspaceSyncReference( row.nativeWorkspaceSync, @@ -1184,6 +1187,7 @@ async function reconcileNativeFinalizationsPass( workspaceFinalizeStatus: barrier.status as "succeeded" | "failed", projectRunStatus: true, skipNativeResponsePresentation: options.skipNativeResponsePresentation, + skipNativeReviewResponse: options.skipNativeReviewResponse, preserveProviderAttempt: Boolean( readNativeWorkspaceSyncReference( row.nativeWorkspaceSync, diff --git a/server/src/services/native-runtime/native-review-response-work.ts b/server/src/services/native-runtime/native-review-response-work.ts new file mode 100644 index 00000000000..b11fb9b0646 --- /dev/null +++ b/server/src/services/native-runtime/native-review-response-work.ts @@ -0,0 +1,88 @@ +import { and, asc, eq, gt, isNotNull, isNull, notExists, sql } from "drizzle-orm"; +import { heartbeatRuns, issueComments, nativeRunFinalizations, statusDecisions, type Db } from "@paperclipai/db"; +import type { createDeliveryWorkCoordinator } from "../delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; +import type { NativeReviewResponseScope } from "../native-review-response-notifications.js"; +import { registerScopedDeadlineWork } from "../scoped-deadline-work.js"; +import { recoverCommittedNativeReviewResponse } from "./native-run-finalizer.js"; + +export function registerNativeReviewResponseWork(db: Db, + coordinator: ReturnType, canRun: () => boolean) { + return nativeReviewResponseWork(db).register(coordinator, canRun); +} + +type Entry = { id: string; companyId: string; issueId: string; agentId: string; dueAt: number }; + +export function nativeReviewResponseWork(db: Db) { + function candidates(scope?: NativeReviewResponseScope, after?: string, runId?: string) { + return db.select({ id: nativeRunFinalizations.runId, companyId: nativeRunFinalizations.companyId, + issueId: nativeRunFinalizations.issueId, agentId: heartbeatRuns.agentId, + decisionId: nativeRunFinalizations.decisionId, resultId: nativeRunFinalizations.resultId, + assessmentId: nativeRunFinalizations.assessmentId, nextAttemptAt: nativeRunFinalizations.nextAttemptAt, + }).from(nativeRunFinalizations) + .innerJoin(heartbeatRuns, and(eq(heartbeatRuns.id, nativeRunFinalizations.runId), + eq(heartbeatRuns.companyId, nativeRunFinalizations.companyId), eq(heartbeatRuns.nativeIssueId, nativeRunFinalizations.issueId))) + .innerJoin(statusDecisions, and(eq(statusDecisions.id, nativeRunFinalizations.decisionId), + eq(statusDecisions.companyId, nativeRunFinalizations.companyId), eq(statusDecisions.issueId, nativeRunFinalizations.issueId), + eq(statusDecisions.runId, nativeRunFinalizations.runId), eq(statusDecisions.assessmentId, nativeRunFinalizations.assessmentId))) + .where(and(eq(nativeRunFinalizations.phase, "committed"), isNull(nativeRunFinalizations.leaseOwner), + isNotNull(nativeRunFinalizations.resultId), isNotNull(nativeRunFinalizations.assessmentId), + eq(heartbeatRuns.runtimeMode, "native"), eq(heartbeatRuns.status, "succeeded"), + eq(statusDecisions.reasonCode, "governed_response_waiting"), eq(statusDecisions.applicationState, "applied"), + sql`${statusDecisions.decisionJson}->'externalChatReviewPresentation'->>'schema' = 'paperclip.chat_review_response_presentation.v1'`, + notExists(db.select({ id: issueComments.id }).from(issueComments).where(and( + eq(issueComments.companyId, nativeRunFinalizations.companyId), eq(issueComments.issueId, nativeRunFinalizations.issueId), + eq(issueComments.createdByRunId, nativeRunFinalizations.runId), eq(issueComments.authorAgentId, heartbeatRuns.agentId), + eq(issueComments.authorType, "agent"), sql`${issueComments.metadata}->>'authorizationReason' = 'allow_chat_run_presentation'`, + sql`${heartbeatRuns.resultJson}->'externalChatReviewPresentation'->>'decisionId' = ${nativeRunFinalizations.decisionId}::text`, + ))), + scope ? eq(nativeRunFinalizations.companyId, scope.companyId) : undefined, + scope?.issueId ? eq(nativeRunFinalizations.issueId, scope.issueId) : undefined, + scope?.runId ? eq(nativeRunFinalizations.runId, scope.runId) : undefined, + scope?.agentId ? eq(heartbeatRuns.agentId, scope.agentId) : undefined, + runId ? eq(nativeRunFinalizations.runId, runId) : undefined, + after ? gt(nativeRunFinalizations.runId, after) : undefined, + )).orderBy(asc(nativeRunFinalizations.runId)).limit(100); + } + async function load(scope?: NativeReviewResponseScope, signal?: AbortSignal, canRun = () => true): Promise { + const entries: Entry[] = []; + let after: string | undefined; + while (canRun() && !signal?.aborted) { + const rows = await candidates(scope, after); + for (const row of rows) entries.push({ id: row.id, companyId: row.companyId, + issueId: row.issueId, agentId: row.agentId, dueAt: row.nextAttemptAt?.getTime() ?? 0 }); + if (rows.length < 100) return entries; + after = rows[rows.length - 1].id; + } + throw new Error("Governed review response discovery stopped"); + } + async function check(entry: Entry, signal?: AbortSignal): Promise { + if (signal?.aborted) throw new Error("Governed review response recovery stopped"); + const scope = { companyId: entry.companyId, issueId: entry.issueId }; + const [current] = await candidates(scope, undefined, entry.id); + if (!current) return null; + if (current.nextAttemptAt && current.nextAttemptAt.getTime() > Date.now()) return { ...entry, dueAt: current.nextAttemptAt.getTime() }; + if (signal?.aborted) throw new Error("Governed review response recovery stopped"); + const recoveryStartedAt = Date.now(); + await recoverCommittedNativeReviewResponse(db, { ...scope, runId: entry.id, + decisionId: current.decisionId!, resultId: current.resultId!, assessmentId: current.assessmentId!, onlyWhenDue: true }); + const [remaining] = await candidates(scope, undefined, entry.id); + return remaining?.nextAttemptAt && remaining.nextAttemptAt.getTime() > recoveryStartedAt + ? { ...entry, dueAt: remaining.nextAttemptAt.getTime() } : null; + } + function register(coordinator: ReturnType, canRun: () => boolean) { + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.nativeReviewResponse, + decodeScope: (hint: string): NativeReviewResponseScope | undefined => hint === "*" ? undefined : JSON.parse(hint), + matches: (entry: Entry, scope?: NativeReviewResponseScope) => !scope || entry.companyId === scope.companyId + && (!scope.issueId || entry.issueId === scope.issueId) && (!scope.runId || entry.id === scope.runId) + && (!scope.agentId || entry.agentId === scope.agentId), + load: (scope, signal) => load(scope, signal, canRun), + check: (entry, signal) => { + if (!canRun()) throw new Error("Governed review response recovery paused"); + return check(entry, signal); + }, + }, canRun); + } + return { load, check, register }; +} diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index 9df2211a496..a1a6ac00c4e 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -723,65 +723,62 @@ async function projectCommittedRun(input: { } } -async function materializeCommittedReviewResponse(db: Db, runId: string) { - try { - const [run] = await db - .select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, - nativeIssueId: heartbeatRuns.nativeIssueId, resultJson: heartbeatRuns.resultJson }) - .from(heartbeatRuns) - .where(eq(heartbeatRuns.id, runId)) - .limit(1); - if (!run?.nativeIssueId || !run.resultJson?.externalChatReviewPresentation) - return; - const decisionId = record( - run.resultJson.externalChatReviewPresentation, - ).decisionId; - if ( - typeof decisionId !== "string" || - (await hasMaterializedNativeReviewResponse(db, { - companyId: run.companyId, - issueId: run.nativeIssueId, - runId, - decisionId, - })) - ) - return; - const summary = record(run.resultJson.nativeResult).summary; - if ( - typeof summary !== "string" || - !(await authorizeNativeChatReviewPresentation(db, { - companyId: run.companyId, - issueId: run.nativeIssueId, - runId, - resultJson: run.resultJson, - })) - ) - return; - // The coordinator transaction is already committed. addComment owns a - // fresh issue transaction, rechecks the proof and carries selected files; - // its same-run/text and publication keys make recovery replay idempotent. - await issueService(db).addComment( - run.nativeIssueId, - summary, - { agentId: run.agentId, runId }, - { - authorizationReason: "allow_chat_run_presentation", - }, - ); - publishChatPublicationCommitSignal({ +async function materializeNativeReviewResponse(db: Db, runId: string, dbOrTx: Db = db) { + const [run] = await dbOrTx + .select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + nativeIssueId: heartbeatRuns.nativeIssueId, resultJson: heartbeatRuns.resultJson }) + .from(heartbeatRuns) + .where(eq(heartbeatRuns.id, runId)) + .limit(1); + if (!run?.nativeIssueId || !run.resultJson?.externalChatReviewPresentation) + return null; + const decisionId = record( + run.resultJson.externalChatReviewPresentation, + ).decisionId; + if ( + typeof decisionId !== "string" || + (await hasMaterializedNativeReviewResponse(dbOrTx, { companyId: run.companyId, issueId: run.nativeIssueId, - runId: run.id, - agentId: run.agentId, - eventType: "run.presentation.resolved", - }); + runId, + decisionId, + })) + ) + return null; + const summary = record(run.resultJson.nativeResult).summary; + if ( + typeof summary !== "string" || + !(await authorizeNativeChatReviewPresentation(dbOrTx, { + companyId: run.companyId, + issueId: run.nativeIssueId, + runId, + resultJson: run.resultJson, + })) + ) + return null; + await issueService(db).addComment( + run.nativeIssueId, + summary, + { agentId: run.agentId, runId }, + { + authorizationReason: "allow_chat_run_presentation", + }, + dbOrTx, + ); + return run; +} + +function publishNativeReviewResponse(run: Pick) { + publishChatPublicationCommitSignal({ companyId: run.companyId, issueId: run.nativeIssueId!, + runId: run.id, agentId: run.agentId, eventType: "run.presentation.resolved" }); +} + +async function materializeCommittedReviewResponse(db: Db, runId: string) { + try { + const run = await materializeNativeReviewResponse(db, runId); + if (run) publishNativeReviewResponse(run); } catch (error) { - // The durable committed proof remains retryable by the next reconciler - // sweep. Presentation failure must not reclassify a successful native run. - logger.warn( - { runId, error }, - "Committed chat review response is awaiting presentation retry", - ); + logger.warn({ runId, error }, "Committed chat review response is awaiting presentation retry"); } } @@ -1183,6 +1180,28 @@ export async function repairCommittedNativeReviewResponse( ); if (restored) await materializeCommittedReviewResponse(db, input.runId); } +export async function recoverCommittedNativeReviewResponse(db: Db, input: { + companyId: string; issueId: string; runId: string; decisionId: string; + resultId: string; assessmentId: string; onlyWhenDue?: boolean; +}): Promise { + const materialized = await db.transaction(async transaction => { + const tx = transaction as unknown as Db; + + + const [coordinator] = await tx.select({ id: nativeRunFinalizations.runId }).from(nativeRunFinalizations) + .where(and(eq(nativeRunFinalizations.companyId, input.companyId), + eq(nativeRunFinalizations.issueId, input.issueId), eq(nativeRunFinalizations.runId, input.runId))).for("update"); + if (!coordinator) return null; + const [issue] = await tx.select({ id: issues.id }).from(issues) + .where(and(eq(issues.companyId, input.companyId), eq(issues.id, input.issueId))).for("update"); + if (!issue || await hasMaterializedNativeReviewResponse(tx, input)) return null; + if (!await restoreNativeChatReviewPresentationInTransaction(tx, input)) return null; + return materializeNativeReviewResponse(db, input.runId, tx); + }); + if (materialized) publishNativeReviewResponse(materialized); + return materialized !== null; +} + export async function finalizeNativeRun(input: { db: Db; runId: string; @@ -1190,6 +1209,7 @@ export async function finalizeNativeRun(input: { /** Reconciliation owns terminal run projection; the live heartbeat does it afterward. */ projectRunStatus?: boolean; skipNativeResponsePresentation?: boolean; + skipNativeReviewResponse?: boolean; /** Workspace-only replay must not consume the provider recovery budget. */ preserveProviderAttempt?: boolean; failpoint?: NativeStatusCommitFailpoint; @@ -1214,14 +1234,14 @@ export async function finalizeNativeRun(input: { const coordinator = claim.coordinator; if (!claim.leaseOwner && coordinator.phase === "committed") { if (isNativeRunnerOwnershipHeld(admittedRun)) return coordinator; - const presentationAlreadyMaterialized = coordinator.decisionId + const presentationAlreadyMaterialized = input.skipNativeReviewResponse || (coordinator.decisionId ? await hasMaterializedNativeReviewResponse(input.db, { companyId: admittedRun.companyId, issueId: coordinator.issueId, runId: admittedRun.id, decisionId: coordinator.decisionId, }) - : false; + : false); if ( !presentationAlreadyMaterialized && coordinator.decisionId && diff --git a/server/src/services/satisfied-connection-intents.ts b/server/src/services/satisfied-connection-intents.ts index 62c29daf942..ba8c74205d4 100644 --- a/server/src/services/satisfied-connection-intents.ts +++ b/server/src/services/satisfied-connection-intents.ts @@ -1,3 +1,4 @@ +import { notifyNativeReviewResponse } from "./native-review-response-notifications.js"; import { notifyChatConfirmationResolution } from "./chat-confirmation-notifications.js"; import { enqueueTerminalIssueInteractionChatPublications } from "./chat-interaction-publications.js"; import { connectionManagementService } from "./connection-management.js"; @@ -155,6 +156,7 @@ export function satisfiedConnectionIntentService(db: Db) { } const connection = await findSatisfiedToolConnection(txDb, task, intent); if (!connection) return false; + await notifyNativeReviewResponse(tx, candidate); const now = new Date(); // Retire the obsolete request without inventing a human approval. await notifyChatConfirmationResolution(tx, intent); diff --git a/server/src/services/tool-access/oauth/cloud-callback.ts b/server/src/services/tool-access/oauth/cloud-callback.ts index af2872ce343..3c777fc7969 100644 --- a/server/src/services/tool-access/oauth/cloud-callback.ts +++ b/server/src/services/tool-access/oauth/cloud-callback.ts @@ -1,3 +1,4 @@ +import { notifyNativeReviewResponse } from "../../native-review-response-notifications.js"; import { notifyToolConnectionHealth } from "../../tool-connection-health-notifications.js"; import { managedConnectorProfile } from "./client-settings.js"; import type { ActorInfo } from "../types.js"; @@ -408,18 +409,19 @@ export function createCloudOAuthCallback(db: Db, dependencies: Dependencies) { personalCredential || agentCredential ? credentialSecretRefs : [], tx, ); - const linkedInteractionKind = stateRow.interactionId + const linkedInteraction = stateRow.interactionId ? await tx - .select({ kind: issueThreadInteractions.kind }) + .select({ kind: issueThreadInteractions.kind, issueId: issueThreadInteractions.issueId }) .from(issueThreadInteractions) .where(eq(issueThreadInteractions.id, stateRow.interactionId)) .limit(1) - .then((rows) => rows[0]?.kind ?? null) + .then((rows) => rows[0] ?? null) : null; if ( stateRow.interactionId && - linkedInteractionKind === "request_confirmation" + linkedInteraction?.kind === "request_confirmation" ) { + await notifyNativeReviewResponse(tx, { companyId: connection.companyId, issueId: linkedInteraction.issueId }); await tx .update(issueThreadInteractions) .set({ diff --git a/server/src/services/tool-access/oauth/direct-callback.ts b/server/src/services/tool-access/oauth/direct-callback.ts index 9029e35438c..e3f68f3ba40 100644 --- a/server/src/services/tool-access/oauth/direct-callback.ts +++ b/server/src/services/tool-access/oauth/direct-callback.ts @@ -1,3 +1,4 @@ +import { notifyNativeReviewResponse } from "../../native-review-response-notifications.js"; import { notifyToolConnectionHealth } from "../../tool-connection-health-notifications.js"; import { credentialScope } from "./client-settings.js"; import { withoutOAuthRefreshLease } from "./refresh-leases.js"; @@ -374,18 +375,19 @@ export function createDirectOAuthCallback(db: Db, dependencies: Dependencies) { updatedAt: new Date(), }) .where(eq(toolApplications.id, connection.applicationId)); - const linkedInteractionKind = stateRow.interactionId + const linkedInteraction = stateRow.interactionId ? await tx - .select({ kind: issueThreadInteractions.kind }) + .select({ kind: issueThreadInteractions.kind, issueId: issueThreadInteractions.issueId }) .from(issueThreadInteractions) .where(eq(issueThreadInteractions.id, stateRow.interactionId)) .limit(1) - .then((rows) => rows[0]?.kind ?? null) + .then((rows) => rows[0] ?? null) : null; if ( stateRow.interactionId && - linkedInteractionKind === "request_confirmation" + linkedInteraction?.kind === "request_confirmation" ) { + await notifyNativeReviewResponse(tx, { companyId: connection.companyId, issueId: linkedInteraction.issueId }); await tx .update(issueThreadInteractions) .set({ diff --git a/server/src/services/tool-access/oauth/state.ts b/server/src/services/tool-access/oauth/state.ts index 2ab81656edc..efa92bc139f 100644 --- a/server/src/services/tool-access/oauth/state.ts +++ b/server/src/services/tool-access/oauth/state.ts @@ -1,3 +1,4 @@ +import { notifyNativeReviewResponse } from "../../native-review-response-notifications.js"; import type { ActorInfo } from "../types.js"; import { and, eq } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; @@ -124,17 +125,19 @@ async function rejectPendingOAuthInteraction( stateRow: typeof toolOauthStates.$inferSelect, actor: ActorInfo | undefined, ) { - if (!stateRow.interactionId) return; + const interactionId = stateRow.interactionId; + if (!interactionId) return; const now = new Date(); const linked = await db .select({ kind: issueThreadInteractions.kind, + issueId: issueThreadInteractions.issueId, payload: issueThreadInteractions.payload, }) .from(issueThreadInteractions) .where( and( - eq(issueThreadInteractions.id, stateRow.interactionId), + eq(issueThreadInteractions.id, interactionId), eq(issueThreadInteractions.companyId, stateRow.companyId), ), ) @@ -152,36 +155,40 @@ async function rejectPendingOAuthInteraction( }) .where( and( - eq(issueThreadInteractions.id, stateRow.interactionId), + eq(issueThreadInteractions.id, interactionId), eq(issueThreadInteractions.companyId, stateRow.companyId), eq(issueThreadInteractions.status, "pending"), ), ); return; } - await db - .update(issueThreadInteractions) - .set({ - status: "rejected", - result: { - version: 1, - outcome: "rejected", - // Paperclip's own words: the provider's explanation is untrusted and - // this reason is rendered in the thread (PAP-17108). - reason: - "Authorization was declined or cancelled in the provider's window", - }, - resolvedByUserId: actor?.actorType === "user" ? actor.actorId : null, - resolvedAt: now, - updatedAt: now, - }) - .where( - and( - eq(issueThreadInteractions.id, stateRow.interactionId), - eq(issueThreadInteractions.companyId, stateRow.companyId), - eq(issueThreadInteractions.status, "pending"), - ), - ); + if (!linked) return; + await db.transaction(async tx => { + await notifyNativeReviewResponse(tx, { companyId: stateRow.companyId, issueId: linked.issueId }); + await tx + .update(issueThreadInteractions) + .set({ + status: "rejected", + result: { + version: 1, + outcome: "rejected", + + + reason: + "Authorization was declined or cancelled in the provider's window", + }, + resolvedByUserId: actor?.actorType === "user" ? actor.actorId : null, + resolvedAt: now, + updatedAt: now, + }) + .where( + and( + eq(issueThreadInteractions.id, interactionId), + eq(issueThreadInteractions.companyId, stateRow.companyId), + eq(issueThreadInteractions.status, "pending"), + ), + ); + }); } return { peekOAuthState, validateOAuthState, consumeOAuthState, rejectPendingOAuthInteraction }; } diff --git a/server/src/services/tool-gateway.ts b/server/src/services/tool-gateway.ts index fb023ba7f6d..540668d69b0 100644 --- a/server/src/services/tool-gateway.ts +++ b/server/src/services/tool-gateway.ts @@ -2689,7 +2689,7 @@ export function createToolGatewayService( approvalId: approval.id, linkedByAgentId: input.session.agentId, }) - .onConflictDoNothing()); + .onConflictDoNothing(), { issueIds: [approvalIssueId] }); } const interaction = await interactions.create(