diff --git a/doc/idle-tool-review-projection.md b/doc/idle-tool-review-projection.md new file mode 100644 index 00000000000..802c1a4105e --- /dev/null +++ b/doc/idle-tool-review-projection.md @@ -0,0 +1,106 @@ +# 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 now reconciles at startup and after committed changes. +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 only `sweep` +and the returned retry obligation. No provider payload enters shared scheduling. +No table or migration is added. + +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 reconciliation; they do not create a recurring deadline. + +| 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 | +| Generic confirmation accept/reject, deleted-addressee cancellation | Projection notification before replacing card state in the owning transaction | +| 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. +The joins inspect existing 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. 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 4339c9f387e..2bb3418992e 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), checks passed at `d41590ac0e`; gateway credentials: [#17450](https://github.com/paperclipai/paperclip/pull/17450), all checks passed, Greptile 5/5 at `c5c53fc43f`; pending tool-review expiry: [#17453](https://github.com/paperclipai/paperclip/pull/17453), all checks passed, Greptile 5/5 at `ce6a260a83`; approved recovery: [#17454](https://github.com/paperclipai/paperclip/pull/17454), Greptile 5/5 at `7a1cf03e62`, CI running; interrupted claims: [#17458](https://github.com/paperclipai/paperclip/pull/17458), local checks passed, fresh CI/review running; projection pending | +| 07 | Secret proposals, tool reviews, gateway tokens | Separate expiry/scrub/recovery owners | Secret proposals: [#17446](https://github.com/paperclipai/paperclip/pull/17446), checks passed at `d41590ac0e`; gateway credentials: [#17450](https://github.com/paperclipai/paperclip/pull/17450), all checks passed, Greptile 5/5 at `c5c53fc43f`; pending tool-review expiry: [#17453](https://github.com/paperclipai/paperclip/pull/17453), all checks passed, Greptile 5/5 at `ce6a260a83`; approved recovery: [#17454](https://github.com/paperclipai/paperclip/pull/17454), all checks passed, Greptile 5/5 at `7a1cf03e62`; interrupted claims: [#17458](https://github.com/paperclipai/paperclip/pull/17458), all checks passed, Greptile 5/5 at `85003a7ac4`; projection implemented, PR pending | | 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 | @@ -255,3 +255,32 @@ 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. + +Fresh parent evidence: #17454 at `7a1cf03e62` and #17458 at `85003a7ac4` have +passing CI test/build matrices and Greptile 5/5 with no unresolved findings. +Their additional Paperclip Review checks are still queued/running. The first +six PRs remain open; the full inventory is still in scope. Master inspected +at `410e5c6eed` retains these recurring duties and has no migration-number +collision (its latest journal entry is 0377). + +Slice 07 final local verification: 815 selected tests pass, workspace typecheck +and build pass. 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. diff --git a/server/src/__tests__/server-startup-feedback-export.test.ts b/server/src/__tests__/server-startup-feedback-export.test.ts index c055bf3cb7f..3c9ce399651 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -69,6 +69,7 @@ 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 })), sweepActionReviews: vi.fn(async () => ({ scanned: 0 })), sweepExpiredActionReviews: vi.fn(async () => ({ scanned: 0 })), nextActionReviewExpiryAt: vi.fn(async () => null), diff --git a/server/src/__tests__/tool-gateway-service.test.ts b/server/src/__tests__/tool-gateway-service.test.ts index 8c25757a09f..4010bf77fc7 100644 --- a/server/src/__tests__/tool-gateway-service.test.ts +++ b/server/src/__tests__/tool-gateway-service.test.ts @@ -1,3 +1,5 @@ +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 { subscribeDeliveryWork } from "../services/delivery-work-notifications.js"; @@ -277,6 +279,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([]); @@ -342,6 +346,155 @@ 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 }, () => !draining).ready; + } + async function savedOutcome(pending: Awaited>, source = db, + status: "executed" | "failed" | "expired" | "cancelled" = "executed") { + await writeToolReviewChange(source, 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.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("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(); + }); + + 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("reactivates when an administrative card mutation overwrites a settled projection", async () => { + const pending = await request(); + 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 c02b2a252c5..dd54b7a393b 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"; @@ -1458,7 +1459,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 }, + () => !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, @@ -1883,8 +1887,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 35073a92fb6..3de970b5f33 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -8,6 +8,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.ts b/server/src/services/issue-thread-interactions.ts index f29473ed6e3..8c4ac914f13 100644 --- a/server/src/services/issue-thread-interactions.ts +++ b/server/src/services/issue-thread-interactions.ts @@ -2240,6 +2240,7 @@ export function issueThreadInteractionService( }) : undefined; + if (lockedCurrent.kind === "request_confirmation") await notifyDeliveryWork(tx, DELIVERY_QUEUES.toolReviewProjection); const [updated] = await tx .update(issueThreadInteractions) .set({ @@ -2517,6 +2518,7 @@ export function issueThreadInteractionService( now, }); + if (lockedCurrent.kind === "request_confirmation") await notifyDeliveryWork(tx, DELIVERY_QUEUES.toolReviewProjection); const [resolved] = await tx .update(issueThreadInteractions) .set({ @@ -3246,6 +3248,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 notifyDeliveryWork(tx, DELIVERY_QUEUES.toolReviewProjection); const [resolved] = await tx .update(issueThreadInteractions) .set({ @@ -3330,6 +3333,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 notifyDeliveryWork(tx, DELIVERY_QUEUES.toolReviewProjection); const [updatedRow] = await tx .update(issueThreadInteractions) .set({ @@ -3749,6 +3753,7 @@ export function issueThreadInteractionService( data.kind === "ask_user_questions" ? buildSupersededByNewerInteractionResult(row.id) : buildSupersededByNewerRequestResult(row.id); + if (data.kind === "request_confirmation") await notifyDeliveryWork(tx, DELIVERY_QUEUES.toolReviewProjection); const supersededRows = await tx .update(issueThreadInteractions) .set({ @@ -4400,6 +4405,7 @@ export function issueThreadInteractionService( .delete(toolOauthStates) .where(eq(toolOauthStates.interactionId, row.id)); } + if (row.kind === "request_confirmation") await notifyDeliveryWork(tx, DELIVERY_QUEUES.toolReviewProjection); const [resolved] = await tx .update(issueThreadInteractions) .set({ @@ -4568,6 +4574,7 @@ export function issueThreadInteractionService( ); if (!sampleConfirmationRow) continue; const updatedInteractions = await db.transaction(async (tx) => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.toolReviewProjection); const updatedRows = await tx .update(issueThreadInteractions) .set({ diff --git a/server/src/services/tool-gateway.ts b/server/src/services/tool-gateway.ts index 9a66bea7975..6d4f2dd5274 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 { claimToolTestApproval, persistToolTestOutcome, startToolTestInvocation, type ToolTestApprovalClaim } from "./tool-test-approval.js"; import { recoverApprovedToolReviews } from "./tool-review-recovery.js"; @@ -2326,6 +2327,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 @@ -7916,10 +7922,10 @@ export function createToolGatewayService( return { projectId: issue.projectId }; } const expiredAt = new Date(); - await db + await writeToolReviewChange(db, 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", @@ -8626,6 +8632,7 @@ export function createToolGatewayService( }; } + const reviewProjection = toolReviewProjectionService(db, reflectToolActionInteractionLifecycle); const reviewInterruption = toolReviewInterruptionService(db, async settled => { void emitConnectionInvoked(db, settled.invocationId); await reflectToolActionInteractionLifecycle(settled); @@ -9649,6 +9656,7 @@ export function createToolGatewayService( return buildTestCallStatus(actionRequest, invocation); }, + repairActionReviewProjections: reviewProjection.sweep, settleInterruptedActionReviews: reviewInterruption.sweep, nextActionReviewInterruptionAt: reviewInterruption.next, sweepExpiredActionReviews: reviewExpiry.sweep, @@ -9679,40 +9687,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..09f454b080c --- /dev/null +++ b/server/src/services/tool-review-projection-work.ts @@ -0,0 +1,20 @@ +import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; +import { registerDeadlineWork } from "./deadline-work.js"; +import type { toolReviewProjectionService } from "./tool-review-projection.js"; + +export function registerToolReviewProjectionWork( + coordinator: ReturnType, + service: ReturnType, + canRun: () => boolean, +) { + let next: number | null = null; + return registerDeadlineWork(coordinator, DELIVERY_QUEUES.toolReviewProjection, { + retryMs: 30_000, canRun, + run: async () => { + const result = await service.sweep(); + next = result.retry ? Date.now() + 30_000 : null; + }, + next: async () => next, + }); +} diff --git a/server/src/services/tool-review-projection.ts b/server/src/services/tool-review-projection.ts new file mode 100644 index 00000000000..ee0e88daf49 --- /dev/null +++ b/server/src/services/tool-review-projection.ts @@ -0,0 +1,46 @@ +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 = (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, cursor ? gt(toolActionRequests.id, cursor) : undefined)) + .orderBy(asc(toolActionRequests.id)); + + return { + async sweep() { + let cursor: string | undefined; + let scanned = 0; + const failures: unknown[] = []; + for (;;) { + const rows = await candidates(cursor).limit(100); + for (const { request, invocation } of rows) { + 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 += rows.length; + if (rows.length < 100) break; + cursor = rows[rows.length - 1].request.id; + } + if (failures.length) throw new AggregateError(failures, "Tool review projection repair failed"); + return { scanned, retry: (await candidates().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 728fad6d15f..e67350b45cd 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) { await notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewExpiry); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewRecovery); await notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewInterruption); + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewProjection); } export function writeToolReviewChange(db: Db, write: (transaction: Db) => Promise): Promise {