mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-28 20:21:50 +01:00
fix(server): recover approved tool actions from committed changes (#17454)
## Thinking Path > - Paperclip manages agent work and its governed tool calls. > - A human approval can persist before the approving process dispatches the tool. > - Recovery currently scans approved requests on every maintenance tick. > - Completed and deliberately inert approvals do not owe another execution. > - Test-tab approval also needs a shared execution claim before recovery can safely race the foreground path. > - This PR wakes recovery from committed changes and retries only unresolved approvals. > - Idle history becomes quiet while existing approval and execution checks remain with the gateway. ## Linked Issues or Issue Description Refs #17453. This is the fifth slice of the recurring-server sleep migration, stacked on `codex/sleep-04-tool-review-expiry`. **Problem** Approved tool requests are scanned periodically even when no executable work remains. The Test-tab foreground path lacks the request claim needed to admit concurrent recovery safely. **Proposed Solution** Use the existing transaction-aware review notification boundary and shared deadline worker. Keep signed-payload interpretation, current policy and target checks, and provider dispatch in the gateway. Share a claim between Test-tab foreground and recovery, then persist both authoritative outcome rows atomically. **Alternatives Considered** A longer interval still scans idle history. A new durable queue would duplicate existing request/invocation state. Dispatching Test-tab approvals without a claim risks duplicate external calls. ## What Changed - Recheck both principals for delegated Test-tab approvals. Save their rate receipts in the same claim transaction, share only the same bucket, and roll back new charges on denial. - Preserve delegated gateway authorization and both issuer/target rate receipts inside the existing claimed-invocation transaction; retain the durable actor during approved replay. - Expire stale managed connector payloads before consuming approved-tool admission. - Persist current rate-policy charges and their invocation receipts atomically before dispatch, preserving exact approval claim fencing and read-only preflight. - Recover approved requests at startup and after committed review changes; retry only unresolved approvals. - Leave legacy agent approvals without execute-on-approve intent inert. Repair completed Test-tab history without another provider call. - Fence Test-tab execution and outcome settlement with an atomic claim shared by both callers. Revalidate the approved target and current policy before dispatch. - Keep provider calls outside database transactions. Retain quarantine for interrupted execution and unknown outcomes. - Add a concurrent, idempotent partial index for approved-request pagination. Reuse the parent's company-scoped review signals and shared scoped deadline worker. - Remove approved scanning from the aggregate tick. Preserve the public combined sweep and scheduler-disabled startup behavior. - Document producer ownership, quiet states, lifecycle behavior, and remaining recovery duties. ## Verification Latest published head: `23868c1751c100a27f58087f7ce16b93c52d4eba`, targeting master. The review-found delegated gateway pool deadlock is repaired by passing the admission transaction through token and issuer-authority reads. Foreground and recovery each reproduced the deadlock with a one-connection pool; all three constrained-pool cases, including revoked authority, now pass. All 81 HTTP integration cases pass. The historical foreground rate-slot regression also reproduced before its repair; verified preview hashes are accepted only by read-only preflight, while consuming admission still requires the complete identity under the exact claim. Five full suites passed 329 cases; the gateway service then passed all 141 cases after two additional mixed historical hash controls. Production types/build passed before the final normal integration of master `35820cf398` (including independent Grok/Runner changes); integration validation is running. Initial fixture/assertion errors and the prior E2E restoration/cancellation deadlock remain recorded in [the owning audit](doc/idle-approved-tool-recovery.md#gateway-admission-uses-its-transaction-connection). Fresh-head CI, Greptile 5/5, and passing Ponytail are still required before merge. - Master landing reconciliation at `8dfe3958db6e4e8a0b7499f3f5be0a396ca77361` after #17453 merged: production and server test files are byte-identical to the accepted 320-case repair. The landed migration journal prefix is unchanged, and #17928's independent eval reporting changes are preserved. Fresh CI and reviews on this exact head are required before merge. - Historical-identity repair published at `096b7c7198c3ca864aca9377e261cbaa0a8aad6f` (production/test source identical to integrated acceptance `81231824d773a79fc50c835eccf7d9b3977e79fc`): all 320 accepted gateway, policy, startup, preparation-timing and authenticated HTTP management cases pass. Server types and the production server build pass. Four real historical sensitive-argument regressions failed before repair; invalid-signature, foreign-hash and stale-claim controls remain denied without dispatch or identity changes. Both agent and Test-tab calls deliver the original signed arguments once, and new rate policies still apply. - The first integrated combined run passed 319 cases and hit a Slack bot ownership collision left in a reused HTTP test database. The guard correctly returned 409. The full 78-case HTTP suite passed on a fresh dedicated fixture without product or assertion changes. Earlier fixture-construction errors are distinguished from the four real regressions in the audit document. Fresh remote CI and reviews remain required. - Parent reconciliation at `736e7f56097055e21ce13b4e45d9033391483af8`: 299 cases pass (221 full gateway/policy/startup cases and 78 assembled-server connection management cases), with no skipped cases. Server types pass. This includes real delegated Test-tab approval admission for newly added shared and caller rate rules. Fresh remote CI and Greptile review remain required. Approved delegated Test-tab calls now evaluate caller and target policies in the existing claim transaction. Real HTTP regressions first showed a missing caller charge and shared-limit self-rejection. All 78 connection-management cases and 92 gateway cases now pass on the owning branch, along with server production typechecking. All 78 connection-management cases also pass at combined tip `d25f830db89c1623bff3547bfc68015c1c66e0cf`. The three new cases cover shared charges, separate charges, rollback on denial, duplicate approval, and rejection of a later authorized call. Intermediate counter/identity fixture failures and a reused-database Slack collision are recorded in [the owning audit](https://github.com/paperclipai/paperclip/blob/bcca5b006976a471f4afd1f70c4031aa40a2b3f7/doc/idle-approved-tool-recovery.md#delegated-test-tab-admission-after-policy-changes). The earlier combined checks below remain evidence for their stated revisions. Current-head CI and review remain required. Current head: `bcca5b006976a471f4afd1f70c4031aa40a2b3f7`. This remains a single-duty PR stacked on #17453. - Integrated pinned master `8548173fc9efdc0fa07644fb92dcc0b6851a95e5` with normal merges. Every branch passed generated-schema checks and its target-branch migration-order gate. Existing SQL is unchanged; the 25 stack migrations follow master’s new migration. - Duty-specific producer coverage, lifecycle tests, and earlier failure records: [2026-10-24-recurring-server-sleep-stack.md](https://github.com/paperclipai/paperclip/blob/a021467f04d22be42ee5b0c87b6583b2300a62a9/doc/plans/2026-10-24-recurring-server-sleep-stack.md). - Combined tip `51e6cbe67979b55b340ea9f0f434c5b5c682ccd6`: **355 startup/cancellation/recovery/dispatch/preservation tests**, **75 connection-management tests**, and production server typechecking pass. These combined checks do not mean every intermediate PR independently ran the full repository suite. - Real master-to-stack database upgrade passes: **379 unchanged prior migration receipts plus 25 new migrations**, with the gateway responsibility fixture preserved. - Owning-branch checks: 84 runner installation/settlement cases plus runner typechecking and generated TypeScript/Rust profile agreement; 253 GitHub review workflow cases. Stop-attribution and held-worker readiness regressions failed before their fixes. - Failure accounting: the initial combined run passed 349 cases and failed six idle assertions in a reused preservation database containing old cancelled runs. Repeating the combined command with a fresh preservation database passed. Earlier invalid project selection, merge import/type errors, and the interrupted full local suite are not counted as passing checks. - **Latest-head CI and review remain required.** Prior CI findings include GitHub idle observation, route teardown, attention cleanup, touch-picker E2E, workspace health, and voice-cleanup timing; local passes do not erase those failures. No full local repository-suite, deployed sleep, fleet cost, or live Copilot qualification claim is made. ## Risks - Notifications cover the existing database-owner, single-active-writer contract. Independent SQL writers and other replicas need an explicit wake path. - Unknown provider outcomes cannot be replayed safely. Failed result persistence retains an executing claim and the existing stale-execution quarantine. - Interrupted execution and terminal interaction projection repair remain periodic until their separate slices. This PR changes only approved recovery. - A pending-only review write can trigger one empty approved-recovery pass. Settled idle work schedules no further scan. - The partial index uses the existing top-level concurrent migration executor. Snapshot pruning follows repository policy. - Cloud sleep eligibility and external alarm coverage remain unchanged. Worker quiet is not proof of deployed instance suspension. ## Model Used OpenAI Codex, GPT-6 family, with reasoning and code execution. The exact serving model identifier and context window were not exposed. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
35820cf398
commit
77295faebc
23 files changed
+2313
-506
No files matched your search
@@ -1,10 +1,12 @@
|
||||
import { createToolGatewayService } from "../services/tool-gateway.js";
|
||||
import { commitToolActionReview } from "../services/tool-action-review.js";
|
||||
import { googleWorkspaceSelectedScopes, defaultGitHubReviewPolicy, SLACK_CHAT_BOT_SCOPES, SLACK_BOT_TOOL_SCOPES } from "@paperclipai/shared";
|
||||
import { randomUUID, generateKeyPairSync } from "node:crypto";
|
||||
import { join } from "node:path";
|
||||
import { createServer } from "node:http";
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { beforeAll, afterAll, describe, expect, it, vi } from "vitest";
|
||||
import { activityLog, agentApiKeys, companies, companySecrets, companySecretVersions, chatConversations, chatDeliveries, chatEndpoints, chatExternalPrincipals, chatIdentityLinks, toolPolicies, connectionGrants, connectionGrantDelegations, toolAccessAuditEvents, toolOauthStates, issueThreadInteractions, connectionIntentDeliveries, agents, companyMemberships, heartbeatRuns, issues, principalPermissionGrants, projects, toolConnections, toolMcpGateways, toolMcpGatewayTokens, toolActionRequests, toolCatalogEntries, toolInvocations, toolCallEvents, toolProfileBindings, plugins, adapterAuthSessions, environments, routines, chatGitHubReviews } from "@paperclipai/db";
|
||||
import { createDb, activityLog, agentApiKeys, companies, companySecrets, companySecretVersions, chatConversations, chatDeliveries, chatEndpoints, chatExternalPrincipals, chatIdentityLinks, toolPolicies, toolRateLimitCounters, connectionGrants, connectionGrantDelegations, toolAccessAuditEvents, toolOauthStates, issueThreadInteractions, connectionIntentDeliveries, agents, companyMemberships, heartbeatRuns, issues, principalPermissionGrants, projects, toolConnections, toolMcpGateways, toolMcpGatewayTokens, toolActionRequests, toolCatalogEntries, toolInvocations, toolCallEvents, toolProfileBindings, plugins, adapterAuthSessions, environments, routines, chatGitHubReviews } from "@paperclipai/db";
|
||||
import { findSatisfiedToolConnection, satisfiedConnectionIntentService } from "../services/satisfied-connection-intents.js";
|
||||
import { connectionManagementService, continuingConnectionManagement } from "../services/connection-management.js";
|
||||
import { redactRunnerApiPayloads } from "../services/native-runtime/runner-api-log-redaction.js";
|
||||
@@ -234,7 +236,7 @@ describe("connection management with production default agents", () => {
|
||||
expect(bindings.map(binding => binding.targetId)).toEqual([f.agentId]);
|
||||
});
|
||||
|
||||
it.each(["none", "other-agent", "other-agent-no-access", "other-agent-allow-policy", "other-agent-explicit-grant", "other-agent-caller-ask", "other-agent-caller-rate", "other-agent-shared-rate", "other-agent-credential", "other-agent-credential-none", "other-agent-caller-revoked", "private-status", "agent", "user", "cancelled", "project-block", "rate-limit", "agent-rate-approval", "authorized-policy-edit"] as const)("keeps delegated test policy and approval continuation bounded (%s)", async revocation => {
|
||||
it.each(["none", "other-agent", "other-agent-no-access", "other-agent-allow-policy", "other-agent-explicit-grant", "other-agent-caller-ask", "other-agent-new-rate", "other-agent-new-shared-rate", "other-agent-new-rate-target-block", "other-agent-caller-rate", "other-agent-shared-rate", "other-agent-credential", "other-agent-credential-none", "other-agent-caller-revoked", "private-status", "agent", "user", "cancelled", "project-block", "rate-limit", "agent-rate-approval", "authorized-policy-edit"] as const)("keeps delegated test policy and approval continuation bounded (%s)", async revocation => {
|
||||
const f = await fixture();
|
||||
let targetAgentId = f.agentId;
|
||||
if (revocation.startsWith("other-agent")) {
|
||||
@@ -247,7 +249,7 @@ describe("connection management with production default agents", () => {
|
||||
await lifecycle.process(targetAgentId);
|
||||
await vi.waitFor(async () => expect((await agentService(server.db).getById(targetAgentId))?.status).toBe("idle"));
|
||||
}
|
||||
const allowed = revocation === "none" || revocation === "other-agent" || revocation === "other-agent-caller-ask";
|
||||
const allowed = revocation === "none" || revocation === "other-agent" || revocation === "other-agent-caller-ask" || ["other-agent-new-rate", "other-agent-new-shared-rate"].includes(revocation);
|
||||
let calls = 0;
|
||||
const remote = createServer(async (req, res) => {
|
||||
const chunks: Buffer[] = [];
|
||||
@@ -264,7 +266,7 @@ describe("connection management with production default agents", () => {
|
||||
const address = remote.address();
|
||||
if (!address || typeof address === "string") throw new Error("Fixture did not listen");
|
||||
const mcpUrl = `http://127.0.0.1:${address.port}/mcp`;
|
||||
const freezeRateClock = ["other-agent-caller-rate", "other-agent-shared-rate", "rate-limit", "agent-rate-approval"].includes(revocation);
|
||||
const freezeRateClock = revocation.startsWith("other-agent-new-") || ["other-agent-caller-rate", "other-agent-shared-rate", "rate-limit", "agent-rate-approval"].includes(revocation);
|
||||
if (freezeRateClock) vi.useFakeTimers({ toFake: ["Date"] });
|
||||
try {
|
||||
const created = await f.request("POST", `/companies/${f.companyId}/tools/connections`, { name: "Governed test", transport: "mcp_remote", authKind: "none", transportConfig: { url: mcpUrl } });
|
||||
@@ -381,6 +383,16 @@ describe("connection management with production default agents", () => {
|
||||
if (revocation === "cancelled") await server.db.update(issues).set({ status: "cancelled" }).where(eq(issues.id, f.issueId));
|
||||
if (revocation === "other-agent-caller-revoked") await server.db.insert(toolPolicies).values({ companyId: f.companyId, name: "Caller access revoked", policyType: "block", selectors: { connectionId: created.body.id, agentId: f.agentId }, priority: 0 });
|
||||
if (revocation === "project-block") await server.db.insert(toolPolicies).values({ companyId: f.companyId, name: "New project block", policyType: "block", selectors: { connectionId: created.body.id, projectId: f.projectId }, priority: 0 });
|
||||
if (revocation.startsWith("other-agent-new-")) {
|
||||
await server.db.insert(toolPolicies).values({ companyId: f.companyId, name: "New delegated test admission", policyType: "rate_limit",
|
||||
selectors: { connectionId: created.body.id, ...(revocation === "other-agent-new-shared-rate" ? {} : { agentId: f.agentId }) },
|
||||
priority: -2, config: { limit: 1, windowSeconds: 60, keyBy: revocation === "other-agent-new-shared-rate" ? ["company", "tool"] : ["agent", "tool"] } });
|
||||
if (revocation === "other-agent-new-rate") await server.db.insert(toolPolicies).values({ companyId: f.companyId, name: "New test target admission", policyType: "rate_limit",
|
||||
selectors: { connectionId: created.body.id, agentId: targetAgentId }, priority: -2,
|
||||
config: { limit: 1, windowSeconds: 60, keyBy: ["agent", "tool"] } });
|
||||
if (revocation === "other-agent-new-rate-target-block") await server.db.insert(toolPolicies).values({ companyId: f.companyId, name: "New test target block", policyType: "block",
|
||||
selectors: { connectionId: created.body.id, agentId: targetAgentId }, priority: -1 });
|
||||
}
|
||||
const cookie = await server.humanSession(f.userId);
|
||||
const approve = () => fetch(server.apiUrl + "/api" + approvalPath, { method: "POST", headers: { cookie, origin: server.apiUrl, "Content-Type": "application/json" }, body: JSON.stringify({ companyId: f.companyId }) });
|
||||
const response = await approve();
|
||||
@@ -390,6 +402,20 @@ describe("connection management with production default agents", () => {
|
||||
else expect(settled, JSON.stringify(settled)).toMatchObject({ status: allowed ? "succeeded" : "failed" });
|
||||
expect(calls).toBe(allowed ? 1 : 0);
|
||||
if (allowed) { expect((await approve()).status).toBe(200); expect(calls).toBe(1); }
|
||||
if (revocation.startsWith("other-agent-new-")) {
|
||||
const counters = await server.db.select().from(toolRateLimitCounters).where(eq(toolRateLimitCounters.companyId, f.companyId));
|
||||
expect(counters).toHaveLength(allowed ? revocation === "other-agent-new-shared-rate" ? 1 : 2 : 0);
|
||||
expect(counters.every(counter => counter.remaining === 0)).toBe(true);
|
||||
if (allowed) {
|
||||
expect(settled.policyExplanation?.consumedRateAdmissions).toHaveLength(1);
|
||||
expect(settled.policyExplanation?.delegatedCallerConsumedRateAdmissions).toHaveLength(1);
|
||||
const [nextRun] = await server.db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId,
|
||||
status: "running", responsibleUserId: f.userId, contextSnapshot: { issueId: f.issueId } }).returning();
|
||||
const nextJwt = createLocalAgentJwt(f.agentId, f.companyId, "paperclip_runner", nextRun.id, f.userId);
|
||||
expect(await f.request("POST", path, body, nextJwt)).toMatchObject({ status: 200, body: { decision: "off", error: { reasonCode: "rate_limited" } } });
|
||||
expect(calls).toBe(1);
|
||||
}
|
||||
}
|
||||
} finally { if (freezeRateClock) vi.useRealTimers(); remote.closeAllConnections(); await new Promise<void>((resolve, reject) => remote.close(error => error ? reject(error) : resolve())); }
|
||||
});
|
||||
|
||||
@@ -1229,7 +1255,7 @@ describe("connection management with production default agents", () => {
|
||||
expect((await use()).status).toBeGreaterThanOrEqual(400);
|
||||
});
|
||||
|
||||
it.each(["block", "ask", "ask-board", "ask-revoked", "ask-rate", "ask-token-revoked", "ask-token-deleted", "ask-token-scope", "ask-grant-revoked", "rate", "shared-rate", "target-block", "credential"] as const)("intersects gateway issuer and target rules (%s)", async mode => {
|
||||
it.each(["block", "ask", "ask-board", "ask-revoked", "ask-rate", "ask-new-rate", "ask-new-shared-rate", "ask-new-rate-target-block", "ask-token-revoked", "ask-token-deleted", "ask-token-scope", "ask-grant-revoked", "ask-one-connection", "ask-one-connection-recovery", "ask-one-connection-revoked", "rate", "shared-rate", "target-block", "credential"] as const)("intersects gateway issuer and target rules (%s)", async mode => {
|
||||
const f = await fixture();
|
||||
const cookie = await server.humanSession(f.userId);
|
||||
const createdTarget = await fetch(`${server.apiUrl}/api/companies/${f.companyId}/agents`, { method: "POST", headers: { cookie, origin: server.apiUrl, "Content-Type": "application/json" },
|
||||
@@ -1302,16 +1328,59 @@ describe("connection management with production default agents", () => {
|
||||
if (!mode.startsWith("ask")) return;
|
||||
const [action] = await server.db.select().from(toolActionRequests).where(eq(toolActionRequests.invocationId, invocation.id));
|
||||
expect(action.requestedByAgentId).toBe(mode === "ask-board" ? null : f.agentId);
|
||||
if (mode.startsWith("ask-new-")) {
|
||||
await server.db.insert(toolPolicies).values({ companyId: f.companyId, name: "New admission after review", policyType: "rate_limit",
|
||||
selectors: { connectionId: connection.body.id, ...(mode === "ask-new-shared-rate" ? {} : { agentId: f.agentId }) },
|
||||
priority: -2, config: { limit: 1, windowSeconds: 60, keyBy: mode === "ask-new-shared-rate" ? ["company", "tool"] : ["agent", "tool"] } });
|
||||
if (mode === "ask-new-rate") await server.db.insert(toolPolicies).values({ companyId: f.companyId, name: "New target admission", policyType: "rate_limit",
|
||||
selectors: { connectionId: connection.body.id, agentId: target.id }, priority: -1,
|
||||
config: { limit: 1, windowSeconds: 60, keyBy: ["agent", "tool"] } });
|
||||
if (mode === "ask-new-rate-target-block") await server.db.insert(toolPolicies).values({ companyId: f.companyId, name: "New target block", policyType: "block",
|
||||
selectors: { connectionId: connection.body.id, agentId: target.id }, priority: -1 });
|
||||
}
|
||||
if (mode === "ask-revoked") await server.db.insert(toolPolicies).values({ companyId: f.companyId, name: "Issuer blocked after review", policyType: "block", selectors: { connectionId: connection.body.id, agentId: f.agentId }, priority: -1 });
|
||||
if (mode === "ask-token-revoked") await server.db.update(toolMcpGatewayTokens).set({ revokedAt: new Date() }).where(eq(toolMcpGatewayTokens.id, token.body.id));
|
||||
if (mode === "ask-token-deleted") await server.db.delete(toolMcpGatewayTokens).where(eq(toolMcpGatewayTokens.id, token.body.id));
|
||||
if (mode === "ask-token-scope") await server.db.update(toolMcpGatewayTokens).set({ allowedActions: ["tools/list"] }).where(eq(toolMcpGatewayTokens.id, token.body.id));
|
||||
if (mode === "ask-grant-revoked") await server.db.delete(principalPermissionGrants).where(and(eq(principalPermissionGrants.principalId, f.agentId), eq(principalPermissionGrants.permissionKey, "tools:manage_connections")));
|
||||
const approvalDenied = ["ask-revoked", "ask-token-revoked", "ask-token-deleted", "ask-token-scope", "ask-grant-revoked"].includes(mode);
|
||||
const response = await fetch(`${server.apiUrl}/api/tool-gateway/action-requests/${action.id}/approve`, { method: "POST", headers: { cookie, origin: server.apiUrl, "Content-Type": "application/json" }, body: JSON.stringify({ companyId: f.companyId }) });
|
||||
expect(response.status, JSON.stringify(await response.json())).toBe(approvalDenied ? 403 : 200);
|
||||
if (mode === "ask-grant-revoked" || mode === "ask-one-connection-revoked") await server.db.delete(principalPermissionGrants).where(and(eq(principalPermissionGrants.principalId, f.agentId), eq(principalPermissionGrants.permissionKey, "tools:manage_connections")));
|
||||
const approvalDenied = ["ask-revoked", "ask-new-rate-target-block", "ask-token-revoked", "ask-token-deleted", "ask-token-scope", "ask-grant-revoked", "ask-one-connection-revoked"].includes(mode);
|
||||
if (mode.startsWith("ask-one-connection")) {
|
||||
const limited = createDb(server.databaseUrl, { maxConnections: 1, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 });
|
||||
const service = createToolGatewayService(limited);
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
let attempt: Promise<unknown> | undefined;
|
||||
try {
|
||||
if (mode === "ask-one-connection-recovery") await commitToolActionReview(server.db, {
|
||||
companyId: f.companyId, actionRequestId: action.id, decision: "approved", actor: { userId: f.userId },
|
||||
});
|
||||
attempt = mode === "ask-one-connection-recovery" ? service.recoverApprovedActionReviews(f.companyId)
|
||||
: service.approveActionRequest({ companyId: f.companyId, actionRequestId: action.id, actor: { userId: f.userId } });
|
||||
void attempt.catch(() => undefined);
|
||||
const completed = Promise.race([attempt, new Promise<never>((_, reject) => {
|
||||
timer = setTimeout(() => reject(new Error("Single-connection gateway approval did not settle")), 5000);
|
||||
})]);
|
||||
if (approvalDenied) await expect(completed).rejects.toMatchObject({ status: 403 });
|
||||
else await completed;
|
||||
expect((await limited.execute(sql`select 1 as available`))[0].available).toBe(1);
|
||||
} finally {
|
||||
clearTimeout(timer); await limited.$client.end({ timeout: 1 }); await attempt?.catch(() => undefined);
|
||||
}
|
||||
} else {
|
||||
const response = await fetch(`${server.apiUrl}/api/tool-gateway/action-requests/${action.id}/approve`, { method: "POST", headers: { cookie, origin: server.apiUrl, "Content-Type": "application/json" }, body: JSON.stringify({ companyId: f.companyId }) });
|
||||
expect(response.status, JSON.stringify(await response.json())).toBe(approvalDenied ? 403 : 200);
|
||||
}
|
||||
expect(calls).toBe(approvalDenied ? 0 : 1);
|
||||
if (mode === "ask-rate") {
|
||||
if (mode.startsWith("ask-new-")) {
|
||||
const counters = await server.db.select().from(toolRateLimitCounters).where(eq(toolRateLimitCounters.companyId, f.companyId));
|
||||
expect(counters).toHaveLength(approvalDenied ? 0 : mode === "ask-new-shared-rate" ? 1 : 2);
|
||||
expect(counters.every(counter => counter.remaining === 0)).toBe(true);
|
||||
if (!approvalDenied) {
|
||||
const [settled] = await server.db.select().from(toolInvocations).where(eq(toolInvocations.id, invocation.id));
|
||||
expect(settled.policyExplanation?.consumedRateAdmissions).toHaveLength(1);
|
||||
expect(settled.policyExplanation?.delegatedCallerConsumedRateAdmissions).toHaveLength(1);
|
||||
}
|
||||
}
|
||||
if (["ask-rate", "ask-new-rate", "ask-new-shared-rate"].includes(mode)) {
|
||||
await rpc("tools/call", { name: tool.name, arguments: { another: true } });
|
||||
expect(calls).toBe(1);
|
||||
}
|
||||
|
||||
@@ -60,6 +60,7 @@ export async function startRunnerApiTestServer(options: {
|
||||
const apiUrl = `http://127.0.0.1:${address.port}`;
|
||||
setupRunnerPrpWebSocketServer(http, { apiUrl });
|
||||
return {
|
||||
databaseUrl: temporary.connectionString,
|
||||
db, root, apiUrl, storage,
|
||||
async humanSession(userId: string) {
|
||||
if (!auth) throw new Error("Enable humanSessions for this fixture");
|
||||
|
||||
@@ -69,6 +69,8 @@ const {
|
||||
stop: vi.fn(async () => {}),
|
||||
},
|
||||
toolGateway: {
|
||||
recoverApprovedActionReviews: vi.fn(async () => ({ scanned: 0, retry: false })),
|
||||
approvedActionReviewCompanies: vi.fn(async () => []),
|
||||
sweepActionReviews: vi.fn(async () => ({ scanned: 0 })),
|
||||
sweepExpiredActionReviews: vi.fn(async () => ({ scanned: 0 })),
|
||||
nextActionReviewExpiryAt: vi.fn(async () => null),
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { eq, sql } from "drizzle-orm";
|
||||
import { afterAll, afterEach, beforeAll, describe, expect, it } from "vitest";
|
||||
import {
|
||||
activityLog,
|
||||
@@ -34,8 +34,9 @@ import { toolAccessService } from "../services/tool-access.js";
|
||||
import { createToolGatewayService, ToolGatewayHttpError } from "../services/tool-gateway.js";
|
||||
import { canonicalToolArguments, signToolArguments } from "../services/tool-content-guards.js";
|
||||
|
||||
const externalTestDatabaseUrl = process.env.PAPERCLIP_TOOL_POLICY_TEST_DATABASE_URL;
|
||||
const embeddedPostgresSupport = externalTestDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
const externalDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_TOOL_POLICY_TEST_DATABASE_URL");
|
||||
const embeddedPostgresSupport = externalDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
|
||||
|
||||
function legacySideEffectKey(invocation: typeof toolInvocations.$inferSelect, argumentsHash: string) {
|
||||
@@ -216,8 +217,11 @@ describeEmbeddedPostgres("tool access policy service", () => {
|
||||
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
||||
|
||||
beforeAll(async () => {
|
||||
if (!externalTestDatabaseUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-tool-access-policy-");
|
||||
db = createDb(externalTestDatabaseUrl ?? tempDb!.connectionString);
|
||||
if (externalDatabaseUrl) db = createDb(externalDatabaseUrl);
|
||||
else {
|
||||
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-tool-access-policy-");
|
||||
db = createDb(tempDb.connectionString);
|
||||
}
|
||||
}, 20_000);
|
||||
|
||||
afterEach(async () => {
|
||||
@@ -1115,11 +1119,103 @@ describeEmbeddedPostgres("tool access policy service", () => {
|
||||
expect(await svc.decide(input, { admittedInvocationId: invocation.id })).toMatchObject({ allowed: true });
|
||||
expect(await db.select().from(toolRateLimitCounters)).toEqual(counters);
|
||||
expect((await db.select().from(toolRateLimitCounters))[0].remaining).toBe(0);
|
||||
// Neither a new invocation nor an accidental consuming recheck can borrow
|
||||
// another invocation's already-spent slot.
|
||||
// A fresh invocation cannot borrow another invocation's already-spent slot.
|
||||
// Execution revalidation retains its own durable admission.
|
||||
expect(await svc.decide(input, { admittedInvocationId: randomUUID() })).toMatchObject({ decision: "rate_limited" });
|
||||
expect(await svc.decide({ ...input, consumeRateLimit: true }, { admittedInvocationId: invocation.id }))
|
||||
.toMatchObject({ decision: "rate_limited" });
|
||||
expect(await svc.admitInvocation(input, invocation.id)).toMatchObject({ allowed: true });
|
||||
expect(await db.select().from(toolRateLimitCounters)).toEqual(counters);
|
||||
});
|
||||
|
||||
it("atomically charges a new rule once across concurrent retries of the same invocation", async () => {
|
||||
const { company, input, invocation, svc } = await admittedRateLimitedInvocation();
|
||||
const added = await svc.createPolicy(company.id, { name: "New admission", policyType: "rate_limit", priority: 0,
|
||||
selectors: { toolName: "send_email" }, config: { limit: 1, windowSeconds: 3600, keyBy: ["agent", "tool"] } });
|
||||
const oldCounters = await db.select().from(toolRateLimitCounters);
|
||||
expect(await svc.decide(input, { admittedInvocationId: invocation.id })).toMatchObject({ allowed: true });
|
||||
expect(await db.select().from(toolRateLimitCounters)).toEqual(oldCounters);
|
||||
const results = await Promise.all(Array.from({ length: 4 }, () => svc.admitInvocation(input, invocation.id)));
|
||||
expect(results.every(result => result.allowed)).toBe(true);
|
||||
expect(await db.select().from(toolRateLimitCounters).where(eq(toolRateLimitCounters.policyId, added.id)))
|
||||
.toMatchObject([{ limit: 1, remaining: 0 }]);
|
||||
const [saved] = await db.select().from(toolInvocations).where(eq(toolInvocations.id, invocation.id));
|
||||
expect(saved.policyExplanation?.consumedRateAdmissions).toHaveLength(2);
|
||||
const { consumedRateAdmissions: oldAdmissions, ...oldExplanation } = invocation.policyExplanation!;
|
||||
expect(saved.policyExplanation).toMatchObject(oldExplanation);
|
||||
expect(saved.policyExplanation?.consumedRateAdmissions).toEqual(expect.arrayContaining(oldAdmissions as unknown[]));
|
||||
});
|
||||
|
||||
it("allows only one of two admitted invocations to spend a newly added final slot", async () => {
|
||||
const { company, input, invocation, svc } = await admittedRateLimitedInvocation(2);
|
||||
const initial = await svc.decide({ ...input, consumeRateLimit: true });
|
||||
expect(initial.allowed).toBe(true);
|
||||
const { invocation: second } = await svc.recordInvocation(input, initial);
|
||||
await db.update(toolInvocations).set({ status: "executing" }).where(eq(toolInvocations.id, second.id));
|
||||
const added = await svc.createPolicy(company.id, { name: "One competing slot", policyType: "rate_limit", priority: 0,
|
||||
selectors: { toolName: "send_email" }, config: { limit: 1, windowSeconds: 3600, keyBy: ["agent", "tool"] } });
|
||||
const results = await Promise.all([invocation.id, second.id].map(id => svc.admitInvocation(input, id)));
|
||||
expect(results.filter(result => result.allowed)).toHaveLength(1);
|
||||
expect(results.filter(result => result.decision === "rate_limited")).toHaveLength(1);
|
||||
expect(await db.select().from(toolRateLimitCounters).where(eq(toolRateLimitCounters.policyId, added.id)))
|
||||
.toMatchObject([{ limit: 1, remaining: 0 }]);
|
||||
});
|
||||
|
||||
it("rolls back new rate charges when a later policy rejects the invocation", async () => {
|
||||
const { company, input, invocation, svc } = await admittedRateLimitedInvocation();
|
||||
await svc.createPolicy(company.id, { name: "Before rejection", policyType: "rate_limit", priority: 0,
|
||||
selectors: { toolName: "send_email" }, config: { limit: 1, windowSeconds: 3600, keyBy: ["agent", "tool"] } });
|
||||
await svc.createPolicy(company.id, { name: "Later rejection", policyType: "block", priority: 10,
|
||||
selectors: { toolName: "send_email" } });
|
||||
const counters = await db.select().from(toolRateLimitCounters);
|
||||
expect(await svc.admitInvocation(input, invocation.id)).toMatchObject({ decision: "deny" });
|
||||
expect(await db.select().from(toolRateLimitCounters)).toEqual(counters);
|
||||
expect((await db.select().from(toolInvocations).where(eq(toolInvocations.id, invocation.id)))[0].policyExplanation)
|
||||
.toEqual(invocation.policyExplanation);
|
||||
});
|
||||
|
||||
it("rolls back rate charges when saving the admission receipt fails and admits a retry once", async () => {
|
||||
const { company, input, invocation, svc } = await admittedRateLimitedInvocation();
|
||||
const added = await svc.createPolicy(company.id, { name: "Receipt failure", policyType: "rate_limit", priority: 0,
|
||||
selectors: { toolName: "send_email" }, config: { limit: 1, windowSeconds: 3600, keyBy: ["agent", "tool"] } });
|
||||
await db.execute(sql`create function sleep_admission_fail() returns trigger language plpgsql as $$ begin
|
||||
raise exception 'receipt write unavailable'; end $$`);
|
||||
await db.execute(sql`create trigger sleep_admission_fail before update of policy_explanation on tool_invocations
|
||||
for each row execute function sleep_admission_fail()`);
|
||||
try {
|
||||
await expect(svc.admitInvocation(input, invocation.id)).rejects.toThrow();
|
||||
expect(await db.select().from(toolRateLimitCounters).where(eq(toolRateLimitCounters.policyId, added.id))).toEqual([]);
|
||||
expect((await db.select().from(toolInvocations).where(eq(toolInvocations.id, invocation.id)))[0].policyExplanation)
|
||||
.toEqual(invocation.policyExplanation);
|
||||
} finally {
|
||||
await db.execute(sql`drop trigger sleep_admission_fail on tool_invocations`);
|
||||
await db.execute(sql`drop function sleep_admission_fail()`);
|
||||
}
|
||||
expect(await svc.admitInvocation(input, invocation.id)).toMatchObject({ allowed: true });
|
||||
expect(await svc.admitInvocation(input, invocation.id)).toMatchObject({ allowed: true });
|
||||
expect(await db.select().from(toolRateLimitCounters).where(eq(toolRateLimitCounters.policyId, added.id)))
|
||||
.toMatchObject([{ limit: 1, remaining: 0 }]);
|
||||
});
|
||||
|
||||
it.each([2, 4])("preserves spent slots when a rate limit changes from three to %s", async limit => {
|
||||
const { company, input, ratePolicy, svc } = await admittedRateLimitedInvocation(3);
|
||||
await svc.updatePolicy({ companyId: company.id, policyId: ratePolicy.id,
|
||||
body: { config: { limit, windowSeconds: 3600, keyBy: ["agent", "tool"] } } });
|
||||
const results = await Promise.all(Array.from({ length: limit }, () => svc.decide({ ...input, consumeRateLimit: true })));
|
||||
expect(results.filter(result => result.allowed)).toHaveLength(limit - 1);
|
||||
expect(results.filter(result => result.decision === "rate_limited")).toHaveLength(1);
|
||||
expect(await db.select().from(toolRateLimitCounters)).toMatchObject([{ limit, remaining: 0 }]);
|
||||
});
|
||||
|
||||
it.each(["pending", "approved"] as const)("retains a waiting invocation's consumed slot without bypassing %s approval policy", async approvalState => {
|
||||
const { company, input, invocation, svc } = await admittedRateLimitedInvocation();
|
||||
await db.update(toolInvocations).set({ status: "awaiting_approval", policyDecision: "require_approval", approvalState })
|
||||
.where(eq(toolInvocations.id, invocation.id));
|
||||
await db.insert(toolPolicies).values({ companyId: company.id, name: "Review admitted sends",
|
||||
policyType: "require_approval", priority: 2, selectors: { toolName: "send_email" } });
|
||||
const counters = await db.select().from(toolRateLimitCounters);
|
||||
expect(await svc.decide(input, { admittedInvocationId: invocation.id }))
|
||||
.toMatchObject({ allowed: false, decision: "require_approval" });
|
||||
expect(await db.select().from(toolRateLimitCounters)).toEqual(counters);
|
||||
expect(await svc.decide(input)).toMatchObject({ decision: "rate_limited" });
|
||||
});
|
||||
|
||||
it("rechecks a newly added rate policy rather than borrowing an older policy's consumed slot", async () => {
|
||||
@@ -1172,15 +1268,19 @@ describeEmbeddedPostgres("tool access policy service", () => {
|
||||
{ actorType: "system" }, { actorId: randomUUID() }, { agentId: null }, { runId: null }, { issueId: null }, { gatewayId: null },
|
||||
{ applicationId: null }, { connectionId: null }, { catalogEntryId: null }, { toolName: "different_tool" },
|
||||
{ providerType: null }, { applicationKey: null }, { upstreamToolName: null }, { argumentsHash: "different" },
|
||||
{ status: "authorized" }, { policyDecision: "deny" }, { policyDecision: "require_approval", approvalState: "pending" },
|
||||
{ status: "authorized" }, { status: "succeeded" }, { status: "awaiting_approval" },
|
||||
{ status: "awaiting_approval", policyDecision: "require_approval", approvalState: "rejected" }, { policyDecision: "deny" }, { policyDecision: "require_approval", approvalState: "pending" },
|
||||
];
|
||||
for (const mutation of mutations) {
|
||||
await db.update(toolInvocations).set(mutation).where(eq(toolInvocations.id, invocation.id));
|
||||
expect(await svc.decide(input, options), JSON.stringify(mutation)).toMatchObject({ decision: "rate_limited" });
|
||||
expect(await svc.admitInvocation(input, invocation.id), JSON.stringify(mutation)).toMatchObject({ decision: "deny" });
|
||||
await db.update(toolInvocations).set({ ...invocation, status: "executing" }).where(eq(toolInvocations.id, invocation.id));
|
||||
}
|
||||
expect(await svc.decide({ ...input, request: { ...input.request, arguments: { to: "another@example.test" } } }, options))
|
||||
.toMatchObject({ decision: "rate_limited" });
|
||||
expect(await svc.admitInvocation({ ...input, request: { ...input.request, arguments: { to: "another@example.test" } } }, invocation.id))
|
||||
.toMatchObject({ decision: "deny" });
|
||||
});
|
||||
|
||||
it("rejects unsupported policy semantics at create and update time", async () => {
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
|
||||
import * as fs from "node:fs/promises";
|
||||
import { createHash, generateKeyPairSync, randomUUID } from "node:crypto";
|
||||
import { createServer } from "node:http";
|
||||
@@ -97,8 +98,8 @@ import { HttpError } from "../errors.js";
|
||||
import type { VercelConnectClient } from "../services/vercel-connect.js";
|
||||
import { invalidatePaperclipCloudConnectorCapabilities, PaperclipCloudConnectorError, type PaperclipCloudConnector } from "../services/paperclip-cloud-connector.js";
|
||||
|
||||
const externalDatabaseUrl = process.env.PAPERCLIP_TOOL_ACCESS_TEST_DATABASE_URL?.trim();
|
||||
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
|
||||
const externalDatabaseUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_TOOL_ACCESS_TEST_DATABASE_URL");
|
||||
const embeddedPostgresSupport = externalDatabaseUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = embeddedPostgresSupport.supported
|
||||
? describe
|
||||
: describe.skip;
|
||||
@@ -4119,8 +4120,9 @@ describeEmbeddedPostgres("tool access service", () => {
|
||||
.from(toolActionRequests)
|
||||
.where(eq(toolActionRequests.companyId, company.id));
|
||||
expect(requests).toHaveLength(2);
|
||||
expect(globalThis.fetch).toHaveBeenCalledTimes(1);
|
||||
expect(requests.map((row) => row.status).sort()).toEqual([
|
||||
"approved",
|
||||
"executed",
|
||||
"pending",
|
||||
]);
|
||||
});
|
||||
|
||||
@@ -35,12 +35,15 @@ import {
|
||||
toolGatewaySessions,
|
||||
toolInvocations,
|
||||
toolPolicies,
|
||||
toolRateLimitCounters,
|
||||
} from "@paperclipai/db";
|
||||
import type { PluginToolDispatcher } from "../services/plugin-tool-dispatcher.js";
|
||||
import type { VercelConnectClient } from "../services/vercel-connect.js";
|
||||
import { initializeRunIdentity, reserveSteeredIdentity, reconcileSteeredIdentity } from "../services/run-identity.js";
|
||||
import { issueService } from "../services/issues.js";
|
||||
import { approvalService } from "../services/approvals.js";
|
||||
import { commitToolActionReview } from "../services/tool-action-review.js";
|
||||
import { bindVerifiedToolApprovalArguments, claimToolTestApproval, releaseToolApproval, startToolInvocation } from "../services/tool-approval-execution.js";
|
||||
import { materializeNativeInteractionResponses } from "../services/native-runtime/native-interaction-bridge.js";
|
||||
import { toolActionDeliveryService } from "../services/tool-action-delivery.js";
|
||||
import { secretService } from "../services/secrets.js";
|
||||
@@ -49,7 +52,7 @@ import {
|
||||
createToolGatewayService,
|
||||
ToolGatewayHttpError,
|
||||
} from "../services/tool-gateway.js";
|
||||
import { canonicalToolArguments, signToolArguments } from "../services/tool-content-guards.js";
|
||||
import { canonicalToolArguments, readSignedToolArgumentsPayload, signToolArguments, summarizeToolValue, toolRequestIdentityHash } from "../services/tool-content-guards.js";
|
||||
import {
|
||||
getEmbeddedPostgresTestSupport,
|
||||
startEmbeddedPostgresTestDatabase,
|
||||
@@ -57,6 +60,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";
|
||||
|
||||
@@ -210,7 +214,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-");
|
||||
@@ -221,7 +225,7 @@ describeEmbeddedPostgres("tool gateway service", () => {
|
||||
}, 20_000);
|
||||
|
||||
afterEach(async () => {
|
||||
await stopExpiryWorker?.(); stopExpiryWorker = undefined;
|
||||
await stopReviewWorker?.(); stopReviewWorker = undefined;
|
||||
await clearFixtures();
|
||||
});
|
||||
|
||||
@@ -268,7 +272,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() {
|
||||
@@ -276,6 +280,7 @@ describeEmbeddedPostgres("tool gateway service", () => {
|
||||
vi.useRealTimers(); vi.restoreAllMocks();
|
||||
await db.execute(sql`drop trigger if exists sleep_review_fail on tool_invocations`);
|
||||
await db.execute(sql`drop function if exists sleep_review_fail()`);
|
||||
await db.execute(sql`drop sequence if exists sleep_review_failure_attempt`);
|
||||
expect(errors).toEqual([]);
|
||||
}
|
||||
async function start() {
|
||||
@@ -313,6 +318,710 @@ 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("does not scan another company's approved backlog after review writes", async () => {
|
||||
const other = await createRunFixture(db);
|
||||
await db.insert(toolPolicies).values(reviewNotePolicyRow(other.company));
|
||||
const foreign = await request(gateway, db, other);
|
||||
await db.update(toolActionRequests).set({ signedArguments: signToolArguments({
|
||||
invocationId: foreign.invocationId, toolName: "mcp-remote-fixture:update_note",
|
||||
canonicalArguments: canonicalToolArguments({ noteId: "legacy", body: "inert" }),
|
||||
signingSecret: testToolActionSigningSecret,
|
||||
}) }).where(eq(toolActionRequests.id, foreign.id));
|
||||
await commitToolActionReview(db, { companyId: other.company.id, actionRequestId: foreign.id,
|
||||
decision: "approved", actor: { userId: "board" } });
|
||||
await startRecovery(); await quiet();
|
||||
const reads: Array<{ query: string; params: unknown[] }> = [];
|
||||
const debug = db.$client.options.debug;
|
||||
db.$client.options.debug = (_id, query, params) => {
|
||||
if (/^select/i.test(query) && query.includes('"tool_action_requests"') &&
|
||||
query.includes('"tool_action_requests"."status" =') && params.includes("approved"))
|
||||
reads.push({ query, params: [...params] });
|
||||
};
|
||||
try {
|
||||
for (let index = 0; index < 3; index++) {
|
||||
const pending = await request(); await settle(1500);
|
||||
await approve(pending); await settle(1500); await executed(pending);
|
||||
}
|
||||
} finally { db.$client.options.debug = debug; }
|
||||
expect(reads.length).toBeGreaterThan(0);
|
||||
for (const read of reads) {
|
||||
expect(read.query).toContain('"tool_action_requests"."company_id" =');
|
||||
expect(read.params).toContain(fixture.company.id);
|
||||
expect(read.params).not.toContain(other.company.id);
|
||||
}
|
||||
expect((await db.select().from(toolActionRequests).where(eq(toolActionRequests.id, foreign.id)))[0].status).toBe("approved");
|
||||
expect((await db.select().from(toolInvocations).where(eq(toolInvocations.id, foreign.invocationId)))[0].status).toBe("awaiting_approval");
|
||||
expect(await db.select().from(toolCallEvents).where(eq(toolCallEvents.reasonCode, "approved_action_executed"))).toHaveLength(3);
|
||||
await quiet();
|
||||
});
|
||||
|
||||
it.each(["foreground", "recovery"] as const)("preserves the admitted rate slot for an agent action through %s approval", async mode => {
|
||||
await db.insert(toolPolicies).values({ companyId: fixture.company.id, name: "One reviewed note",
|
||||
policyType: "rate_limit", priority: 1, selectors: { toolName: "mcp-remote-fixture:update_note" },
|
||||
config: { limit: 1, windowSeconds: 3600, keyBy: ["agent", "tool"] } });
|
||||
const pending = await request();
|
||||
const counters = await db.select().from(toolRateLimitCounters);
|
||||
expect(counters).toHaveLength(1); expect(counters[0].remaining).toBe(0);
|
||||
if (mode === "foreground") {
|
||||
await gateway.approveActionRequest({ companyId: fixture.company.id, actionRequestId: pending.id, actor: { userId: "board" } });
|
||||
await startRecovery();
|
||||
} else { await approve(pending); await startRecovery(); }
|
||||
await settle(); await executed(pending); await quiet();
|
||||
expect(await db.select().from(toolRateLimitCounters)).toEqual(counters);
|
||||
expect(await db.select().from(toolCallEvents).where(eq(toolCallEvents.reasonCode, "approved_action_executed"))).toHaveLength(1);
|
||||
});
|
||||
|
||||
it.each(["new", "increased"] as const)("spends the newly available slot of a %s rate rule during approved recovery", async change => {
|
||||
const rate = { companyId: fixture.company.id, name: "Changed reviewed-note limit",
|
||||
policyType: "rate_limit" as const, priority: 1, selectors: { toolName: "mcp-remote-fixture:update_note" },
|
||||
config: { limit: 2, windowSeconds: 3600, keyBy: ["agent", "tool"] } };
|
||||
const original = change === "increased" ? (await db.insert(toolPolicies).values(rate).returning())[0] : null;
|
||||
const first = await request(), second = await request();
|
||||
if (original) {
|
||||
await db.update(toolPolicies).set({ config: { ...rate.config, limit: 3 }, updatedAt: new Date() })
|
||||
.where(eq(toolPolicies.id, original.id));
|
||||
} else {
|
||||
await db.insert(toolPolicies).values({ ...rate, config: { ...rate.config, limit: 1 } });
|
||||
}
|
||||
await approve(first); await approve(second);
|
||||
await startRecovery(); await settle(); await quiet();
|
||||
const requests = await db.select().from(toolActionRequests);
|
||||
expect(requests.filter(row => row.status === "executed")).toHaveLength(1);
|
||||
expect(requests.filter(row => row.status === "failed")).toHaveLength(1);
|
||||
const invocations = await db.select().from(toolInvocations);
|
||||
expect(invocations.filter(row => row.errorCode === "rate_limited")).toHaveLength(1);
|
||||
expect(await db.select().from(toolCallEvents).where(eq(toolCallEvents.reasonCode, "approved_action_executed"))).toHaveLength(1);
|
||||
const counters = await db.select().from(toolRateLimitCounters);
|
||||
expect(counters).toHaveLength(1);
|
||||
expect(counters[0]).toMatchObject({ limit: change === "increased" ? 3 : 1, remaining: 0 });
|
||||
});
|
||||
|
||||
it.each([
|
||||
...["agent", "test-tab"].flatMap(origin => ["valid", "new-rate", "invalid-signature", "foreign-hash", "foreground-rate", "foreground-invalid-signature", "foreground-foreign-hash", "foreground-modern-review-hash"]
|
||||
.map(state => ({ origin, state }))),
|
||||
])("recovers historical preview-hash approval ($origin, $state)", async ({ origin, state }) => {
|
||||
const remoteCall = providerFixture(); remoteCall.release();
|
||||
const parameters = { value: "reviewed", apiKey: "synthetic-historical-approval-key" };
|
||||
const foreground = state.startsWith("foreground-");
|
||||
if (foreground) await db.insert(toolPolicies).values({ companyId: fixture.company.id,
|
||||
name: "Already admitted historical action", policyType: "rate_limit", priority: 1,
|
||||
selectors: {}, config: { limit: 1, windowSeconds: 3600, keyBy: ["company", "tool"] } });
|
||||
const pending = origin === "agent" ? await agentRequest(remoteCall.provider, parameters)
|
||||
: (await testRequest(remoteCall.provider, false, parameters)).pending;
|
||||
const [originalInvocation] = await db.select().from(toolInvocations).where(eq(toolInvocations.id, pending.invocationId));
|
||||
const payload = readSignedToolArgumentsPayload({ signedArguments: pending.signedArguments,
|
||||
invocationId: pending.invocationId, toolName: originalInvocation.toolName, signingSecret: testToolActionSigningSecret })!;
|
||||
expect(payload).not.toBeNull();
|
||||
const legacyHash = summarizeToolValue(payload.arguments).sha256!;
|
||||
const completeHash = toolRequestIdentityHash(payload.arguments, testToolActionSigningSecret);
|
||||
expect(legacyHash).not.toBe(completeHash);
|
||||
const storedHash = state.endsWith("foreign-hash") ? "e".repeat(64) : legacyHash;
|
||||
await db.update(toolInvocations).set({ argumentsHash: storedHash }).where(eq(toolInvocations.id, pending.invocationId));
|
||||
await db.update(toolActionRequests).set({ canonicalArgumentsHash: state === "foreground-modern-review-hash" ? completeHash : legacyHash,
|
||||
...(["invalid-signature", "foreground-invalid-signature"].includes(state) ? { signedArguments: "unreadable-old-signature" } : {}),
|
||||
}).where(eq(toolActionRequests.id, pending.id));
|
||||
if (state === "new-rate") await db.insert(toolPolicies).values({ companyId: fixture.company.id,
|
||||
name: "New admission for historical approval", policyType: "rate_limit", priority: 1,
|
||||
selectors: {}, config: { limit: 1, windowSeconds: 3600, keyBy: ["company", "tool"] } });
|
||||
const counters = foreground ? await db.select().from(toolRateLimitCounters) : [];
|
||||
if (foreground) {
|
||||
expect(counters).toEqual([expect.objectContaining({ remaining: 0 })]);
|
||||
const approval = gateway.approveActionRequest({ companyId: fixture.company.id,
|
||||
actionRequestId: pending.id, actor: { userId: "board" } });
|
||||
if (state === "foreground-invalid-signature") await expect(approval).rejects.toMatchObject({ reasonCode: "action_request_invalidated" });
|
||||
else if (state === "foreground-foreign-hash" && origin === "agent") await expect(approval).rejects.toMatchObject({ reasonCode: "rate_limited" });
|
||||
else await approval;
|
||||
} else await approve(pending);
|
||||
await startRecovery(); await settle();
|
||||
if (state === "valid" || state === "new-rate" || state === "foreground-rate" || state === "foreground-modern-review-hash") {
|
||||
await executed(pending);
|
||||
expect((await db.select().from(toolInvocations))[0].argumentsHash).toBe(completeHash);
|
||||
expect(remoteCall.provider).toHaveBeenCalledOnce();
|
||||
expect(JSON.parse(String(remoteCall.provider.mock.calls[0][1].body)).params.arguments).toEqual(parameters);
|
||||
if (state === "new-rate") expect(await db.select().from(toolRateLimitCounters))
|
||||
.toEqual([expect.objectContaining({ limit: 1, remaining: 0 })]);
|
||||
} else {
|
||||
expect((await db.select().from(toolActionRequests))[0].status)
|
||||
.toBe(["invalid-signature", "foreground-invalid-signature"].includes(state) ? "cancelled"
|
||||
: state === "foreground-foreign-hash" && origin === "agent" ? "pending" : "failed");
|
||||
expect((await db.select().from(toolInvocations))[0].argumentsHash).toBe(storedHash);
|
||||
expect(remoteCall.provider).not.toHaveBeenCalled();
|
||||
}
|
||||
if (foreground) expect(await db.select().from(toolRateLimitCounters)).toEqual(counters);
|
||||
await quiet();
|
||||
});
|
||||
|
||||
it("does not rebind historical arguments after the execution claim is replaced", async () => {
|
||||
const remoteCall = providerFixture(); remoteCall.release();
|
||||
const pending = await agentRequest(remoteCall.provider, { value: "reviewed", apiKey: "synthetic-stale-claim-key" });
|
||||
const [invocation] = await db.select().from(toolInvocations).where(eq(toolInvocations.id, pending.invocationId));
|
||||
const payload = readSignedToolArgumentsPayload({ signedArguments: pending.signedArguments,
|
||||
invocationId: invocation.id, toolName: invocation.toolName, signingSecret: testToolActionSigningSecret })!;
|
||||
const preview = summarizeToolValue(payload.arguments).sha256;
|
||||
const verified = toolRequestIdentityHash(payload.arguments, testToolActionSigningSecret);
|
||||
await db.update(toolInvocations).set({ argumentsHash: preview }).where(eq(toolInvocations.id, invocation.id));
|
||||
await db.update(toolActionRequests).set({ canonicalArgumentsHash: preview }).where(eq(toolActionRequests.id, pending.id));
|
||||
await approve(pending);
|
||||
const claim = (await claimToolTestApproval(db, pending.id, invocation.id))!;
|
||||
expect(claim).not.toBeNull();
|
||||
await db.update(toolActionRequests).set({ updatedAt: new Date(claim.claimedAt.getTime() + 1) })
|
||||
.where(eq(toolActionRequests.id, pending.id));
|
||||
expect(await bindVerifiedToolApprovalArguments(db, claim, { preview, verified })).toBe(false);
|
||||
expect((await db.select().from(toolInvocations))[0].argumentsHash).toBe(preview);
|
||||
expect(remoteCall.provider).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
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("preserves a failed company's retry across writes while another company executes", async () => {
|
||||
const pending = await request(); await approve(pending);
|
||||
const original = gateway.approveActionRequest.bind(gateway);
|
||||
const dispatch = vi.spyOn(gateway, "approveActionRequest").mockImplementation(async input => {
|
||||
if (input.companyId === fixture.company.id) throw new Error("Company dependency unavailable");
|
||||
return original(input);
|
||||
});
|
||||
await startRecovery();
|
||||
expect(dispatch).toHaveBeenCalledTimes(1);
|
||||
const retryAt = coordinator!.nextWakeAt();
|
||||
expect(retryAt).not.toBeNull();
|
||||
const later = await request(); await approve(later);
|
||||
const other = await createRunFixture(db);
|
||||
await db.insert(toolPolicies).values(reviewNotePolicyRow(other.company));
|
||||
const foreign = await request(gateway, db, other);
|
||||
await commitToolActionReview(db, { companyId: other.company.id, actionRequestId: foreign.id,
|
||||
decision: "approved", actor: { userId: "board" } });
|
||||
await settle(5000); await executed(foreign);
|
||||
expect(dispatch.mock.calls.filter(([input]) => input.companyId === fixture.company.id)).toHaveLength(1);
|
||||
expect(coordinator!.nextWakeAt()).toBe(retryAt);
|
||||
dispatch.mockRestore();
|
||||
await settle(); await executed(pending); await executed(later); await quiet();
|
||||
expect(await db.select().from(toolCallEvents).where(eq(toolCallEvents.reasonCode, "approved_action_executed"))).toHaveLength(3);
|
||||
});
|
||||
|
||||
it("disarms a failed company's recovery after company deletion commits", async () => {
|
||||
const pending = await request(); await approve(pending);
|
||||
const dispatch = vi.spyOn(gateway, "approveActionRequest").mockRejectedValue(new Error("Company dependency unavailable"));
|
||||
await startRecovery(); expect(coordinator!.nextWakeAt()).not.toBeNull();
|
||||
await db.update(heartbeatRuns).set({ status: "succeeded", finishedAt: new Date() })
|
||||
.where(eq(heartbeatRuns.id, fixture.run.id));
|
||||
await db.update(agents).set({ status: "terminated", lifecycleState: "terminated" })
|
||||
.where(eq(agents.id, fixture.agent.id));
|
||||
await deleteCompany(db, fixture.company.id);
|
||||
await settle(1500);
|
||||
expect(coordinator!.nextWakeAt()).toBeNull();
|
||||
expect(dispatch).toHaveBeenCalledTimes(1); await quiet();
|
||||
});
|
||||
|
||||
it.each([null, "unreadable-old-signature"])("settles an approved invalid signature (%s) without recurring recovery", async signedArguments => {
|
||||
const pending = await request(); await approve(pending);
|
||||
await db.update(toolActionRequests).set({ signedArguments }).where(eq(toolActionRequests.id, pending.id));
|
||||
const dispatch = vi.spyOn(gateway, "approveActionRequest");
|
||||
await startRecovery(); await settle();
|
||||
expect((await db.select().from(toolActionRequests))[0].status).toBe("cancelled");
|
||||
expect((await db.select().from(toolInvocations))[0]).toMatchObject({
|
||||
status: "failed", errorCode: "action_request_invalidated",
|
||||
});
|
||||
expect(dispatch).not.toHaveBeenCalled(); await quiet();
|
||||
});
|
||||
|
||||
it.each(["rejected", "cancelled"])("settles a %s formal approval without recurring recovery", async status => {
|
||||
const pending = await request(); await approve(pending);
|
||||
const [formal] = await db.insert(approvals).values({ companyId: fixture.company.id,
|
||||
type: "request_board_approval", status, payload: {} }).returning();
|
||||
await db.update(toolActionRequests).set({ approvalId: formal.id }).where(eq(toolActionRequests.id, pending.id));
|
||||
const dispatch = vi.spyOn(gateway, "approveActionRequest");
|
||||
await startRecovery(); await settle();
|
||||
expect((await db.select().from(toolActionRequests))[0].status).toBe("cancelled");
|
||||
expect((await db.select().from(toolInvocations))[0]).toMatchObject({
|
||||
status: "denied", errorCode: "formal_approval_unavailable",
|
||||
});
|
||||
expect(dispatch).not.toHaveBeenCalled(); await quiet();
|
||||
});
|
||||
|
||||
it.each([false, true])("parks a formal approval until its decision commits (rollback %s)", async rollback => {
|
||||
const pending = await request(); await approve(pending);
|
||||
const [formal] = await db.insert(approvals).values({ companyId: fixture.company.id,
|
||||
type: "request_board_approval", status: "pending", payload: {} }).returning();
|
||||
await db.update(toolActionRequests).set({ approvalId: formal.id }).where(eq(toolActionRequests.id, pending.id));
|
||||
await startRecovery(); await quiet();
|
||||
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("formal approval rollback fixture");
|
||||
const writing = db.transaction(async tx => {
|
||||
await approvalService(tx as unknown as typeof db).approve(formal.id, "board");
|
||||
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("approved");
|
||||
expect((await db.select().from(toolInvocations))[0].status).toBe("awaiting_approval");
|
||||
} finally { release(); }
|
||||
await writing; await settle();
|
||||
if (rollback) {
|
||||
expect((await db.select().from(toolActionRequests))[0].status).toBe("approved"); await quiet();
|
||||
await approvalService(db).approve(formal.id, "board"); await settle();
|
||||
}
|
||||
await executed(pending); await quiet();
|
||||
expect(await db.select().from(toolCallEvents).where(eq(toolCallEvents.reasonCode, "approved_action_executed"))).toHaveLength(1);
|
||||
});
|
||||
|
||||
it.each(["reject", "cancel"] as const)("wakes a parked revision-requested formal approval on %s", async decision => {
|
||||
const pending = await request(); await approve(pending);
|
||||
const [formal] = await db.insert(approvals).values({ companyId: fixture.company.id,
|
||||
type: "request_board_approval", status: "revision_requested", payload: {} }).returning();
|
||||
await db.update(toolActionRequests).set({ approvalId: formal.id }).where(eq(toolActionRequests.id, pending.id));
|
||||
await startRecovery(); await quiet();
|
||||
const service = approvalService(db);
|
||||
if (decision === "cancel") await service.cancel(formal.id, "Cancelled by user");
|
||||
else await service.reject(formal.id, "board");
|
||||
await settle();
|
||||
expect((await db.select().from(toolActionRequests))[0].status).toBe("cancelled");
|
||||
expect((await db.select().from(toolInvocations))[0]).toMatchObject({ status: "denied", errorCode: "formal_approval_unavailable" });
|
||||
await quiet();
|
||||
});
|
||||
|
||||
it("retains approved work when the signing key is temporarily unavailable", async () => {
|
||||
const pending = await request(); await approve(pending);
|
||||
vi.stubEnv("PAPERCLIP_TOOL_ACTION_SIGNING_SECRET", "");
|
||||
vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "invalid-master-key");
|
||||
gateway = createToolGatewayService(db);
|
||||
await startRecovery();
|
||||
expect(coordinator!.nextWakeAt()).not.toBeNull();
|
||||
expect((await db.select().from(toolActionRequests))[0].status).toBe("approved");
|
||||
expect((await db.select().from(toolInvocations))[0].status).toBe("awaiting_approval");
|
||||
vi.stubEnv("PAPERCLIP_TOOL_ACTION_SIGNING_SECRET", testToolActionSigningSecret);
|
||||
await settle(); await executed(pending); await quiet();
|
||||
});
|
||||
|
||||
it("does not accept another company's formal approval", async () => {
|
||||
const pending = await request(); await approve(pending);
|
||||
const other = await createRunFixture(db);
|
||||
const [formal] = await db.insert(approvals).values({ companyId: other.company.id,
|
||||
type: "request_board_approval", status: "approved", payload: {} }).returning();
|
||||
await db.update(toolActionRequests).set({ approvalId: formal.id }).where(eq(toolActionRequests.id, pending.id));
|
||||
await startRecovery(); await settle();
|
||||
expect((await db.select().from(toolActionRequests))[0].status).toBe("cancelled");
|
||||
expect((await db.select().from(toolInvocations))[0]).toMatchObject({ status: "denied", errorCode: "formal_approval_unavailable" });
|
||||
expect((await db.select().from(approvals))[0]).toEqual(formal);
|
||||
await quiet();
|
||||
});
|
||||
|
||||
it("quarantines a mismatched request without modifying another company's invocation", async () => {
|
||||
const pending = await request(); await approve(pending);
|
||||
const [invocation] = await db.select().from(toolInvocations);
|
||||
const other = await createRunFixture(db);
|
||||
await db.update(toolActionRequests).set({ companyId: other.company.id }).where(eq(toolActionRequests.id, pending.id));
|
||||
await startRecovery(); await settle();
|
||||
expect((await db.select().from(toolActionRequests))[0].status).toBe("failed");
|
||||
expect((await db.select().from(toolInvocations))[0]).toEqual(invocation);
|
||||
await quiet();
|
||||
});
|
||||
|
||||
it("repairs completed agent history without executing the action again", async () => {
|
||||
const pending = await request();
|
||||
await gateway.approveActionRequest({ companyId: fixture.company.id, actionRequestId: pending.id, actor: { userId: "board" } });
|
||||
await executed(pending);
|
||||
const [invocation] = await db.select().from(toolInvocations);
|
||||
await db.update(toolActionRequests).set({ status: "approved", signedArguments: "unreadable-old-signature" })
|
||||
.where(eq(toolActionRequests.id, pending.id));
|
||||
await startRecovery(); await settle(); await executed(pending); await quiet();
|
||||
expect((await db.select().from(toolInvocations))[0]).toEqual(invocation);
|
||||
expect(await db.select().from(toolCallEvents).where(eq(toolCallEvents.reasonCode, "approved_action_executed"))).toHaveLength(1);
|
||||
});
|
||||
|
||||
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"]>, rateLimited = false, parameters: Record<string, unknown> = { value: "reviewed" }) {
|
||||
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 } });
|
||||
if (rateLimited) await db.insert(toolPolicies).values({ companyId: fixture.company.id, name: "One reviewed test call",
|
||||
policyType: "rate_limit", priority: 1, selectors: { connectionId: remote.connection.id },
|
||||
config: { limit: 1, windowSeconds: 3600, keyBy: ["agent", "tool"] } });
|
||||
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 });
|
||||
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)("retries a Test-tab %s approval after a real pre-dispatch database timeout", async mode => {
|
||||
const remoteCall = providerFixture(); remoteCall.release();
|
||||
const { pending } = await testRequest(remoteCall.provider); await approve(pending);
|
||||
const preparationDb = createDb(externalDatabaseUrl ?? tempDb!.connectionString, {
|
||||
maxConnections: 1, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0,
|
||||
});
|
||||
const preparingGateway = createTestToolGatewayService(preparationDb, { remoteHttpRequest: remoteCall.provider });
|
||||
let unlock!: () => void, locked!: () => void;
|
||||
const gate = new Promise<void>(resolve => { unlock = resolve; });
|
||||
const acquired = new Promise<void>(resolve => { locked = resolve; });
|
||||
const lock = db.transaction(async tx => {
|
||||
await tx.execute(sql`lock table tool_policies in access exclusive mode`);
|
||||
locked(); await gate;
|
||||
});
|
||||
await acquired;
|
||||
try {
|
||||
await preparationDb.execute(sql`set lock_timeout = '2s'`);
|
||||
const attempting = mode === "foreground"
|
||||
? preparingGateway.approveActionRequest({ companyId: fixture.company.id, actionRequestId: pending.id, actor: { userId: "board" } })
|
||||
: preparingGateway.recoverApprovedActionReviews(fixture.company.id);
|
||||
// Observe the real committed claim while its subsequent policy read
|
||||
// is blocked. Provider execution has not started.
|
||||
void attempting.catch(() => undefined);
|
||||
await vi.waitFor(async () => {
|
||||
expect((await db.select().from(toolActionRequests).where(eq(toolActionRequests.id, pending.id)))[0].status).toBe("executing");
|
||||
});
|
||||
if (mode === "foreground") await expect(attempting).rejects.toMatchObject({ cause: { code: "55P03" } });
|
||||
else expect(await attempting).toEqual({ scanned: 1, retry: true });
|
||||
expect((await db.select().from(toolActionRequests).where(eq(toolActionRequests.id, pending.id)))[0])
|
||||
.toMatchObject({ status: "approved", signedArguments: pending.signedArguments });
|
||||
expect((await db.select().from(toolInvocations).where(eq(toolInvocations.id, pending.invocationId)))[0])
|
||||
.toMatchObject({ status: "awaiting_approval", startedAt: null });
|
||||
expect(remoteCall.provider).not.toHaveBeenCalled();
|
||||
} finally { unlock(); await lock; await preparationDb.$client.end({ timeout: 1 }); }
|
||||
await startRecovery(); await settle(); await executed(pending); await quiet();
|
||||
expect(remoteCall.provider).toHaveBeenCalledOnce();
|
||||
}, 15_000);
|
||||
|
||||
async function agentRequest(provider: NonNullable<ToolGatewayServiceOptions["remoteHttpRequest"]>, parameters: Record<string, unknown> = { value: "reviewed" }) {
|
||||
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 agent writes",
|
||||
policyType: "require_approval", selectors: { connectionId: remote.connection.id } });
|
||||
gateway = createTestToolGatewayService(db, { remoteHttpRequest: provider });
|
||||
const session = await gateway.createSession(runSessionInput(fixture.company, fixture.agent, fixture.run));
|
||||
const tool = (await gateway.listToolsForSession(session.token)).find(t => t.connectionId === remote.connection.id)!;
|
||||
await expect(gateway.executeTool({ sessionToken: session.token, tool: tool.name, parameters }))
|
||||
.rejects.toMatchObject({ reasonCode: "approval_required" });
|
||||
const [pending] = await db.select().from(toolActionRequests);
|
||||
return pending;
|
||||
}
|
||||
|
||||
it.each(["foreground", "recovery"] as const)("retries an agent-origin %s approval after a real pre-dispatch database timeout", async mode => {
|
||||
const remoteCall = providerFixture(); remoteCall.release();
|
||||
const pending = await agentRequest(remoteCall.provider); await approve(pending);
|
||||
const preparationDb = createDb(externalDatabaseUrl ?? tempDb!.connectionString, {
|
||||
maxConnections: 1, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0,
|
||||
});
|
||||
const preparingGateway = createTestToolGatewayService(preparationDb, { remoteHttpRequest: remoteCall.provider });
|
||||
let unlock!: () => void, locked!: () => void;
|
||||
const gate = new Promise<void>(resolve => { unlock = resolve; });
|
||||
const acquired = new Promise<void>(resolve => { locked = resolve; });
|
||||
const lock = db.transaction(async tx => {
|
||||
await tx.execute(sql`lock table tool_policies in access exclusive mode`);
|
||||
locked(); await gate;
|
||||
});
|
||||
await acquired;
|
||||
try {
|
||||
await preparationDb.execute(sql`set lock_timeout = '2s'`);
|
||||
const attempting = mode === "foreground"
|
||||
? preparingGateway.approveActionRequest({ companyId: fixture.company.id, actionRequestId: pending.id, actor: { userId: "board" } })
|
||||
: preparingGateway.recoverApprovedActionReviews(fixture.company.id);
|
||||
// Observe the real committed claim while its subsequent policy read
|
||||
// is blocked. Provider execution has not started.
|
||||
void attempting.catch(() => undefined);
|
||||
await vi.waitFor(async () => {
|
||||
expect((await db.select().from(toolActionRequests).where(eq(toolActionRequests.id, pending.id)))[0].status).toBe("executing");
|
||||
});
|
||||
if (mode === "foreground") expect(await attempting).toMatchObject({ status: "approved" });
|
||||
else expect(await attempting).toEqual({ scanned: 1, retry: true });
|
||||
expect((await db.select().from(toolActionRequests).where(eq(toolActionRequests.id, pending.id)))[0])
|
||||
.toMatchObject({ status: "approved", signedArguments: pending.signedArguments });
|
||||
expect((await db.select().from(toolInvocations).where(eq(toolInvocations.id, pending.invocationId)))[0])
|
||||
.toMatchObject({ status: "awaiting_approval", startedAt: null });
|
||||
expect(remoteCall.provider).not.toHaveBeenCalled();
|
||||
} finally { unlock(); await lock; await preparationDb.$client.end({ timeout: 1 }); }
|
||||
await startRecovery(); await settle(); await executed(pending); await quiet();
|
||||
expect(remoteCall.provider).toHaveBeenCalledOnce();
|
||||
}, 15_000);
|
||||
|
||||
it("refuses agent dispatch after preparation loses its claim", async () => {
|
||||
const remoteCall = providerFixture(); remoteCall.release();
|
||||
const pending = await agentRequest(remoteCall.provider); await approve(pending);
|
||||
let unlock!: () => void, locked!: () => void;
|
||||
const gate = new Promise<void>(resolve => { unlock = resolve; });
|
||||
const acquired = new Promise<void>(resolve => { locked = resolve; });
|
||||
const lock = db.transaction(async tx => {
|
||||
// Pause tool discovery before admission locks the invocation. Revocation
|
||||
// cannot legitimately take that lock from an in-flight admission.
|
||||
await tx.execute(sql`lock table tool_catalog_entries in access exclusive mode`);
|
||||
locked(); await gate;
|
||||
});
|
||||
await acquired;
|
||||
const attempting = gateway.approveActionRequest({ companyId: fixture.company.id,
|
||||
actionRequestId: pending.id, actor: { userId: "board" } });
|
||||
void attempting.catch(() => undefined);
|
||||
let successor!: NonNullable<Awaited<ReturnType<typeof claimToolTestApproval>>>;
|
||||
try {
|
||||
await vi.waitFor(async () => {
|
||||
expect((await db.select().from(toolActionRequests).where(eq(toolActionRequests.id, pending.id)))[0].status).toBe("executing");
|
||||
});
|
||||
const [claimed] = await db.select().from(toolActionRequests).where(eq(toolActionRequests.id, pending.id));
|
||||
expect(await releaseToolApproval(db, { actionRequestId: pending.id,
|
||||
invocationId: pending.invocationId, claimedAt: claimed.updatedAt })).toBe(true);
|
||||
await vi.advanceTimersByTimeAsync(10);
|
||||
successor = (await claimToolTestApproval(db, pending.id, pending.invocationId))!;
|
||||
expect(successor).not.toBeNull();
|
||||
expect(successor.claimedAt.getTime()).toBeGreaterThan(claimed.updatedAt.getTime());
|
||||
} finally { unlock(); await lock; }
|
||||
expect(await attempting).toMatchObject({ status: "executing" });
|
||||
expect(remoteCall.provider).not.toHaveBeenCalled();
|
||||
expect((await db.select().from(toolInvocations).where(eq(toolInvocations.id, pending.invocationId)))[0])
|
||||
.toMatchObject({ status: "awaiting_approval", startedAt: null });
|
||||
expect(await releaseToolApproval(db, successor)).toBe(true);
|
||||
await startRecovery(); await settle(); await executed(pending); await quiet();
|
||||
expect(remoteCall.provider).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("does not retry an agent provider error as a preparation failure", async () => {
|
||||
const provider = vi.fn(async () => { throw new Error("Provider outcome is unknown"); });
|
||||
const pending = await agentRequest(provider); await approve(pending);
|
||||
await gateway.approveActionRequest({ companyId: fixture.company.id, actionRequestId: pending.id, actor: { userId: "board" } });
|
||||
expect((await db.select().from(toolActionRequests).where(eq(toolActionRequests.id, pending.id)))[0].status).toBe("failed");
|
||||
expect((await db.select().from(toolInvocations).where(eq(toolInvocations.id, pending.invocationId)))[0].startedAt).not.toBeNull();
|
||||
await startRecovery(); await quiet();
|
||||
expect(provider).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it.each(["started", "reclaimed", "mismatched", "prior_dispatch"] as const)("refuses to release a Test-tab preparation claim with %s ownership evidence", async state => {
|
||||
const remoteCall = providerFixture(); remoteCall.release();
|
||||
const { pending } = await testRequest(remoteCall.provider); await approve(pending);
|
||||
const claim = await claimToolTestApproval(db, pending.id, pending.invocationId);
|
||||
expect(claim).not.toBeNull();
|
||||
if (state === "started") expect(await startToolInvocation(db, pending.invocationId, claim!)).toBe(true);
|
||||
if (state === "reclaimed") await db.update(toolActionRequests)
|
||||
.set({ updatedAt: new Date(claim!.claimedAt.getTime() + 1000) }).where(eq(toolActionRequests.id, pending.id));
|
||||
if (state === "prior_dispatch") await db.update(toolInvocations)
|
||||
.set({ startedAt: new Date() }).where(eq(toolInvocations.id, pending.invocationId));
|
||||
const beforeRequest = (await db.select().from(toolActionRequests).where(eq(toolActionRequests.id, pending.id)))[0];
|
||||
const beforeInvocation = (await db.select().from(toolInvocations).where(eq(toolInvocations.id, pending.invocationId)))[0];
|
||||
expect(await releaseToolApproval(db, state === "mismatched" ? { ...claim!, invocationId: randomUUID() } : claim!)).toBe(false);
|
||||
expect((await db.select().from(toolActionRequests).where(eq(toolActionRequests.id, pending.id)))[0]).toEqual(beforeRequest);
|
||||
expect((await db.select().from(toolInvocations).where(eq(toolInvocations.id, pending.invocationId)))[0]).toEqual(beforeInvocation);
|
||||
expect(remoteCall.provider).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("rolls a Test-tab preparation release back with its transaction and invalidates the released claim", async () => {
|
||||
const remoteCall = providerFixture(); remoteCall.release();
|
||||
const { pending } = await testRequest(remoteCall.provider); await approve(pending);
|
||||
const claim = await claimToolTestApproval(db, pending.id, pending.invocationId);
|
||||
expect(claim).not.toBeNull();
|
||||
await expect(db.transaction(async tx => {
|
||||
expect(await releaseToolApproval(tx as unknown as typeof db, claim!)).toBe(true);
|
||||
throw new Error("release rollback");
|
||||
})).rejects.toThrow("release rollback");
|
||||
expect((await db.select().from(toolActionRequests).where(eq(toolActionRequests.id, pending.id)))[0].status).toBe("executing");
|
||||
expect(await releaseToolApproval(db, claim!)).toBe(true);
|
||||
expect(await startToolInvocation(db, pending.invocationId, claim!)).toBe(false);
|
||||
await startRecovery(); await settle(); await executed(pending); await quiet();
|
||||
expect(remoteCall.provider).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
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.each(["foreground", "recovery"] as const)("preserves the admitted rate slot for a Test-tab action through %s approval", async mode => {
|
||||
const remoteCall = providerFixture(); remoteCall.release();
|
||||
const { pending } = await testRequest(remoteCall.provider, true);
|
||||
const counters = await db.select().from(toolRateLimitCounters);
|
||||
expect(counters).toHaveLength(1); expect(counters[0].remaining).toBe(0);
|
||||
if (mode === "foreground") {
|
||||
await gateway.approveActionRequest({ companyId: fixture.company.id, actionRequestId: pending.id, actor: { userId: "board" } });
|
||||
await startRecovery();
|
||||
} else { await approve(pending); await startRecovery(); }
|
||||
await settle(); await executed(pending); await quiet();
|
||||
expect(await db.select().from(toolRateLimitCounters)).toEqual(counters);
|
||||
expect(remoteCall.provider).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
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.each(["pending", "authorized", "cancelled"])("settles inconsistent Test-tab %s history without replay", async status => {
|
||||
const remoteCall = providerFixture(); remoteCall.release();
|
||||
const { pending } = await testRequest(remoteCall.provider); await approve(pending);
|
||||
await db.update(toolInvocations).set({ status }).where(eq(toolInvocations.id, pending.invocationId));
|
||||
await startRecovery(); await settle();
|
||||
expect((await db.select().from(toolActionRequests))[0].status).toBe(status === "cancelled" ? "cancelled" : "failed");
|
||||
expect((await db.select().from(toolInvocations))[0]).toMatchObject(status === "cancelled"
|
||||
? { status: "cancelled" }
|
||||
: { status: "failed", errorCode: "tool_recovery_state_invalid" });
|
||||
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.each(["always", "once"])("does not replay a Test-tab call when persisting its external result fails %s", async mode => {
|
||||
const remoteCall = providerFixture(); remoteCall.release();
|
||||
const { pending } = await testRequest(remoteCall.provider);
|
||||
await approve(pending);
|
||||
await db.execute(sql`create sequence sleep_review_failure_attempt`);
|
||||
await db.execute(sql.raw(`create function sleep_review_fail() returns trigger language plpgsql as $$
|
||||
begin
|
||||
if ${mode === "always" ? "true" : "nextval('sleep_review_failure_attempt') = 1"} then
|
||||
raise exception 'approval result fixture failure';
|
||||
end if;
|
||||
return new;
|
||||
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();
|
||||
});
|
||||
});
|
||||
|
||||
async function requestWithExpiry(expiresAt: Date, context = fixture) {
|
||||
return db.transaction(async tx => {
|
||||
const source = tx as unknown as typeof db;
|
||||
@@ -876,13 +1585,21 @@ 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) => ({
|
||||
const [originalInvocation] = await db.select().from(toolInvocations).where(eq(toolInvocations.id, request.invocationId));
|
||||
const blockedInvocations = Array.from({ length: 101 }, (_, i) => ({ ...originalInvocation,
|
||||
id: `10000000-0000-4000-8000-${String(i).padStart(12, "0")}`, approvalState: "approved", idempotencyKey: null,
|
||||
}));
|
||||
await db.insert(toolInvocations).values(blockedInvocations);
|
||||
await db.insert(toolActionRequests).values(blockedInvocations.map((invocation, i) => ({
|
||||
...request, id: `00000000-0000-4000-8000-${String(i).padStart(12, "0")}`,
|
||||
interactionId: null, status: "approved" as const,
|
||||
invocationId: invocation.id, interactionId: null, status: "approved" as const,
|
||||
signedArguments: signToolArguments({ invocationId: invocation.id, toolName: invocation.toolName,
|
||||
canonicalArguments: canonicalToolArguments({ noteId: "blocked" }), executionOnApprove: true,
|
||||
signingSecret: testToolActionSigningSecret }),
|
||||
})));
|
||||
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
@@ -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";
|
||||
@@ -1472,7 +1473,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,
|
||||
@@ -1903,7 +1910,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) {
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { writeToolReviewChange } from "./tool-review-work-notifications.js";
|
||||
import { approvalResolutionRecords, resolvableApprovalStatuses as resolvableStatuses } from "../lib/approval-records.js";
|
||||
import { notifyHireApproved } from "./hire-hook.js";
|
||||
|
||||
@@ -12,7 +13,15 @@ import { instanceSettingsService } from "./instance-settings.js";
|
||||
|
||||
export function approvalRecords(db: Db) {
|
||||
const instanceSettings = instanceSettingsService(db);
|
||||
const { getExistingApproval, resolveApproval } = approvalResolutionRecords(db);
|
||||
const { getExistingApproval, resolveApproval: resolveApprovalRecord } = approvalResolutionRecords(db);
|
||||
|
||||
// Tool reviews waiting for a formal decision stay parked until this producer
|
||||
// commits. The shared record owner retains approval validation and CAS rules.
|
||||
async function resolveApproval(id: string, status: "approved" | "rejected", userId: string,
|
||||
note: string | null | undefined, database: Db = db) {
|
||||
const existing = await getExistingApproval(id, database);
|
||||
return writeToolReviewChange(database, existing.companyId, tx => resolveApprovalRecord(id, status, userId, note, tx));
|
||||
}
|
||||
|
||||
function redactApprovalComment<T extends { body: string }>(comment: T, censorUsernameInLogs: boolean): T {
|
||||
return {
|
||||
@@ -64,7 +73,9 @@ export function approvalRecords(db: Db) {
|
||||
// cleanup. Idempotent: a no-op on already-resolved approvals.
|
||||
cancel: async (id: string, reason?: string | null) => {
|
||||
const now = new Date();
|
||||
const updated = await db
|
||||
const [existing] = await db.select({ companyId: approvals.companyId }).from(approvals).where(eq(approvals.id, id));
|
||||
if (!existing) return null;
|
||||
const updated = await writeToolReviewChange(db, existing.companyId, tx => tx
|
||||
.update(approvals)
|
||||
.set({
|
||||
status: "cancelled",
|
||||
@@ -74,7 +85,7 @@ export function approvalRecords(db: Db) {
|
||||
})
|
||||
.where(and(eq(approvals.id, id), inArray(approvals.status, resolvableStatuses)))
|
||||
.returning()
|
||||
.then((rows) => rows[0] ?? null);
|
||||
.then((rows) => rows[0] ?? null));
|
||||
return updated;
|
||||
},
|
||||
|
||||
|
||||
@@ -9,6 +9,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",
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
import type { ConnectionManagementContext } from "./connection-management.js";
|
||||
import { createHash } from "node:crypto";
|
||||
import { boundRunIssueIdSql } from "./run-task-binding.js";
|
||||
import { and, asc, desc, eq, gt, inArray, isNull, ne, sql } from "drizzle-orm";
|
||||
import type { ToolApprovalClaim } from "./tool-approval-execution.js";
|
||||
import { and, asc, desc, eq, inArray, isNull, ne, sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import {
|
||||
agents,
|
||||
@@ -53,6 +54,12 @@ import {
|
||||
import { recordToolRuntimeAuditWriteFailure } from "./tool-runtime-metrics.js";
|
||||
import { canonicalToolArguments, readSignedToolArgumentsPayload, toolRequestIdentityHash } from "./tool-content-guards.js";
|
||||
|
||||
class RejectedInvocationAdmission extends Error {
|
||||
constructor(readonly decision: ToolAccessDecision) {
|
||||
super(decision.explanation);
|
||||
}
|
||||
}
|
||||
|
||||
type ToolAccessContext = {
|
||||
companyId: string;
|
||||
actorType: "agent" | "user" | "system" | "plugin";
|
||||
@@ -717,7 +724,7 @@ function assertSupportedGenericPolicyShape(
|
||||
}
|
||||
}
|
||||
|
||||
async function getGenericPolicyRow(db: Db, companyId: string, policyId: string) {
|
||||
async function getGenericPolicyRow(db: Pick<Db, "select">, companyId: string, policyId: string) {
|
||||
const [policy] = await db
|
||||
.select()
|
||||
.from(toolPolicies)
|
||||
@@ -785,7 +792,7 @@ function scopeAllowsTool(scope: Record<string, unknown> | null, ctx: ToolAccessC
|
||||
return selectorMatches(selectors, ctx);
|
||||
}
|
||||
|
||||
export function toolAccessPolicyService(db: Db, options: { toolActionSigningSecret?: string } = {}) {
|
||||
export function toolAccessPolicyService(db: Pick<Db, "select" | "insert" | "update" | "delete" | "transaction">, options: { toolActionSigningSecret?: string } = {}) {
|
||||
// Policy administration includes authorized self-targeted changes. It cannot
|
||||
// resolve governed action requests or modify principal permission grants.
|
||||
async function listPolicies(companyId: string) {
|
||||
@@ -1301,11 +1308,12 @@ export function toolAccessPolicyService(db: Db, options: { toolActionSigningSecr
|
||||
],
|
||||
set: {
|
||||
limit: rule.limit,
|
||||
remaining: sql`greatest(0, least(${toolRateLimitCounters.remaining}, ${rule.limit}) - 1)`,
|
||||
// Preserve the number already spent when an operator changes the limit.
|
||||
remaining: sql`greatest(0, ${rule.limit} - (${toolRateLimitCounters.limit} - ${toolRateLimitCounters.remaining}) - 1)`,
|
||||
resetAt,
|
||||
updatedAt: now,
|
||||
},
|
||||
setWhere: gt(toolRateLimitCounters.remaining, 0),
|
||||
setWhere: sql`${toolRateLimitCounters.limit} - ${toolRateLimitCounters.remaining} < ${rule.limit}`,
|
||||
})
|
||||
.returning({ remaining: toolRateLimitCounters.remaining });
|
||||
if (!counter) {
|
||||
@@ -1376,7 +1384,7 @@ export function toolAccessPolicyService(db: Db, options: { toolActionSigningSecr
|
||||
});
|
||||
}
|
||||
|
||||
async function admittedRateLimitAdmissions(ctx: ToolAccessContext, invocationId: string): Promise<ConsumedRateAdmission[]> {
|
||||
async function admittedRateLimitAdmissions(ctx: ToolAccessContext, invocationId: string, previewHash?: string): Promise<ConsumedRateAdmission[] | null> {
|
||||
const [invocation] = await db.select({
|
||||
actorType: toolInvocations.actorType, actorId: toolInvocations.actorId, agentId: toolInvocations.agentId,
|
||||
runId: toolInvocations.runId, issueId: toolInvocations.issueId, gatewayId: toolInvocations.gatewayId,
|
||||
@@ -1387,6 +1395,9 @@ export function toolAccessPolicyService(db: Db, options: { toolActionSigningSecr
|
||||
status: toolInvocations.status, policyDecision: toolInvocations.policyDecision, approvalState: toolInvocations.approvalState,
|
||||
policyExplanation: toolInvocations.policyExplanation,
|
||||
}).from(toolInvocations).where(and(eq(toolInvocations.companyId, ctx.companyId), eq(toolInvocations.id, invocationId)));
|
||||
// A parked request already spent its slot before asking for review. Reusing
|
||||
// that admission only skips the same rate rule; current policy and the
|
||||
// gateway's approval/claim owners still decide whether it may execute.
|
||||
const delegated = invocation?.policyExplanation?.delegatedConnectionTest as { agentId?: string } | undefined;
|
||||
const gatewayIssuer = invocation?.gatewayId ? invocation.policyExplanation?.delegatedGatewayIssuer : undefined;
|
||||
const isDelegatedCaller = invocation?.actorType === "agent" && invocation.actorId === ctx.agentId
|
||||
@@ -1401,13 +1412,81 @@ export function toolAccessPolicyService(db: Db, options: { toolActionSigningSecr
|
||||
&& invocation.catalogEntryId === ctx.catalogEntryId && invocation.toolName === ctx.toolName
|
||||
&& invocation.providerType === ctx.providerType && invocation.applicationKey === ctx.applicationKey
|
||||
&& invocation.upstreamToolName === ctx.upstreamToolName
|
||||
&& invocation.argumentsHash === ctx.argumentsHash;
|
||||
const stored = authorized ? invocation.policyExplanation?.[isDelegatedCaller ? "delegatedCallerConsumedRateAdmissions" : "consumedRateAdmissions"] : null;
|
||||
&& (invocation.argumentsHash === ctx.argumentsHash || invocation.argumentsHash === previewHash);
|
||||
if (!authorized) return null;
|
||||
const stored = invocation.policyExplanation?.[isDelegatedCaller ? "delegatedCallerConsumedRateAdmissions" : "consumedRateAdmissions"];
|
||||
return Array.isArray(stored) ? stored.filter((item): item is ConsumedRateAdmission => isRecord(item)
|
||||
&& typeof item.policyId === "string" && CANONICAL_UUID_RE.test(item.policyId)
|
||||
&& typeof item.versionHash === "string" && /^[a-f0-9]{64}$/.test(item.versionHash)) : [];
|
||||
}
|
||||
|
||||
/** Revalidate claimed work and persist new rate charges before provider dispatch. */
|
||||
async function admitInvocation(
|
||||
input: ToolAccessDecisionInput,
|
||||
invocationId: string,
|
||||
claim?: ToolApprovalClaim,
|
||||
evaluate?: (
|
||||
policy: Pick<ReturnType<typeof toolAccessPolicyService>, "decide" | "writeAudit">,
|
||||
input: ToolAccessDecisionInput,
|
||||
transaction: Db,
|
||||
) => Promise<ToolAccessDecision>,
|
||||
): Promise<ToolAccessDecision> {
|
||||
try {
|
||||
return await db.transaction(async tx => {
|
||||
const [invocation] = await tx.select().from(toolInvocations)
|
||||
.where(and(eq(toolInvocations.companyId, input.companyId), eq(toolInvocations.id, invocationId)))
|
||||
.for("update");
|
||||
// Historical reviewed requests can still have a pending invocation.
|
||||
// Accept that state only with the exact executing claim owned by the
|
||||
// gateway, locked in the same invocation-then-request order as dispatch.
|
||||
let ownsClaim = false;
|
||||
if (claim && claim.invocationId === invocationId) {
|
||||
const [request] = await tx.select().from(toolActionRequests).where(and(
|
||||
eq(toolActionRequests.id, claim.actionRequestId),
|
||||
eq(toolActionRequests.invocationId, invocationId),
|
||||
eq(toolActionRequests.companyId, input.companyId),
|
||||
)).for("update");
|
||||
ownsClaim = request?.status === "executing" && request.updatedAt.getTime() === claim.claimedAt.getTime();
|
||||
}
|
||||
if (!invocation || (claim && !ownsClaim)
|
||||
|| !(invocation.status === "executing" && invocation.policyDecision === "allow"
|
||||
|| ["awaiting_approval", "executing"].includes(invocation.status)
|
||||
&& invocation.policyDecision === "require_approval" && invocation.approvalState === "approved"
|
||||
|| invocation.status === "awaiting_approval" && invocation.policyDecision === "require_approval"
|
||||
&& invocation.approvalState === "pending" && ownsClaim)) {
|
||||
return decision("deny", "deny_policy_block", "Tool invocation admission is unavailable.", [], []);
|
||||
}
|
||||
const admissionPolicy = toolAccessPolicyService(tx, options);
|
||||
const admissionInput = { ...input, consumeRateLimit: true };
|
||||
// Composed caller/target policy must share this claim transaction so
|
||||
// either all new charges and receipts commit or all roll back.
|
||||
const result = evaluate
|
||||
? await evaluate(admissionPolicy, admissionInput, tx as unknown as Db)
|
||||
: await admissionPolicy.decide(admissionInput, { admittedInvocationId: invocationId });
|
||||
if (!result.allowed && result.decision !== "require_approval") {
|
||||
// A later policy can reject after an earlier rate rule spent a slot.
|
||||
// Roll back both the new charges and their receipt on any rejection.
|
||||
throw new RejectedInvocationAdmission(result);
|
||||
}
|
||||
const admissions = result.policyExplanation?.consumedRateAdmissions;
|
||||
const delegatedAdmissions = result.policyExplanation?.delegatedCallerConsumedRateAdmissions;
|
||||
if (Array.isArray(admissions) || Array.isArray(delegatedAdmissions)) {
|
||||
await tx.update(toolInvocations).set({
|
||||
policyExplanation: {
|
||||
...invocation.policyExplanation,
|
||||
...(Array.isArray(admissions) ? { consumedRateAdmissions: admissions } : {}),
|
||||
...(Array.isArray(delegatedAdmissions) ? { delegatedCallerConsumedRateAdmissions: delegatedAdmissions } : {}),
|
||||
},
|
||||
}).where(and(eq(toolInvocations.companyId, input.companyId), eq(toolInvocations.id, invocationId)));
|
||||
}
|
||||
return result;
|
||||
});
|
||||
} catch (error) {
|
||||
if (error instanceof RejectedInvocationAdmission) return error.decision;
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Decides one tool access request. Pass the same `cache` to every call of a
|
||||
* batch that shares one actor and run context (for example, one tools/list
|
||||
@@ -1415,7 +1494,7 @@ export function toolAccessPolicyService(db: Db, options: { toolActionSigningSecr
|
||||
*/
|
||||
async function decide(
|
||||
input: ToolAccessDecisionInput,
|
||||
options: { cache?: ToolAccessDecisionCache; admittedInvocationId?: string; sharedRateDecision?: ToolAccessDecision } = {},
|
||||
options: { cache?: ToolAccessDecisionCache; admittedInvocationId?: string; admittedPreviewHash?: string; sharedRateDecision?: ToolAccessDecision } = {},
|
||||
): Promise<ToolAccessDecision> {
|
||||
const consumedRateAdmissions: ConsumedRateAdmission[] = [];
|
||||
const result = await decideInternal(input, options, consumedRateAdmissions);
|
||||
@@ -1426,7 +1505,7 @@ export function toolAccessPolicyService(db: Db, options: { toolActionSigningSecr
|
||||
|
||||
async function decideInternal(
|
||||
input: ToolAccessDecisionInput,
|
||||
options: { cache?: ToolAccessDecisionCache; admittedInvocationId?: string; sharedRateDecision?: ToolAccessDecision },
|
||||
options: { cache?: ToolAccessDecisionCache; admittedInvocationId?: string; admittedPreviewHash?: string; sharedRateDecision?: ToolAccessDecision },
|
||||
consumedRateAdmissions: ConsumedRateAdmission[],
|
||||
): Promise<ToolAccessDecision> {
|
||||
// A decision that consumes a rate limit can also record a trust-rule hit
|
||||
@@ -1438,9 +1517,14 @@ export function toolAccessPolicyService(db: Db, options: { toolActionSigningSecr
|
||||
// A final handoff rechecks current permissions for work already admitted by
|
||||
// the gateway. Its own last consumed slot cannot reject that same work.
|
||||
// Only durable consumption of an unchanged rule retains that slot; new or
|
||||
// edited rules still apply. This option never approves or consumes a call.
|
||||
const admittedRates = input.consumeRateLimit === false && options.admittedInvocationId
|
||||
? await admittedRateLimitAdmissions(ctx, options.admittedInvocationId) : [];
|
||||
// edited rules still apply. admitInvocation owns atomic consuming rechecks;
|
||||
// ordinary preflight and browser handoff callers remain read-only.
|
||||
const admittedRates = options.admittedInvocationId
|
||||
? await admittedRateLimitAdmissions(ctx, options.admittedInvocationId, input.consumeRateLimit === true ? undefined : options.admittedPreviewHash) : [];
|
||||
if (options.admittedInvocationId && input.consumeRateLimit === true && admittedRates === null) {
|
||||
return decision("deny", "deny_policy_block", "Tool invocation no longer matches this admission.", [], []);
|
||||
}
|
||||
if (input.consumeRateLimit === true) consumedRateAdmissions.push(...(admittedRates ?? []));
|
||||
const profileState = await effectiveProfiles(ctx, cache);
|
||||
const effectiveProfileIds = profileState.profiles.map((profile) => profile.id);
|
||||
const permittedByProfile = profileState.profiles.some((profile) => {
|
||||
@@ -1513,7 +1597,7 @@ export function toolAccessPolicyService(db: Db, options: { toolActionSigningSecr
|
||||
);
|
||||
}
|
||||
const versionHash = ratePolicyVersionHash(policy);
|
||||
if (admittedRates.some(admission => admission.policyId === policy.id && admission.versionHash === versionHash)) continue;
|
||||
if (admittedRates?.some(admission => admission.policyId === policy.id && admission.versionHash === versionHash)) continue;
|
||||
// Paired caller/target checks for one delegated test share a slot only
|
||||
// when the exact policy version and counter bucket are identical.
|
||||
const shared = options.sharedRateDecision?.policyExplanation?.consumedRateAdmissions;
|
||||
@@ -1548,7 +1632,7 @@ export function toolAccessPolicyService(db: Db, options: { toolActionSigningSecr
|
||||
{ redactionPlan: redaction.redactionPlan, policyExplanation },
|
||||
);
|
||||
}
|
||||
if (input.consumeRateLimit === true) {
|
||||
if (input.consumeRateLimit === true && !options.admittedInvocationId) {
|
||||
await recordTrustRuleHit(policy, ctx, redaction);
|
||||
}
|
||||
return decision("allow", "allow_trust_rule", policy.description ?? "Tool access allowed by trust rule.", effectiveProfileIds, [policy.id], { redactionPlan: redaction.redactionPlan, policyExplanation });
|
||||
@@ -2178,6 +2262,7 @@ export function toolAccessPolicyService(db: Db, options: { toolActionSigningSecr
|
||||
|
||||
return {
|
||||
decide,
|
||||
admitInvocation,
|
||||
writeAudit,
|
||||
recordInvocation,
|
||||
summarizeAndRedact,
|
||||
|
||||
@@ -0,0 +1,119 @@
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import { toolActionRequests, toolInvocations, type Db } from "@paperclipai/db";
|
||||
import type { ToolRedactedValueSummary } from "@paperclipai/shared";
|
||||
import { notifyToolReviewChange } from "./tool-review-work-notifications.js";
|
||||
|
||||
export type ToolApprovalClaim = { 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<ToolApprovalClaim | null> {
|
||||
return db.transaction(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;
|
||||
await notifyToolReviewChange(tx, request.companyId);
|
||||
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 };
|
||||
});
|
||||
}
|
||||
|
||||
/** Return only this still-undispatched claim to its existing approval queue. */
|
||||
export async function releaseToolApproval(db: Db, claim: ToolApprovalClaim) {
|
||||
return db.transaction(async tx => {
|
||||
const [invocation] = await tx.select().from(toolInvocations)
|
||||
.where(eq(toolInvocations.id, claim.invocationId)).for("update");
|
||||
const [request] = await tx.select().from(toolActionRequests)
|
||||
.where(and(eq(toolActionRequests.id, claim.actionRequestId),
|
||||
eq(toolActionRequests.invocationId, claim.invocationId))).for("update");
|
||||
if (invocation?.status !== "awaiting_approval" || invocation.startedAt !== null ||
|
||||
request?.status !== "executing" || request.companyId !== invocation.companyId ||
|
||||
request.updatedAt.getTime() !== claim.claimedAt.getTime()) return false;
|
||||
await notifyToolReviewChange(tx, request.companyId);
|
||||
await tx.update(toolActionRequests).set({ status: "approved", updatedAt: new Date() })
|
||||
.where(eq(toolActionRequests.id, request.id));
|
||||
return true;
|
||||
});
|
||||
}
|
||||
|
||||
export async function bindVerifiedToolApprovalArguments(db: Db, claim: ToolApprovalClaim, hashes: { preview: string; verified: string }) {
|
||||
return db.transaction(async tx => {
|
||||
const [invocation] = await tx.select().from(toolInvocations)
|
||||
.where(eq(toolInvocations.id, claim.invocationId)).for("update");
|
||||
const [request] = await tx.select().from(toolActionRequests)
|
||||
.where(and(eq(toolActionRequests.id, claim.actionRequestId),
|
||||
eq(toolActionRequests.invocationId, claim.invocationId))).for("update");
|
||||
if (invocation?.status !== "awaiting_approval" || invocation.startedAt !== null ||
|
||||
request?.status !== "executing" || request.companyId !== invocation.companyId ||
|
||||
request.updatedAt.getTime() !== claim.claimedAt.getTime() ||
|
||||
![hashes.preview, hashes.verified].includes(request.canonicalArgumentsHash) ||
|
||||
![hashes.preview, hashes.verified].includes(invocation.argumentsHash ?? "")) return false;
|
||||
if (invocation.argumentsHash !== hashes.verified) await tx.update(toolInvocations)
|
||||
.set({ argumentsHash: hashes.verified, updatedAt: new Date() }).where(eq(toolInvocations.id, claim.invocationId));
|
||||
return true;
|
||||
});
|
||||
}
|
||||
|
||||
export async function startToolInvocation(db: Db, invocationId: string, claim?: ToolApprovalClaim) {
|
||||
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().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 (invocationId !== claim.invocationId || invocation?.status !== "awaiting_approval" || invocation.startedAt !== null ||
|
||||
request?.status !== "executing" || request.invocationId !== invocationId || request.companyId !== invocation.companyId ||
|
||||
request.updatedAt.getTime() !== claim.claimedAt.getTime()) return false;
|
||||
await tx.update(toolInvocations).set({ status: "executing", approvalState: "approved", startedAt: now, updatedAt: now })
|
||||
.where(eq(toolInvocations.id, invocationId));
|
||||
return true;
|
||||
});
|
||||
}
|
||||
|
||||
export async function persistToolTestOutcome(db: Db, invocationId: string, outcome: ToolTestOutcome, claim?: ToolApprovalClaim) {
|
||||
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 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 || !["awaiting_approval", "executing"].includes(invocation.status) ||
|
||||
request?.status !== "executing" || request.updatedAt.getTime() !== claim.claimedAt.getTime()) return false;
|
||||
await notifyToolReviewChange(tx, request.companyId);
|
||||
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;
|
||||
});
|
||||
}
|
||||
+402
-447
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,33 @@
|
||||
import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
|
||||
import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
|
||||
import { registerScopedDeadlineWork } from "./scoped-deadline-work.js";
|
||||
import type { createToolGatewayService } from "./tool-gateway.js";
|
||||
|
||||
export function registerToolReviewRecoveryWork(
|
||||
coordinator: ReturnType<typeof createDeliveryWorkCoordinator>,
|
||||
gateway: Pick<ReturnType<typeof createToolGatewayService>, "recoverApprovedActionReviews" | "approvedActionReviewCompanies">,
|
||||
canRun: () => boolean,
|
||||
) {
|
||||
// Dispatch failures are reported by the domain owner, not thrown. Preserve
|
||||
// their retry deadline when another write reloads the same company's work.
|
||||
const retries = new Map<string, number>();
|
||||
return registerScopedDeadlineWork(coordinator, {
|
||||
queue: DELIVERY_QUEUES.toolReviewRecovery,
|
||||
batchSize: 1,
|
||||
decodeScope: hint => hint === "*" ? undefined : hint,
|
||||
matches: (entry, companyId?: string) => companyId === undefined || entry.companyId === companyId,
|
||||
load: async (companyId?: string) => {
|
||||
const rows = await gateway.approvedActionReviewCompanies(companyId);
|
||||
const retained = new Set(rows.map(row => row.companyId));
|
||||
for (const id of retries.keys()) if ((companyId === undefined || id === companyId) && !retained.has(id)) retries.delete(id);
|
||||
return rows.map(row => ({ ...row, id: row.companyId, dueAt: retries.get(row.companyId) ?? Date.now() }));
|
||||
},
|
||||
check: async entry => {
|
||||
const result = await gateway.recoverApprovedActionReviews(entry.companyId);
|
||||
if (!result.retry) { retries.delete(entry.companyId); return null; }
|
||||
const dueAt = Date.now() + 30_000;
|
||||
retries.set(entry.companyId, dueAt);
|
||||
return { ...entry, dueAt };
|
||||
},
|
||||
}, canRun);
|
||||
}
|
||||
@@ -0,0 +1,119 @@
|
||||
import { and, asc, eq, gt } from "drizzle-orm";
|
||||
import { approvals, toolActionRequests, toolInvocations, type Db } from "@paperclipai/db";
|
||||
|
||||
import { readSignedToolArgumentsPayload } from "./tool-content-guards.js";
|
||||
import { notifyToolReviewChange } from "./tool-review-work-notifications.js";
|
||||
|
||||
type ReviewProjection = {
|
||||
actionRequestId: string;
|
||||
status: "executed" | "executing" | "failed" | "cancelled";
|
||||
errorCode?: string | null;
|
||||
errorMessage?: string | null;
|
||||
resultSummary?: string | null;
|
||||
};
|
||||
|
||||
/** Reconcile saved history before dispatch; never infer a safe provider replay. */
|
||||
export async function prepareApprovedToolReview(db: Db, input: {
|
||||
actionRequestId: string;
|
||||
invocationId: string;
|
||||
signingSecret?: string;
|
||||
}) {
|
||||
return db.transaction(async tx => {
|
||||
// Match Test-tab claims/outcome persistence. Recheck after the locks because
|
||||
// a foreground approval may have claimed the row since the recovery scan.
|
||||
const [invocation] = await tx.select().from(toolInvocations)
|
||||
.where(eq(toolInvocations.id, input.invocationId)).for("update");
|
||||
const [request] = await tx.select().from(toolActionRequests)
|
||||
.where(and(eq(toolActionRequests.id, input.actionRequestId),
|
||||
eq(toolActionRequests.invocationId, input.invocationId))).for("update");
|
||||
if (!invocation || request?.status !== "approved") return { kind: "skip" } as const;
|
||||
const now = new Date();
|
||||
|
||||
async function settle(projection: ReviewProjection, failure?: {
|
||||
status: "failed" | "denied";
|
||||
errorCode: string;
|
||||
errorMessage: string;
|
||||
}) {
|
||||
await notifyToolReviewChange(tx, request.companyId);
|
||||
if (failure) await tx.update(toolInvocations).set({ ...failure, completedAt: now, updatedAt: now })
|
||||
.where(eq(toolInvocations.id, invocation.id));
|
||||
await tx.update(toolActionRequests).set({ status: projection.status,
|
||||
...(projection.status === "executing"
|
||||
? { updatedAt: invocation.startedAt ?? invocation.updatedAt }
|
||||
: { resolvedAt: invocation.completedAt ?? now, updatedAt: now }),
|
||||
}).where(eq(toolActionRequests.id, request.id));
|
||||
return { kind: "settled", projection } as const;
|
||||
}
|
||||
function fail(status: "failed" | "cancelled", invocationStatus: "failed" | "denied", errorCode: string, errorMessage: string) {
|
||||
return settle({ actionRequestId: request.id, status, errorCode, errorMessage },
|
||||
{ status: invocationStatus, errorCode, errorMessage });
|
||||
}
|
||||
if (invocation.companyId !== request.companyId) {
|
||||
// Quarantine the request without modifying another company's invocation.
|
||||
return settle({ actionRequestId: request.id, status: "failed",
|
||||
errorCode: "action_context_mismatch", errorMessage: "Tool action and invocation companies do not match." });
|
||||
}
|
||||
if (["succeeded", "failed", "denied", "cancelled", "timed_out", "rate_limited", "executing"].includes(invocation.status)) {
|
||||
return settle({ actionRequestId: request.id,
|
||||
status: invocation.status === "succeeded" ? "executed"
|
||||
: invocation.status === "executing" ? "executing"
|
||||
: invocation.status === "cancelled" ? "cancelled" : "failed",
|
||||
errorCode: invocation.errorCode, errorMessage: invocation.errorMessage,
|
||||
resultSummary: invocation.resultSummary?.summary });
|
||||
}
|
||||
if (invocation.status !== "awaiting_approval") {
|
||||
return fail("failed", "failed", "tool_recovery_state_invalid",
|
||||
"Approved action has inconsistent invocation history. Inspect the provider before retrying; recovery did not replay it.");
|
||||
}
|
||||
|
||||
// Missing-key/configuration errors deliberately propagate as retryable;
|
||||
// only a decoded invalid signature is a permanent request failure.
|
||||
const payload = readSignedToolArgumentsPayload({ signedArguments: request.signedArguments,
|
||||
invocationId: invocation.id, toolName: invocation.toolName, signingSecret: input.signingSecret });
|
||||
if (!payload) return fail("cancelled", "failed", "action_request_invalidated",
|
||||
"Stored tool arguments are no longer verifiable; submit a new action for review.");
|
||||
if (request.approvalId) {
|
||||
const [formal] = await tx.select({ status: approvals.status }).from(approvals)
|
||||
.where(and(eq(approvals.id, request.approvalId), eq(approvals.companyId, request.companyId)));
|
||||
if (!formal || formal.status === "rejected" || formal.status === "cancelled") {
|
||||
return fail("cancelled", "denied", "formal_approval_unavailable",
|
||||
"Required formal board approval is unavailable or was declined; submit a new action for review.");
|
||||
}
|
||||
// Human input has no deadline. The approval owner publishes a committed
|
||||
// resolution/cancellation to this queue; startup also rediscovers the row.
|
||||
if (formal.status === "pending" || formal.status === "revision_requested") {
|
||||
return { kind: "skip" } as const;
|
||||
}
|
||||
}
|
||||
return { kind: "ready", request, invocation, payload } as const;
|
||||
});
|
||||
}
|
||||
|
||||
export function approvedToolReviewCompanies(db: Db, companyId?: string) {
|
||||
return db.select({ companyId: toolActionRequests.companyId }).from(toolActionRequests)
|
||||
.innerJoin(toolInvocations, eq(toolInvocations.id, toolActionRequests.invocationId))
|
||||
.where(and(eq(toolActionRequests.status, "approved"),
|
||||
companyId ? eq(toolActionRequests.companyId, companyId) : undefined))
|
||||
.groupBy(toolActionRequests.companyId);
|
||||
}
|
||||
|
||||
export async function recoverApprovedToolReviews(db: Db, recover: (input: {
|
||||
request: typeof toolActionRequests.$inferSelect;
|
||||
invocation: typeof toolInvocations.$inferSelect;
|
||||
}) => Promise<boolean>, companyId?: string) {
|
||||
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"),
|
||||
companyId ? eq(toolActionRequests.companyId, companyId) : undefined,
|
||||
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, companyId: string) {
|
||||
return notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewExpiry, companyId);
|
||||
export async function notifyToolReviewChange(transaction: object, companyId: string) {
|
||||
await notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewExpiry, companyId);
|
||||
await notifyDeliveryWork(transaction, DELIVERY_QUEUES.toolReviewRecovery, companyId);
|
||||
}
|
||||
|
||||
export function writeToolReviewChange<T>(db: Db, companyId: string, write: (transaction: Db) => Promise<T>): Promise<T> {
|
||||
|
||||
@@ -11,7 +11,7 @@ function minuteBucket(date: Date): Date {
|
||||
}
|
||||
|
||||
export async function incrementToolRuntimeMetricCounter(
|
||||
db: Db,
|
||||
db: Pick<Db, "insert">,
|
||||
input: {
|
||||
companyId: string;
|
||||
metric: string;
|
||||
@@ -42,7 +42,7 @@ export async function incrementToolRuntimeMetricCounter(
|
||||
});
|
||||
}
|
||||
|
||||
export async function recordToolRuntimeAuditWriteFailure(db: Db, companyId: string) {
|
||||
export async function recordToolRuntimeAuditWriteFailure(db: Pick<Db, "insert">, companyId: string) {
|
||||
try {
|
||||
await incrementToolRuntimeMetricCounter(db, {
|
||||
companyId,
|
||||
|
||||
Reference in new issue
Block a user