feat(plugins): synchronize standing connection assignments (#16603)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - Lifecycle plugins prepare agent runtimes outside the control plane.
> - They need committed connection assignment changes without polling.
> - A reader and its revision hooks must cover the same fields.
> - This pull request adds a bounded standing-assignment snapshot and
its matching hooks.
> - Plugins can react to assignment changes while execution keeps its
existing access checks.

## Linked Issues or Issue Description

**What happened?**

The agent runtime reader does not expose standing connection
assignments. Changes to connection installs, active profiles, and
connector feature switches do not request synchronization. Adding
company-wide invalidation also exposes lock inversions in assignment
writers and AI adoption.

**Expected behavior**

The owning lifecycle plugin can read a bounded assignment snapshot.
Every committed change to its stored fields advances the runtime
revision. Rollback removes both the change and its event. Assignment
writers finish without a lock-order cycle.

**Steps to reproduce**

1. Install a connection or bind an active profile to an agent.
2. Read its runtime inputs through the owning lifecycle plugin.
3. Rename or disable the connection, revoke the assignment, or switch
connector features.
4. Expect a committed runtime event and an updated snapshot without
periodic polling.

**Additional context**

Extracted from #16503. Catalog publication #16602 and the separate test
maintenance PRs #16601 and #16610 are merged. This PR now targets master
and is the active review. Tool-definition hooks #16604 and
access/credential hooks #16605 remain drafts without review requests.

## What Changed

- Add SDK types and an owning-plugin runtime snapshot for installed
connections, active profile IDs, and effective connector feature
switches.
- Reject more than 512 connections or 128 active profiles instead of
returning partial data.
- Add migration 0354 for the exact stored fields used by the snapshot,
with transaction-coalesced invalidation. Agent-targeted bindings and
installs notify only their old/new target agents;
project/routine/issue/gateway bindings are outside the snapshot and
produce no notification.
- Use ordered agent locks that are compatible with foreign-key
references and exclude runtime readers. Prelock complete old/new target
sets before replacing assignments or restoring managed profiles.
- Record one company baseline at delivery startup for process-managed
feature overlays.
- Lock assignment inputs before install saves, task completion, and
email assignment. Setup finalization reads the locked connection state
and prelocks company agents before activating, re-enabling, or reviving
a profile.
- Make profile binding and reassignment writes transactional and lock
parent profiles first. Keep target validation and profile/detail reads
on the reserved transaction connection, including with a one-connection
pool.
- Lock company agents in order before validated AI adoption. Keep
adoption and shared configuration writes at `FOR NO KEY UPDATE`,
avoiding an upgrade that can deadlock foreign-key inserts during gateway
creation. Configuration fields cannot change agent identity; lifecycle
transitions and deletion retain their existing locks.
- Document supported coverage and the inputs reserved for later PRs.

## Verification

- Thirteen affected suites passed: 757 tests covering runtime
hooks/assignments/skills, concurrent skill publication, tool
access/catalog, task connection intents, lifecycle delivery, email, AI
accounts, migration histories, snapshot drift, and the agent lifecycle
module. This includes the merged catalog PR's OAuth and
overlapping-refresh regressions.
- The assignment suite passes all 24 cases, including bounded snapshots,
company/agent isolation, rollback, health-field no-ops, process-start
feature overlays, and real lock contention.
- Four one-connection PostgreSQL regressions reproduce the earlier
helper-read hangs before the correction; bind, unbind, delete, and
reassign complete with it. The earlier adoption/gateway probe exposed
both lock upgrades: changing only validation still deadlocks in shared
configuration. Current scoped coverage verifies adoption with
agent-targeted binding writes; gateway-only bindings no longer
invalidate this snapshot.
- The original ordered-adoption regression reproduces PostgreSQL `40P01`
without ordered company-agent locking. Scoped-recipient tests reproduce
unrelated-peer journaling before the trigger correction and verify
repeated bind/unbind operations leave peers untouched, ignored binding
types produce no events, both reassignment recipients are reached, and
coalescing remains intact.
- Six replacement contention cases cover installs, existing setup,
activation, re-enabling, profile revival, and profile reassignment.
Negative controls reproduce `40P01` without the corresponding
target/company prelocks; all cases pass with the guards. Activation
decisions use the connection row reread under lock.
- An initial contention fixture failed because its settings connection
lacked an MCP URL. Supplying the fixture URL and controlled lookup
corrected that test setup error.
- Server typecheck, database migration numbering/safety, five-snapshot
retention, feature-map structural validation, and `git diff --check`
passed. Migration metadata preserves historical journal entries and adds
no schema change. The complete diff was reviewed for secrets and
personal data before push.
- Earlier hosted PR CI runs 38983440861 and 38985530069 passed; the
superseded promotion run 38983440470 was cancelled. Superagent
acknowledged the recipient fix and accepted the documented company-wide
cost at `359b5b2f30a8a0248177367e458663d5379810d1`. Its two standard
Greptile summaries were conditional on a later setup correction and were
not counted as independent acceptance; the setup correction is now
included here.
- At `7da246aab46274db1a120a0546abbb404c5427c7`, hosted PR run
38986958988 passed: all 55 checks succeeded, with two optional Storybook
checks skipped. Superagent passed, all review threads are resolved, and
standard Greptile independently returned 5/5 with no blocking findings
(review completed 2026-10-17 00:40 UTC). Apex independently returned 5/5
with no actionable findings (completed 2026-10-17 00:50 UTC). CI was
rechecked after Apex and remains green; the branch is mergeable against
master. Full local build and repository-wide tests were not run.

## Risks

Migration 0354 adds database functions and triggers without changing
table schema. Apply it through the normal migration flow. Agent-targeted
bindings and installs only notify their affected agents. Explicit
company-wide assignments, connection metadata, profile
activation/removal, and connector feature switches retain synchronous,
transaction-atomic company fanout. Each such transaction takes ordered
row locks and inserts up to one connector runtime event per eligible
company agent: O(number of eligible agents) work and journal storage,
with no company-size cap or cross-transaction mutation quota. This
remains an intentional scope tradeoff for this PR; deferred fanout is
not implemented. Many of these conservative changes can request
preparation even if an individual snapshot stays unchanged. Snapshot
limits fail closed. This PR provides desired assignments, not
credentials or task authorization. Later PRs must add mutation coverage
before consumers can rely on immediate synchronization of tool
definitions or credentials.

## Model Used

OpenAI Codex, GPT-6, with reasoning, repository inspection, code
execution, and testing. The exact deployment model ID and context window
are not exposed in this session.

## 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
- [x] All Paperclip CI gates are green
- [x] 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:
Nicky LeachandPaperclip authored and GitHub committed 2026-10-16 17:55:47 -07:00
1 parent 190080642e
commit 5f5602bba7
17 files changed
+5469 -4668

No files matched your search

+7
View File
@@ -980,3 +980,10 @@ session-compatible database connection; do not enable it through a transaction
pooler. Reconnection scans the durable journal. Existing process-local work
signals continue to serve other delivery queues. See
[the plugin contract](plugins/AGENT_RUNTIME_SYNC.md).
Migration 0354 extends this stream to standing connection assignments. Statement
triggers compare the fields returned by the assignment snapshot and coalesce
company invalidation within the mutation transaction. Agent runtime journaling
uses ordered `FOR NO KEY UPDATE` locks, which remain compatible with foreign-key
references while excluding runtime readers' `FOR SHARE` locks. Connector feature
switches use the same journal; process-start overlays receive a delivery baseline.
+39
View File
@@ -94,3 +94,42 @@ Pinned assignments receive conservative content invalidations too. Their resolve
version and bytes stay pinned when pin support is enabled. Consumers can compare
versions or destination hashes and avoid rewriting unchanged trees. Task-only
skill mentions and a running task's saved version selection remain task-owned.
## Standing connection assignments
`getRuntime` also returns `connectionAssignments`: effective `publicMcp` and
`chatConnectors` feature switches, installed connections with `id`, `name`,
`transport`, `enabled`, and `status`, and active profile IDs. It includes both
company assignments and assignments for the requested agent. It excludes other
agents' assignments and has no credential values, grant details, or tool schemas.
The owning-plugin checks for the runtime reader apply to this field too.
Migration 0354 records committed changes to every stored field used by this
snapshot. Agent-targeted installation and profile binding edits invalidate only
their old and new target agents. Company-targeted assignments invalidate company
agent runtimes. Project, routine, issue, and gateway bindings are outside this
snapshot and do not invalidate it. Connection metadata, profile activation or
removal, and connector feature switches still use conservative company-wide
invalidation. Empty writes and changes to health bookkeeping do not. Multiple
assignment statements in one transaction coalesce to one update per agent.
Journal writes and their triggering changes commit or roll back together.
Company-wide invalidation remains synchronous: it locks eligible company agents
in order and journals one update per agent. Its work grows with company size;
there is no fanout ceiling or cross-transaction mutation quota.
The reader orders and deduplicates assignments. It rejects more than 512
connections or 128 active profiles with status 422 instead of returning a partial
snapshot. It rechecks the revision after reading assignments and skills, with the
same three-attempt conflict behavior described above.
Each delivery-service startup records one company baseline when an enabled
runtime subscriber becomes active. This covers feature overlays supplied at
process startup. It does not add a recurring scan. Plugins must compare desired
state and destination hashes before rewriting external files.
This assignment contract does not yet emit connector-specific invalidations for
tool catalog contents, profile entries, policies, connection configuration,
grants, credentials, or endpoint configuration. Those inputs need their own
mutation coverage before a consumer can rely on their immediate synchronization.
Existing agent and skill invalidations still apply independently. Standing
assignments express desired state; they do not prove task-specific authorization.
+7
View File
@@ -34,6 +34,13 @@ own policy and access review.
Gaps: Granting access is distinct from using access. No equivalent agent management path
exists in these routes.
The owning lifecycle plugin can read bounded standing connection and active
profile assignments with `ctx.agents.getRuntime`. It requires the existing
`agents.lifecycle.manage` capability and agent ownership checks. Committed
assignment changes use the opt-in `agent_runtime` stream. These inputs contain
no grant credentials and do not authorize a task's tool calls. See the
[runtime contract](../../doc/plugins/AGENT_RUNTIME_SYNC.md#standing-connection-assignments).
## `action-policy`
Review: `baseline`.
+6
View File
@@ -36,6 +36,12 @@ Preconditions: follow the [baseline](./README.md#before-driving-a-journey). Prep
Automated: [app detail](../../ui/src/pages/apps/AppDetail.test.tsx) covers agent/action controls, quarantined actions, and navigation using mocked requests.
[Runtime assignment tests](../../server/src/__tests__/agent-runtime-assignments.test.ts)
use PostgreSQL to verify committed assignment notifications, bounded snapshots,
rollback, and concurrent AI adoption. The owning lifecycle plugin can read
standing assignments through `ctx.agents.getRuntime`; this does not grant tool
execution access or replace task-specific authorization.
Manual: Grant one agent access, leave another excluded, and exercise off, ask, and allowed states for a harmless action. Reload between changes. Verify actual denied calls stay denied and ask-first creates a decision rather than executing early.
### `test-and-maintain`
@@ -0,0 +1,184 @@
CREATE OR REPLACE FUNCTION paperclip_record_agent_runtime(p_company uuid, p_agent uuid) RETURNS bigint
LANGUAGE plpgsql SET search_path = public, pg_temp AS $$
DECLARE revision bigint;
BEGIN
PERFORM 1 FROM agents WHERE company_id = p_company AND id = p_agent
AND status NOT IN ('pending_approval', 'terminated') FOR NO KEY UPDATE;
IF FOUND THEN
INSERT INTO resource_lifecycle_events(company_id, resource_type, resource_id, action)
VALUES (p_company, 'agent_runtime', p_agent, 'update') RETURNING id INTO revision;
END IF;
RETURN revision;
END;
$$;
--> statement-breakpoint
CREATE OR REPLACE FUNCTION paperclip_invalidate_company_agent_runtimes(p_company uuid) RETURNS void
LANGUAGE plpgsql SET search_path = public, pg_temp AS $$
DECLARE candidate uuid;
BEGIN
FOR candidate IN SELECT id FROM agents WHERE company_id = p_company
AND status NOT IN ('pending_approval', 'terminated') ORDER BY id FOR NO KEY UPDATE LOOP
PERFORM paperclip_record_agent_runtime(p_company, candidate);
END LOOP;
END;
$$;
--> statement-breakpoint
CREATE OR REPLACE FUNCTION paperclip_invalidate_skill_source_runtimes(p_company uuid, sources uuid[]) RETURNS void
LANGUAGE plpgsql SET search_path = public, pg_temp AS $$
DECLARE candidate uuid;
BEGIN
FOR candidate IN SELECT a.id FROM agents a WHERE a.company_id = p_company
AND a.status NOT IN ('pending_approval', 'terminated') AND EXISTS (
SELECT 1 FROM company_skills s WHERE s.company_id = p_company AND s.access_source_id = ANY(sources)
AND paperclip_agent_selects_runtime_skill(a.adapter_config, a.adapter_type, s.key, s.slug))
ORDER BY a.id FOR NO KEY UPDATE LOOP
PERFORM paperclip_record_agent_runtime(p_company, candidate);
END LOOP;
END;
$$;
--> statement-breakpoint
CREATE OR REPLACE FUNCTION paperclip_skill_runtime_changed() RETURNS trigger
LANGUAGE plpgsql SET search_path = public, pg_temp AS $$
DECLARE previous jsonb; current jsonb; keys text[]; company uuid; candidate uuid;
BEGIN
IF TG_OP <> 'INSERT' THEN previous := to_jsonb(OLD); END IF;
IF TG_OP <> 'DELETE' THEN current := to_jsonb(NEW); END IF;
IF TG_OP = 'UPDATE' AND
jsonb_build_array(OLD.key, OLD.slug, OLD.markdown, OLD.source_type, OLD.source_locator, OLD.source_ref,
OLD.file_inventory, OLD.current_version_id, OLD.access_source_id, OLD.compatibility, OLD.trust_level, OLD.metadata)
IS NOT DISTINCT FROM
jsonb_build_array(NEW.key, NEW.slug, NEW.markdown, NEW.source_type, NEW.source_locator, NEW.source_ref,
NEW.file_inventory, NEW.current_version_id, NEW.access_source_id, NEW.compatibility, NEW.trust_level, NEW.metadata)
THEN RETURN NEW; END IF;
company := coalesce((current->>'company_id')::uuid, (previous->>'company_id')::uuid);
SELECT array_agg(DISTINCT lower(reference)) INTO keys FROM (
SELECT row->>'key' AS reference FROM unnest(ARRAY[previous, current]) row
UNION ALL SELECT regexp_replace(row->>'key', '.*/', '') FROM unnest(ARRAY[previous, current]) row
UNION ALL SELECT CASE WHEN row->>'key' LIKE 'paperclipai/paperclip/%' THEN row->>'slug'
ELSE (row->>'slug') || '--' || left(encode(sha256(convert_to(row->>'key', 'UTF8')), 'hex'), 10) END
FROM unnest(ARRAY[previous, current]) row
) refs WHERE reference IS NOT NULL;
FOR candidate IN SELECT a.id FROM agents a WHERE a.company_id = company
AND a.status NOT IN ('pending_approval', 'terminated') AND (
(a.adapter_type <> 'paperclip_runner' AND keys && ARRAY['paperclipai/paperclip/paperclip', 'paperclipai/paperclip/complain', 'paperclipai/paperclip/suggestion-box']) OR
EXISTS (SELECT 1 FROM jsonb_array_elements(CASE WHEN jsonb_typeof(a.adapter_config#>'{paperclipSkillSync,desiredSkills}') = 'array'
THEN a.adapter_config#>'{paperclipSkillSync,desiredSkills}' ELSE '[]'::jsonb END) choice
WHERE lower(trim(CASE WHEN jsonb_typeof(choice) = 'string' THEN choice#>>'{}' ELSE choice->>'key' END)) = ANY(keys)))
ORDER BY a.id FOR NO KEY UPDATE LOOP
PERFORM paperclip_record_agent_runtime(company, candidate);
END LOOP;
IF TG_OP = 'DELETE' THEN RETURN OLD; END IF;
RETURN NEW;
END;
$$;
--> statement-breakpoint
CREATE OR REPLACE FUNCTION paperclip_invalidate_connector_agent(p_company uuid, p_agent uuid) RETURNS void
LANGUAGE plpgsql SET search_path = public, pg_temp AS $$
DECLARE setting text := 'paperclip.runtime_agent_' || replace(p_agent::text, '-', '_'); xid text := pg_current_xact_id()::text;
BEGIN
IF current_setting('paperclip.runtime_' || replace(p_company::text, '-', '_'), true) IS NOT DISTINCT FROM xid
OR current_setting(setting, true) IS NOT DISTINCT FROM xid THEN RETURN; END IF;
PERFORM paperclip_record_agent_runtime(p_company, p_agent);
PERFORM set_config(setting, xid, true);
END;
$$;
--> statement-breakpoint
CREATE OR REPLACE FUNCTION paperclip_invalidate_connector_company(p_company uuid) RETURNS void
LANGUAGE plpgsql SET search_path = public, pg_temp AS $$
DECLARE setting text := 'paperclip.runtime_' || replace(p_company::text, '-', '_'); xid text := pg_current_xact_id()::text; candidate uuid;
BEGIN
IF current_setting(setting, true) IS NOT DISTINCT FROM xid THEN RETURN; END IF;
FOR candidate IN SELECT id FROM agents WHERE company_id = p_company
AND status NOT IN ('pending_approval', 'terminated') ORDER BY id FOR NO KEY UPDATE LOOP
PERFORM paperclip_invalidate_connector_agent(p_company, candidate);
END LOOP;
PERFORM set_config(setting, xid, true);
END;
$$;
--> statement-breakpoint
CREATE OR REPLACE FUNCTION paperclip_connector_runtime_changed() RETURNS trigger
LANGUAGE plpgsql SET search_path = public, pg_temp AS $$
DECLARE company uuid; query text; old_projection text; new_projection text; target record; candidate uuid;
BEGIN
IF TG_TABLE_NAME = 'tool_profiles' AND TG_OP = 'INSERT' THEN RETURN NULL; END IF;
IF TG_OP = 'UPDATE' THEN
SELECT 'jsonb_build_array(' || string_agg(format('to_jsonb(o)->%L', field), ',') || ')',
'jsonb_build_array(' || string_agg(format('to_jsonb(n)->%L', field), ',') || ')'
INTO old_projection, new_projection FROM unnest(TG_ARGV[1:TG_NARGS - 1]) field;
END IF;
IF TG_TABLE_NAME IN ('tool_profile_bindings', 'tool_connection_installs') THEN
IF TG_OP = 'INSERT' THEN query := 'SELECT company_id, target_type, target_id FROM new_rows';
ELSIF TG_OP = 'DELETE' THEN query := 'SELECT company_id, target_type, target_id FROM old_rows';
ELSE query := format('WITH changed AS (
SELECT o.company_id AS old_company, o.target_type AS old_type, o.target_id AS old_target,
n.company_id AS new_company, n.target_type AS new_type, n.target_id AS new_target
FROM old_rows o FULL JOIN new_rows n ON o.%I = n.%I WHERE %s IS DISTINCT FROM %s)
SELECT old_company AS company_id, old_type AS target_type, old_target AS target_id FROM changed WHERE old_company IS NOT NULL UNION
SELECT new_company AS company_id, new_type AS target_type, new_target AS target_id FROM changed WHERE new_company IS NOT NULL',
TG_ARGV[0], TG_ARGV[0], old_projection, new_projection);
END IF;
FOR target IN EXECUTE 'SELECT DISTINCT company_id, target_type, target_id FROM (' || query || ') targets
WHERE target_type IN (''agent'', ''company'')
ORDER BY company_id, target_type DESC, target_id' LOOP
IF target.target_type = 'company' AND target.target_id = target.company_id::text THEN
PERFORM paperclip_invalidate_connector_company(target.company_id);
ELSIF target.target_type = 'agent' AND target.target_id ~ '^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$' THEN
SELECT id INTO candidate FROM agents WHERE company_id = target.company_id AND id = target.target_id::uuid;
IF FOUND THEN PERFORM paperclip_invalidate_connector_agent(target.company_id, candidate); END IF;
END IF;
END LOOP;
RETURN NULL;
END IF;
IF TG_OP = 'INSERT' THEN query := 'SELECT DISTINCT company_id FROM new_rows ORDER BY company_id';
ELSIF TG_OP = 'DELETE' THEN query := 'SELECT DISTINCT company_id FROM old_rows ORDER BY company_id';
ELSE
query := format('WITH changed AS (
SELECT o.company_id AS old_company, n.company_id AS new_company FROM old_rows o FULL JOIN new_rows n ON o.%I = n.%I
WHERE %s IS DISTINCT FROM %s)
SELECT old_company AS company_id FROM changed WHERE old_company IS NOT NULL UNION
SELECT new_company AS company_id FROM changed WHERE new_company IS NOT NULL ORDER BY company_id', TG_ARGV[0], TG_ARGV[0], old_projection, new_projection);
END IF;
FOR company IN EXECUTE query LOOP
PERFORM paperclip_invalidate_connector_company(company);
END LOOP;
RETURN NULL;
END;
$$;
--> statement-breakpoint
DO $$
DECLARE relation text; arguments text;
BEGIN
FOREACH relation IN ARRAY ARRAY['tool_connections', 'tool_connection_installs', 'tool_profiles', 'tool_profile_bindings'] LOOP
arguments := CASE relation
WHEN 'tool_connections' THEN '''id'', ''company_id'', ''name'', ''transport'', ''enabled'', ''status'''
WHEN 'tool_connection_installs' THEN '''id'', ''company_id'', ''connection_id'', ''target_type'', ''target_id'''
WHEN 'tool_profiles' THEN '''id'', ''company_id'', ''status'''
ELSE '''id'', ''company_id'', ''profile_id'', ''target_type'', ''target_id''' END;
EXECUTE format('DROP TRIGGER IF EXISTS paperclip_connector_runtime_insert ON %I', relation);
EXECUTE format('DROP TRIGGER IF EXISTS paperclip_connector_runtime_update ON %I', relation);
EXECUTE format('DROP TRIGGER IF EXISTS paperclip_connector_runtime_delete ON %I', relation);
EXECUTE format('CREATE TRIGGER paperclip_connector_runtime_insert AFTER INSERT ON %I REFERENCING NEW TABLE AS new_rows FOR EACH STATEMENT EXECUTE FUNCTION paperclip_connector_runtime_changed(%s)', relation, arguments);
EXECUTE format('CREATE TRIGGER paperclip_connector_runtime_update AFTER UPDATE ON %I REFERENCING OLD TABLE AS old_rows NEW TABLE AS new_rows FOR EACH STATEMENT EXECUTE FUNCTION paperclip_connector_runtime_changed(%s)', relation, arguments);
EXECUTE format('CREATE TRIGGER paperclip_connector_runtime_delete AFTER DELETE ON %I REFERENCING OLD TABLE AS old_rows FOR EACH STATEMENT EXECUTE FUNCTION paperclip_connector_runtime_changed(%s)', relation, arguments);
END LOOP;
END;
$$;
--> statement-breakpoint
CREATE OR REPLACE FUNCTION paperclip_connector_flags_runtime_changed() RETURNS trigger
LANGUAGE plpgsql SET search_path = public, pg_temp AS $$
DECLARE company uuid; previous jsonb := coalesce(to_jsonb(OLD)->'experimental', '{}'::jsonb);
current jsonb := coalesce(to_jsonb(NEW)->'experimental', '{}'::jsonb);
BEGIN
IF jsonb_build_array(coalesce(previous->'enablePublicMcp', 'false'::jsonb), coalesce(previous->'enableChatConnectors', 'false'::jsonb))
IS DISTINCT FROM jsonb_build_array(coalesce(current->'enablePublicMcp', 'false'::jsonb), coalesce(current->'enableChatConnectors', 'false'::jsonb)) THEN
FOR company IN SELECT DISTINCT company_id FROM agents ORDER BY company_id LOOP
PERFORM paperclip_invalidate_connector_company(company);
END LOOP;
END IF;
RETURN NULL;
END;
$$;
--> statement-breakpoint
DROP TRIGGER IF EXISTS paperclip_connector_flags_runtime_changed ON instance_settings;
CREATE TRIGGER paperclip_connector_flags_runtime_changed AFTER INSERT OR UPDATE OR DELETE ON instance_settings
FOR EACH ROW EXECUTE FUNCTION paperclip_connector_flags_runtime_changed();
File diff suppressed because it is too large. Load diff
+7
View File
@@ -2465,6 +2465,13 @@
"when": 1792172416248,
"tag": "0353_skinny_gamma_corps",
"breakpoints": true
},
{
"idx": 354,
"version": "7",
"when": 1792187602372,
"tag": "0354_agent_connection_assignment_hooks",
"breakpoints": true
}
]
}
+1
View File
@@ -271,6 +271,7 @@ export type {
PluginIssueSummariesClient,
PluginAgentsClient,
PluginAgentRuntimeState,
PluginAgentConnectionAssignments,
PluginAgentRuntimeSkill,
PluginAccessClient,
PluginAccessMembersClient,
+7
View File
@@ -1697,7 +1697,14 @@ export interface PluginAgentRuntimeSkill {
missingDetail?: string | null;
}
export interface PluginAgentConnectionAssignments {
features: { publicMcp: boolean; chatConnectors: boolean };
connections: Array<{ id: string; name: string; transport: string; enabled: boolean; status: string }>;
profileIds: string[];
}
export interface PluginAgentRuntimeState {
connectionAssignments: PluginAgentConnectionAssignments;
skills: PluginAgentRuntimeSkill[];
revision: string;
agent: Agent;
@@ -0,0 +1,382 @@
import { randomUUID } from "node:crypto";
import { EventEmitter } from "node:events";
import { mkdtemp, realpath, rm } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { and, eq, sql } from "drizzle-orm";
import { agents, heartbeatRuns, issues, companies, companyMemberships, createDb, instanceSettings, pluginConfig, plugins, projects, toolApplications, toolConnections, toolConnectionInstalls, toolProfiles, toolMcpGateways, toolProfileBindings, resourceLifecycleEvents, type Db } from "@paperclipai/db";
import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
import { pluginAgentRuntimeState } from "../services/plugin-agent-runtime-state.js";
import { pluginLifecycleInbox } from "../services/plugin-lifecycle-inbox.js";
import { registerPluginLifecycleDelivery } from "../services/plugin-lifecycle-delivery.js";
import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js";
import { toolAccessService } from "../services/tool-access.js";
import { createToolGatewayService } from "../services/tool-gateway.js";
import { connectionIntentService } from "../services/connection-intents.js";
import { aiConnectionService } from "../services/ai-connections.js";
vi.mock("../services/subscription-refresh.js", () => ({ refreshSubscriptionConnection: vi.fn().mockResolvedValue(undefined) }));
const support = await getEmbeddedPostgresTestSupport();
const describeDatabase = support.supported ? describe : describe.skip;
describeDatabase("connection assignment synchronization", () => {
let database: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
let db: Db;
let companyId: string;
let agentId: string;
let peerId: string;
let pluginId: string;
let applicationId: string;
let connectionId: string;
beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("paperclip-runtime-assignments-"); db = createDb(database.connectionString); }, 90_000);
afterAll(async () => { await database?.cleanup(); });
afterEach(() => vi.unstubAllEnvs());
beforeEach(async () => {
companyId = randomUUID(); agentId = randomUUID(); peerId = randomUUID(); pluginId = randomUUID();
await db.insert(companies).values({ id: companyId, name: "Connector runtime fixture", issuePrefix: `C${companyId.slice(0, 6)}` });
await db.insert(plugins).values({ id: pluginId, pluginKey: pluginId, packageName: "runtime-fixture", version: "1.0.0", status: "ready", manifestJson: { resourceLifecycle: ["agent_runtime"], capabilities: ["events.subscribe", "agents.lifecycle.manage"] } as never });
await db.insert(agents).values([agentId, peerId].map(id => ({ id, companyId, name: id, adapterType: "paperclip_runner", lifecycleRequiredPluginIds: [pluginId] })));
const [application] = await db.insert(toolApplications).values({ companyId, name: "Fixture MCP", type: "mcp_http" }).returning(); applicationId = application.id;
const [connection] = await db.insert(toolConnections).values({ companyId, applicationId, name: "Fixture connection", uid: randomUUID(), transport: "mcp_remote", authKind: "none", enabled: true, status: "active" }).returning(); connectionId = connection.id;
});
const events = () => db.select().from(resourceLifecycleEvents).where(and(eq(resourceLifecycleEvents.companyId, companyId), eq(resourceLifecycleEvents.resourceType, "agent_runtime"))).orderBy(resourceLifecycleEvents.id);
it("creates gateways without invalidating unrelated standing assignments", async () => {
const [profile] = await db.insert(toolProfiles).values({ companyId, profileKey: "gateway-fixture", name: "Gateway fixture" }).returning();
const [lower, higher] = [agentId, peerId].sort();
let locked!: () => void; let release!: () => void;
const held = new Promise<void>(resolve => { locked = resolve; });
const resume = new Promise<void>(resolve => { release = resolve; });
const blocker = db.transaction(async tx => {
await tx.select().from(agents).where(eq(agents.id, lower)).for("no key update");
locked(); await resume;
});
await held;
const before = await events();
const saves = [higher, lower].map((target, index) => db.transaction(async tx => {
await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
return createToolGatewayService(tx as unknown as Db).createNamedGateway({ companyId, body: { name: `Gateway ${index}`, profileId: profile.id, agentId: target } });
}));
const completed = Promise.allSettled(saves);
try {
const results = await completed;
for (const result of results) if (result.status === "rejected") throw result.reason;
expect(await events()).toEqual(before);
} finally { release(); await blocker; }
const results = await completed;
for (const result of results) if (result.status === "rejected") throw result.reason;
expect(results.map(result => result.status)).toEqual(["fulfilled", "fulfilled"]);
expect(await db.select().from(toolMcpGateways).where(eq(toolMcpGateways.companyId, companyId))).toHaveLength(2);
expect(await db.select().from(toolProfileBindings).where(eq(toolProfileBindings.companyId, companyId))).toHaveLength(2);
}, 15000);
it("captures profile assignment and revocation without exposing another agent's assignments", async () => {
const [profile] = await db.insert(toolProfiles).values({ companyId, profileKey: "fixture", name: "Fixture profile" }).returning();
const [binding] = await db.insert(toolProfileBindings).values({ companyId, profileId: profile.id, targetType: "agent", targetId: agentId }).returning();
const [install] = await db.insert(toolConnectionInstalls).values({ companyId, connectionId, targetType: "agent", targetId: agentId }).returning();
const owner = await pluginAgentRuntimeState(db, pluginId, companyId, agentId);
expect(owner!.connectionAssignments).toMatchObject({ connections: [expect.objectContaining({ id: connectionId })], profileIds: [profile.id] });
expect((await pluginAgentRuntimeState(db, pluginId, companyId, peerId))!.connectionAssignments).toMatchObject({ connections: [], profileIds: [] });
const before = (await events()).length;
await db.delete(toolProfileBindings).where(eq(toolProfileBindings.id, binding.id));
await db.delete(toolConnectionInstalls).where(eq(toolConnectionInstalls.id, install.id));
expect(await events()).toHaveLength(before + 2);
expect((await pluginAgentRuntimeState(db, pluginId, companyId, agentId))!.connectionAssignments).toMatchObject({ connections: [], profileIds: [] });
});
it("journals only snapshot recipients when bindings and installs change", async () => {
const service = toolAccessService(db);
const [profile] = await db.insert(toolProfiles).values({ companyId, profileKey: "scoped", name: "Scoped profile" }).returning();
const [project] = await db.insert(projects).values({ companyId, name: "Project binding fixture" }).returning();
for (let round = 0; round < 2; round++) {
let before = (await events()).length;
await service.bindProfile(profile.id, { targetType: "agent", targetId: agentId });
expect((await events()).slice(before).map(event => event.resourceId)).toEqual([agentId]);
before = (await events()).length;
await service.unbindProfile(profile.id, { targetType: "agent", targetId: agentId });
expect((await events()).slice(before).map(event => event.resourceId)).toEqual([agentId]);
}
const beforeProject = await events();
await service.bindProfile(profile.id, { targetType: "project", targetId: project.id });
await service.unbindProfile(profile.id, { targetType: "project", targetId: project.id });
expect(await events()).toEqual(beforeProject);
let before = (await events()).length;
await db.transaction(async tx => {
const [binding] = await tx.insert(toolProfileBindings).values({ companyId, profileId: profile.id, targetType: "agent", targetId: agentId }).returning();
await tx.delete(toolProfileBindings).where(eq(toolProfileBindings.id, binding.id));
});
expect((await events()).slice(before).map(event => event.resourceId)).toEqual([agentId]);
before = (await events()).length;
const [install] = await db.insert(toolConnectionInstalls).values({ companyId, connectionId, targetType: "agent", targetId: agentId }).returning();
expect((await events()).slice(before).map(event => event.resourceId)).toEqual([agentId]);
before = (await events()).length;
await db.update(toolConnectionInstalls).set({ targetId: peerId }).where(eq(toolConnectionInstalls.id, install.id));
expect((await events()).slice(before).map(event => event.resourceId).sort()).toEqual([agentId, peerId].sort());
before = (await events()).length;
await service.bindProfile(profile.id, { targetType: "company", targetId: companyId });
expect((await events()).slice(before).map(event => event.resourceId).sort()).toEqual([agentId, peerId].sort());
});
it("keeps connector changes and their events atomic and company-scoped", async () => {
const before = await events();
await expect(db.transaction(async tx => {
await tx.update(toolConnections).set({ enabled: false }).where(eq(toolConnections.id, connectionId));
throw new Error("rollback");
})).rejects.toThrow("rollback");
expect(await events()).toEqual(before);
await db.execute(sql`ALTER TABLE resource_lifecycle_events ADD CONSTRAINT connector_fixture_failure CHECK (resource_type <> 'agent_runtime') NOT VALID`);
try { await expect(db.update(toolConnections).set({ enabled: false }).where(eq(toolConnections.id, connectionId))).rejects.toThrow(); }
finally { await db.execute(sql`ALTER TABLE resource_lifecycle_events DROP CONSTRAINT connector_fixture_failure`); }
expect((await db.select().from(toolConnections).where(eq(toolConnections.id, connectionId)))[0].enabled).toBe(true);
const other = randomUUID();
await db.insert(companies).values({ id: other, name: "Other company", issuePrefix: `X${other.slice(0, 6)}` });
await db.insert(toolApplications).values({ companyId: other, name: "Other MCP", type: "mcp_http" });
expect(await events()).toEqual(before);
});
it("fails closed when standing assignments exceed the bounded snapshot", async () => {
const added = await db.insert(toolConnections).values(Array.from({ length: 513 }, (_, index) => ({ companyId, applicationId, name: `bounded-${index}`, uid: randomUUID(), transport: "mcp_remote" as const, authKind: "none" as const }))).returning({ id: toolConnections.id });
await db.insert(toolConnectionInstalls).values(added.map(connection => ({ companyId, connectionId: connection.id, targetType: "agent" as const, targetId: agentId })));
await expect(pluginAgentRuntimeState(db, pluginId, companyId, agentId)).rejects.toMatchObject({ status: 422 });
});
it("invalidates connector feature switches and returns their effective standing state", async () => {
await pluginAgentRuntimeState(db, pluginId, companyId, agentId);
const before = (await events()).length;
await db.update(instanceSettings).set({ experimental: { enableChatConnectors: true } });
expect(await events()).toHaveLength(before + 2);
expect((await pluginAgentRuntimeState(db, pluginId, companyId, agentId))!.connectionAssignments.features).toEqual({ publicMcp: false, chatConnectors: true });
});
it("refreshes acknowledged runtime state after managed feature overlays change across restarts", async () => {
await pluginAgentRuntimeState(db, pluginId, companyId, agentId);
await db.insert(pluginConfig).values({ pluginId, companyId, configJson: {} });
const inbox = pluginLifecycleInbox(db, pluginId);
for (const event of await events()) await inbox.acknowledge(companyId, String(event.id));
for (const enabled of [true, false]) {
vi.stubEnv("PAPERCLIP_MANAGED_CONFIG", JSON.stringify({ v: 1, mode: "cloud", catalogVersion: "2026.720.0", features: { enablePublicMcp: enabled, enableChatConnectors: enabled }, plugins: { autoInstall: [] } }));
const before = (await events()).length;
const snapshots: Array<{ publicMcp: boolean; chatConnectors: boolean }> = [];
const call = vi.fn(async (_plugin, _method, event) => {
const state = await pluginAgentRuntimeState(db, pluginId, companyId, event.resourceId);
snapshots.push(state!.connectionAssignments.features);
return { eventId: event.id, status: "acknowledged" };
});
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError: error => { throw error; } });
const delivery = registerPluginLifecycleDelivery(coordinator, db, { call, isRunning: () => true, onConfigApplied: () => () => {} } as never, new EventEmitter() as never);
try {
await delivery.ready;
await vi.waitFor(() => expect(snapshots).toEqual(Array(2).fill({ publicMcp: enabled, chatConnectors: enabled })), { timeout: 5000 });
await vi.waitFor(async () => expect(await inbox.list(companyId, 100, undefined, ["agent_runtime"])).toEqual([]), { timeout: 5000 });
expect(await events()).toHaveLength(before + 2);
} finally { delivery.dispose(); await coordinator.stop(); }
}
});
it("versions every returned connection field and skips unrelated bookkeeping", async () => {
await db.insert(toolConnectionInstalls).values({ companyId, connectionId, targetType: "company", targetId: companyId });
const initial = await pluginAgentRuntimeState(db, pluginId, companyId, agentId);
const before = await events();
await db.update(toolConnections).set({ healthCheckedAt: new Date(), lastError: "fixture", config: { fixture: true }, updatedAt: new Date() }).where(eq(toolConnections.id, connectionId));
expect(await events()).toEqual(before);
await db.transaction(async tx => {
await tx.update(toolConnections).set({ name: "Renamed", enabled: false }).where(eq(toolConnections.id, connectionId));
await tx.update(toolConnections).set({ status: "disabled", transport: "local_stdio" }).where(eq(toolConnections.id, connectionId));
});
expect(await events()).toHaveLength(before.length + 2);
const current = await pluginAgentRuntimeState(db, pluginId, companyId, agentId);
expect(current!.revision).not.toBe(initial!.revision);
expect(current!.connectionAssignments.connections).toEqual([{ id: connectionId, name: "Renamed", transport: "local_stdio", enabled: false, status: "disabled" }]);
const [profile] = await db.insert(toolProfiles).values({ companyId, profileKey: "status-fixture", name: "Status fixture" }).returning();
await db.insert(toolProfileBindings).values({ companyId, profileId: profile.id, targetType: "company", targetId: companyId });
expect((await pluginAgentRuntimeState(db, pluginId, companyId, agentId))!.connectionAssignments.profileIds).toContain(profile.id);
const active = await events();
await db.update(toolProfiles).set({ name: "Cosmetic rename" }).where(eq(toolProfiles.id, profile.id));
expect(await events()).toEqual(active);
await db.update(toolProfiles).set({ status: "archived" }).where(eq(toolProfiles.id, profile.id));
expect(await events()).toHaveLength(active.length + 2);
expect((await pluginAgentRuntimeState(db, pluginId, companyId, agentId))!.connectionAssignments.profileIds).not.toContain(profile.id);
});
it("rejects excessive active profile assignments", async () => {
const profiles = await db.insert(toolProfiles).values(Array.from({ length: 129 }, (_, index) => ({ companyId, profileKey: `bound-${index}`, name: `Bound ${index}` }))).returning();
await db.insert(toolProfileBindings).values(profiles.map(profile => ({ companyId, profileId: profile.id, targetType: "company" as const, targetId: companyId })));
await expect(pluginAgentRuntimeState(db, pluginId, companyId, agentId)).rejects.toMatchObject({ status: 422 });
});
it.each(["bind", "unbind", "delete", "reassign"] as const)("completes profile %s with a single database connection", async change => {
const [source, target] = await db.insert(toolProfiles).values([
{ companyId, profileKey: "single-source", name: "Single source" },
{ companyId, profileKey: "single-target", name: "Single target" },
]).returning();
if (change !== "bind") await db.insert(toolProfileBindings).values({ companyId, profileId: source.id, targetType: "agent", targetId: agentId });
const single = createDb(database.connectionString, { maxConnections: 1 });
const service = toolAccessService(single);
let timer: ReturnType<typeof setTimeout> | undefined;
const work = change === "bind" ? service.bindProfile(source.id, { targetType: "agent", targetId: agentId })
: change === "unbind" ? service.unbindProfile(source.id, { targetType: "agent", targetId: agentId })
: service.deleteProfile(source.id, change === "reassign" ? { reassignToProfileId: target.id } : { force: true });
const settled = Promise.allSettled([work]);
try {
await Promise.race([work, new Promise<never>((_, reject) => { timer = setTimeout(() => reject(new Error("Profile operation waited for another pooled connection")), 3000); })]);
if (change === "bind") expect(await db.select().from(toolProfileBindings).where(eq(toolProfileBindings.profileId, source.id))).toHaveLength(1);
else if (change === "unbind") expect(await db.select().from(toolProfileBindings).where(eq(toolProfileBindings.profileId, source.id))).toHaveLength(0);
else {
expect(await db.select().from(toolProfiles).where(eq(toolProfiles.id, source.id))).toHaveLength(0);
expect(await db.select().from(toolProfileBindings).where(eq(toolProfileBindings.profileId, target.id))).toHaveLength(change === "reassign" ? 1 : 0);
}
} finally { clearTimeout(timer); await single.$client.end({ timeout: 1 }); await settled; }
}, 10000);
it.each(["bind", "unbind", "reassign"] as const)("orders profile %s with setup finalization before assignment fanout", async change => {
const targetAgent = [agentId, peerId].sort()[0];
const [profile] = await db.insert(toolProfiles).values({ companyId, profileKey: `app:${connectionId}`, name: "Managed profile" }).returning();
await db.insert(toolProfileBindings).values({ companyId, profileId: profile.id, targetType: "company", targetId: companyId });
if (change === "unbind") await db.insert(toolProfileBindings).values({ companyId, profileId: profile.id, targetType: "agent", targetId: targetAgent });
const [source] = change === "reassign" ? await db.insert(toolProfiles).values({ companyId, profileKey: "source", name: "Source profile" }).returning() : [];
if (source) await db.insert(toolProfileBindings).values({ companyId, profileId: source.id, targetType: "agent", targetId: targetAgent });
let locked!: () => void;
let release!: () => void;
const held = new Promise<void>(resolve => { locked = resolve; });
const resume = new Promise<void>(resolve => { release = resolve; });
const blocker = db.transaction(async tx => {
await tx.select().from(agents).where(eq(agents.id, [agentId, peerId].sort()[0])).for("no key update");
locked(); await resume;
});
await held;
const pids: number[] = [];
const run = (work: (service: ReturnType<typeof toolAccessService>) => Promise<unknown>) => db.transaction(async tx => {
await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
const [backend] = await tx.execute(sql`SELECT pg_backend_pid() AS pid`);
pids.push(Number(backend.pid));
return work(toolAccessService(tx as unknown as Db));
});
const waitForLock = (index: number) => vi.waitFor(async () => {
expect(pids.length).toBeGreaterThan(index);
const [activity] = await db.execute(sql`SELECT wait_event_type FROM pg_stat_activity WHERE pid = ${pids[index]}`);
expect(activity.wait_event_type).toBe("Lock");
}, { timeout: 5000 });
const mutation = run(writer => change === "bind"
? writer.bindProfile(profile.id, { targetType: "agent", targetId: targetAgent })
: change === "unbind" ? writer.unbindProfile(profile.id, { targetType: "agent", targetId: targetAgent })
: writer.deleteProfile(source!.id, { reassignToProfileId: profile.id }));
const mutationResult = Promise.allSettled([mutation]);
let finishResult: Promise<PromiseSettledResult<unknown>[]> | undefined;
try {
await waitForLock(0);
finishResult = Promise.allSettled([run(writer => writer.finishGalleryAppConnection(companyId, connectionId, { enabledCatalogEntryIds: [], askFirstCatalogEntryIds: [], access: "all_agents" }))]);
await waitForLock(1);
} finally { release(); await blocker; }
for (const result of [...await mutationResult, ...await finishResult!]) if (result.status === "rejected") throw result.reason;
expect((await db.select().from(toolProfiles).where(eq(toolProfiles.id, profile.id)))[0].status).toBe("active");
if (source) expect(await db.select().from(toolProfiles).where(eq(toolProfiles.id, source.id))).toEqual([]);
}, 15000);
it.each(["connection settings", "profile binding"] as const)("orders AI adoption with concurrent %s using foreign-key-compatible locks", async change => {
const home = await realpath(await mkdtemp(join(tmpdir(), "paperclip-adoption-fixture-")));
vi.stubEnv("PAPERCLIP_HOME", home);
vi.stubEnv("PAPERCLIP_INSTANCE_ID", "adoption-fixture");
const [lower, higher] = [agentId, peerId].sort();
const userId = `adoption-${randomUUID()}`;
const issueId = randomUUID();
const runId = randomUUID();
let release!: () => void;
try {
await db.update(toolConnections).set({ config: { url: "https://mcp.example.test/mcp" } }).where(eq(toolConnections.id, connectionId));
await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, status: "active", membershipRole: "member" });
await db.update(agents).set({ adapterType: "paperclip_runner", adapterConfig: { provider: "codex" } }).where(eq(agents.id, higher));
await db.insert(issues).values({ id: issueId, companyId, title: "Legacy AI adoption", status: "in_progress", assigneeAgentId: higher, responsibleUserId: userId });
await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId: higher, nativeIssueId: issueId, status: "failed", errorCode: "acpx_auth_required", responsibleUserId: userId, contextSnapshot: {} });
const intents = connectionIntentService(db);
const card = await intents.requestForRunAuthFailure(runId);
const setup = await intents.setupOptions(card!.interactionId!);
expect(setup.aiConnectionRequiresAdoption).toBe(true);
const account = await aiConnectionService(db).save(companyId, userId, { provider: "openai", method: "api_key", ownership: "personal", name: "Adopted account", apiKey: "fixture", allAgents: true, agentIds: [] }, "fixture-api-key");
const [bindingProfile] = change === "profile binding" ? await db.insert(toolProfiles).values({ companyId, profileKey: "adoption-binding", name: "Adoption binding" }).returning() : [];
const [agent] = await db.select().from(agents).where(eq(agents.id, higher));
let locked!: () => void;
const held = new Promise<void>(resolve => { locked = resolve; });
const resume = new Promise<void>(resolve => { release = resolve; });
const blocker = db.transaction(async tx => {
await tx.select().from(agents).where(eq(agents.id, lower)).for("no key update");
locked(); await resume;
});
await held;
const pids: number[] = [];
const run = (work: (client: Db) => Promise<unknown>) => db.transaction(async tx => {
await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
const [backend] = await tx.execute(sql`SELECT pg_backend_pid() AS pid`);
pids.push(Number(backend.pid));
return work(tx as unknown as Db);
});
const waitForLock = (index: number) => vi.waitFor(async () => {
expect(pids.length).toBeGreaterThan(index);
const [activity] = await db.execute(sql`SELECT wait_event_type FROM pg_stat_activity WHERE pid = ${pids[index]}`);
expect(activity.wait_event_type).toBe("Lock");
}, { timeout: 5000 });
const adopt = (client: Db) => connectionIntentService(client).complete(card!.interactionId!, account.connectionId, userId, { validatedAdoption: { agentUpdatedAt: agent.updatedAt, binding: setup.aiConnection! } });
const first = run(change === "connection settings" ? client => toolAccessService(client, { remoteHttpEndpointLookup: async () => [{ address: "8.8.8.8", family: 4 }] }).updateConnection(connectionId, { name: "Concurrent settings" }) : adopt);
let second: Promise<PromiseSettledResult<unknown>[]> | undefined;
const firstResult = Promise.allSettled([first]);
try {
await waitForLock(0);
second = Promise.allSettled([run(change === "connection settings" ? adopt : client => toolAccessService(client).bindProfile(bindingProfile!.id, { targetType: "agent", targetId: lower }, { actorType: "agent", actorId: higher }))]);
await waitForLock(1);
} finally { release(); await blocker; }
for (const result of [...await firstResult, ...await second!]) if (result.status === "rejected") throw result.reason;
expect((await db.select().from(agents).where(eq(agents.id, higher)))[0].runtimeConfig.aiConnection).toEqual(setup.aiConnection);
if (change === "connection settings") expect((await db.select().from(toolConnections).where(eq(toolConnections.id, connectionId)))[0].name).toBe("Concurrent settings");
else expect(await db.select().from(toolProfileBindings).where(eq(toolProfileBindings.profileId, bindingProfile!.id))).toEqual(expect.arrayContaining([expect.objectContaining({ targetType: "agent", targetId: lower })]));
expect(await toolAccessService(db).listConnectionInstalls(account.connectionId, companyId)).toEqual(expect.arrayContaining([expect.objectContaining({ targetType: "agent", targetId: higher })]));
expect((await intents.setupOptions(card!.interactionId!)).interaction.status).toBe("accepted");
} finally { release?.(); await rm(home, { recursive: true, force: true }); }
}, 20000);
it.each(["installs", "finish", "activate", "enable", "revive", "reassign"] as const)("orders %s replacement before scoped and company runtime locks", async change => {
const [lower, higher] = [agentId, peerId].sort();
await db.update(toolConnections).set({ config: { url: "https://mcp.example.test/mcp" } }).where(eq(toolConnections.id, connectionId));
const [app] = await db.insert(toolApplications).values({ companyId, name: "Replacement app", type: "mcp_http" }).returning();
const [connection] = await db.insert(toolConnections).values({ companyId, applicationId: app.id, name: "Replacement connection", uid: randomUUID(), transport: "mcp_remote", status: change === "activate" ? "draft" : "active", enabled: change !== "enable" }).returning();
const [source, target] = await db.insert(toolProfiles).values([
{ companyId, profileKey: `app:${connection.id}`, name: "Replacement source", status: change === "revive" ? "archived" : "active" },
{ companyId, profileKey: "replacement-target", name: "Replacement target" },
]).returning();
await db.insert(toolProfileBindings).values({ companyId, profileId: source.id, targetType: "agent", targetId: higher });
if (change === "installs") await db.insert(toolConnectionInstalls).values({ companyId, connectionId: connection.id, targetType: "agent", targetId: higher });
let locked!: () => void; let release!: () => void;
const held = new Promise<void>(resolve => { locked = resolve; });
const resume = new Promise<void>(resolve => { release = resolve; });
const blocker = db.transaction(async tx => {
await tx.select().from(agents).where(eq(agents.id, lower)).for("no key update");
locked(); await resume;
});
await held;
const pids: number[] = [];
const run = (work: (service: ReturnType<typeof toolAccessService>) => Promise<unknown>) => db.transaction(async tx => {
await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
const [backend] = await tx.execute(sql`SELECT pg_backend_pid() AS pid`);
pids.push(Number(backend.pid));
return work(toolAccessService(tx as unknown as Db, { remoteHttpEndpointLookup: async () => [{ address: "8.8.8.8", family: 4 }] }));
});
const waitForLock = (index: number) => vi.waitFor(async () => {
expect(pids.length).toBeGreaterThan(index);
const [activity] = await db.execute(sql`SELECT wait_event_type FROM pg_stat_activity WHERE pid = ${pids[index]}`);
expect(activity.wait_event_type).toBe("Lock");
}, { timeout: 5000 });
const settings = Promise.allSettled([run(service => service.updateConnection(connectionId, { name: "Ordered replacement settings" }))]);
let replacement: Promise<PromiseSettledResult<unknown>[]> | undefined;
try {
await waitForLock(0);
replacement = Promise.allSettled([run(service => change === "installs"
? service.putConnectionInstalls(connection.id, { installs: [{ targetType: "agent", targetId: lower }] })
: change === "reassign" ? service.deleteProfile(source.id, { reassignToProfileId: target.id })
: service.finishGalleryAppConnection(companyId, connection.id, { enabledCatalogEntryIds: [], askFirstCatalogEntryIds: [], access: { agentIds: [change === "finish" ? lower : higher] } }))]);
await waitForLock(1);
} finally { release(); await blocker; }
for (const result of [...await settings, ...await replacement!]) if (result.status === "rejected") throw result.reason;
if (change === "installs") expect(await toolAccessService(db).listConnectionInstalls(connection.id, companyId)).toEqual([expect.objectContaining({ targetType: "agent", targetId: lower })]);
else expect(await db.select().from(toolProfileBindings).where(eq(toolProfileBindings.profileId, change === "reassign" ? target.id : source.id))).toEqual([expect.objectContaining({ targetType: "agent", targetId: change === "finish" ? lower : higher })]);
if (change === "activate" || change === "enable" || change === "revive") {
expect((await db.select().from(toolConnections).where(eq(toolConnections.id, connection.id)))[0].status).toBe("active");
expect((await db.select().from(toolConnections).where(eq(toolConnections.id, connection.id)))[0].enabled).toBe(true);
expect((await db.select().from(toolProfiles).where(eq(toolProfiles.id, source.id)))[0].status).toBe("active");
}
}, 15000);
});
@@ -9,7 +9,7 @@ import type { LifecycleEffects } from "./effects.js";
export async function updateAgentConfiguration(tx: Db, effects: LifecycleEffects, id: string, data: AgentConfigurationPatch, options?: UpdateAgentOptions, publications: ActivityPublication[] = []) {
if (!("nestedIndex" in tx)) throw new AgentLifecycleConflict("Configuration integration requires a database transaction");
assertConfigurationFields(data);
const [current] = await tx.select().from(agents).where(eq(agents.id, id)).for("update");
const [current] = await tx.select().from(agents).where(eq(agents.id, id)).for("no key update");
if (!current) return null;
const patch = await effects.prepareConfiguration(tx, current, data, options);
assertConfigurationFields(patch);
+6 -3
View File
@@ -51,7 +51,7 @@ import { instanceSettingsService } from "./instance-settings.js";
import { conflict, forbidden, notFound, unprocessable } from "../errors.js";
import type { RuntimeToolsTokenClaims } from "../runtime-tools-token.js";
import { issueThreadInteractionService } from "./issue-thread-interactions.js";
import { toolAccessService } from "./tool-access.js";
import { lockToolConnectionRuntimeInputs, toolAccessService } from "./tool-access.js";
import { toolAccessPolicyService } from "./tool-access-policy.js";
import { captureRunIdentity } from "./run-identity.js";
import { resolveManagedGitHubIdentitySelection } from "./git-credentials.js";
@@ -994,7 +994,7 @@ export function connectionIntentService(db: Db) {
if (payload.accessRequest.connectionId !== connectionId) throw conflict("Use the connection named in this access request");
throw forbidden("Granting agent tool access requires connection-management authority");
}
await tx.select({ id: toolConnections.id }).from(toolConnections).where(and(eq(toolConnections.id, connectionId), eq(toolConnections.companyId, loaded.issue.companyId))).for("update");
await lockToolConnectionRuntimeInputs(tx, loaded.issue.companyId, connectionId);
let selectedConnection = await txAccess.getConnection(connectionId, loaded.issue.companyId);
const selectedApplication = await txAccess.getApplication(
selectedConnection.applicationId,
@@ -1031,7 +1031,10 @@ export function connectionIntentService(db: Db) {
if (payload.purpose !== "ai") throw conflict("AI authentication cannot satisfy a tool connection request");
let managed;
if (options.validatedAdoption) {
const [agent] = await tx.select().from(agents).where(and(eq(agents.id, payload.requestingAgentId), eq(agents.companyId, loaded.issue.companyId))).for("update");
await tx.select({ id: agents.id }).from(agents)
.where(and(eq(agents.companyId, loaded.issue.companyId), sql`${agents.status} NOT IN ('pending_approval', 'terminated')`))
.orderBy(agents.id).for("no key update");
const [agent] = await tx.select().from(agents).where(and(eq(agents.id, payload.requestingAgentId), eq(agents.companyId, loaded.issue.companyId))).for("no key update");
if (!agent || agent.runtimeConfig.aiConnection || agent.updatedAt.getTime() !== options.validatedAdoption.agentUpdatedAt.getTime()) {
throw conflict("The agent changed during validation. Reload the task and try again.");
}
+2 -11
View File
@@ -11,7 +11,7 @@ import {
import type { EmailConnectionInput, EmailCredentialOption } from "@paperclipai/shared";
import { badRequest, forbidden, HttpError, notFound } from "../errors.js";
import { secretService } from "./secrets.js";
import { toolAccessService } from "./tool-access.js";
import { lockToolConnectionRuntimeInputs, toolAccessService } from "./tool-access.js";
import { agentmailApi } from "./agentmail-api.js";
import { logActivity } from "./activity-log.js";
import type { EmailActor } from "./email-channels.js";
@@ -321,16 +321,7 @@ export function emailConnectionService(
) {
return db.transaction(async (tx) => {
const db = tx as unknown as Db;
await db
.select()
.from(toolConnections)
.where(
and(
eq(toolConnections.id, id),
eq(toolConnections.companyId, companyId),
),
)
.for("update");
await lockToolConnectionRuntimeInputs(db, companyId, id);
const connection = await emailConnectionService(db, fetchImpl).get(companyId, id, actor);
const tools = toolAccessService(db);
const installs = await db
@@ -0,0 +1,24 @@
import { and, eq, or } from "drizzle-orm";
import { toolConnections, toolConnectionInstalls, toolProfiles, toolProfileBindings, type Db } from "@paperclipai/db";
import type { PluginAgentConnectionAssignments } from "@paperclipai/plugin-sdk";
import { unprocessable } from "../errors.js";
export async function pluginAgentConnectionAssignments(db: Db, companyId: string, agentId: string, features: PluginAgentConnectionAssignments["features"]): Promise<PluginAgentConnectionAssignments> {
const [connections, profiles] = await Promise.all([
db.selectDistinct({ id: toolConnections.id, name: toolConnections.name, transport: toolConnections.transport,
enabled: toolConnections.enabled, status: toolConnections.status })
.from(toolConnectionInstalls).innerJoin(toolConnections, and(eq(toolConnections.id, toolConnectionInstalls.connectionId), eq(toolConnections.companyId, companyId)))
.where(and(eq(toolConnectionInstalls.companyId, companyId), or(
and(eq(toolConnectionInstalls.targetType, "agent"), eq(toolConnectionInstalls.targetId, agentId)),
and(eq(toolConnectionInstalls.targetType, "company"), eq(toolConnectionInstalls.targetId, companyId)))))
.orderBy(toolConnections.id).limit(513),
db.selectDistinct({ id: toolProfiles.id }).from(toolProfileBindings)
.innerJoin(toolProfiles, and(eq(toolProfiles.id, toolProfileBindings.profileId), eq(toolProfiles.companyId, companyId), eq(toolProfiles.status, "active")))
.where(and(eq(toolProfileBindings.companyId, companyId), or(
and(eq(toolProfileBindings.targetType, "agent"), eq(toolProfileBindings.targetId, agentId)),
and(eq(toolProfileBindings.targetType, "company"), eq(toolProfileBindings.targetId, companyId)))))
.orderBy(toolProfiles.id).limit(129),
]);
if (connections.length > 512 || profiles.length > 128) throw unprocessable("Agent connection assignments exceed the synchronization limit");
return { features, connections, profileIds: profiles.map(profile => profile.id) };
}
@@ -4,6 +4,7 @@ import type { PluginAgentRuntimeState } from "@paperclipai/plugin-sdk";
import { agentRecordQueries } from "../lib/agent-records.js";
import { conflict, forbidden } from "../errors.js";
import { agentInstructionsService } from "./agent-instructions.js";
import { pluginAgentConnectionAssignments } from "./plugin-agent-connections.js";
import { companySkillService } from "./company-skills.js";
import { withSkillActor } from "./skill-access.js";
import { readPaperclipSkillSyncPreference } from "@paperclipai/adapter-utils/server-utils";
@@ -11,7 +12,7 @@ import { skillVersionSelectionMap } from "./runtime-skill-selections.js";
import { instanceSettingsService } from "./instance-settings.js";
import { agentInstanceSecretService } from "./agent-instance-secrets.js";
async function readState(db: Db, pluginId: string, companyId: string, agentId: string): Promise<Omit<PluginAgentRuntimeState, "skills"> | null> {
async function readState(db: Db, pluginId: string, companyId: string, agentId: string): Promise<Omit<PluginAgentRuntimeState, "skills" | "connectionAssignments"> | null> {
return db.transaction(async tx => {
const [agent] = await tx.select().from(agents)
.where(and(eq(agents.companyId, companyId), eq(agents.id, agentId))).for("share");
@@ -48,9 +49,10 @@ export async function pluginAgentRuntimeState(db: Db, pluginId: string, companyI
selectionConfig: state.agent.adapterConfig, includeLegacy: state.agent.adapterType !== "paperclip_runner", snapshotLatest: true,
versionSelections: skillVersionSelectionMap(preference.desiredSkillEntries, { versionPinsEnabled: settings.enableBetaSkills }),
}));
const connectionAssignments = await pluginAgentConnectionAssignments(db, companyId, agentId, { publicMcp: settings.enablePublicMcp, chatConnectors: settings.enableChatConnectors });
const current = await readState(db, pluginId, companyId, agentId);
if (!current) return null;
if (current.revision === state.revision) return { ...current, skills };
if (current.revision === state.revision) return { ...current, skills, connectionAssignments };
}
throw conflict("Agent runtime changed while resolving synchronization inputs; retry");
}
@@ -1,4 +1,4 @@
import { and, asc, eq, isNull, or } from "drizzle-orm";
import { and, asc, eq, isNull, or, sql } from "drizzle-orm";
import { pluginCompanySettings, pluginConfig, plugins, type Db } from "@paperclipai/db";
import { logger } from "../middleware/logger.js";
import { pluginLifecycleInbox } from "./plugin-lifecycle-inbox.js";
@@ -21,6 +21,7 @@ export function registerPluginLifecycleDelivery(
let pending = true;
let disposed = false;
const retryAt = new Map<string, number>();
const refreshedRuntimeCompanies = new Set<string>();
const retryMs = 30_000;
const delivery = coordinator.register(DELIVERY_QUEUES.pluginLifecycle, {
retryMs,
@@ -50,6 +51,10 @@ export function registerPluginLifecycleDelivery(
const [disabled] = await db.select({ id: pluginCompanySettings.id }).from(pluginCompanySettings)
.where(and(eq(pluginCompanySettings.pluginId, plugin.id), eq(pluginCompanySettings.companyId, companyId), eq(pluginCompanySettings.enabled, false)));
if (disabled) continue;
if (types.includes("agent_runtime") && !refreshedRuntimeCompanies.has(companyId)) {
await db.execute(sql`SELECT paperclip_invalidate_company_agent_runtimes(${companyId}::uuid)`);
refreshedRuntimeCompanies.add(companyId);
}
const inbox = pluginLifecycleInbox(db, plugin.id);
let afterId: string | undefined;
// Visit independent resources past a failed or deferred head.
+78 -34
View File
@@ -3073,6 +3073,29 @@ async function lockManagedAppPolicies(client: ToolAccessMutationDb, companyId: s
)).orderBy(toolPolicies.id).for("no key update");
}
async function lockToolAssignmentAgents(client: ToolAccessMutationDb, companyId: string, targets: ReadonlyArray<{ targetType: string; targetId: string }>) {
const allAgents = targets.some(target => target.targetType === "company" && target.targetId === companyId);
const ids = targets.filter(target => target.targetType === "agent").map(target => target.targetId);
if (!allAgents && !ids.length) return;
await client.select({ id: agents.id }).from(agents)
.where(and(eq(agents.companyId, companyId), sql`${agents.status} NOT IN ('pending_approval', 'terminated')`,
allAgents ? undefined : inArray(agents.id, ids)))
.orderBy(agents.id).for("no key update");
}
export async function lockToolConnectionRuntimeInputs(client: ToolAccessMutationDb, companyId: string, connectionId: string) {
await client.select({ id: toolApplications.id }).from(toolApplications)
.innerJoin(toolConnections, and(eq(toolConnections.applicationId, toolApplications.id), eq(toolConnections.companyId, toolApplications.companyId)))
.where(and(eq(toolConnections.id, connectionId), eq(toolConnections.companyId, companyId)))
.for("update", { of: toolApplications });
await client.select({ id: toolConnections.id }).from(toolConnections)
.where(and(eq(toolConnections.id, connectionId), eq(toolConnections.companyId, companyId))).for("update");
await client.select({ id: toolProfiles.id }).from(toolProfiles)
.where(and(eq(toolProfiles.companyId, companyId), eq(toolProfiles.profileKey, `app:${connectionId}`)))
.for("no key update");
await lockManagedAppPolicies(client, companyId, connectionId);
}
export function toolAccessService(
db: Db,
options: ToolAccessServiceOptions = {},
@@ -5495,6 +5518,7 @@ export function toolAccessService(
companyId: string,
targetType: CreateToolProfileBindingForProfile["targetType"],
targetId: string,
dbClient: ToolAccessMutationDb = db,
) {
if (targetType === "company") {
if (targetId !== companyId)
@@ -5504,7 +5528,7 @@ export function toolAccessService(
return;
}
if (targetType === "agent") {
const [row] = await db
const [row] = await dbClient
.select({ id: agents.id })
.from(agents)
.where(and(eq(agents.id, targetId), eq(agents.companyId, companyId)));
@@ -5515,7 +5539,7 @@ export function toolAccessService(
return;
}
if (targetType === "project") {
const [row] = await db
const [row] = await dbClient
.select({ id: projects.id })
.from(projects)
.where(
@@ -5528,7 +5552,7 @@ export function toolAccessService(
return;
}
if (targetType === "routine") {
const [row] = await db
const [row] = await dbClient
.select({ id: routines.id })
.from(routines)
.where(
@@ -5541,7 +5565,7 @@ export function toolAccessService(
return;
}
if (targetType === "issue") {
const [row] = await db
const [row] = await dbClient
.select({ id: issues.id })
.from(issues)
.where(and(eq(issues.id, targetId), eq(issues.companyId, companyId)));
@@ -5552,7 +5576,7 @@ export function toolAccessService(
return;
}
if (targetType === "gateway") {
const [row] = await db
const [row] = await dbClient
.select({ id: toolMcpGateways.id })
.from(toolMcpGateways)
.where(
@@ -5661,6 +5685,7 @@ export function toolAccessService(
await lockManagedAppPolicies(dbClient, input.connection.companyId, input.connection.id);
const createdProfile = !profile;
const resetToRecommendedDefaults = !profile || profile.status !== "active";
if (resetToRecommendedDefaults) await lockToolAssignmentAgents(dbClient, input.connection.companyId, [{ targetType: "company", targetId: input.connection.companyId }]);
if (!profile) {
const [sameName] = await dbClient
.select({ id: toolProfiles.id })
@@ -6003,14 +6028,14 @@ export function toolAccessService(
return created;
}
async function getProfileRow(profileId: string, companyId?: string) {
async function getProfileRow(profileId: string, companyId?: string, dbClient: ToolAccessMutationDb = db) {
const where = companyId
? and(
eq(toolProfiles.id, profileId),
eq(toolProfiles.companyId, companyId),
)
: eq(toolProfiles.id, profileId);
const [row] = await db.select().from(toolProfiles).where(where);
const [row] = await dbClient.select().from(toolProfiles).where(where);
if (!row) throw notFound("Tool profile not found");
return row;
}
@@ -6018,8 +6043,9 @@ export function toolAccessService(
async function profileDetails(
profileId: string,
companyId?: string,
dbClient: ToolAccessMutationDb = db,
): Promise<ToolProfileWithDetails> {
const profile = await getProfileRow(profileId, companyId);
const profile = await getProfileRow(profileId, companyId, dbClient);
const [
entries,
bindings,
@@ -6028,7 +6054,7 @@ export function toolAccessService(
applications,
connections,
] = await Promise.all([
db
dbClient
.select()
.from(toolProfileEntries)
.where(
@@ -6038,7 +6064,7 @@ export function toolAccessService(
),
)
.orderBy(asc(toolProfileEntries.createdAt)),
db
dbClient
.select()
.from(toolProfileBindings)
.where(
@@ -6051,7 +6077,7 @@ export function toolAccessService(
asc(toolProfileBindings.priority),
asc(toolProfileBindings.createdAt),
),
db
dbClient
.select()
.from(toolCatalogEntries)
.where(
@@ -6060,15 +6086,15 @@ export function toolAccessService(
eq(toolCatalogEntries.status, "active"),
),
),
db
dbClient
.select({ id: agents.id })
.from(agents)
.where(eq(agents.companyId, profile.companyId)),
db
dbClient
.select()
.from(toolApplications)
.where(eq(toolApplications.companyId, profile.companyId)),
db
dbClient
.select()
.from(toolConnections)
.where(eq(toolConnections.companyId, profile.companyId)),
@@ -14022,8 +14048,8 @@ export function toolAccessService(
await tx.select({ id: toolApplications.id }).from(toolApplications)
.where(and(eq(toolApplications.id, connection.applicationId), eq(toolApplications.companyId, companyId)))
.for("update");
await tx
.select({ id: toolConnections.id })
const [lockedConnection] = await tx
.select({ id: toolConnections.id, status: toolConnections.status, enabled: toolConnections.enabled })
.from(toolConnections)
.where(
and(
@@ -14032,6 +14058,7 @@ export function toolAccessService(
),
)
.for("update");
if (!lockedConnection) throw notFound("Tool connection not found");
const [existingProfile] = await tx
.select()
.from(toolProfiles)
@@ -14043,6 +14070,13 @@ export function toolAccessService(
)
.limit(1).for("no key update");
await lockManagedAppPolicies(tx, companyId, connection.id);
const existingBindings = existingProfile ? await tx.select().from(toolProfileBindings)
.where(and(eq(toolProfileBindings.companyId, companyId), eq(toolProfileBindings.profileId, existingProfile.id))) : [];
const existingInstalls = await tx.select().from(toolConnectionInstalls)
.where(and(eq(toolConnectionInstalls.companyId, companyId), eq(toolConnectionInstalls.connectionId, connection.id)));
const companyTargets = lockedConnection.status !== "active" || !lockedConnection.enabled || (existingProfile && existingProfile.status !== "active")
? [{ targetType: "company", targetId: companyId }] : [];
await lockToolAssignmentAgents(tx, companyId, [...existingBindings, ...bindingInputs, ...existingInstalls, ...companyTargets]);
// A renamed connection can leave its old profile name in use. Resolve
// names against profiles too, including when finalizing an installed draft.
const otherProfiles = await tx
@@ -14059,11 +14093,7 @@ export function toolAccessService(
let profileId: string;
if (existingProfile) {
if (input.preserveExistingAccess) {
const priorBindings = await tx
.select()
.from(toolProfileBindings)
.where(eq(toolProfileBindings.profileId, existingProfile.id));
for (const prior of priorBindings)
for (const prior of existingBindings)
if (
!bindingInputs.some(
(binding) =>
@@ -14187,9 +14217,6 @@ export function toolAccessService(
// These connectors expose one agent-access choice. Commit installation
// reach and permission bindings together, including an empty selection.
if (isRemoteMcpConnectorMethod(connection.config.sourceTemplateKey, connection.config.connectionMethodKey)) {
const existingInstalls = await tx.select().from(toolConnectionInstalls).where(and(
eq(toolConnectionInstalls.companyId, companyId), eq(toolConnectionInstalls.connectionId, connection.id),
));
const desired = new Map(bindingInputs.flatMap((binding) => binding.targetType === "company" || binding.targetType === "agent"
? [[`${binding.targetType}:${binding.targetId}`, { targetType: binding.targetType, targetId: binding.targetId }] as const] : []));
const removed = existingInstalls.filter((install) => !desired.has(`${install.targetType}:${install.targetId}`));
@@ -18847,6 +18874,7 @@ export function toolAccessService(
profileId: string;
}> = [];
await db.transaction(async (tx) => {
await lockToolConnectionRuntimeInputs(tx, connection.companyId, connection.id);
if (markMcpAccessConfigured) {
// Preserve an explicit zero-agent selection through OAuth. Merge the
// marker in SQL so concurrent credential/config changes are retained.
@@ -18873,6 +18901,10 @@ export function toolAccessService(
(install) => `${install.targetType}:${install.targetId}`,
),
);
const bindings = await tx.select({ targetType: toolProfileBindings.targetType, targetId: toolProfileBindings.targetId })
.from(toolProfileBindings).innerJoin(toolProfiles, eq(toolProfiles.id, toolProfileBindings.profileId))
.where(and(eq(toolProfileBindings.companyId, connection.companyId), eq(toolProfiles.companyId, connection.companyId), eq(toolProfiles.profileKey, `app:${connection.id}`)));
await lockToolAssignmentAgents(tx, connection.companyId, [...existing, ...requested.values(), ...bindings]);
const removals = existing.filter(
(install) =>
!requested.has(`${install.targetType}:${install.targetId}`),
@@ -19753,14 +19785,19 @@ export function toolAccessService(
summary: ToolProfileSummary;
reassignedToProfileId: string | null;
reassignedBindingCount: number;
}> => {
const existing = await getProfileRow(profileId);
}> => db.transaction(async tx => {
const existing = await getProfileRow(profileId, undefined, tx);
const db = tx as unknown as Db;
await tx.select({ id: toolProfiles.id }).from(toolProfiles)
.where(and(eq(toolProfiles.companyId, existing.companyId), inArray(toolProfiles.id, [profileId, ...(input.reassignToProfileId ? [input.reassignToProfileId] : [])])))
.orderBy(toolProfiles.id).for("update");
await lockToolAssignmentAgents(tx, existing.companyId, [{ targetType: "company", targetId: existing.companyId }]);
if (input.force && input.reassignToProfileId) {
throw badRequest(
"Use either force or reassignToProfileId when deleting a tool profile, not both",
);
}
const details = await profileDetails(existing.id, existing.companyId);
const details = await profileDetails(existing.id, existing.companyId, tx);
if (
details.summary.isCompanyDefault &&
!input.force &&
@@ -19782,6 +19819,7 @@ export function toolAccessService(
const target = await getProfileRow(
input.reassignToProfileId,
existing.companyId,
tx,
);
if (target.status !== "active") {
throw unprocessable(
@@ -19838,7 +19876,7 @@ export function toolAccessService(
reassignedToProfileId: input.reassignToProfileId ?? null,
reassignedBindingCount,
};
},
}),
addProfileEntry: async (
profileId: string,
@@ -19936,12 +19974,15 @@ export function toolAccessService(
profileId: string,
input: CreateToolProfileBindingForProfile,
actor?: ActorInfo,
): Promise<ToolProfileBinding> => {
const profile = await getProfileRow(profileId);
): Promise<ToolProfileBinding> => db.transaction(async tx => {
const db = tx as unknown as Db;
const [profile] = await tx.select().from(toolProfiles).where(eq(toolProfiles.id, profileId)).for("no key update");
if (!profile) throw notFound("Tool profile not found");
await assertTargetExists(
profile.companyId,
input.targetType,
input.targetId,
tx,
);
const [row] = await db
.insert(toolProfileBindings)
@@ -19963,17 +20004,20 @@ export function toolAccessService(
.set({ updatedAt: new Date() })
.where(eq(toolProfiles.id, profile.id));
return toProfileBinding(row);
},
}),
unbindProfile: async (
profileId: string,
input: UnbindToolProfileBinding,
): Promise<{ unbound: number }> => {
const profile = await getProfileRow(profileId);
): Promise<{ unbound: number }> => db.transaction(async tx => {
const db = tx as unknown as Db;
const [profile] = await tx.select().from(toolProfiles).where(eq(toolProfiles.id, profileId)).for("no key update");
if (!profile) throw notFound("Tool profile not found");
await assertTargetExists(
profile.companyId,
input.targetType,
input.targetId,
tx,
);
const rows = await db
.delete(toolProfileBindings)
@@ -19993,7 +20037,7 @@ export function toolAccessService(
.where(eq(toolProfiles.id, profile.id));
}
return { unbound: rows.length };
},
}),
getEffectiveProfilesForAgent: async (
companyId: string,