diff --git a/doc/idle-approved-tool-recovery.md b/doc/idle-approved-tool-recovery.md new file mode 100644 index 00000000000..f7fe02cc90f --- /dev/null +++ b/doc/idle-approved-tool-recovery.md @@ -0,0 +1,82 @@ +# Approved tool-action recovery + +Approved tool actions recover at startup and after committed review changes. +The aggregate maintenance tick no longer scans approved requests. The existing +request and invocation rows remain authoritative; this change adds no queue table. + +## Work and quiet states + +The gateway owns the approved-work predicate. A signed agent action with +`executionOnApprove: true` owes execution through the normal approval path. +A legacy signed action without that flag remains deliberately inert. A Test-tab +approval can owe execution even when no agent run exists. Recovery uses the saved +origin and current authorization/target checks; the scheduler sees only whether +another attempt is needed. + +Recovery pages approved requests in ID order using a partial index. It does not +let a full batch of unavailable approvals starve later work. A failed read or +unresolved executable approval retains a 30-second retry. Once a pass leaves no +such obligation, there is no recurring approved-recovery query. An empty system, +configured idle tools, completed history, and deliberately inert approvals are +quiet after reconciliation. + +## Producer and transaction ownership + +The existing tool-review notification boundary now signals both pending expiry +and approved recovery. This covers human review transactions, gateway approval +and accepted-interaction inference, administrative interaction resolution, review +queue invalidation, and the gateway's approval claims and managed-argument +invalidation. The notification is awaited before SQL in the existing transaction; +caller-owned outer transactions retain commit/rollback ownership. + +A pending-only change can cause one empty approved-recovery pass. It does not +create a periodic scan. No per-route recovery bookkeeping or database listener +is added. Signals use the existing database owner's single-active-writer model. +Independent SQL writers or another process need a separate notification contract. + +## Single execution and outcome ownership + +Agent actions retain their existing approved-to-executing claim. Test-tab +foreground approval and recovery now share an equivalent claim before provider +dispatch. The invocation then starts only if the request still holds that claim. +Provider calls and target/policy lookup remain outside SQL transactions. + +The Test-tab invocation outcome and terminal request status commit together. +Claim timestamps fence late settlement. A later audit failure cannot overwrite +an already committed outcome. Recovery repairs legacy approved Test-tab requests +from a saved terminal invocation without decoding old arguments or calling the +provider. A legacy executing invocation moves to the existing interrupted-work +path; it is not dispatched again. + +If the provider may have run but its result cannot be persisted, the executing +claim remains authoritative. The existing stale-execution recovery quarantines +unknown outcomes. This slice does not promise exactly-once external effects or +replay an uncertain provider call. + +## Lifecycle and remaining duties + +The shared deadline worker coalesces notifications, preserves changes during a +pass, and retains skipped work across idle drain/resume. Startup reconstructs +saved approvals. Shutdown stops its coordinator registration. With heartbeat +scheduling disabled, startup keeps the existing one-off recovery behavior. + +Pending expiry is owned by the preceding slice. Interrupted execution and +interaction projection repair remain in the old sweep until their own migrations. +The public combined sweep retains all duties for manual/recovery callers. +No Cloud eligibility or external-wake proof changes here; worker quiet does not +establish deployed database or host suspension. + +## Verification + +Real PostgreSQL service/startup tests cover producer commit and rollback, +startup reconstruction, empty/configured/history quiet states, concrete retries, +first-write drain/resume, legacy inert approvals, and multi-page recovery. +Controlled Test-tab provider tests cover both foreground/recovery race winners, +completed legacy outcomes, interrupted legacy calls, target changes, and a failed +atomic outcome write followed by quarantine without another provider call. + +The service/startup selection passes 122 tests. The existing gateway acceptance +suite passes 92 tests. The strengthened paging and outcome-failure cases pass +again after their final edits. Workspace typecheck and build pass. Full CI and +review remain separate PR gates; no real external provider was invoked by these +fixtures. 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 efe48c16152..89a9e404a99 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -162,3 +162,20 @@ Current parent CI note: #17450 at `c5c53fc43f` has Greptile 5/5. Its server shar All 16 login-cleanup tests passed locally on recheck. The run remains active; an attempted single-job rerun was rejected as not rerunnable. Recheck after the run finishes before requesting its one justified retry. Do not report it green. + + +Slice 05 implementation now separates approved recovery from the aggregate tick. +The existing review notification boundary wakes it after outer commit. Legacy +inert approvals and completed Test-tab history become quiet; unresolved work +retains a retry. Test-tab foreground and recovery share a claim and atomic outcome +settlement. No provider call occurs inside that transaction. See +[approved recovery](../idle-approved-tool-recovery.md). Local service/startup +verification passes 122 tests, gateway acceptance passes 92, and workspace +typecheck/build pass. Paging and outcome-failure tests passed again after their +final strengthening. Full CI/review is not yet complete. + +Parent CI follow-up: #17450's completed run was retried once for its unchanged +login-cleanup timeout after all 16 local tests passed. #17453 has a Runner +prepared-launch close failure. Seven focused local Runner cases also fail, so +this is under investigation rather than assumed flaky. No retry of that run has +been requested. Neither parent is reported green without fresh evidence. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 97f1dd3d7c8..5dac4f14540 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -506,6 +506,7 @@ const MIGRATIONS_WITH_TOP_LEVEL_STATEMENTS = new Set([ "0369_late_logan.sql", "0378_mean_scream.sql", "0379_narrow_the_executioner.sql", + "0380_funny_adam_warlock.sql", ]); async function applyPendingMigrationsManually( diff --git a/packages/db/src/migrations/0380_funny_adam_warlock.sql b/packages/db/src/migrations/0380_funny_adam_warlock.sql new file mode 100644 index 00000000000..c4f62d14e82 --- /dev/null +++ b/packages/db/src/migrations/0380_funny_adam_warlock.sql @@ -0,0 +1 @@ +CREATE INDEX CONCURRENTLY IF NOT EXISTS "tool_action_requests_approved_idx" ON "tool_action_requests" USING btree ("id") WHERE "tool_action_requests"."status" = 'approved'; diff --git a/packages/db/src/migrations/meta/0375_snapshot.json b/packages/db/src/migrations/meta/0380_snapshot.json similarity index 99% rename from packages/db/src/migrations/meta/0375_snapshot.json rename to packages/db/src/migrations/meta/0380_snapshot.json index 7bd1861bd2c..7b91e0b5909 100644 --- a/packages/db/src/migrations/meta/0375_snapshot.json +++ b/packages/db/src/migrations/meta/0380_snapshot.json @@ -1,6 +1,6 @@ { - "id": "3e17203b-ec95-41e9-8363-685982450199", - "prevId": "945c31ed-f39f-4272-a91d-85de925f5361", + "id": "e78284d4-7983-44c9-a285-0b68f22c3326", + "prevId": "b8e3cf47-6bf6-487f-adee-5642c0f97ea8", "version": "7", "dialect": "postgresql", "tables": { @@ -52787,6 +52787,44 @@ "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_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": [ @@ -55197,6 +55235,21 @@ "method": "btree", "with": {} }, + "tool_gateway_sessions_expiry_idx": { + "name": "tool_gateway_sessions_expiry_idx", + "columns": [ + { + "expression": "expires_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + }, "tool_gateway_sessions_run_idx": { "name": "tool_gateway_sessions_run_idx", "columns": [ diff --git a/packages/db/src/migrations/meta/_journal.json b/packages/db/src/migrations/meta/_journal.json index 37bb3bf609e..f4191cc18e2 100644 --- a/packages/db/src/migrations/meta/_journal.json +++ b/packages/db/src/migrations/meta/_journal.json @@ -2647,6 +2647,13 @@ "when": 1792861791949, "tag": "0379_narrow_the_executioner", "breakpoints": true + }, + { + "idx": 380, + "version": "7", + "when": 1792863024087, + "tag": "0380_funny_adam_warlock", + "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 000922351bb..f6b87c17913 100644 --- a/packages/db/src/schema/tool_access.ts +++ b/packages/db/src/schema/tool_access.ts @@ -830,6 +830,7 @@ export const toolActionRequests = pgTable( index("tool_action_requests_company_status_idx").on(table.companyId, table.status), index("tool_action_requests_pending_expiry_idx").on(table.expiresAt, table.id) .where(sql`${table.status} = 'pending'`), + 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_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 fc66f95154c..790fd64ab54 100644 --- a/server/src/__tests__/server-startup-feedback-export.test.ts +++ b/server/src/__tests__/server-startup-feedback-export.test.ts @@ -66,6 +66,7 @@ const { stop: vi.fn(async () => {}), }, toolGateway: { + recoverApprovedActionReviews: 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 7d6023926ef..166f8c3dbaa 100644 --- a/server/src/__tests__/tool-gateway-service.test.ts +++ b/server/src/__tests__/tool-gateway-service.test.ts @@ -56,6 +56,7 @@ import { import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; +import { registerToolReviewRecoveryWork } from "../services/tool-review-recovery-work.js"; import { registerToolReviewExpiryWork } from "../services/tool-review-expiry-work.js"; import { idleWorkSnapshot } from "../services/task-admission.js"; @@ -209,7 +210,7 @@ function runSessionInput(company: IdRef, agent: IdRef, run: IdRef): Parameters { let db!: ReturnType; let tempDb: Awaited> | null = null; - let stopExpiryWorker: (() => Promise) | undefined; + let stopReviewWorker: (() => Promise) | undefined; beforeAll(async () => { if (!externalDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-tool-gateway-"); @@ -220,7 +221,7 @@ describeEmbeddedPostgres("tool gateway service", () => { }, 20_000); afterEach(async () => { - await stopExpiryWorker?.(); stopExpiryWorker = undefined; + await stopReviewWorker?.(); stopReviewWorker = undefined; await clearFixtures(); }); @@ -267,7 +268,7 @@ describeEmbeddedPostgres("tool gateway service", () => { await db.insert(toolPolicies).values(reviewNotePolicyRow(fixture.company)); gateway = createTestToolGatewayService(db); draining = false; errors.length = 0; - stopExpiryWorker = stopWorker; + stopReviewWorker = stopWorker; vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout", "setInterval", "clearInterval"], shouldAdvanceTime: true }); }); async function stopWorker() { @@ -309,6 +310,191 @@ describeEmbeddedPostgres("tool gateway service", () => { expect(await db.select().from(toolCallEvents).where(eq(toolCallEvents.reasonCode, "approved_action_executed"))).toEqual([]); } + describe("approved review recovery worker", () => { + async function startRecovery() { + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => !draining, + canReconcile: () => true, onError: error => errors.push(error) }); + await registerToolReviewRecoveryWork(coordinator, gateway, () => !draining).ready; + } + const approve = (pending: Awaited>, source = db) => commitToolActionReview(source, { + companyId: fixture.company.id, actionRequestId: pending.id, decision: "approved", actor: { userId: "board" }, + }); + async function executed(pending: Awaited>) { + expect((await db.select().from(toolActionRequests).where(eq(toolActionRequests.id, pending.id)))[0].status).toBe("executed"); + expect((await db.select().from(toolInvocations).where(eq(toolInvocations.id, pending.invocationId)))[0].status).toBe("succeeded"); + } + + it("is quiet while configured idle and executes a committed approval once", async () => { + await startRecovery(); await quiet(); + const pending = await request(); await approve(pending); await settle(); + await executed(pending); await quiet(); + expect(await db.select().from(toolCallEvents).where(eq(toolCallEvents.reasonCode, "approved_action_executed"))).toHaveLength(1); + }); + + it("reconstructs a saved approved action after restart", async () => { + const pending = await request(); await approve(pending); + await startRecovery(); await settle(); await executed(pending); await quiet(); + }); + + it("retries a failed approval dispatch only while the approved action remains unresolved", async () => { + const pending = await request(); await approve(pending); + const recover = vi.spyOn(gateway, "approveActionRequest").mockRejectedValue(new Error("Recovery dependency unavailable")); + await startRecovery(); + expect(recover).toHaveBeenCalledTimes(1); expect(coordinator!.nextWakeAt()).not.toBeNull(); + await settle(10_000); expect(recover).toHaveBeenCalledTimes(1); + recover.mockRestore(); await settle(); await executed(pending); await quiet(); + }); + + it("keeps deliberately inert legacy approvals quiet", async () => { + const pending = await request(); + await db.update(toolActionRequests).set({ signedArguments: signToolArguments({ + invocationId: pending.invocationId, toolName: "mcp-remote-fixture:update_note", + canonicalArguments: canonicalToolArguments({ noteId: "legacy", body: "inert" }), + signingSecret: testToolActionSigningSecret, + }) }).where(eq(toolActionRequests.id, pending.id)); + await approve(pending); await startRecovery(); await quiet(); + expect((await db.select().from(toolActionRequests))[0].status).toBe("approved"); + expect((await db.select().from(toolInvocations))[0].status).toBe("awaiting_approval"); + }); + + it.each([false, true])("waits for the outer approval transaction (rollback %s)", async rollback => { + await startRecovery(); const pending = await request(); + let release!: () => void, staged!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const ready = new Promise(resolve => { staged = resolve; }); + const aborted = new Error("approval rollback fixture"); + const writing = db.transaction(async tx => { + await approve(pending, tx as unknown as typeof db); staged(); await gate; + if (rollback) throw aborted; + }).catch(error => { if (error !== aborted) throw error; }); + await Promise.race([ready, writing]); + try { + await vi.advanceTimersByTimeAsync(60_000); + expect((await db.select().from(toolActionRequests))[0].status).toBe("pending"); + } finally { release(); } + await writing; await settle(); + if (rollback) expect((await db.select().from(toolActionRequests))[0].status).toBe("pending"); + else await executed(pending); + await quiet(); + }); + + it("retains the first approval during drain without passive queries", async () => { + await startRecovery(); await quiet(); draining = true; + const pending = await request(); await approve(pending); await settle(); + const reads = vi.spyOn(db, "select"); + await settle(60 * 60_000); expect(reads).not.toHaveBeenCalled(); reads.mockRestore(); + draining = false; await settle(); await executed(pending); await quiet(); + }); + + async function testRequest(provider: NonNullable) { + await db.insert(companyMemberships).values({ companyId: fixture.company.id, principalType: "user", + principalId: "board", status: "active", membershipRole: "owner" }); + const remote = await createRemoteMcpToolFixture(db, fixture.company.id); + await db.update(toolCatalogEntries).set({ isReadOnly: false, riskLevel: "write" }) + .where(eq(toolCatalogEntries.id, remote.catalogEntry.id)); + await db.insert(toolPolicies).values({ companyId: fixture.company.id, name: "Review connected writes", + policyType: "require_approval", priority: 10, selectors: { connectionId: remote.connection.id } }); + gateway = createTestToolGatewayService(db, { remoteHttpRequest: provider }); + const result = await gateway.executeTestCall({ companyId: fixture.company.id, connectionId: remote.connection.id, + agentId: fixture.agent.id, userId: "board", toolName: "needs_input", parameters: { value: "reviewed" } }); + expect(result.decision).toBe("ask_first"); + const [pending] = await db.select().from(toolActionRequests); + return { pending, remote }; + } + function providerFixture() { + let release!: () => void, entered!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + const started = new Promise(resolve => { entered = resolve; }); + const provider = vi.fn(async (_url, init) => { + const payload = JSON.parse(String(init.body)); + expect(payload.method).toBe("tools/call"); + entered(); await gate; + return Response.json({ jsonrpc: "2.0", id: payload.id, + result: { content: [{ type: "text", text: "Saved once" }] } }); + }); + return { provider, release, started }; + } + + it.each(["foreground", "recovery"] as const)("claims a Test-tab action once when %s wins", async winner => { + const remoteCall = providerFixture(); + const { pending } = await testRequest(remoteCall.provider); + let running: Promise; + if (winner === "foreground") { + await startRecovery(); + running = gateway.approveActionRequest({ companyId: fixture.company.id, actionRequestId: pending.id, actor: { userId: "board" } }); + } else { + await approve(pending); + running = startRecovery(); + } + await Promise.race([remoteCall.started, running]); + try { + await vi.advanceTimersByTimeAsync(60_000); + await gateway.approveActionRequest({ companyId: fixture.company.id, actionRequestId: pending.id, actor: { userId: "board" } }); + expect(remoteCall.provider).toHaveBeenCalledTimes(1); + } finally { remoteCall.release(); } + await running; await settle(); await executed(pending); await quiet(); + expect(remoteCall.provider).toHaveBeenCalledTimes(1); + }); + + it("repairs completed Test-tab history without sending the request again", async () => { + const remoteCall = providerFixture(); remoteCall.release(); + const { pending } = await testRequest(remoteCall.provider); + await gateway.approveActionRequest({ companyId: fixture.company.id, actionRequestId: pending.id, actor: { userId: "board" } }); + await executed(pending); + await db.update(toolActionRequests).set({ status: "approved", signedArguments: "unreadable-old-signature" }) + .where(eq(toolActionRequests.id, pending.id)); + remoteCall.provider.mockClear(); + await startRecovery(); await settle(); await executed(pending); await quiet(); + expect(remoteCall.provider).not.toHaveBeenCalled(); + }); + + it("does not replay an interrupted legacy Test-tab invocation", async () => { + const remoteCall = providerFixture(); remoteCall.release(); + const { pending } = await testRequest(remoteCall.provider); + await approve(pending); + await db.update(toolInvocations).set({ status: "executing", startedAt: new Date(Date.now() - 11 * 60_000) }) + .where(eq(toolInvocations.id, pending.invocationId)); + await startRecovery(); await settle(); + expect((await db.select().from(toolActionRequests))[0].status).toBe("executing"); + await gateway.sweepActionReviews({ includeApproved: false, includeExpiry: false }); + expect((await db.select().from(toolInvocations))[0]).toMatchObject({ status: "failed", errorCode: "tool_execution_outcome_unknown" }); + await quiet(); expect(remoteCall.provider).not.toHaveBeenCalled(); + }); + + it("does not replay a Test-tab call when persisting its external result fails", async () => { + const remoteCall = providerFixture(); remoteCall.release(); + const { pending } = await testRequest(remoteCall.provider); + await approve(pending); + await db.execute(sql`create function sleep_review_fail() returns trigger language plpgsql as $$ + begin raise exception 'approval result fixture failure'; end $$`); + await db.execute(sql`create trigger sleep_review_fail before update on tool_action_requests + for each row when (new.status in ('executed', 'failed')) execute function sleep_review_fail()`); + try { + await startRecovery(); + expect((await db.select().from(toolActionRequests))[0].status).toBe("executing"); + expect((await db.select().from(toolInvocations))[0].status).toBe("executing"); + expect(remoteCall.provider).toHaveBeenCalledTimes(1); + } finally { + await db.execute(sql`drop trigger sleep_review_fail on tool_action_requests`); + } + await settle(); + await vi.advanceTimersByTimeAsync(11 * 60_000); + await gateway.sweepActionReviews({ includeApproved: false, includeExpiry: false }); + expect((await db.select().from(toolInvocations))[0]).toMatchObject({ status: "failed", errorCode: "tool_execution_outcome_unknown" }); + await quiet(); expect(remoteCall.provider).toHaveBeenCalledTimes(1); + }); + + it("revalidates the reviewed Test-tab target before recovery dispatch", async () => { + const remoteCall = providerFixture(); remoteCall.release(); + const { pending, remote } = await testRequest(remoteCall.provider); + await approve(pending); + await db.update(toolCatalogEntries).set({ versionHash: "changed-after-review" }).where(eq(toolCatalogEntries.id, remote.catalogEntry.id)); + await startRecovery(); await settle(); + expect((await db.select().from(toolInvocations))[0]).toMatchObject({ status: "failed", errorCode: "approved_tool_target_changed" }); + await quiet(); expect(remoteCall.provider).not.toHaveBeenCalled(); + }); + }); + it("is quiet when configured idle, wakes on signing, expires the review and card, then stays quiet with history", async () => { await start(); await quiet(); const pending = await request(); @@ -726,13 +912,13 @@ describeEmbeddedPostgres("tool gateway service", () => { await expect(gateway.executeTool({ sessionToken: session.token, tool: "mcp-remote-fixture:update_note", parameters: { noteId: "expire" } })).rejects.toMatchObject({ reasonCode: "approval_required" }); const [request] = await db.select().from(toolActionRequests); await db.update(toolActionRequests).set({ expiresAt: new Date(Date.now() - 1000) }).where(eq(toolActionRequests.id, request.id)); - await db.insert(toolActionRequests).values(Array.from({ length: 100 }, (_, i) => ({ + await db.insert(toolActionRequests).values(Array.from({ length: 101 }, (_, i) => ({ ...request, id: `00000000-0000-4000-8000-${String(i).padStart(12, "0")}`, interactionId: null, status: "approved" as const, }))); const recover = vi.spyOn(gateway, "approveActionRequest").mockRejectedValue(new Error("Unavailable approval dependency")); - expect(await gateway.sweepActionReviews()).toEqual({ scanned: 101 }); - expect(recover).toHaveBeenCalledTimes(100); + expect(await gateway.sweepActionReviews()).toEqual({ scanned: 102 }); + expect(recover).toHaveBeenCalledTimes(101); const [settled] = await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, request.interactionId!)); expect(settled.status).toBe("expired"); }); diff --git a/server/src/index.ts b/server/src/index.ts index d3515c7a84e..163eb8e5cbc 100644 --- a/server/src/index.ts +++ b/server/src/index.ts @@ -1,3 +1,4 @@ +import { registerToolReviewRecoveryWork } from "./services/tool-review-recovery-work.js"; import { registerToolReviewExpiryWork } from "./services/tool-review-expiry-work.js"; import { registerGatewayCredentialExpiryWork } from "./services/gateway-credential-expiry-work.js"; import { registerSecretProposalExpiryWork } from "./services/secret-proposal-expiry-work.js"; @@ -1444,7 +1445,13 @@ async function startServerWithDatabaseTeardown( } else { await app.locals.toolGateway.sweepExpiredActionReviews(); } - return app.locals.toolGateway.sweepActionReviews({ includeExpiry: false }); + if (config.heartbeatSchedulerEnabled) { + await registerToolReviewRecoveryWork(deliveryWork, app.locals.toolGateway, + () => !isIdleTaskDrainActive() && !heartbeatSchedulerStopped && !isWarmStandby()).ready; + } else { + await app.locals.toolGateway.recoverApprovedActionReviews(); + } + return app.locals.toolGateway.sweepActionReviews({ includeExpiry: false, includeApproved: false }); }).catch((err: unknown) => logger.error({ err }, "startup tool review recovery failed")); await timeStartupRecovery("gateway_sessions", () => config.heartbeatSchedulerEnabled ? registerGatewayCredentialExpiryWork(deliveryWork, app.locals.toolGateway, @@ -1869,7 +1876,7 @@ async function startServerWithDatabaseTeardown( logger.error({ err }, "periodic tool connection health sweep failed"); })); - trackHeartbeatSchedulerWork(app.locals.toolGateway.sweepActionReviews({ includeExpiry: false }).catch((err: unknown) => logger.error({ err }, "tool review recovery failed"))); + trackHeartbeatSchedulerWork(app.locals.toolGateway.sweepActionReviews({ includeExpiry: false, includeApproved: false }).catch((err: unknown) => logger.error({ err }, "tool review recovery failed"))); if (isIdleTaskDrainActive() || heartbeatSchedulerStopped) return; if (!(await heartbeat.resolveSchedulingSuppression()).suppressed) { diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 22f272a0400..d77c801ca5b 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", + toolReviewRecovery: "tool-review-recovery", toolReviewExpiry: "tool-review-expiry", gatewayCredentialExpiry: "gateway-credential-expiry", secretProposalExpiry: "secret-proposal-expiry", diff --git a/server/src/services/tool-gateway.ts b/server/src/services/tool-gateway.ts index f43f2e950d6..4846e53bdf1 100644 --- a/server/src/services/tool-gateway.ts +++ b/server/src/services/tool-gateway.ts @@ -1,3 +1,5 @@ +import { claimToolTestApproval, persistToolTestOutcome, startToolTestInvocation, type ToolTestApprovalClaim } from "./tool-test-approval.js"; +import { recoverApprovedToolReviews } from "./tool-review-recovery.js"; import { toolReviewExpiryService } from "./tool-review-expiry.js"; import { writeToolReviewChange } from "./tool-review-work-notifications.js"; import { restConnectorFor } from "./connectors/rest-registry.js"; @@ -7416,6 +7418,7 @@ export function createToolGatewayService( matchedPolicyIds: string[]; timeoutMs?: number; signal?: AbortSignal; + approvedClaim?: ToolTestApprovalClaim; }): Promise< | { decision: "allowed"; invocationId: string; result: unknown } | { @@ -7424,14 +7427,9 @@ export function createToolGatewayService( error: { message: string; reasonCode: string }; } > { - await db - .update(toolInvocations) - .set({ - status: "executing", - startedAt: new Date(), - updatedAt: new Date(), - }) - .where(eq(toolInvocations.id, args.invocationId)); + if (!(await startToolTestInvocation(db, args.invocationId, args.approvedClaim))) { + throw new ToolGatewayHttpError(409, "Tool action execution claim is no longer current", "action_already_consumed"); + } await writeAudit({ session: args.session, companyId: args.companyId, @@ -7498,17 +7496,9 @@ export function createToolGatewayService( throw new ToolGatewayHttpError(502, String(providerResult.content || providerResult.error), "tool_error", { execution: connectedMcpExecution.execution }); } - await db - .update(toolInvocations) - .set({ - status: "succeeded", - resultHash: resultValidation.summary.sha256 ?? null, - resultSummary: resultValidation.summary, - resultSizeBytes: resultValidation.summary.sizeBytes ?? null, - completedAt: new Date(), - updatedAt: new Date(), - }) - .where(eq(toolInvocations.id, args.invocationId)); + await persistToolTestOutcome(db, args.invocationId, { + status: "succeeded", result: resultValidation.summary, + }, args.approvedClaim); await writeToolCallEvent({ invocationId: args.invocationId, session: args.session, @@ -7552,9 +7542,6 @@ export function createToolGatewayService( execution: connectedMcpExecution.execution, }, }); - // After the bookkeeping above: a throw there is caught below, overwrites - // the row to failed, and emits — an earlier success emit would make one - // execution report both outcomes. void emitConnectionInvoked(db, args.invocationId); return { decision: "allowed" as const, @@ -7570,21 +7557,10 @@ export function createToolGatewayService( ? err.reasonCode : "tool_execution_failed"; const message = err instanceof Error ? err.message : String(err); - await db - .update(toolInvocations) - .set({ - status: - status === 504 - ? "timed_out" - : status === 429 - ? "rate_limited" - : "failed", - errorCode: reasonCode, - errorMessage: message, - completedAt: new Date(), - updatedAt: new Date(), - }) - .where(eq(toolInvocations.id, args.invocationId)); + await persistToolTestOutcome(db, args.invocationId, { + status: status === 504 ? "timed_out" : status === 429 ? "rate_limited" : "failed", + errorCode: reasonCode, errorMessage: message, + }, args.approvedClaim); void emitConnectionInvoked(db, args.invocationId); await writeToolCallEvent({ invocationId: args.invocationId, @@ -7655,103 +7631,46 @@ export function createToolGatewayService( */ async function runApprovedTestInvocation( invocation: typeof toolInvocations.$inferSelect, - parameters: unknown, + signedPayload: NonNullable>, actionRequestId: string, ): Promise { - const agentId = invocation.agentId; - if (!invocation.connectionId || !agentId) return; - const userId = invocation.actorId ?? "board"; - const session: ToolGatewaySession = { - id: "test-call", - token: "test-call", - companyId: invocation.companyId, - agentId, - runId: null, - issueId: null, - projectId: null, - actorType: "user", - actorId: userId, - responsibleUserId: userId, - createdAt: new Date(), - expiresAt: new Date(Date.now() + DEFAULT_SESSION_TTL_MS), - }; - let tool: ToolGatewayDescriptor | undefined; + const claim = await claimToolTestApproval(db, actionRequestId, invocation.id); + if (!claim) return; try { - tool = ( - await connectedMcpToolsForConnection( - invocation.companyId, - invocation.connectionId, - ) - ).find( - (candidate) => - candidate.name === invocation.toolName || - candidate.upstreamToolName === invocation.toolName, - ); - } catch { - tool = undefined; - } - if (!tool) { - await db - .update(toolInvocations) - .set({ - status: "failed", - errorCode: "tool_not_found", - errorMessage: `Tool "${invocation.toolName}" is no longer connected`, - completedAt: new Date(), - updatedAt: new Date(), - }) - .where(eq(toolInvocations.id, invocation.id)); - void emitConnectionInvoked(db, invocation.id); - await reflectToolActionInteractionLifecycle({ - actionRequestId, - status: "failed", - errorCode: "tool_not_found", - errorMessage: `Tool "${invocation.toolName}" is no longer connected`, - }); - return; - } - const argumentsSummary = (await validateToolContent({ - value: parameters, - direction: "arguments", - sensitiveMode: "redact", - promptInjectionMode: "ignore", - })).summary; - try { - await runTestToolInvocation({ - session, - tool, - parameters, - invocationId: invocation.id, - companyId: invocation.companyId, - agentId, - userId, - argumentsSummary, - reasonCode: "approval_granted", - matchedPolicyIds: invocation.matchedPolicyIds ?? [], - }); - await reflectToolActionInteractionLifecycle({ - actionRequestId, - status: "executed", - }); + const agentId = invocation.agentId; + if (!invocation.connectionId || !agentId) { + throw new ToolGatewayHttpError(409, "Test action context is no longer available", "action_origin_invalid"); + } + const userId = invocation.actorId ?? "board"; + const session: ToolGatewaySession = { + id: "test-call", token: "test-call", companyId: invocation.companyId, agentId, + runId: null, issueId: null, projectId: null, actorType: "user", actorId: userId, + responsibleUserId: userId, createdAt: new Date(), expiresAt: new Date(Date.now() + DEFAULT_SESSION_TTL_MS), + }; + const tool = (await connectedMcpToolsForConnection(invocation.companyId, invocation.connectionId)) + .find(candidate => candidate.name === invocation.toolName || candidate.upstreamToolName === invocation.toolName); + if (!tool) throw new ToolGatewayHttpError(404, `Tool "${invocation.toolName}" is no longer connected`, "tool_not_found"); + if (!approvalSnapshotsMatch(signedPayload.approvalSnapshot, await connectedRemoteApprovalSnapshot(session, tool))) { + throw new ToolGatewayHttpError(409, "Tool definition or connection changed; request a new review", "approved_tool_target_changed"); + } + if (!(await approvedManagedArgumentsRemainCurrent(session, tool, signedPayload.arguments))) { + throw new ToolGatewayHttpError(409, "Approved tool action managed arguments changed after review", "approved_tool_managed_arguments_changed"); + } + const access = await policyService.decide(policyInputForTool({ session, tool, parameters: signedPayload.arguments })); + if (!access.allowed && access.decision !== "require_approval") { + throw new ToolGatewayHttpError(403, access.explanation, access.reasonCode); + } + const argumentsSummary = (await validateToolContent({ value: signedPayload.arguments, + direction: "arguments", sensitiveMode: "redact", promptInjectionMode: "ignore" })).summary; + await runTestToolInvocation({ session, tool, parameters: signedPayload.arguments, invocationId: invocation.id, + companyId: invocation.companyId, agentId, userId, argumentsSummary, reasonCode: "approval_granted", + approvedClaim: claim, matchedPolicyIds: invocation.matchedPolicyIds ?? [] }); } catch (err) { const message = err instanceof Error ? err.message : String(err); - await db - .update(toolInvocations) - .set({ - status: "failed", - errorCode: "tool_execution_failed", - errorMessage: message, - completedAt: new Date(), - updatedAt: new Date(), - }) - .where(eq(toolInvocations.id, invocation.id)); - void emitConnectionInvoked(db, invocation.id); - await reflectToolActionInteractionLifecycle({ - actionRequestId, - status: "failed", - errorCode: "tool_execution_failed", - errorMessage: message, - }); + const errorCode = err instanceof ToolGatewayHttpError || err instanceof ToolContentValidationError + ? err.reasonCode : "tool_execution_failed"; + const settled = await persistToolTestOutcome(db, invocation.id, { status: "failed", errorCode, errorMessage: message }, claim); + if (settled) void emitConnectionInvoked(db, invocation.id); } } @@ -7939,7 +7858,7 @@ export function createToolGatewayService( ); const now = new Date(); await options.beforeManagedArgumentDriftExpiry?.(); - const [expired] = await db + const [expired] = await writeToolReviewChange(db, tx => tx .update(toolActionRequests) .set({ status: "expired", resolvedAt: now, updatedAt: now }) .where( @@ -7950,7 +7869,7 @@ export function createToolGatewayService( : eq(toolActionRequests.status, "approved"), ), ) - .returning({ id: toolActionRequests.id }); + .returning({ id: toolActionRequests.id })); if (!expired) return error; await db .update(toolInvocations) @@ -8059,7 +7978,7 @@ export function createToolGatewayService( ); } - const [claimed] = await db + const [claimed] = await writeToolReviewChange(db, tx => tx .update(toolActionRequests) .set({ status: "executing", updatedAt: new Date() }) .where( @@ -8068,7 +7987,7 @@ export function createToolGatewayService( eq(toolActionRequests.status, "approved"), ), ) - .returning(); + .returning()); if (!claimed) { const settled = await waitForActionRequestExecution(actionRequest.id); const [settledInvocation] = await db @@ -9730,7 +9649,31 @@ export function createToolGatewayService( sweepExpiredActionReviews: reviewExpiry.sweep, nextActionReviewExpiryAt: reviewExpiry.next, - async sweepActionReviews(input: { includeExpiry?: boolean } = {}) { + async recoverApprovedActionReviews() { + return recoverApprovedToolReviews(db, async ({ request, invocation }) => { + try { + if (isTestOriginInvocation(invocation) && invocation.status !== "awaiting_approval") { + await claimToolTestApproval(db, request.id, invocation.id); + const [remaining] = await db.select({ status: toolActionRequests.status }).from(toolActionRequests) + .where(eq(toolActionRequests.id, request.id)); + return remaining?.status === "approved"; + } + const payload = readSignedToolArgumentsPayload({ signedArguments: request.signedArguments, + invocationId: invocation.id, toolName: invocation.toolName, signingSecret: options.toolActionSigningSecret }); + if (payload && !isTestOriginInvocation(invocation) && payload.executionOnApprove !== true) return false; + await this.approveActionRequest({ companyId: request.companyId, actionRequestId: request.id, + actor: { userId: request.decidedByUserId ?? request.resolvedByUserId } }); + const [remaining] = await db.select({ status: toolActionRequests.status }).from(toolActionRequests) + .where(eq(toolActionRequests.id, request.id)); + return remaining?.status === "approved"; + } catch (err) { + logger.warn({ err, actionRequestId: request.id }, "Could not recover approved tool action"); + return true; + } + }); + }, + + async sweepActionReviews(input: { includeExpiry?: boolean; includeApproved?: 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 @@ -9769,38 +9712,21 @@ export function createToolGatewayService( const staleAt = new Date(now.getTime() - 10 * 60_000); let cursor: string | undefined; let scanned = input.includeExpiry === false ? 0 : (await reviewExpiry.sweep(now)).scanned; + if (input.includeApproved !== false) scanned += (await this.recoverApprovedActionReviews()).scanned; for (;;) { const rows = await db .select() .from(toolActionRequests) .where( and( - or( - eq(toolActionRequests.status, "approved"), - and( - eq(toolActionRequests.status, "executing"), - lte(toolActionRequests.updatedAt, staleAt), - ), - ), + eq(toolActionRequests.status, "executing"), + lte(toolActionRequests.updatedAt, staleAt), cursor ? gt(toolActionRequests.id, cursor) : undefined, ), ) .orderBy(asc(toolActionRequests.id)) .limit(100); for (const row of rows) { - if (row.status === "approved") { - await this.approveActionRequest({ - companyId: row.companyId, - actionRequestId: row.id, - actor: { userId: row.decidedByUserId ?? row.resolvedByUserId }, - }).catch((error) => - logger.warn( - { err: error, actionRequestId: row.id }, - "Could not recover approved tool action", - ), - ); - continue; - } const status = "failed"; const errorCode = "tool_execution_outcome_unknown"; const errorMessage = "Execution was interrupted; the external outcome is unknown. Inspect the provider before retrying."; @@ -9992,10 +9918,12 @@ export function createToolGatewayService( actionRequestId: actionRequest.id, status: "approved", }); - if ( - !isTestOriginInvocation(invocation) && - signedPayload.executionOnApprove === true - ) { + if (isTestOriginInvocation(invocation)) { + await runApprovedTestInvocation(invocation, signedPayload, actionRequest.id); + const [settled] = await db.select().from(toolActionRequests).where(eq(toolActionRequests.id, actionRequest.id)); + return actionRequestResolution(settled ?? actionRequest); + } + if (signedPayload.executionOnApprove === true) { try { await executeApprovedAgentInvocation({ actionRequest, invocation }); } catch { @@ -10093,7 +10021,7 @@ export function createToolGatewayService( if (isTestOriginInvocation(invocation)) { await runApprovedTestInvocation( { ...invocation, approvalState: "approved" }, - signedPayload.arguments, + signedPayload, updated.id, ); } else if (signedPayload.executionOnApprove === true) { @@ -10656,7 +10584,7 @@ export function createToolGatewayService( ); await options.beforeLegacyApprovedActionClaim?.(); const claimedAt = new Date(); - const [claimed] = await db + const [claimed] = await writeToolReviewChange(db, tx => tx .update(toolActionRequests) .set({ status: "executing", @@ -10669,7 +10597,7 @@ export function createToolGatewayService( eq(toolActionRequests.status, "approved"), ), ) - .returning(); + .returning()); if (!claimed) { throw new ToolGatewayHttpError( 409, @@ -10755,7 +10683,7 @@ export function createToolGatewayService( }); } const claimedAt = new Date(); - const [claimed] = await db + const [claimed] = await writeToolReviewChange(db, tx => tx .update(toolActionRequests) .set({ status: "executing", @@ -10768,7 +10696,7 @@ export function createToolGatewayService( eq(toolActionRequests.status, "approved"), ), ) - .returning(); + .returning()); if (!claimed) { throw new ToolGatewayHttpError( 409, diff --git a/server/src/services/tool-review-recovery-work.ts b/server/src/services/tool-review-recovery-work.ts new file mode 100644 index 00000000000..1322b5f50a0 --- /dev/null +++ b/server/src/services/tool-review-recovery-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 { createToolGatewayService } from "./tool-gateway.js"; + +export function registerToolReviewRecoveryWork( + coordinator: ReturnType, + gateway: Pick, "recoverApprovedActionReviews">, + canRun: () => boolean, +) { + let next: number | null = null; + return registerDeadlineWork(coordinator, DELIVERY_QUEUES.toolReviewRecovery, { + retryMs: 30_000, canRun, + run: async () => { + const result = await gateway.recoverApprovedActionReviews(); + next = result.retry ? Date.now() + 30_000 : null; + }, + next: async () => next, + }); +} diff --git a/server/src/services/tool-review-recovery.ts b/server/src/services/tool-review-recovery.ts new file mode 100644 index 00000000000..334f0538177 --- /dev/null +++ b/server/src/services/tool-review-recovery.ts @@ -0,0 +1,21 @@ +import { and, asc, eq, gt } from "drizzle-orm"; +import { toolActionRequests, toolInvocations, type Db } from "@paperclipai/db"; + +export async function recoverApprovedToolReviews(db: Db, recover: (input: { + request: typeof toolActionRequests.$inferSelect; + invocation: typeof toolInvocations.$inferSelect; +}) => Promise) { + let cursor: string | undefined; + let scanned = 0, retry = false; + for (;;) { + const rows = await db.select({ request: toolActionRequests, invocation: toolInvocations }) + .from(toolActionRequests).innerJoin(toolInvocations, eq(toolInvocations.id, toolActionRequests.invocationId)) + .where(and(eq(toolActionRequests.status, "approved"), cursor ? gt(toolActionRequests.id, cursor) : undefined)) + .orderBy(asc(toolActionRequests.id)).limit(100); + for (const row of rows) retry = await recover(row) || retry; + scanned += rows.length; + if (rows.length < 100) break; + cursor = rows[rows.length - 1].request.id; + } + return { scanned, retry }; +} diff --git a/server/src/services/tool-review-work-notifications.ts b/server/src/services/tool-review-work-notifications.ts index 6c13b75c7c6..dd36fbb30af 100644 --- a/server/src/services/tool-review-work-notifications.ts +++ b/server/src/services/tool-review-work-notifications.ts @@ -1,8 +1,9 @@ import type { Db } from "@paperclipai/db"; import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; -export function notifyToolReviewChange(transaction: object) { - return notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewExpiry); +export async function notifyToolReviewChange(transaction: object) { + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewExpiry); + await notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewRecovery); } export function writeToolReviewChange(db: Db, write: (transaction: Db) => Promise): Promise { diff --git a/server/src/services/tool-test-approval.ts b/server/src/services/tool-test-approval.ts new file mode 100644 index 00000000000..20e4b6f6524 --- /dev/null +++ b/server/src/services/tool-test-approval.ts @@ -0,0 +1,80 @@ +import { and, eq } from "drizzle-orm"; +import { toolActionRequests, toolInvocations, type Db } from "@paperclipai/db"; +import type { ToolRedactedValueSummary } from "@paperclipai/shared"; +import { writeToolReviewChange } from "./tool-review-work-notifications.js"; + +export type ToolTestApprovalClaim = { actionRequestId: string; invocationId: string; claimedAt: Date }; +export type ToolTestOutcome = { + status: "succeeded" | "failed" | "timed_out" | "rate_limited"; + result?: ToolRedactedValueSummary; + errorCode?: string; + errorMessage?: string; +}; + +export async function claimToolTestApproval(db: Db, actionRequestId: string, invocationId: string): Promise { + return writeToolReviewChange(db, async tx => { + const [invocation] = await tx.select().from(toolInvocations) + .where(eq(toolInvocations.id, invocationId)).for("update"); + const [request] = await tx.select().from(toolActionRequests) + .where(and(eq(toolActionRequests.id, actionRequestId), eq(toolActionRequests.invocationId, invocationId))).for("update"); + if (!invocation || request?.status !== "approved") return null; + const now = new Date(); + if (invocation.status !== "awaiting_approval") { + if (["succeeded", "failed", "denied", "timed_out", "rate_limited"].includes(invocation.status)) { + await tx.update(toolActionRequests).set({ status: invocation.status === "succeeded" ? "executed" : "failed", + resolvedAt: invocation.completedAt ?? now, updatedAt: now }).where(eq(toolActionRequests.id, actionRequestId)); + } else if (invocation.status === "executing") { + await tx.update(toolActionRequests).set({ status: "executing", updatedAt: invocation.startedAt ?? invocation.updatedAt }) + .where(eq(toolActionRequests.id, actionRequestId)); + } + return null; + } + await tx.update(toolActionRequests).set({ status: "executing", updatedAt: now }) + .where(eq(toolActionRequests.id, actionRequestId)); + return { actionRequestId, invocationId, claimedAt: now }; + }); +} + +export async function startToolTestInvocation(db: Db, invocationId: string, claim?: ToolTestApprovalClaim) { + const now = new Date(); + if (!claim) { + await db.update(toolInvocations).set({ status: "executing", startedAt: now, updatedAt: now }) + .where(eq(toolInvocations.id, invocationId)); + return true; + } + return db.transaction(async tx => { + const [invocation] = await tx.select({ status: toolInvocations.status }).from(toolInvocations) + .where(eq(toolInvocations.id, invocationId)).for("update"); + const [request] = await tx.select().from(toolActionRequests) + .where(eq(toolActionRequests.id, claim.actionRequestId)).for("update"); + if (invocation?.status !== "awaiting_approval" || request?.status !== "executing" || + request.updatedAt.getTime() !== claim.claimedAt.getTime()) return false; + await tx.update(toolInvocations).set({ status: "executing", startedAt: now, updatedAt: now }) + .where(eq(toolInvocations.id, invocationId)); + return true; + }); +} + +export async function persistToolTestOutcome(db: Db, invocationId: string, outcome: ToolTestOutcome, claim?: ToolTestApprovalClaim) { + const now = new Date(); + const values = { status: outcome.status, ...(outcome.result ? { + resultHash: outcome.result.sha256 ?? null, resultSummary: outcome.result, + resultSizeBytes: outcome.result.sizeBytes ?? null, + } : {}), errorCode: outcome.errorCode, errorMessage: outcome.errorMessage, completedAt: now, updatedAt: now }; + if (!claim) { + await db.update(toolInvocations).set(values).where(eq(toolInvocations.id, invocationId)); + return true; + } + return writeToolReviewChange(db, async tx => { + const [invocation] = await tx.select({ status: toolInvocations.status }).from(toolInvocations) + .where(eq(toolInvocations.id, invocationId)).for("update"); + const [request] = await tx.select().from(toolActionRequests) + .where(eq(toolActionRequests.id, claim.actionRequestId)).for("update"); + if (!invocation || !["awaiting_approval", "executing"].includes(invocation.status) || + request?.status !== "executing" || request.updatedAt.getTime() !== claim.claimedAt.getTime()) return false; + await tx.update(toolInvocations).set(values).where(eq(toolInvocations.id, invocationId)); + await tx.update(toolActionRequests).set({ status: outcome.status === "succeeded" ? "executed" : "failed", + resolvedAt: now, updatedAt: now }).where(eq(toolActionRequests.id, claim.actionRequestId)); + return true; + }); +}