fix(server): recover approved tool actions from committed changes

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-24 10:38:39 -07:00
1 parent 38c4e28135
commit 5dcd193d35
16 files changed
+582 -175

No files matched your search

+82
View File
@@ -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.
@@ -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.
+1
View File
@@ -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(
@@ -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';
@@ -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": [
+7
View File
@@ -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
}
]
}
+1
View File
@@ -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),
],
@@ -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),
@@ -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<R
describeEmbeddedPostgres("tool gateway service", () => {
let db!: ReturnType<typeof createDb>;
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
let stopExpiryWorker: (() => Promise<void>) | undefined;
let stopReviewWorker: (() => Promise<void>) | 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<ReturnType<typeof request>>, source = db) => commitToolActionReview(source, {
companyId: fixture.company.id, actionRequestId: pending.id, decision: "approved", actor: { userId: "board" },
});
async function executed(pending: Awaited<ReturnType<typeof request>>) {
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<void>(resolve => { release = resolve; });
const ready = new Promise<void>(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<ToolGatewayServiceOptions["remoteHttpRequest"]>) {
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<void>(resolve => { release = resolve; });
const started = new Promise<void>(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<unknown>;
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");
});
+9 -2
View File
@@ -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) {
@@ -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",
+91 -163
View File
@@ -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<ReturnType<typeof readSignedToolArgumentsPayload>>,
actionRequestId: string,
): Promise<void> {
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,
@@ -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<typeof createDeliveryWorkCoordinator>,
gateway: Pick<ReturnType<typeof createToolGatewayService>, "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,
});
}
@@ -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<boolean>) {
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 };
}
@@ -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<T>(db: Db, write: (transaction: Db) => Promise<T>): Promise<T> {
+80
View File
@@ -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<ToolTestApprovalClaim | null> {
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;
});
}