diff --git a/doc/idle-tool-review-projection.md b/doc/idle-tool-review-projection.md new file mode 100644 index 00000000000..f1d431ea920 --- /dev/null +++ b/doc/idle-tool-review-projection.md @@ -0,0 +1,229 @@ +# Tool-review interaction repair + +The tool-action request and invocation are authoritative. The interaction card +is a projection of their outcome. Previously the aggregate maintenance tick +joined these tables repeatedly, even when every card already matched. + +The projection worker reconstructs affected companies at startup and reconciles +after committed changes. Producer hints carry the company ID. Subsequent +candidate discovery and paginated repair queries filter by that company; review +writes cannot drive a scan of another company's terminal history. The shared +scoped worker dispatches one company per pass with one-second minimum spacing. +It retains a 30-second retry only for a mismatch or failed reconciliation. +Configured idle tools, pending reviews without terminal outcomes, and matching +terminal history have no recurring projection SQL or provider calls. This is +worker quiet, not whole-instance suspension. + +## Ownership and trigger contract + +`tool-review-projection.ts` owns candidate discovery and pagination. The gateway +retains card interpretation, current-state checks, locks, activity publication, +and the existing continuation callback. The scheduler consumes company discovery, +`sweep(companyId)`, and the returned retry obligation. No provider payload enters shared scheduling. +No table is added. A concurrent company/interaction lookup index supports the +existing link checks and linked-request updates. + +The existing review notification helper now also signals projection repair. +Notifications are awaited before writes and inherit the outer transaction's +commit, rollback, and uncertainty tracking. Unrelated review changes can cause +one empty company-scoped reconciliation; they do not create a recurring deadline. +Interaction-only writes first check for an actual linked tool request using the +company and affected card IDs. Generic confirmation kinds alone do not signal +any tool-review queue. This also preserves recovery for legacy linked cards +without current tool payload flags. These reads happen in the mutation owner, +not a background poll. Sibling selection shares the update predicate and runs +under the existing issue lock before publishing the notification. + +| Producer | Notification ownership | +| --- | --- | +| Gateway signing, approval inference, execution claims, provider outcomes, and invalidation | Existing `writeToolReviewChange`; the closed-issue expiry path now uses it too | +| Human approval/rejection | `commitToolActionReview` signals in its existing transaction | +| Pending expiry, approved recovery, interrupted claims, Test-tab settlement | Existing review-write boundary | +| Review-queue invalidation | Existing review-write boundary | +| Interaction withdrawal, skip, terminal-issue expiry | Linked-request resolution signals before the mutation | +| Linked tool-card accept/reject, deleted-addressee cancellation | Projection notification before replacing card state in the owning transaction; unlinked confirmations do not notify | +| Sibling supersession, startup duplicate cleanup, new/historical comment supersession | Projection notification before card replacement, including cases where the request is already terminal | +| Comment-based confirmation resolution | Calls the interaction owner inside the same outer transaction; its later comment-ID update shares that signal | + +The audited other interaction writers do not create this mismatch: native bridge +and status-committer summary/status-only updates preserve the projection; native +review retirement owns separate completion-review cards; connection and secret +flows own their separate card types. Document-staleness and plan-resume paths +use document targets, while tool cards use `tool-action:` custom targets. +Reflection consent uses its instruction/profile/skill target keys, not tool-action +keys. A future writer that replaces tool-card result state must use the same +transaction notification boundary. A future writer that changes authoritative +tool outcomes must signal the review boundary. + +## Repair, concurrency, and lifecycle + +The mismatch predicate preserves the previous policy: terminal executed, failed, +expired, or cancelled requests with a linked interaction whose displayed tool +status differs. Cancelled requests display `expired`. Invocation error and result +summaries come from saved state. Repair never repeats a provider action. + +Candidate queries page in request-ID order, 100 rows at a time. A card that cannot +yet be projected does not starve later pages. Per-row failures do not prevent +attempting the remaining candidates; failures keep the retry. A final mismatch +check retains retries for no-op repairs rather than silently declaring them done. +Failed companies retain their own backoff across notifications while other +companies progress. An unresolved no-op projection retains a 30-second deadline; +a new relevant company change can make it actionable sooner. A company pass +still drains all its pages, preserving progress past an unprojectable card. +The joins inspect that company's history on activation. This change removes repeated +idle scans; it does not introduce an indexed projection-debt table or claim that +an activated reconciliation has constant cost independent of history size. + +The gateway locks the interaction and request, then checks the current request +status. If another repair already wrote the same status and outcome fields, +it avoids a second write and duplicate activity. Different saved details still +use the existing update path. Missing projection state remains repairable. + +The shared deadline worker retains first writes during drain, failed retries, +and changes arriving during reconciliation. Shutdown joins coordinator work. +Startup reconstructs debt from existing rows. With heartbeat scheduling disabled, +startup still performs one pass, preserving the existing disabled-mode contract. +The public/manual combined review sweep remains available; server maintenance +no longer calls it periodically. + +## Continuation and sleep limits + +Card state is not proof that continuation delivery finished. Existing +`tool_action_deliveries` receipts and their coordinator remain responsible for +that work. Activity publication still occurs after the card transaction; this +slice does not add exactly-once activity delivery or a new outbox. + +Signals remain process-local to the supported database owner. Independent SQL +writers and another active server require a separate notification/reconciliation +contract. Cloud eligibility still conservatively blocks retained tool-action +requests. No external wake or sleep-safety guard is relaxed here. + +## Verification + +Twelve new real-PostgreSQL cases exercise normal approval and single execution, +quiet configured and terminal-history states, four persisted outcome states, +outer commit/rollback, first write during drain, failure retry through drain, +foreground/worker repair contention, multi-page progress past an unprojectable +card, and an administrative card overwrite that reactivates repair. Quiet checks +observe absence of database reads across an hour of controlled time. + +The first focused run caught a missing-projection null access in the new +idempotency check; optional access fixes it. The pagination fixture also attempted +to change a referenced request ID; it now orders the additional candidates without +mutating that key. The focused rerun passed all eleven then-existing tests. +After adding the concurrent repair case, all 148 service/startup tests and 92 +gateway acceptance tests passed. Workspace typechecking and build passed. All 469 tool-access and 106 interaction +service tests also pass (815 tests across the final selections). The first +interaction run passed 105 cases but timed out in one cleanup hook during a +53-minute local run; the isolated full rerun passed in nine seconds. No production +change or timeout increase was made to force that result. Full repository tests +will run in CI; the local selections do not claim a complete repository test run. +Fresh PR CI/review remains required. + +Company-scoping follow-up reproduced two regressions: captured recovery SQL lacked +a company predicate, and ordinary confirmation creation/replacement/cancellation +published three unnecessary tool-projection wakes. Both regressions now pass +through real producers. Additional coverage keeps failed-company retry timing +independent, proves ordinary mutations still complete without recovery scans, +and exercises administrative repair of legacy cards identified by their saved +link rather than payload flags. + +Company-scoping follow-up verification: all 180 gateway and startup cases and all 152 interaction and secret-proposal route cases passed against dedicated PostgreSQL test databases. Server typechecking passed. These are local checks; stack CI and deployed sleep qualification remain separate. + + +## Indexed interaction links + +Review found that the new link check had no supporting interaction index. An +ordinary unlinked confirmation could scan all tool-action history for its company. +The generated `0386_steady_epoch.sql` migration adds the non-unique +`(company_id, interaction_id)` index concurrently, and the existing top-level +migration executor handles its creation and interrupted-build recovery. The +schema and generated snapshot describe the same index; older SQL and journal +entries are preserved and snapshot pruning retains the five newest snapshots. + +A real PostgreSQL check applied the migration to a fresh dedicated test database +and seeded 30,000 tool-action rows. The ordinary missing-link query returned no +rows in both treatments: without the index it performed a sequential scan, +filtered 30,000 rows and touched 429 shared buffers; with the index it used the +new index and touched two buffers. These are local planner measurements, not a +fleet latency or cost claim. The baseline index removal was rolled back in that +dedicated database, and the index remained valid afterwards. An initial probe +failed while binding a JSON fixture parameter before any comparison; correcting +the parameter encoding made the complete probe pass. That failed fixture is not +counted as a product regression. + +Migration numbering/safety checks, generated snapshot drift, server types and +server build pass. All 214 cases in the full gateway/projection/startup/scoped- +deadline selection pass against a separate freshly migrated database, preserving +notification quiet, company isolation, transaction settlement, and interruption +behavior. Latest-head CI and automated reviews remain required merge gates. + +## Bounded projection and lifecycle interruption + +A review found that one company could drain an unlimited backlog in a single +worker pass. Five real PostgreSQL regressions reproduced 501 rows being consumed +instead of the 500-row budget, missing abort/drain handling, and the absence of +continuation past a full page of unprojectable or failed cards. + +The worker now gives its existing service a 500-row budget, the coordinator's +abort signal, and its admission predicate. The service checks admission between +rows and reads at most 100 candidates per page. It retains the last attempted +request ID for each unfinished company, including after a failed pass, so a +poisoned early page cannot hide later healthy cards. A truncated pass schedules +a one-second continuation; the existing shared per-company fairness and failure +backoff remain in force. A completed pass that still has an unprojectable card +backs off from thirty seconds to five minutes. Fresh committed company changes +reset that delay and wake the affected scope promptly. These are real unresolved projections, +not an unconditional empty-history scan. + +Reconstruction clears continuation state for companies that no longer have +mismatches. A sixth real regression reproduced an abandoned cursor skipping the +first new earlier card after an empty reconstruction; clearing that cursor makes +the first new pass repair it immediately. The maps contain no durable truth: +restart reconstructs from the existing rows, and completed rows stop matching. +There is no cursor table, independent timer or per-producer recovery wrapper. + +Legacy manual/global and scheduler-disabled one-shot repair calls retain their +complete-sweep behavior unless they request a budget. The normal coordinator +worker always supplies the bounded options. Cancellation is observed between +rows and pages; an already issued database query or reflection finishes before +the service stops. A sweep never replays a provider tool action. + +All 220 cases across the full gateway/projection/interruption, startup and scoped +deadline suites pass, with no skipped cases. The five initial regressions and +sixth cursor regression failed before their fixes; server types and build pass. +The earlier 219-case acceptance remains evidence for the preceding revision. +Fresh integrated-head CI and both automated reviews remain merge gates. + + +After normal reconciliation with master `f6bc653f7f`, all 320 integrated cases +pass, including the 81 authenticated gateway HTTP cases and the upstream narrowed +identity-read tests. Server types and build pass. Thirty-five independent master +performance files are preserved exactly; the gateway's narrowed identity read +is preserved alongside the projection extraction. The pre-integration 220-case +selection above remains evidence for its stated revision. + + +## Review efficiency follow-up + +A company-scoped hint now enters the bounded owner directly rather than first +repeating company discovery. Startup/global reconstruction still discovers +companies from existing rows. An empty sweep with no saved cursor returns after +one candidate query; a resumed cursor or attempted reflection still needs the +final query to detect unresolved earlier cards. The reflection callback returns +no settlement result, so an attempted reflection cannot establish that a card +was repaired. + +Returned unresolved work now backs off through 30, 60, 120, 240 and 300 seconds, +matching the shared error-backoff cap. Truncated backlog continuation remains +one second. Settlement removes the retry; a new committed hint resets backoff. +No pending row is discarded, and no independent timer or table is added. + +The fixed-cadence regression failed before repair and now verifies the full +backoff cap, producer reactivation and quiet after settlement. An initial scan +assertion incorrectly assumed that a complete approval emitted only one wake; +its separate legitimate writes caused two scans even after optimization. The +corrected case isolates one committed hint and verifies one candidate scan with +matching history, followed by an empty direct sweep with one query. All 222 +cases in the full gateway, startup and scoped-deadline suites pass without skips; +server typecheck and build pass. Fresh-head CI and both reviews are still required. 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 b8976fb34fb..25d3ebf70bc 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -18,7 +18,7 @@ Do not mark an item done from a timer removal or a green check alone. | 04 | Decision retention and notifications | Retention deadlines and notification delivery are separate reasons | Pending | | 05 | Decision expiry and interrupted effects | Expiry, effect recovery, continuation recovery | Pending | | 06 | Tool connection health | Configuration/use activation and defined proactive health deadlines | Pending | -| 07 | Secret proposals, tool reviews, gateway tokens | Separate expiry/scrub/recovery owners | Secret proposals: [#17446](https://github.com/paperclipai/paperclip/pull/17446), merged; gateway credentials: [#17450](https://github.com/paperclipai/paperclip/pull/17450), merged; pending tool-review expiry: [#17453](https://github.com/paperclipai/paperclip/pull/17453), merged; approved recovery: [#17454](https://github.com/paperclipai/paperclip/pull/17454), merged after current-head CI and both 5/5 reviews; interrupted claims: [#17458](https://github.com/paperclipai/paperclip/pull/17458), integration verified and awaiting fresh CI/review; projection: [#17468](https://github.com/paperclipai/paperclip/pull/17468), interaction-link index verified and awaiting fresh CI/review | +| 07 | Secret proposals, tool reviews, gateway tokens | Separate expiry/scrub/recovery owners | Secret proposals: [#17446](https://github.com/paperclipai/paperclip/pull/17446), merged; gateway credentials: [#17450](https://github.com/paperclipai/paperclip/pull/17450), merged; pending tool-review expiry: [#17453](https://github.com/paperclipai/paperclip/pull/17453), merged; approved recovery: [#17454](https://github.com/paperclipai/paperclip/pull/17454), merged after current-head CI and both 5/5 reviews; interrupted claims: [#17458](https://github.com/paperclipai/paperclip/pull/17458), merged after current-head CI and both 5/5 reviews; projection: [#17468](https://github.com/paperclipai/paperclip/pull/17468), interaction-link index verified and awaiting fresh CI/review | | 08 | GitHub connection events and continuity | Event discovery/replay and credential continuity separately | Pending | | 09 | Merged-PR confirmations | Candidate writes, webhook activation where supported, outstanding-candidate fallback | Pending | | 10 | External-object refresh | Configuration changes, refresh deadlines, source events | Pending | @@ -280,3 +280,58 @@ initial concurrent-run timeout is retained above as evidence, not hidden. Workspace typecheck/build and migration safety/numbering pass. The first four PRs are mergeable and have all checks passed at their recorded heads. #17454 is mergeable with fresh checks and review in progress at `7a1cf03e62`. + +Slice 06 is [#17458](https://github.com/paperclipai/paperclip/pull/17458), head +`85003a7ac4`, based on #17454. Greptile review was requested and the PR is +attached to this chat. Slice 05 has fresh Greptile 5/5 at `7a1cf03e62`; CI is +running with no failure in the latest snapshot. The next branch is +`codex/sleep-07-tool-review-projection`, based on #17458. No seventh implementation +or readiness is claimed yet. The full remaining inventory stays in scope. + + +## Slice 07: tool-review interaction projection + +The implementation removes the last combined review sweep from the main tick. +It pages terminal mismatches and retains retries only for actual debt or failures. +Gateway reflection keeps locks, current-state checks, activity, and continuation +ownership; concurrent identical repairs do not write twice. The producer audit +also covers administrative card result replacements, before the mutation in the +existing transaction. See [projection repair](../idle-tool-review-projection.md). + +Twelve new lifecycle cases are included in 148 passing service/startup tests. +Gateway acceptance passes all 92 tests and workspace typechecking passes. The +initial missing-state bug and referenced-key fixture failure are recorded in the +behavior document. Build and additional consumer regressions are running; no +seventh PR is claimed ready yet. + +Historical parent evidence at the initial migration: #17454 at `7a1cf03e62` +and #17458 at `85003a7ac4` passed the recorded CI matrices and Greptile review. +Those snapshots preceded their current-head review and landing. The first six +stack PRs have now merged. The full remaining inventory is still in scope. + +Initial slice 07 local verification: 815 selected tests passed, with workspace +typecheck and build passing. One initial interaction cleanup-hook timeout is recorded in +the behavior document; the complete isolated suite then passed. Both parent +PRs #17454 and #17458 now have every check passed and are mergeable, including +the additional Paperclip Review checks. Full CI/review for slice 07 is next. + + +Slice 07 current integration: master `f6bc653f7f` contains the first six landed +slices and the independent identity/resume/sidebar performance work. This slice +adds generated concurrent interaction-link index migration `0386_steady_epoch` +and retains five schema snapshots. Projection worker passes stop at 500 rows, +honor abort/drain between rows, and resume past failed or unprojectable pages; +empty reconstruction clears obsolete cursors. Six real regressions failed before +the fixes. All 320 integrated gateway/projection/startup/deadline/identity/HTTP +cases, server types and build pass; 35 independent master files are preserved +exactly. Current-head CI and Greptile/Ponytail review remain required. Earlier +case counts above describe the broader selections at their original revisions. + + +Slice 07 follow-up: scoped hints avoid redundant discovery, empty sweeps avoid a +second candidate read, and unresolved-card retries back off to five minutes. +Fresh writes reactivate promptly and settlement is quiet. The full affected +gateway/startup/deadline selection passes 222 cases; server types/build pass. +The initial invalid one-wake-per-approval assertion and corrected isolated-hint +measurement are retained in the behavior document. Current-head CI/reviews remain +required before this slice can merge. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 3e4cce15e2e..342256ac462 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -504,6 +504,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0381_mean_scream.sql", "0382_narrow_the_executioner.sql", "0385_open_madame_web.sql", + "0386_steady_epoch.sql", "0384_funny_adam_warlock.sql", "0383_tool_review_company_expiry.sql", ]); diff --git a/packages/db/src/migrations/0386_steady_epoch.sql b/packages/db/src/migrations/0386_steady_epoch.sql new file mode 100644 index 00000000000..06356e30ced --- /dev/null +++ b/packages/db/src/migrations/0386_steady_epoch.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "tool_action_requests_company_interaction_idx" ON "tool_action_requests" USING btree ("company_id","interaction_id"); diff --git a/packages/db/src/migrations/meta/0381_snapshot.json b/packages/db/src/migrations/meta/0386_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0381_snapshot.json rename to packages/db/src/migrations/meta/0386_snapshot.json index 23802aaeece..8dd26eef4fb 100644 --- a/packages/db/src/migrations/meta/0381_snapshot.json +++ b/packages/db/src/migrations/meta/0386_snapshot.json @@ -1,6 +1,6 @@ { - "id": "2e93deb7-f24c-42d1-8c8e-30b21cff90bc", - "prevId": "bfda87fa-662b-47bb-87c0-dabc814aebd7", + "id": "da3229b5-4eab-4d8b-ba6b-d756e6c42430", + "prevId": "9965d300-e355-4f47-ba0a-951ed3b4b8a2", "version": "7", "dialect": "postgresql", "tables": { @@ -52812,6 +52812,88 @@ "method": "btree", "with": {} }, + "tool_action_requests_pending_expiry_idx": { + "name": "tool_action_requests_pending_expiry_idx", + "columns": [ + { + "expression": "expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"tool_action_requests\".\"status\" = 'pending'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "tool_action_requests_company_pending_expiry_idx": { + "name": "tool_action_requests_company_pending_expiry_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"tool_action_requests\".\"status\" = 'pending'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "tool_action_requests_executing_deadline_idx": { + "name": "tool_action_requests_executing_deadline_idx", + "columns": [ + { + "expression": "updated_at", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"tool_action_requests\".\"status\" = 'executing'", + "concurrently": false, + "method": "btree", + "with": {} + }, + "tool_action_requests_approved_idx": { + "name": "tool_action_requests_approved_idx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "where": "\"tool_action_requests\".\"status\" = 'approved'", + "concurrently": false, + "method": "btree", + "with": {} + }, "tool_action_requests_invocation_idx": { "name": "tool_action_requests_invocation_idx", "columns": [ @@ -52827,6 +52909,27 @@ "method": "btree", "with": {} }, + "tool_action_requests_company_interaction_idx": { + "name": "tool_action_requests_company_interaction_idx", + "columns": [ + { + "expression": "company_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "interaction_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, "tool_action_requests_issue_idx": { "name": "tool_action_requests_issue_idx", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index f4cfed69f11..578d8027ed9 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2689,6 +2689,13 @@ "when": 1793060791729, "tag": "0385_open_madame_web", "breakpoints": true + }, + { + "idx": 386, + "version": "7", + "when": 1793135146893, + "tag": "0386_steady_epoch", + "breakpoints": true } ] -} +} \ No newline at end of file diff --git a/packages/db/src/schema/tool_access.ts b/packages/db/src/schema/tool_access.ts index 9f79743d6c3..b8b317ddb91 100644 --- a/packages/db/src/schema/tool_access.ts +++ b/packages/db/src/schema/tool_access.ts @@ -839,6 +839,7 @@ export const toolActionRequests = pgTable( index("tool_action_requests_executing_deadline_idx").on(table.updatedAt, table.id).where(sql`${table.status} = 'executing'`), index("tool_action_requests_approved_idx").on(table.id).where(sql`${table.status} = 'approved'`), index("tool_action_requests_invocation_idx").on(table.invocationId), + index("tool_action_requests_company_interaction_idx").on(table.companyId, table.interactionId), index("tool_action_requests_issue_idx").on(table.companyId, table.issueId), ], ); diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index 1a9fcf51576..cb7da55e61a 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -72,6 +72,8 @@ const { settleInterruptedActionReviews: vi.fn(async () => ({ scanned: 0 })), nextActionReviewInterruptionAt: vi.fn(async () => null), recoverApprovedActionReviews: vi.fn(async () => ({ scanned: 0, retry: false })), + repairActionReviewProjections: vi.fn(async () => ({ scanned: 0, retry: false })), + actionReviewProjectionCompanies: vi.fn(async () => []), approvedActionReviewCompanies: vi.fn(async () => []), sweepActionReviews: vi.fn(async () => ({ scanned: 0 })), sweepExpiredActionReviews: vi.fn(async () => ({ scanned: 0 })), diff --git a/server/src/__tests__/tool-gateway-service.test.ts b/server/src/__tests__/tool-gateway-service.test.ts index 35dabccbe22..0d277ee4dbf 100644 --- a/server/src/__tests__/tool-gateway-service.test.ts +++ b/server/src/__tests__/tool-gateway-service.test.ts @@ -1,3 +1,6 @@ +import { toolReviewProjectionService } from "../services/tool-review-projection.js"; +import { registerToolReviewProjectionWork } from "../services/tool-review-projection-work.js"; +import { issueThreadInteractionService } from "../services/issue-thread-interactions.js"; import { registerToolReviewInterruptionWork } from "../services/tool-review-interruption-work.js"; import { writeToolReviewChange } from "../services/tool-review-work-notifications.js"; import { deleteCompany } from "../services/company-deletion.js"; @@ -281,6 +284,8 @@ describeEmbeddedPostgres("tool gateway service", () => { await coordinator?.stop(); coordinator = undefined; vi.useRealTimers(); vi.restoreAllMocks(); await db.execute(sql`drop trigger if exists sleep_review_fail on tool_invocations`); + await db.execute(sql`drop trigger if exists sleep_projection_fail on issue_thread_interactions`); + await db.execute(sql`drop function if exists sleep_projection_fail()`); await db.execute(sql`drop function if exists sleep_review_fail()`); await db.execute(sql`drop sequence if exists sleep_review_failure_attempt`); expect(errors).toEqual([]); @@ -352,6 +357,357 @@ describeEmbeddedPostgres("tool gateway service", () => { return { provider, release, started }; } + describe("tool review projection worker", () => { + async function startProjection() { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, + canReconcile: () => true, onError: error => errors.push(error) }); + await registerToolReviewProjectionWork(coordinator, { sweep: gateway.repairActionReviewProjections, companies: gateway.actionReviewProjectionCompanies }, () => !draining).ready; + } + async function savedOutcome(pending: Awaited>, source = db, + status: "executed" | "failed" | "expired" | "cancelled" = "executed") { + await writeToolReviewChange(source, pending.companyId, async tx => { + await tx.update(toolInvocations).set({ status: status === "executed" ? "succeeded" : "failed", + errorCode: status === "executed" ? null : "saved_failure", errorMessage: null, + resultSummary: { summary: "Durable result" }, completedAt: new Date() }) + .where(eq(toolInvocations.id, pending.invocationId)); + await tx.update(toolActionRequests).set({ status, updatedAt: new Date() }) + .where(eq(toolActionRequests.id, pending.id)); + }); + } + async function acceptedCard(pending: Awaited>) { + expect(pending.interactionId).not.toBeNull(); + await db.update(issueThreadInteractions).set({ status: "accepted", result: { version: 1, outcome: "accepted" } }) + .where(eq(issueThreadInteractions.id, pending.interactionId!)); + } + async function projection(pending: Awaited>) { + const [row] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, pending.interactionId!)); + return (row.result as { toolAction?: { status: string; resultSummary: string; errorCode: string | null } } | null)?.toolAction; + } + + it("becomes quiet with configuration and repairs a real foreground approval without replay", async () => { + await startProjection(); await quiet(); + const pending = await request(); + await gateway.approveActionRequest({ companyId: fixture.company.id, actionRequestId: pending.id, actor: { userId: "board" } }); + await quiet(); expect(await projection(pending)).toMatchObject({ status: "executed" }); + expect(await db.select().from(toolCallEvents).where(eq(toolCallEvents.reasonCode, "approved_action_executed"))).toHaveLength(1); + const before = (await db.select().from(activityLog).where(eq(activityLog.action, "issue.thread_interaction_updated"))).length; + await gateway.repairActionReviewProjections(); + expect((await db.select().from(activityLog).where(eq(activityLog.action, "issue.thread_interaction_updated"))).length).toBe(before); + }); + + it("keeps projection scans scoped to the company whose review changed", async () => { + const other = await createRunFixture(db); + await db.insert(toolPolicies).values(reviewNotePolicyRow(other.company)); + const foreign = await request(gateway, db, other); + await gateway.approveActionRequest({ companyId: other.company.id, actionRequestId: foreign.id, actor: { userId: "board" } }); + await startProjection(); await quiet(); + const before = await projection(foreign); + const reads: Array<{ query: string; params: unknown[] }> = []; + const debug = db.$client.options.debug; + db.$client.options.debug = (_id, query, params) => { + if (/^select/i.test(query) && query.includes('"tool_action_requests"') && query.includes("coalesce(")) + reads.push({ query, params: [...params] }); + }; + try { + for (let index = 0; index < 3; index++) { + const pending = await request(); + await gateway.approveActionRequest({ companyId: fixture.company.id, actionRequestId: pending.id, actor: { userId: "board" } }); + await settle(1500); expect(await projection(pending)).toMatchObject({ status: "executed" }); + } + } finally { db.$client.options.debug = debug; } + expect(reads.length).toBeGreaterThan(0); + for (const read of reads) { + expect(read.query).toContain('"tool_action_requests"."company_id" ='); + expect(read.params).toContain(fixture.company.id); + expect(read.params).not.toContain(other.company.id); + } + expect(await projection(foreign)).toEqual(before); await quiet(); + }); + + it("uses one candidate scan for a committed company hint with matching retained history", async () => { + const pending = await request(); + await gateway.approveActionRequest({ companyId: fixture.company.id, actionRequestId: pending.id, actor: { userId: "board" } }); + await startProjection(); await quiet(); + const reads: string[] = []; + const debug = db.$client.options.debug; + db.$client.options.debug = (_id, query) => { + if (/^select/i.test(query) && query.includes('"tool_action_requests"') && query.includes("coalesce(")) reads.push(query); + }; + try { + await writeToolReviewChange(db, pending.companyId, async tx => { + await tx.update(toolActionRequests).set({ updatedAt: new Date() }).where(eq(toolActionRequests.id, pending.id)); + }); + await settle(); + expect(reads).toHaveLength(1); + await quiet(); + expect(reads).toHaveLength(1); + reads.length = 0; + expect(await gateway.repairActionReviewProjections(fixture.company.id)).toMatchObject({ scanned: 0, retry: false }); + expect(reads).toHaveLength(1); + } finally { db.$client.options.debug = debug; } + }); + + it("backs off unresolved cards to five minutes and clears the retry after settlement", async () => { + const pending = await request(); + await savedOutcome(pending, db, "failed"); + await startProjection(); + for (const delay of [30_000, 60_000, 120_000, 240_000, 300_000, 300_000]) { + const next = coordinator!.nextWakeAt(); + expect(next).not.toBeNull(); + expect(next! - Date.now()).toBeGreaterThan(delay - 1500); + expect(next! - Date.now()).toBeLessThanOrEqual(delay); + await settle(next! - Date.now() + 100); + } + expect(await projection(pending)).toBeUndefined(); + await acceptedCard(pending); + await writeToolReviewChange(db, pending.companyId, async tx => { + await tx.update(toolActionRequests).set({ updatedAt: new Date() }).where(eq(toolActionRequests.id, pending.id)); + }); + await settle(); + expect(await projection(pending)).toMatchObject({ status: "failed" }); + await quiet(); + }); + + it("does not wake projection repair for ordinary confirmation creation, replacement, or cancellation", async () => { + await startProjection(); await quiet(); + const interactions = issueThreadInteractionService(db); + const scope = { id: fixture.issue.id, companyId: fixture.company.id }; + const [recipient] = await db.insert(agents).values({ companyId: fixture.company.id, + name: "Plan recipient", role: "engineer", adapterType: "process" }).returning(); + const input = { kind: "request_confirmation" as const, addresseeAgentId: recipient.id, + payload: { version: 1 as const, prompt: "Approve this plan?" } }; + const signal = vi.fn(); + const unsubscribe = subscribeDeliveryWork(db, DELIVERY_QUEUES.toolReviewProjection, signal); + const reads: string[] = []; + const debug = db.$client.options.debug; + db.$client.options.debug = (_id, query) => { + if (/^select/i.test(query) && query.includes('"tool_action_requests"') && query.includes("coalesce(")) reads.push(query); + }; + try { + const first = await interactions.create(scope, input, { agentId: fixture.agent.id }); + const second = await interactions.create(scope, input, { agentId: fixture.agent.id }); + const cancelled = await interactions.cancelPendingForDeletedAddressee(fixture.company.id, recipient.id); + expect(cancelled.map(row => row.id)).toEqual([second.id]); + expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, first.id)))[0].status).toBe("expired"); + await settle(); + expect(signal).not.toHaveBeenCalled(); + expect(reads).toEqual([]); + await quiet(); + } finally { unsubscribe(); db.$client.options.debug = debug; } + }); + + it.each(["executed", "failed", "expired", "cancelled"] as const)("reconstructs a saved %s result after restart and leaves terminal history quiet", async status => { + const pending = await request(); await acceptedCard(pending); await savedOutcome(pending, db, status); + await startProjection(); await quiet(); + expect(await projection(pending)).toMatchObject({ status: status === "cancelled" ? "expired" : status, + resultSummary: "Durable result", errorCode: status === "executed" ? null : "saved_failure" }); + expect(await db.select().from(toolCallEvents).where(eq(toolCallEvents.reasonCode, "approved_action_executed"))).toEqual([]); + }); + + it("retains a first terminal write during drain without passive scans", async () => { + const pending = await request(); await acceptedCard(pending); + await startProjection(); await quiet(); draining = true; + await savedOutcome(pending); await settle(); + const reads = vi.spyOn(db, "select"); + await settle(60 * 60_000); expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + expect(await projection(pending)).toBeUndefined(); + draining = false; await settle(); await quiet(); + expect(await projection(pending)).toMatchObject({ status: "executed" }); + }); + + it.each([false, true])("waits for the outer outcome transaction (rollback %s)", async rollback => { + const pending = await request(); await acceptedCard(pending); await startProjection(); await quiet(); + let entered!: () => void, release!: () => void; + const ready = new Promise(resolve => { entered = resolve; }); + const gate = new Promise(resolve => { release = resolve; }); + const aborted = new Error("fixture rollback"); + const running = db.transaction(async tx => { + await savedOutcome(pending, tx as unknown as typeof db); + entered(); await gate; if (rollback) throw aborted; + }).catch(error => { if (error !== aborted) throw error; }); + await ready; + try { await vi.advanceTimersByTimeAsync(60_000); expect(await projection(pending)).toBeUndefined(); } + finally { release(); await running; } + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(30_000); + expect(idleWorkSnapshot().active).toBe(0); + if (rollback) expect(await projection(pending)).toBeUndefined(); + else expect(await projection(pending)).toMatchObject({ status: "executed" }); + }, { timeout: 5000 }); + await quiet(); + }); + + it("retries a failed projection through drain without executing the saved action", async () => { + const pending = await request(); await acceptedCard(pending); await savedOutcome(pending); + await db.execute(sql`create function sleep_projection_fail() returns trigger language plpgsql as $$ + begin raise exception 'projection fixture failure'; end $$`); + await db.execute(sql`create trigger sleep_projection_fail before update on issue_thread_interactions + for each row when (new.result->'toolAction'->>'status' = 'executed') execute function sleep_projection_fail()`); + await startProjection(); expect(errors.length).toBeGreaterThan(0); errors.length = 0; + expect(await projection(pending)).toBeUndefined(); + draining = true; + await db.execute(sql`drop trigger sleep_projection_fail on issue_thread_interactions`); + await settle(60 * 60_000); draining = false; + await vi.waitFor(async () => { + await vi.advanceTimersByTimeAsync(30_000); + expect(await projection(pending)).toMatchObject({ status: "executed" }); + }, { timeout: 5000 }); + await quiet(); + expect(await db.select().from(toolCallEvents).where(eq(toolCallEvents.reasonCode, "approved_action_executed"))).toEqual([]); + }); + + it("backs off a failing company while another company's projection completes", async () => { + const other = await createRunFixture(db); + await db.insert(toolPolicies).values(reviewNotePolicyRow(other.company)); + const foreign = await request(gateway, db, other); await acceptedCard(foreign); await savedOutcome(foreign, db, "failed"); + await db.execute(sql`create function sleep_projection_fail() returns trigger language plpgsql as $$ + begin raise exception 'company projection fixture failure'; end $$`); + await db.execute(sql`create trigger sleep_projection_fail before update on issue_thread_interactions + for each row when (new.result->'toolAction'->>'status' = 'failed') execute function sleep_projection_fail()`); + await startProjection(); expect(errors).toHaveLength(1); errors.length = 0; + const retryAt = coordinator!.nextWakeAt(); expect(retryAt).not.toBeNull(); + const pending = await request(); await acceptedCard(pending); await savedOutcome(pending); + await request(gateway, db, other); + await settle(5000); + expect(await projection(pending)).toMatchObject({ status: "executed" }); + expect(await projection(foreign)).toBeUndefined(); + expect(errors).toEqual([]); expect(coordinator!.nextWakeAt()).toBe(retryAt); + await db.execute(sql`drop trigger sleep_projection_fail on issue_thread_interactions`); + await settle(); await quiet(); + expect(await projection(foreign)).toMatchObject({ status: "failed" }); + expect(await db.select().from(toolCallEvents).where(eq(toolCallEvents.reasonCode, "approved_action_executed"))).toEqual([]); + }); + + it("rechecks the projection under lock when foreground repair wins the race", async () => { + const pending = await request(); await acceptedCard(pending); await savedOutcome(pending); + const transaction = db.transaction.bind(db); + let entered!: () => void, release!: () => void; + const ready = new Promise(resolve => { entered = resolve; }); + const gate = new Promise(resolve => { release = resolve; }); + const intercepted = vi.spyOn(db, "transaction").mockImplementationOnce(async callback => { + entered(); await gate; return transaction(callback); + }); + const repair = gateway.repairActionReviewProjections(); + await ready; + try { await gateway.repairActionReviewProjections(); } finally { release(); } + await repair; intercepted.mockRestore(); + const updates = await db.select().from(activityLog).where(and(eq(activityLog.companyId, fixture.company.id), + eq(activityLog.action, "issue.thread_interaction_updated"))); + expect(updates).toHaveLength(1); + expect(await projection(pending)).toMatchObject({ status: "executed" }); + await startProjection(); await quiet(); + }); + + async function projectionBacklog(count: number) { + const pending = await request(); await acceptedCard(pending); await savedOutcome(pending, db, "failed"); + const [template] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, pending.interactionId!)); + const cards = Array.from({ length: count - 1 }, () => ({ ...template, id: randomUUID(), idempotencyKey: randomUUID() })); + if (cards.length) await db.insert(issueThreadInteractions).values(cards); + const requests = cards.map((card, index) => ({ ...pending, + id: `ffffffff-ffff-4fff-afff-${String(index).padStart(12, "0")}`, + interactionId: card.id, status: "failed" as const })); + if (requests.length) await db.insert(toolActionRequests).values(requests); + const cardByRequest = new Map([pending, ...requests].map(row => [row.id, row.interactionId!])); + const reflect = async (id: string) => { + await db.update(issueThreadInteractions).set({ result: sql`${issueThreadInteractions.result} + || ${JSON.stringify({ toolAction: { status: "failed" } })}::jsonb` }) + .where(eq(issueThreadInteractions.id, cardByRequest.get(id)!)); + }; + return { cardByRequest, reflect }; + } + + it("bounds a projection worker pass and continues its saved backlog", async () => { + await projectionBacklog(501); + await startProjection(); + const projected = async () => (await db.select().from(issueThreadInteractions)) + .filter(card => (card.result as { toolAction?: { status: string } } | null)?.toolAction?.status === "failed").length; + expect(await projected()).toBe(500); + expect(coordinator!.nextWakeAt()).not.toBeNull(); + await settle(2000); await quiet(); + expect(await projected()).toBe(501); + expect(await db.select().from(toolCallEvents).where(eq(toolCallEvents.reasonCode, "approved_action_executed"))).toEqual([]); + }); + + it("clears an abandoned projection cursor after empty reconstruction before the next write", async () => { + const backlog = await projectionBacklog(3); + const service = toolReviewProjectionService(db, input => backlog.reflect(input.actionRequestId)); + expect(await service.sweep(fixture.company.id, { maxRows: 1 })).toMatchObject({ scanned: 1, truncated: true }); + for (const id of backlog.cardByRequest.keys()) await backlog.reflect(id); + expect(await service.companies(fixture.company.id)).toEqual([]); + const firstId = [...backlog.cardByRequest.keys()].sort()[0]!; + await db.update(issueThreadInteractions).set({ result: sql`${issueThreadInteractions.result} - 'toolAction'` }) + .where(eq(issueThreadInteractions.id, backlog.cardByRequest.get(firstId)!)); + expect(await service.sweep(fixture.company.id, { maxRows: 500 })).toMatchObject({ scanned: 1, retry: false }); + }); + + it.each(["abort", "drain"] as const)("stops projection between rows on %s and resumes without losing cards", async mode => { + const backlog = await projectionBacklog(5), controller = new AbortController(); + const seen: string[] = []; let admitted = true; + const service = toolReviewProjectionService(db, async input => { + seen.push(input.actionRequestId); await backlog.reflect(input.actionRequestId); + if (seen.length === 1) { if (mode === "abort") controller.abort(); else admitted = false; } + }); + expect(await service.sweep(fixture.company.id, { maxRows: 500, signal: controller.signal, shouldContinue: () => admitted })) + .toMatchObject({ scanned: 1, retry: true, truncated: true }); + expect(seen).toHaveLength(1); + admitted = true; + expect(await service.sweep(fixture.company.id, { maxRows: 500 })).toMatchObject({ scanned: 4, retry: false }); + expect(new Set(seen).size).toBe(5); + }); + + it.each(["unprojectable", "failed"] as const)("continues beyond a full %s page before revisiting it", async mode => { + const backlog = await projectionBacklog(501); + const blocked = new Set([...backlog.cardByRequest.keys()].sort().slice(0, 500)), seen: string[] = []; + const service = toolReviewProjectionService(db, async input => { + seen.push(input.actionRequestId); + if (blocked.has(input.actionRequestId)) { if (mode === "failed") throw new Error("saved card unavailable"); return; } + await backlog.reflect(input.actionRequestId); + }); + const first = service.sweep(fixture.company.id, { maxRows: 500 }); + if (mode === "failed") await expect(first).rejects.toBeInstanceOf(AggregateError); + else expect(await first).toMatchObject({ scanned: 500, retry: true, truncated: true }); + expect(seen).toHaveLength(500); seen.length = 0; + expect(await service.sweep(fixture.company.id, { maxRows: 500 })).toMatchObject({ scanned: 1, retry: true, truncated: false }); + expect(seen).toHaveLength(1); expect(blocked.has(seen[0]!)).toBe(false); + }); + + it("does not starve later pages behind a card that cannot yet reflect its outcome", async () => { + const pending = await request(); await savedOutcome(pending, db, "failed"); + const [template] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, pending.interactionId!)); + const copies = Array.from({ length: 101 }, () => ({ ...template, id: randomUUID(), idempotencyKey: randomUUID(), + status: "accepted" as const, result: { version: 1 as const, outcome: "accepted" as const } })); + await db.insert(issueThreadInteractions).values(copies); + await db.insert(toolActionRequests).values(copies.map((card, index) => ({ ...pending, + id: `ffffffff-ffff-4fff-afff-${String(index).padStart(12, "0")}`, + interactionId: card.id, status: "failed" as const }))); + await startProjection(); + expect(coordinator!.nextWakeAt()).not.toBeNull(); + const cards = await db.select().from(issueThreadInteractions); + expect(cards.filter(card => (card.result as { toolAction?: { status: string } } | null)?.toolAction?.status === "failed")).toHaveLength(101); + await db.update(issueThreadInteractions).set({ addresseeAgentId: fixture.agent.id }).where(eq(issueThreadInteractions.id, pending.interactionId!)); + await issueThreadInteractionService(db).cancelPendingForDeletedAddressee(fixture.company.id, fixture.agent.id); + await settle(); await quiet(); + expect(await projection(pending)).toMatchObject({ status: "failed" }); + }); + + it.each([false, true])("reactivates when an administrative card mutation overwrites a settled projection (legacy %s)", async legacy => { + const pending = await request(); + if (legacy) { + const [card] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, pending.interactionId!)); + const payload = { ...card.payload } as Record; + delete payload.toolAction; delete payload.target; + await db.update(issueThreadInteractions).set({ payload }).where(eq(issueThreadInteractions.id, pending.interactionId!)); + } + await db.update(issueThreadInteractions).set({ addresseeAgentId: fixture.agent.id, + result: { version: 1, outcome: "accepted" } }).where(eq(issueThreadInteractions.id, pending.interactionId!)); + await savedOutcome(pending, db, "failed"); await startProjection(); await quiet(); + expect(await projection(pending)).toMatchObject({ status: "failed" }); + await issueThreadInteractionService(db).cancelPendingForDeletedAddressee(fixture.company.id, fixture.agent.id); + await settle(); await quiet(); + expect(await projection(pending)).toMatchObject({ status: "failed", errorCode: "saved_failure" }); + }); + }); + describe("interrupted review recovery worker", () => { async function startInterruption() { coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, diff --git a/server/src/index.ts b/server/src/index.ts index 50627307a54..bef68b441e7 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,3 +1,4 @@ +import { registerToolReviewProjectionWork } from "./services/tool-review-projection-work.js"; import { registerToolReviewInterruptionWork } from "./services/tool-review-interruption-work.js"; import { registerToolReviewRecoveryWork } from "./services/tool-review-recovery-work.js"; import { registerToolReviewExpiryWork } from "./services/tool-review-expiry-work.js"; @@ -1486,7 +1487,10 @@ async function startServerWithDatabaseTeardown( } else { await app.locals.toolGateway.settleInterruptedActionReviews(); } - return app.locals.toolGateway.sweepActionReviews({ includeExpiry: false, includeApproved: false, includeInterrupted: false }); + return config.heartbeatSchedulerEnabled + ? registerToolReviewProjectionWork(deliveryWork, { sweep: app.locals.toolGateway.repairActionReviewProjections, companies: app.locals.toolGateway.actionReviewProjectionCompanies }, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready + : app.locals.toolGateway.repairActionReviewProjections(); }).catch((err: unknown) => logger.error({ err }, "startup tool review recovery failed")); await timeStartupRecovery("gateway_sessions", () => config.heartbeatSchedulerEnabled ? registerGatewayCredentialExpiryWork(deliveryWork, app.locals.toolGateway, @@ -1917,8 +1921,6 @@ async function startServerWithDatabaseTeardown( logger.error({ err }, "periodic tool connection health sweep failed"); })); - trackHeartbeatSchedulerWork(app.locals.toolGateway.sweepActionReviews({ includeExpiry: false, includeApproved: false, includeInterrupted: false }).catch((err: unknown) => logger.error({ err }, "tool review recovery failed"))); - if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; if (!(await heartbeat.resolveSchedulingSuppression()).suppressed) { // Periodically reap orphaned runs (5-min staleness threshold) and make sure diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 9c943677dac..8af5f57babe 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -9,6 +9,7 @@ export const DELIVERY_QUEUES = { executionStatus: "execution-status", executionDisposition: "execution-disposition", environmentCleanup: "environment-cleanup", + toolReviewProjection: "tool-review-projection", toolReviewInterruption: "tool-review-interruption", toolReviewRecovery: "tool-review-recovery", toolReviewExpiry: "tool-review-expiry", diff --git a/server/src/services/issue-thread-interactions.test.ts b/server/src/services/issue-thread-interactions.test.ts index e2c9ac62044..24e1489006f 100644 --- a/server/src/services/issue-thread-interactions.test.ts +++ b/server/src/services/issue-thread-interactions.test.ts @@ -14,7 +14,7 @@ type SelectRow = Record; function createSelectChain(rows: SelectRow[] | ((table: unknown) => SelectRow[])) { return { from(table: unknown) { - const selectedRows = typeof rows === "function" ? rows(table) : rows; + let selectedRows = typeof rows === "function" ? rows(table) : rows; const query = { innerJoin() { return query; @@ -25,6 +25,10 @@ function createSelectChain(rows: SelectRow[] | ((table: unknown) => SelectRow[]) for() { return query; }, + limit(count: number) { + selectedRows = selectedRows.slice(0, count); + return query; + }, then(callback: (rows: SelectRow[]) => unknown) { return Promise.resolve(callback(selectedRows)); }, @@ -292,7 +296,7 @@ describe("issueThreadInteractionService", () => { ]); }); - it("withdraws a pending interaction with attribution and rejects repeats", async () => { + it("withdraws an unlinked pending interaction with attribution and rejects repeats", async () => { const { issueThreadInteractionService } = await import("./issue-thread-interactions.js"); const interactionRow = { id: "interaction-withdraw", companyId: "company-1", issueId: "11111111-1111-4111-8111-111111111111", @@ -308,8 +312,7 @@ describe("issueThreadInteractionService", () => { expect(withdrawn.status).toBe("cancelled"); expect(withdrawn.result).toEqual({ version: 1, outcome: "withdrawn", reason: "Replanning" }); expect(withdrawn.resolvedByAgentId).toBe("agent-1"); - expect(state.toolActionRequestUpdates).toHaveLength(1); - expect(state.toolActionRequestUpdates[0]).toMatchObject({ status: "cancelled", resolvedByAgentId: "agent-1" }); + expect(state.toolActionRequestUpdates).toHaveLength(0); const resolvedState = createFakeDb({ interactionRow: { ...interactionRow, status: "accepted" } }); const resolvedSvc = issueThreadInteractionService(resolvedState.db as never); await expect(resolvedSvc.withdrawInteraction( @@ -388,7 +391,9 @@ describe("issueThreadInteractionService", () => { }, result: null, resolvedAt: null, createdAt: new Date("2026-07-25T10:00:00.000Z"), updatedAt: new Date("2026-07-25T10:00:00.000Z"), }; - const state = createFakeDb({ interactionRow }); + const state = createFakeDb({ interactionRow, parentRows: [{ + id: "33333333-3333-4333-8333-333333333333", companyId: "company-1", interactionId: interactionRow.id, + }] }); const svc = issueThreadInteractionService(state.db as never); const expired = await svc.expirePendingInteractionsForTerminalIssue( { id: interactionRow.issueId, companyId: "company-1", status: "cancelled" }, diff --git a/server/src/services/issue-thread-interactions.ts b/server/src/services/issue-thread-interactions.ts index 48ecf1d9dbe..44eb4391055 100644 --- a/server/src/services/issue-thread-interactions.ts +++ b/server/src/services/issue-thread-interactions.ts @@ -1073,8 +1073,22 @@ class InteractionResolvedConcurrentlyError extends Error { // tool_action_requests row. Administrative resolutions (withdraw, terminal-issue // expiry) must settle that row too, or the parked call stays approvable under // its own one-hour lifecycle after its card is gone. +async function hasLinkedToolActionRequests(db: Pick, companyId: string, interactionIds: string[]) { + if (interactionIds.length === 0) return false; + const [request] = await db.select({ id: toolActionRequests.id }).from(toolActionRequests) + .where(and(eq(toolActionRequests.companyId, companyId), inArray(toolActionRequests.interactionId, interactionIds))) + .limit(1); + return request !== undefined; +} + +async function notifyToolProjectionForInteractions(db: Pick, companyId: string, interactionIds: string[]) { + // The persisted link also covers legacy cards without current payload flags. + if (await hasLinkedToolActionRequests(db, companyId, interactionIds)) + await notifyDeliveryWork(db, DELIVERY_QUEUES.toolReviewProjection, companyId); +} + async function resolveLinkedToolActionRequests( - db: Pick, + db: Pick, interaction: Pick, outcome: { status: "expired" | "cancelled"; @@ -1083,7 +1097,8 @@ async function resolveLinkedToolActionRequests( now: Date; }, ) { - if (interaction.kind !== "request_confirmation") return; + if (interaction.kind !== "request_confirmation" || + !(await hasLinkedToolActionRequests(db, interaction.companyId, [interaction.id]))) return; await notifyToolReviewChange(db, interaction.companyId); await db .update(toolActionRequests) @@ -2240,6 +2255,7 @@ export function issueThreadInteractionService( }) : undefined; + if (lockedCurrent.kind === "request_confirmation") await notifyToolProjectionForInteractions(tx, lockedCurrent.companyId, [lockedCurrent.id]); const [updated] = await tx .update(issueThreadInteractions) .set({ @@ -2517,6 +2533,7 @@ export function issueThreadInteractionService( now, }); + if (lockedCurrent.kind === "request_confirmation") await notifyToolProjectionForInteractions(tx, lockedCurrent.companyId, [lockedCurrent.id]); const [resolved] = await tx .update(issueThreadInteractions) .set({ @@ -3273,6 +3290,7 @@ export function issueThreadInteractionService( const cancelled: IssueThreadInteraction[] = []; for (const row of rows) { const updated = await db.transaction(async (tx) => { + if (row.kind === "request_confirmation") await notifyToolProjectionForInteractions(tx, row.companyId, [row.id]); const [resolved] = await tx .update(issueThreadInteractions) .set({ @@ -3357,6 +3375,7 @@ export function issueThreadInteractionService( const expired: IssueThreadInteraction[] = []; for (const { row, replacementInteractionId } of supersededRows) { const updated = await db.transaction(async (tx) => { + if (row.kind === "request_confirmation") await notifyToolProjectionForInteractions(tx, row.companyId, [row.id]); const [updatedRow] = await tx .update(issueThreadInteractions) .set({ @@ -3776,6 +3795,19 @@ export function issueThreadInteractionService( data.kind === "ask_user_questions" ? buildSupersededByNewerInteractionResult(row.id) : buildSupersededByNewerRequestResult(row.id); + const siblingCondition = and( + eq(issueThreadInteractions.companyId, issue.companyId), + eq(issueThreadInteractions.issueId, issue.id), + eq(issueThreadInteractions.kind, data.kind), + eq(issueThreadInteractions.createdByAgentId, actor.agentId), + eq(issueThreadInteractions.status, "pending"), + ne(issueThreadInteractions.id, row.id), + ); + if (data.kind === "request_confirmation") { + // The issue lock above excludes concurrent interaction creation. + const siblings = await tx.select({ id: issueThreadInteractions.id }).from(issueThreadInteractions).where(siblingCondition); + await notifyToolProjectionForInteractions(tx, issue.companyId, siblings.map(sibling => sibling.id)); + } const supersededRows = await tx .update(issueThreadInteractions) .set({ @@ -3786,17 +3818,7 @@ export function issueThreadInteractionService( resolvedAt: now, updatedAt: now, }) - .where( - and( - eq(issueThreadInteractions.companyId, issue.companyId), - eq(issueThreadInteractions.issueId, issue.id), - eq(issueThreadInteractions.kind, data.kind), - eq(issueThreadInteractions.createdByAgentId, actor.agentId), - eq(issueThreadInteractions.status, "pending"), - ne(issueThreadInteractions.id, row.id), - - ), - ) + .where(siblingCondition) .returning(); for (const supersededRow of supersededRows) { await resolveLinkedToolActionRequests(tx, supersededRow, { @@ -4442,6 +4464,7 @@ export function issueThreadInteractionService( .delete(toolOauthStates) .where(eq(toolOauthStates.interactionId, row.id)); } + if (row.kind === "request_confirmation") await notifyToolProjectionForInteractions(tx, row.companyId, [row.id]); const [resolved] = await tx .update(issueThreadInteractions) .set({ @@ -4610,6 +4633,7 @@ export function issueThreadInteractionService( ); if (!sampleConfirmationRow) continue; const updatedInteractions = await db.transaction(async (tx) => { + await notifyToolProjectionForInteractions(tx, issue.companyId, confirmationRowIds); const updatedRows = await tx .update(issueThreadInteractions) .set({ diff --git a/server/src/services/tool-gateway.ts b/server/src/services/tool-gateway.ts index 2d94a361de2..4cacb30e47e 100644 --- a/server/src/services/tool-gateway.ts +++ b/server/src/services/tool-gateway.ts @@ -1,3 +1,4 @@ +import { toolReviewProjectionService } from "./tool-review-projection.js"; import { toolReviewInterruptionService } from "./tool-review-interruption.js"; import { bindVerifiedToolApprovalArguments, claimToolTestApproval, releaseToolApproval, persistToolTestOutcome, startToolInvocation, type ToolApprovalClaim } from "./tool-approval-execution.js"; import { approvedToolReviewCompanies, prepareApprovedToolReview, recoverApprovedToolReviews } from "./tool-review-recovery.js"; @@ -2398,6 +2399,11 @@ export function createToolGatewayService( ? "stale_target" : null; if (!outcome) return; + const projection = asRecord(currentResult?.toolAction); + if (projection?.status === (input.status === "cancelled" ? "expired" : input.status) + && projection.errorCode === (input.errorCode ?? null) + && projection.errorMessage === (input.errorMessage ?? null) + && projection.resultSummary === (input.resultSummary ?? null)) return false; const now = new Date(); await tx @@ -8125,10 +8131,10 @@ export function createToolGatewayService( return { projectId: issue.projectId }; } const expiredAt = new Date(); - await db + await writeToolReviewChange(db, claimed.companyId, tx => tx .update(toolActionRequests) .set({ status: "expired", resolvedAt: expiredAt, updatedAt: expiredAt }) - .where(eq(toolActionRequests.id, claimed.id)); + .where(eq(toolActionRequests.id, claimed.id))); await reflectToolActionInteractionLifecycle({ actionRequestId: claimed.id, status: "expired", @@ -8820,6 +8826,7 @@ export function createToolGatewayService( }; } + const reviewProjection = toolReviewProjectionService(db, reflectToolActionInteractionLifecycle); const reviewInterruption = toolReviewInterruptionService(db, async settled => { void emitConnectionInvoked(db, settled.invocationId); await reflectToolActionInteractionLifecycle(settled); @@ -9917,6 +9924,8 @@ export function createToolGatewayService( return buildTestCallStatus(actionRequest, invocation); }, + repairActionReviewProjections: reviewProjection.sweep, + actionReviewProjectionCompanies: reviewProjection.companies, settleInterruptedActionReviews: reviewInterruption.sweep, nextActionReviewInterruptionAt: reviewInterruption.next, sweepExpiredActionReviews: reviewExpiry.sweep, @@ -9951,40 +9960,7 @@ export function createToolGatewayService( }, async sweepActionReviews(input: { includeExpiry?: boolean; includeApproved?: boolean; includeInterrupted?: boolean } = {}) { - // A process may stop between persisting a provider outcome and updating the - // feed projection. Reconcile from authoritative rows before delivering it. - const unreflected = await db - .select({ request: toolActionRequests, invocation: toolInvocations }) - .from(toolActionRequests) - .innerJoin( - issueThreadInteractions, - eq(issueThreadInteractions.id, toolActionRequests.interactionId), - ) - .innerJoin( - toolInvocations, - eq(toolInvocations.id, toolActionRequests.invocationId), - ) - .where( - and( - inArray(toolActionRequests.status, [ - "executed", - "failed", - "expired", - "cancelled", - ]), - sql`coalesce(${issueThreadInteractions.result}->'toolAction'->>'status', '') <> case when ${toolActionRequests.status} = 'cancelled' then 'expired' else ${toolActionRequests.status} end`, - ), - ) - .limit(100); - for (const { request, invocation } of unreflected) - await reflectToolActionInteractionLifecycle({ - actionRequestId: request.id, - status: request.status as - "executed" | "failed" | "expired" | "cancelled", - errorCode: invocation.errorCode, - errorMessage: invocation.errorMessage, - resultSummary: invocation.resultSummary?.summary, - }); + await reviewProjection.sweep(); let scanned = input.includeExpiry === false ? 0 : (await reviewExpiry.sweep()).scanned; if (input.includeApproved !== false) scanned += (await this.recoverApprovedActionReviews()).scanned; if (input.includeInterrupted !== false) scanned += (await reviewInterruption.sweep()).scanned; diff --git a/server/src/services/tool-review-projection-work.ts b/server/src/services/tool-review-projection-work.ts new file mode 100644 index 00000000000..b400f19f615 --- /dev/null +++ b/server/src/services/tool-review-projection-work.ts @@ -0,0 +1,33 @@ +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import { registerScopedDeadlineWork } from "./scoped-deadline-work.js"; +import type { toolReviewProjectionService } from "./tool-review-projection.js"; + +export function registerToolReviewProjectionWork( + coordinator: ReturnType, + service: ReturnType, + canRun: () => boolean, +) { + const unresolvedPasses = new Map(); + return registerScopedDeadlineWork(coordinator, { + queue: DELIVERY_QUEUES.toolReviewProjection, + batchSize: 1, + decodeScope: hint => hint === "*" ? undefined : hint, + matches: (entry, companyId?: string) => companyId === undefined || entry.companyId === companyId, + load: async (companyId?: string) => { + const rows = companyId ? [{ companyId }] : await service.companies(); + for (const scope of unresolvedPasses.keys()) { + if (!companyId || scope === companyId) unresolvedPasses.delete(scope); + } + return rows.map(row => ({ ...row, id: row.companyId, dueAt: Date.now() })); + }, + check: async (entry, signal) => { + const result = await service.sweep(entry.companyId, { maxRows: 500, signal, shouldContinue: canRun }); + if (!result.retry) { unresolvedPasses.delete(entry.id); return null; } + if (result.truncated) return { ...entry, dueAt: Date.now() + 1000 }; + const count = Math.min((unresolvedPasses.get(entry.id) ?? 0) + 1, 5); + unresolvedPasses.set(entry.id, count); + return { ...entry, dueAt: Date.now() + Math.min(300_000, 30_000 * 2 ** (count - 1)) }; + }, + }, canRun); +} diff --git a/server/src/services/tool-review-projection.ts b/server/src/services/tool-review-projection.ts new file mode 100644 index 00000000000..32644929cac --- /dev/null +++ b/server/src/services/tool-review-projection.ts @@ -0,0 +1,77 @@ +import { and, asc, eq, gt, inArray, sql } from "drizzle-orm"; +import { issueThreadInteractions, toolActionRequests, toolInvocations, type Db } from "@paperclipai/db"; + +export function toolReviewProjectionService(db: Db, reflect: (input: { + actionRequestId: string; + status: "executed" | "failed" | "expired" | "cancelled"; + errorCode: string | null; + errorMessage: string | null; + resultSummary: string | null; +}) => Promise) { + const mismatch = and( + inArray(toolActionRequests.status, ["executed", "failed", "expired", "cancelled"]), + sql`coalesce(${issueThreadInteractions.result}->'toolAction'->>'status', '') <> case when ${toolActionRequests.status} = 'cancelled' then 'expired' else ${toolActionRequests.status} end`, + ); + const candidates = (companyId?: string, cursor?: string) => db + .select({ request: toolActionRequests, invocation: toolInvocations }) + .from(toolActionRequests) + .innerJoin(issueThreadInteractions, eq(issueThreadInteractions.id, toolActionRequests.interactionId)) + .innerJoin(toolInvocations, eq(toolInvocations.id, toolActionRequests.invocationId)) + .where(and(mismatch, companyId ? eq(toolActionRequests.companyId, companyId) : undefined, + cursor ? gt(toolActionRequests.id, cursor) : undefined)) + .orderBy(asc(toolActionRequests.id)); + + const continuations = new Map(); + return { + async companies(companyId?: string) { + const rows = await db.select({ companyId: toolActionRequests.companyId }).from(toolActionRequests) + .innerJoin(issueThreadInteractions, eq(issueThreadInteractions.id, toolActionRequests.interactionId)) + .innerJoin(toolInvocations, eq(toolInvocations.id, toolActionRequests.invocationId)) + .where(and(mismatch, companyId ? eq(toolActionRequests.companyId, companyId) : undefined)) + .groupBy(toolActionRequests.companyId); + const pending = new Set(rows.map(row => row.companyId)); + for (const scope of continuations.keys()) { + if (companyId && scope !== companyId) continue; + if (scope === "*" ? rows.length === 0 : !pending.has(scope)) continuations.delete(scope); + } + return rows; + }, + async sweep(companyId?: string, options: { + maxRows?: number; + signal?: AbortSignal; + shouldContinue?: () => boolean; + } = {}) { + const maxRows = options.maxRows ?? Infinity; + const scope = companyId ?? "*"; + let cursor = maxRows === Infinity ? undefined : continuations.get(scope); + const resumed = cursor !== undefined; + let scanned = 0, completed = false; + const canContinue = () => !options.signal?.aborted && (options.shouldContinue?.() ?? true); + const failures: unknown[] = []; + while (scanned < maxRows && canContinue()) { + const pageSize = Math.min(100, maxRows - scanned); + const rows = await candidates(companyId, cursor).limit(pageSize); + for (const { request, invocation } of rows) { + if (!canContinue()) break; + try { + await reflect({ actionRequestId: request.id, + status: request.status as "executed" | "failed" | "expired" | "cancelled", + errorCode: invocation.errorCode, errorMessage: invocation.errorMessage, + resultSummary: invocation.resultSummary?.summary ?? null }); + } catch (error) { failures.push(error); } + scanned++; + cursor = request.id; + } + if (!canContinue()) break; + if (rows.length < pageSize) { completed = true; break; } + } + if (maxRows !== Infinity) { + if (completed) continuations.delete(scope); + else if (cursor) continuations.set(scope, cursor); + } + if (failures.length) throw new AggregateError(failures, "Tool review projection repair failed"); + return { scanned, truncated: !completed, + retry: !completed || ((scanned > 0 || resumed) && (await candidates(companyId).limit(1)).length > 0) }; + }, + }; +} diff --git a/server/src/services/tool-review-work-notifications.ts b/server/src/services/tool-review-work-notifications.ts index 44957fefc1b..1ed9dfe7520 100644 --- a/server/src/services/tool-review-work-notifications.ts +++ b/server/src/services/tool-review-work-notifications.ts @@ -5,6 +5,7 @@ export async function notifyToolReviewChange(transaction: object, companyId: str await notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewExpiry, companyId); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewRecovery, companyId); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewInterruption, companyId); + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewProjection, companyId); } export function writeToolReviewChange(db: Db, companyId: string, write: (transaction: Db) => Promise): Promise {