feat(plugins): invalidate runtimes when connector access changes (#16605)

## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Runtime preparation depends on access and credential state as well as tool definitions. > - Grant, membership, secret, and endpoint changes need committed invalidation. > - Their writers must acquire locks before journal fanout can lock agent rows. > - This pull request adds access hooks with the corresponding transaction guards. > - Plugins can refresh preparation immediately while task-time authorization remains authoritative. ## Linked Issues or Issue Description **What happened?** Grant changes, credential rotation and revocation, company access changes, and connector endpoint configuration do not request runtime synchronization. Journal fanout exposes lock inversions between credential persistence, access cleanup, and connection assignment writers. **Expected behavior** Committed access changes request runtime preparation without polling or credential values in the journal. Agent-specific secret bindings and new personal delegations affect their target agent. Company access changes remain conservative. Provider bookkeeping does not create preparation work. **Steps to reproduce** 1. Change an audience, delegation, membership permission, referenced credential, vault configuration, or endpoint instruction. 2. Expect a committed runtime event in the correct company and scope. 3. Run reconnect, credential refresh, SSH setup, or membership cleanup concurrently with related writes. 4. Expect both operations to preserve access checks and finish without a lock-order cycle. **Additional context** Final feature extraction from #16503. The earlier slices, #16604, and the separate delivery-test correction #16871 are merged. This PR is now the active change against master. The rebase preserves the merged assignment and definition recipient rules, Unicode selector semantics, and profile/reference transaction guards. ## What Changed - Add migration 0356 for grants, audiences, delegation, membership/permissions, secret bindings, referenced credentials and vaults, and chat/email endpoint configuration. - Keep events content-free. Ignore health, delivery, publication, Vercel verification, and GitHub webhook bookkeeping while retaining credential and repository-access changes. - Target agent secret bindings and all personal delegation changes to their old and new target agents. Coalesce delegation changes per transaction. - Lock existing grants before connections and credential persistence in OAuth and cleanup writers. - Reject superseded refresh leases before rotating stored credentials. Ignore lease acquisition and release when access metadata is unchanged. - Serialize audience replacement, validate membership, and preserve unchanged audiences. - Lock membership before permission-only saves and grant before Railway SSH setup. Individual delegation revocation locks its grant and target. Whole-grant and membership cleanup lock company recipients before deleting scoped delegations. - Lock grants and connection inputs before static-key reconnect writes. Lock company agents before cascading secret deletion; use a nonblocking secret lock and bounded retries to avoid a new cycle with credential writers. - Add PostgreSQL concurrency coverage for access writers and endpoint publication/configuration behavior. - Preserve the owner lock for plugin instance-secret removal; these secrets cannot use ordinary bindings. - Align Slack search refresh, callbacks, and disconnect with grant-before-connection locking. Recheck newly discovered grants without waiting before provider work. - Retry complete membership transactions when a credential deletion candidate is busy. Cover all four cleanup entry points and process company access in ID order. - Document final invalidation coverage and task-time authorization boundaries. ## Verification - Current head: `78aace55bef7f52bc5742b131e0b3961eb5c0f11`. All 1,061 tests across 23 affected suites pass. The named Slack authorization integration scenario also passes; it exercises authorization, refresh, disconnect, and rejection of a late callback. Server/database typechecks, migration safety and numbering, five-snapshot retention, feature maps, and diff checks pass. The complete PR diff was reviewed for secrets and personal data before push. Final acceptance: [CI run 39308303837, attempt 2](https://github.com/paperclipai/paperclip/actions/runs/39308303837) is green, with 55 successful checks and two optional Storybook skips. The current-head Greptile reassessment returned **Apex 5/5** at 19:16 UTC on 2026-10-19. Its full summary reports no actionable issues. Security passed, all threads are resolved, and GitHub reports no merge conflicts. The reassessment was requested as a standard review after the fixes; Greptile retained the Apex tier. - On this head, [CI run 39308303837](https://github.com/paperclipai/paperclip/actions/runs/39308303837) failed the unchanged adapter-utils HTTP/2 fixture with `EADDRINUSE` before READY. Its closed probe left a port handoff race. That test and its gateway are byte-identical on master. No upstream fix was found. The deterministic fixture correction is isolated in #16888; no adapter test changes are included here. The failed jobs recovered on retry without a feature source change. The original failure remains documented here, and #16888 carries the fixture correction separately. That retry demonstrates recovery; the held-probe negative control establishes the port handoff race. - Rebased onto master `194fc11406`, including merged #16604 and #16871. Replayed only the access-hook commit. Resolved OAuth, test, and documentation conflicts while retaining both sets of tests. Migration 0356 preserves the merged assignment/definition recipient rules, Unicode selector semantics, and profile/reference guards. - Apex returned 3/5 on `0fb65bc27`. Negative controls reproduce its Slack delegation and member-cleanup deadlocks (`40P01`) and unwanted Vercel/GitHub bookkeeping events. Slack grant writers now take their locks before the connection. A separate late-grant control reproduces `40P01` without the post-lock recheck; the complete guard passes. All four membership cleanup commands retry from a fresh transaction when the credential is busy. Tests verify successful cleanup and no journal rows from failed attempts. Credential IDs and repository-access changes still notify runtimes. - Earlier review corrections remain covered: static-key reconnect and cascading secret removal each reproduce `40P01` without their guards. Grant lease acquisition/release is ignored. Secret removal releases agent locks when a credential writer is busy. Instance-secret removal retains the owner lock and passes its public plugin API contention regression. Delegation updates and revocations target old/new agents; unrelated-peer, coalescing, unchanged-update, and rollback cases pass. Three negative controls verify the required whole-grant and membership cleanup guards after narrowing delegation scope. - Preserve prior acceptance failures: the first rebased 20-suite run passed 1,025/1,026; its only failure was a draft uninstall fixture blocking a random peer. Blocking the assigned agent corrected that assumption, and all 44 access/assignment tests passed. Later, all 1,044 tests across 21 suites, [CI run 39302185527](https://github.com/paperclipai/paperclip/actions/runs/39302185527), standard Greptile 5/5, and security passed on `0fb65bc27`. That did not clear the later Apex findings described above. Draft run 38975428081 failed a server shard; its cause was not re-investigated. A later green run does not establish that earlier cause. - Local fixture corrections are recorded separately from product failures: an assignment filename filter initially matched no file; the actual suite was included in full runs. A delegation fixture omitted its required creator, and a guard run omitted a query import. The new global-access cleanup fixture now uses its own user to avoid earlier fixtures' memberships. All were corrected before publishing their final iterations. - Local checks target affected services and real PostgreSQL boundaries. A full local build and repository-wide test run were not performed. Hosted CI supplies broader verification. Provider responses in tests are controlled; live provider behavior was not qualified. ## Risks Migration 0356 adds triggers without changing table schema. Apply it through the normal migration flow. Most access changes retain synchronous company-wide invalidation. Company fanout takes O(eligible company agents) locks and journal rows, without a fanout ceiling or mutation quota. Agent secret bindings and individual delegation changes retain narrower recipients. Events contain no credentials. Physical removal of non-instance secrets takes ordered locks on all company agents before cascading into bindings. This costs O(company agents) locks even for an otherwise unreferenced secret. Instance secrets cannot use ordinary bindings and retain their owner lock. A busy secret lock releases the transaction guards and retries up to three times. Continued contention returns a retryable conflict. Provider deletion stays outside that transaction; a retry resumes removal of an already disabled secret. Membership cleanup uses nonblocking credential guards and retries the full membership transaction. Slack search adds two grant queries, scoped to the subject and connection. Its retries occur before provider work or credential persistence. Each retry loop is limited to three attempts. Invalidation requests preparation; task-time authorization and provider session activation remain authoritative. Provider calls use fixtures; no live provider qualification is claimed. ## 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>
2026-10-21 00:05:49 +02:00 · 2026-10-19 12:49:11 -07:00
1 parent 06d4fe3121
commit 313fb265db
@@ -1034,6 +1034,12 @@ an agent's runtime; its assignment does. Entry changes resolve standing profile
 bindings; policies resolve old and new agent selectors. Both retain scoped
 recipients. Credential-only fields remain outside this migration's coverage.

+Migration 0356 adds access, referenced credentials and vaults, and chat/email
+endpoint configuration to connector runtime invalidation. Grant and membership
+changes retain company scope. Agent secret bindings and new personal delegations
+use agent scope. Credential values remain outside the journal. Endpoint activity,
+vault health, and other bookkeeping fields remain excluded.
+
 ## Retired Pipelines storage

 The Pipelines experiment has been removed from the application, API, CLI, and
@@ -161,6 +161,56 @@ the same transaction.
 The bounded assignment snapshot remains the API described above. An invalidation
 can advance its revision without changing its assignment fields. Consumers must
 use their existing authorized readers for tool definitions and configuration.
-Credential-only changes, grants, secrets, and endpoint configuration still need
-the access hooks supplied by a later migration before consumers can rely on
-their immediate synchronization.
+
+## Access, credentials, and endpoints
+
+Migration 0356 adds connection grants, grant audiences and delegations, company
+membership and permissions, secret bindings, chat/email endpoint configuration,
+and referenced credential rotation or revocation. Vault configuration and status
+changes invalidate companies whose connector credentials reference that vault.
+Vault health checks, OAuth grant refresh leases, Vercel verification timestamps,
+GitHub webhook health, and endpoint publication timestamps do not request work.
+Credential identifiers, expiry, and repository access changes remain covered.
+Events contain no credential values, phone numbers, or provider payloads.
+
+Access changes conservatively invalidate eligible company agents because
+responsible-user and task-time access can vary across runs. Agent secret bindings
+instead invalidate their owning agent. Personal delegation creation, reassignment, and revocation invalidate only
+their old and new target agents. Whole-grant revocation and membership cleanup
+retain company invalidation. Unchanged grant audiences preserve their rows after
+membership validation.
+
+OAuth persistence and membership cleanup lock affected grants before connection
+inputs and credential mutations. Each affected set is locked in ID order. Task
+completion locks existing grants before assignment inputs. Permission-only saves
+lock membership before replacing permissions, as role and status saves do.
+Slack search refresh, callbacks, and disconnect lock only the subject's grants
+before their connection. After locking the connection, they recheck the grants
+without waiting. A newly discovered busy grant rolls back and retries before
+provider work or credential persistence. Three failed attempts return a retryable
+conflict. Each operation adds two grant queries, bounded to the subject and
+connection. The existing connection lock still serializes provider refresh work.
+
+Membership cleanup locks credential deletion candidates without waiting before
+company agent locks. A busy credential rolls back the complete membership
+transaction, including its membership and grant locks. All four cleanup entry
+points re-read their state on retry, with at most three attempts. Company-access
+changes process companies in ID order.
+
+Personal delegation creation locks its grant, connection reference, and target
+agent before insertion. Delegation revocation locks its grant and target before
+deletion. Whole-grant and membership cleanup lock eligible company agents in ID
+order before deleting scoped delegations. Static-key reconnect locks grants and connection inputs
+before credential writes. Secret removal locks company agents in ID order before
+cascading into bindings. Plugin instance secrets cannot use ordinary bindings,
+so their removal retains its owner lock without a company sweep. It tries the secret lock without waiting, releases its
+transaction locks if that lock is busy, and retries up to three times. Continued
+contention returns a retryable conflict. Provider deletion stays outside that
+transaction; retrying resumes removal of the already disabled secret. These
+guards accompany the hooks that require them.
+
+The runtime snapshot still excludes credentials and arbitrary configuration.
+Task-scoped profile bindings, responsible-user consent, gateway sessions,
+automatic connector skill overlays, and connection instructions remain in the
+existing run preparation path. Invalidation does not create a run identity or
+activate a provider session. Transport consumers must preserve those boundaries.
@@ -48,6 +48,13 @@ bookkeeping, profile-entry edits concurrent with setup finalization, and profile
 replacement concurrent with application deletion using PostgreSQL. They do not
 qualify live provider access.

+[Connector access runtime tests](../../server/src/__tests__/agent-runtime-connector-access.test.ts)
+cover grants, delegation, membership cleanup, referenced vault changes, and
+agent-scoped secret bindings. OAuth and Railway tests exercise the corresponding
+credential lock order. Slack search and membership cleanup tests cover lock
+contention and complete transaction retries. Provider calls use fixtures; no live credentials are
+qualified by this coverage.
+
 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,175 @@
+CREATE OR REPLACE FUNCTION paperclip_connector_runtime_changed() RETURNS trigger
+LANGUAGE plpgsql SET search_path = public, pg_temp AS $$
+DECLARE company uuid; query text; target record; binding record; candidate uuid;
+  old_projection text := '(to_jsonb(o) - $1)'; new_projection text := '(to_jsonb(n) - $1)';
+  ignored text[] := ARRAY['updated_at', 'created_at', 'last_used_at', 'last_resolved_at', 'last_health_at',
+    'health_status', 'health_checked_at', 'health_message', 'last_catalog_refresh_at', 'last_seen_at', 'last_error', 'last_event_at', 'last_publication_at',
+    'last_sync_at', 'sync_checkpoint', 'activation_at', 'lease_token', 'lease_expires_at'];
+BEGIN
+  IF TG_TABLE_NAME = 'tool_profiles' AND TG_OP = 'INSERT' THEN RETURN NULL; END IF;
+  IF TG_TABLE_NAME = 'connection_grant_delegations' THEN
+    IF TG_OP = 'INSERT' THEN query := 'SELECT DISTINCT company_id, agent_id FROM new_rows ORDER BY company_id, agent_id';
+    ELSIF TG_OP = 'DELETE' THEN query := 'SELECT DISTINCT company_id, agent_id FROM old_rows ORDER BY company_id, agent_id';
+    ELSE query := 'WITH changed AS (
+      SELECT to_jsonb(o) AS previous, to_jsonb(n) AS current FROM old_rows o FULL JOIN new_rows n ON o.id = n.id
+      WHERE (to_jsonb(o) - $1) IS DISTINCT FROM (to_jsonb(n) - $1)), targets AS (
+      SELECT previous AS target FROM changed UNION SELECT current FROM changed)
+      SELECT DISTINCT (target->>''company_id'')::uuid AS company_id, (target->>''agent_id'')::uuid AS agent_id
+      FROM targets WHERE target IS NOT NULL ORDER BY company_id, agent_id';
+    END IF;
+    FOR binding IN EXECUTE query USING ignored LOOP
+      PERFORM paperclip_invalidate_connector_agent(binding.company_id, binding.agent_id);
+    END LOOP;
+    RETURN NULL;
+  END IF;
+  IF TG_TABLE_NAME = 'company_secret_bindings' THEN
+    IF TG_OP = 'INSERT' THEN query := 'SELECT DISTINCT company_id, target_type, target_id FROM new_rows ORDER BY company_id, target_id';
+    ELSIF TG_OP = 'DELETE' THEN query := 'SELECT DISTINCT company_id, target_type, target_id FROM old_rows ORDER BY company_id, target_id';
+    ELSE query := 'WITH changed AS (
+      SELECT to_jsonb(o) AS previous, to_jsonb(n) AS current FROM old_rows o FULL JOIN new_rows n ON o.id = n.id
+      WHERE (to_jsonb(o) - $1) IS DISTINCT FROM (to_jsonb(n) - $1)), targets AS (
+      SELECT previous AS target FROM changed UNION SELECT current FROM changed)
+      SELECT DISTINCT (target->>''company_id'')::uuid AS company_id, target->>''target_type'' AS target_type, target->>''target_id'' AS target_id
+      FROM targets WHERE target IS NOT NULL ORDER BY company_id, target_id';
+    END IF;
+    FOR binding IN EXECUTE query USING ignored LOOP
+      IF binding.target_type = 'agent' THEN
+        SELECT id INTO candidate FROM agents WHERE company_id = binding.company_id AND id::text = binding.target_id;
+        IF FOUND THEN PERFORM paperclip_record_agent_runtime(binding.company_id, candidate); END IF;
+      ELSE PERFORM paperclip_invalidate_connector_company(binding.company_id);
+      END IF;
+    END LOOP;
+    RETURN NULL;
+  END IF;
+  IF TG_TABLE_NAME = 'connection_grants' THEN
+    old_projection := old_projection || ' #- ''{provider_tenant,oauth,refreshLease}'' #- ''{external_credential,lastVerifiedAt}'' #- ''{provider_tenant,github,lastWebhookAt}'' #- ''{provider_tenant,github,webhookHealth}''';
+    new_projection := new_projection || ' #- ''{provider_tenant,oauth,refreshLease}'' #- ''{external_credential,lastVerifiedAt}'' #- ''{provider_tenant,github,lastWebhookAt}'' #- ''{provider_tenant,github,webhookHealth}''';
+  END IF;
+  IF TG_TABLE_NAME = 'tool_policies' THEN
+    old_projection := old_projection || ' #- ''{config,trustRule,hitCount}'' #- ''{config,trustRule,lastHitAt}''';
+    new_projection := new_projection || ' #- ''{config,trustRule,hitCount}'' #- ''{config,trustRule,lastHitAt}''';
+  END IF;
+  IF TG_TABLE_NAME IN ('tool_profile_bindings', 'tool_connection_installs', 'tool_profile_entries', 'tool_policies') THEN
+    IF TG_TABLE_NAME IN ('tool_profile_entries', 'tool_policies') THEN
+      IF TG_OP = 'INSERT' THEN query := 'SELECT to_jsonb(n) AS row_data FROM new_rows n';
+      ELSIF TG_OP = 'DELETE' THEN query := 'SELECT to_jsonb(o) AS row_data FROM old_rows o';
+      ELSE query := format('WITH changed AS (
+        SELECT to_jsonb(o) AS previous, to_jsonb(n) AS current FROM old_rows o FULL JOIN new_rows n ON o.%I = n.%I
+        WHERE %s IS DISTINCT FROM %s)
+        SELECT previous AS row_data FROM changed WHERE previous IS NOT NULL UNION
+        SELECT current AS row_data FROM changed WHERE current IS NOT NULL', TG_ARGV[0], TG_ARGV[0], old_projection, new_projection);
+      END IF;
+      IF TG_TABLE_NAME = 'tool_profile_entries' THEN
+        query := 'SELECT b.company_id, b.target_type, b.target_id FROM (' || query || ') changed
+          JOIN tool_profile_bindings b ON b.company_id = (row_data->>''company_id'')::uuid
+          AND b.profile_id = (row_data->>''profile_id'')::uuid';
+      ELSE
+        query := 'SELECT (row_data->>''company_id'')::uuid AS company_id, t.target_type, t.target_id
+          FROM (' || query || ') changed CROSS JOIN LATERAL
+          paperclip_connector_policy_targets((row_data->>''company_id'')::uuid, row_data->''selectors'') t';
+      END IF;
+    ELSIF 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' USING ignored 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 USING ignored LOOP
+    PERFORM paperclip_invalidate_connector_company(company);
+  END LOOP;
+  RETURN NULL;
+END;
+$$;
+--> statement-breakpoint
+DO $$
+DECLARE relation text; key_column text;
+BEGIN
+  FOREACH relation IN ARRAY ARRAY['tool_applications', 'tool_connections', 'connection_grants', 'connection_grant_members',
+    'connection_grant_delegations', 'tool_connection_installs', 'tool_catalog_entries', 'tool_profiles', 'tool_profile_entries',
+    'tool_profile_bindings', 'tool_policies', 'tool_mcp_gateways', 'company_memberships', 'principal_permission_grants',
+    'company_secret_bindings', 'chat_endpoints', 'chat_endpoint_resources', 'chat_agent_routes', 'chat_identity_links',
+    'chat_voice_callbacks', 'email_endpoints'] LOOP
+    key_column := CASE WHEN relation = 'email_endpoints' THEN 'endpoint_id' ELSE '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(%L)', relation, key_column);
+    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(%L)', relation, key_column);
+    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(%L)', relation, key_column);
+  END LOOP;
+END;
+$$;
+--> statement-breakpoint
+CREATE OR REPLACE FUNCTION paperclip_connector_uses_secret(p_company uuid, p_secret uuid) RETURNS boolean
+LANGUAGE sql STABLE SET search_path = public, pg_temp AS $$
+  SELECT EXISTS (SELECT 1 FROM tool_connections WHERE company_id = p_company AND (
+      credential_refs @> jsonb_build_array(jsonb_build_object('secretId', p_secret::text)) OR
+      credential_secret_refs @> jsonb_build_array(jsonb_build_object('secretId', p_secret::text)))) OR
+    EXISTS (SELECT 1 FROM connection_grants WHERE company_id = p_company AND
+      credential_secret_refs @> jsonb_build_array(jsonb_build_object('secretId', p_secret::text)));
+$$;
+--> statement-breakpoint
+CREATE OR REPLACE FUNCTION paperclip_connector_secret_changed() RETURNS trigger
+LANGUAGE plpgsql SET search_path = public, pg_temp AS $$
+DECLARE secret uuid; company uuid;
+BEGIN
+  IF TG_OP = 'UPDATE' AND jsonb_build_array(OLD.status, OLD.latest_version, OLD.deleted_at, OLD.external_ref, OLD.provider_config_id, OLD.provider_metadata)
+    IS NOT DISTINCT FROM jsonb_build_array(NEW.status, NEW.latest_version, NEW.deleted_at, NEW.external_ref, NEW.provider_config_id, NEW.provider_metadata) THEN RETURN NEW; END IF;
+  IF TG_OP = 'DELETE' THEN secret := OLD.id; company := OLD.company_id;
+  ELSE secret := NEW.id; company := NEW.company_id; END IF;
+  IF paperclip_connector_uses_secret(company, secret) THEN
+    PERFORM paperclip_invalidate_connector_company(company);
+  END IF;
+  IF TG_OP = 'DELETE' THEN RETURN OLD; END IF;
+  RETURN NEW;
+END;
+$$;
+--> statement-breakpoint
+DROP TRIGGER IF EXISTS paperclip_connector_secret_changed ON company_secrets;
+CREATE TRIGGER paperclip_connector_secret_changed AFTER UPDATE OR DELETE ON company_secrets
+  FOR EACH ROW EXECUTE FUNCTION paperclip_connector_secret_changed();
+--> statement-breakpoint
+CREATE OR REPLACE FUNCTION paperclip_connector_vault_changed() RETURNS trigger
+LANGUAGE plpgsql SET search_path = public, pg_temp AS $$
+DECLARE vault uuid; company uuid;
+BEGIN
+  IF TG_OP = 'UPDATE' AND jsonb_build_array(OLD.provider, OLD.status, OLD.config, OLD.is_default, OLD.disabled_at)
+    IS NOT DISTINCT FROM jsonb_build_array(NEW.provider, NEW.status, NEW.config, NEW.is_default, NEW.disabled_at) THEN RETURN NEW; END IF;
+  IF TG_OP = 'DELETE' THEN vault := OLD.id; ELSE vault := NEW.id; END IF;
+  FOR company IN SELECT DISTINCT s.company_id FROM company_secrets s WHERE s.provider_config_id = vault
+    AND paperclip_connector_uses_secret(s.company_id, s.id) ORDER BY s.company_id LOOP
+    PERFORM paperclip_invalidate_connector_company(company);
+  END LOOP;
+  IF TG_OP = 'DELETE' THEN RETURN OLD; END IF;
+  RETURN NEW;
+END;
+$$;
+--> statement-breakpoint
+DROP TRIGGER IF EXISTS paperclip_connector_vault_changed ON company_secret_provider_configs;
+CREATE TRIGGER paperclip_connector_vault_changed AFTER INSERT OR UPDATE OR DELETE ON company_secret_provider_configs
+  FOR EACH ROW EXECUTE FUNCTION paperclip_connector_vault_changed();
+--> statement-breakpoint
@@ -2479,6 +2479,13 @@
      "when": 1792188236099,
      "tag": "0355_agent_tool_definition_hooks",
      "breakpoints": true
+    },
+    {
+      "idx": 356,
+      "version": "7",
+      "when": 1792188583719,
+      "tag": "0356_agent_connector_access_hooks",
+      "breakpoints": true
    }
  ]
 }
@@ -1,7 +1,7 @@
 import { randomBytes, randomUUID } from "node:crypto";
 import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
-import { eq } from "drizzle-orm";
-import { agents, companies, companySecrets, companySecretVersions, activityLog, plugins, createDb, type Db } from "@paperclipai/db";
+import { eq, sql } from "drizzle-orm";
+import { agents, companies, companySecrets, companySecretVersions, activityLog, plugins, toolApplications, toolConnections, createDb, type Db } from "@paperclipai/db";
 import type { AgentLifecycleRequest } from "@paperclipai/shared";
 import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
 import { agentInstanceSecretService } from "../services/agent-instance-secrets.js";
@@ -32,11 +32,11 @@ const support = await getEmbeddedPostgresTestSupport();
      manifestJson: { agentLifecycle: true, capabilities: ["agents.lifecycle.manage"] } as never });
  }, 90_000);
  afterAll(async () => { await database?.cleanup(); vi.unstubAllEnvs(); });
-  async function fixture(ownerPluginId = pluginId) {
+  async function fixture(ownerPluginId = pluginId, agentId = randomUUID()) {
    const operationId = randomUUID();
    const owner = randomUUID();
    const now = new Date();
-    const [agent] = await db.insert(agents).values({ companyId, name: "Secret owner", lifecycleState: "preparing", lifecycleVersion: 1,
+    const [agent] = await db.insert(agents).values({ id: agentId, companyId, name: "Secret owner", lifecycleState: "preparing", lifecycleVersion: 1,
      lifecycleRequiredPluginIds: [ownerPluginId], lifecycleOperation: { id: operationId, hostComplete: true, completedPluginIds: [], attempts: 0,
        leaseOwner: owner, leaseUntil: new Date(now.getTime() + 120_000).toISOString() } }).returning();
    const request: AgentLifecycleRequest = { companyId, agentId: agent.id, operationId, version: 1, phase: "preparing" };
@@ -66,6 +66,44 @@ const support = await getEmbeddedPostgresTestSupport();
    expect(JSON.stringify(await db.select().from(activityLog).where(eq(activityLog.entityId, f.agent.id)))).not.toContain(replacement);
  });

+  it("removes a plugin instance secret while a company connection save waits for its owner", async () => {
+    const [lower, higher] = [randomUUID(), randomUUID()].sort();
+    await db.insert(agents).values({ id: lower, companyId, name: "Unrelated lower agent" });
+    const f = await fixture(pluginId, higher);
+    await f.service.apply(f.request, pluginId, { runtime });
+    const [application] = await db.insert(toolApplications).values({ companyId, name: "Instance removal fixture", type: "mcp_http" }).returning();
+    const [connection] = await db.insert(toolConnections).values({ companyId, applicationId: application.id, name: "Instance removal fixture", uid: randomUUID(), transport: "mcp_remote", authKind: "none", status: "active" }).returning();
+    let paused!: () => void; let resume!: () => void; let started!: (pid: number) => void;
+    const held = new Promise<void>(resolve => { paused = resolve; });
+    const mayRemove = new Promise<void>(resolve => { resume = resolve; });
+    const settingsStarted = new Promise<number>(resolve => { started = resolve; });
+    const removeProvider = localEncryptedProvider.deleteOrArchive.bind(localEncryptedProvider);
+    const providerSpy = vi.spyOn(localEncryptedProvider, "deleteOrArchive").mockImplementation(async input => {
+      paused(); await mayRemove;
+      return removeProvider(input);
+    });
+    const removal = f.service.apply(f.request, pluginId, { runtime: null });
+    await held;
+    const settings = db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      const [row] = await tx.execute(sql`SELECT pg_backend_pid() AS pid`);
+      started(Number(row.pid));
+      await tx.update(toolConnections).set({ name: "Saved during instance removal" }).where(eq(toolConnections.id, connection.id));
+    });
+    const completed = Promise.allSettled([removal, settings]);
+    try {
+      const pid = await settingsStarted;
+      await vi.waitFor(async () => {
+        const [row] = await db.execute(sql`SELECT wait_event_type FROM pg_stat_activity WHERE pid = ${pid}`);
+        expect(row.wait_event_type).toBe("Lock");
+      }, { timeout: 5000 });
+    } finally { resume(); await completed; providerSpy.mockRestore(); }
+    const results = await completed;
+    for (const result of results) if (result.status === "rejected") throw result.reason;
+    expect(await f.service.read(f.scope)).toEqual({});
+    expect((await db.select().from(toolConnections).where(eq(toolConnections.id, connection.id)))[0].name).toBe("Saved during instance removal");
+  }, 15000);
+
  it("excludes instance secrets from company catalogs, ordinary resolution, bindings and agent access", async () => {
    const f = await fixture(); await f.service.apply(f.request, pluginId, { runtime });
    const [stored] = await db.select().from(companySecrets).where(eq(companySecrets.ownerAgentId, f.agent.id));
@@ -0,0 +1,734 @@
+import { randomUUID } from "node:crypto";
+import { and, eq, sql } from "drizzle-orm";
+import { agents, chatEndpoints, authUsers, heartbeatRuns, issues, companies, companyMemberships, principalPermissionGrants, companySecrets, companySecretBindings, companySecretProviderConfigs, connectionGrants, connectionGrantMembers, connectionGrantDelegations, userSecretDefinitions, createDb, plugins, toolApplications, toolConnections, toolCatalogEntries, toolConnectionInstalls, toolProfiles, 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 { toolAccessService } from "../services/tool-access.js";
+import { updateAgentConfigurationInTransaction } from "../services/agent-configuration-transaction.js";
+import { connectionIntentService } from "../services/connection-intents.js";
+import { emailConnectionService } from "../services/email-connections.js";
+import { accessService } from "../services/access.js";
+import { secretService } from "../services/secrets.js";
+import { slackSearchOAuthService } from "../services/connectors/slack-search-oauth.js";
+
+const support = await getEmbeddedPostgresTestSupport();
+const describeDatabase = support.supported ? describe : describe.skip;
+
+describeDatabase("connector access runtime 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-access-"); 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("captures referenced vault configuration and disablement through the real provider service", async () => {
+    const [vault] = await db.insert(companySecretProviderConfigs).values({ companyId, provider: "aws_secrets_manager", displayName: "Fixture vault", config: { region: "us-east-1" } }).returning();
+    const [secret] = await db.insert(companySecrets).values({ companyId, provider: "aws_secrets_manager", providerConfigId: vault.id, key: "vault", name: "Vault fixture" }).returning();
+    await db.update(toolConnections).set({ credentialSecretRefs: [{ secretId: secret.id, configPath: "token" }] }).where(eq(toolConnections.id, connectionId));
+    const before = (await events()).length;
+    await db.update(companySecretProviderConfigs).set({ healthStatus: "ok", healthCheckedAt: new Date(), healthDetails: { fixture: true } }).where(eq(companySecretProviderConfigs.id, vault.id));
+    expect(await events()).toHaveLength(before);
+    await secretService(db).updateProviderConfig(vault.id, { config: { region: "us-west-2" } });
+    expect(await events()).toHaveLength(before + 2);
+    await secretService(db).disableProviderConfig(vault.id);
+    expect(await events()).toHaveLength(before + 4);
+    const [unrelated] = await db.insert(companySecretProviderConfigs).values({ companyId, provider: "local_encrypted", displayName: "Unreferenced vault" }).returning();
+    await secretService(db).disableProviderConfig(unrelated.id);
+    expect(await events()).toHaveLength(before + 4);
+  });
+  it("skips unchanged grant audiences and coalesces replacements to one company refresh per transaction", async () => {
+    await db.insert(companyMemberships).values(["alice", "bob"].map(principalId => ({ companyId, principalType: "user", principalId, status: "active" })));
+    const [grant] = await db.insert(connectionGrants).values({ companyId, connectionId, kind: "organization" }).returning();
+    const service = toolAccessService(db);
+    await service.replaceConnectionGrantMembers(connectionId, grant.id, ["alice"]);
+    const before = (await events()).length;
+    expect((await service.replaceConnectionGrantMembers(connectionId, grant.id, ["alice", "alice"])).members).toHaveLength(1);
+    expect(await events()).toHaveLength(before);
+    await service.replaceConnectionGrantMembers(connectionId, grant.id, ["bob"]);
+    expect(await events()).toHaveLength(before + 2);
+    await service.replaceConnectionGrantMembers(connectionId, grant.id, []);
+    expect(await events()).toHaveLength(before + 4);
+    await service.replaceConnectionGrantMembers(connectionId, grant.id, []);
+    expect(await events()).toHaveLength(before + 4);
+  });
+  it.each(["uninstall", "permission save"] as const)("serializes %s with finalization or role changes before runtime locks", async change => {
+    let first: (client: Db) => Promise<unknown>;
+    let second: (client: Db) => Promise<unknown>;
+    let memberId: string | undefined;
+    if (change === "uninstall") {
+      await toolAccessService(db).putConnectionInstalls(connectionId, { installs: [{ targetType: "agent", targetId: agentId }] });
+      expect((await toolAccessService(db).getConnection(connectionId, companyId)).config.mcpAgentAccessConfigured).toBe(true);
+      first = client => toolAccessService(client).putConnectionInstalls(connectionId, { installs: [] });
+      second = client => toolAccessService(client).finishGalleryAppConnection(companyId, connectionId, { enabledCatalogEntryIds: [], askFirstCatalogEntryIds: [], access: "all_agents" });
+    } else {
+      const [member] = await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "alice", status: "active", membershipRole: "member" }).returning();
+      memberId = member.id;
+      await db.insert(principalPermissionGrants).values({ companyId, principalType: "user", principalId: "alice", permissionKey: "agents:create" });
+      first = client => accessService(client).updateMemberAndPermissions(companyId, member.id, { membershipRole: "admin", grants: [{ permissionKey: "agents:create" }] }, null);
+      second = client => accessService(client).setMemberPermissions(companyId, member.id, [{ permissionKey: "issues:create" }], null);
+    }
+    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)).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 [row] = await tx.execute(sql`SELECT pg_backend_pid() AS pid`);
+      pids.push(Number(row.pid));
+      return work(tx as unknown as Db);
+    });
+    const waitForLock = (index: number) => vi.waitFor(async () => {
+      expect(pids.length).toBeGreaterThan(index);
+      const [row] = await db.execute(sql`SELECT wait_event_type FROM pg_stat_activity WHERE pid = ${pids[index]}`);
+      expect(row.wait_event_type).toBe("Lock");
+    }, { timeout: 5000 });
+    const initial = run(first);
+    let concurrent: Promise<unknown> | undefined;
+    try {
+      await waitForLock(0);
+      concurrent = run(second);
+      await waitForLock(1);
+    } finally { release(); await blocker; }
+    const results = await Promise.allSettled([initial, concurrent]);
+    for (const result of results) if (result.status === "rejected") throw result.reason;
+    expect(results.map(result => result.status)).toEqual(["fulfilled", "fulfilled"]);
+    if (change === "uninstall") {
+      expect(await db.select().from(toolConnectionInstalls).where(eq(toolConnectionInstalls.connectionId, connectionId))).toHaveLength(0);
+      const bindings = await db.select().from(toolProfileBindings).where(eq(toolProfileBindings.companyId, companyId));
+      expect(bindings.some(binding => binding.metadata?.source === "tool_connection_install")).toBe(false);
+    } else {
+      expect(await accessService(db).getMemberById(companyId, memberId!)).toMatchObject({ membershipRole: "admin" });
+      expect(await db.select().from(principalPermissionGrants).where(eq(principalPermissionGrants.companyId, companyId))).toEqual([expect.objectContaining({ permissionKey: "issues:create" })]);
+    }
+  }, 15000);
+  it.each(["email allowAgent", "connection intent"] as const)("orders outer %s with direct install saves", async caller => {
+    let outer: (client: Db) => Promise<unknown>;
+    if (caller === "email allowAgent") {
+      await db.update(toolConnections).set({ config: { provider: "agentmail", mcpAgentAccessConfigured: true } }).where(eq(toolConnections.id, connectionId));
+      outer = client => emailConnectionService(client).allowAgent(companyId, connectionId, agentId, { localImplicit: true }, "fixture-setup");
+    } else {
+      const userId = randomUUID(), issueId = randomUUID(), runId = randomUUID();
+      await db.insert(authUsers).values({ id: userId, name: "Fixture user", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() });
+      await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, status: "active", membershipRole: "member" });
+      await db.insert(issues).values({ id: issueId, companyId, title: "Use Notion", status: "in_progress", assigneeAgentId: agentId });
+      await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", responsibleUserId: userId, contextSnapshot: { issueId } });
+      const pending = await connectionIntentService(db).request({ sub: agentId, company_id: companyId, run_id: runId, responsible_user_id: userId, scope: "connection_intents", iat: 1, exp: 2, instance_id: "test" }, "notion");
+      expect(pending.interactionId).toEqual(expect.any(String));
+      await db.update(toolConnections).set({ healthStatus: "ok", config: { sourceTemplateKey: "notion", mcpAgentAccessConfigured: true } }).where(eq(toolConnections.id, connectionId));
+      await db.insert(connectionGrants).values({ companyId, connectionId, kind: "organization", status: "active", isDefault: true });
+      await db.insert(toolCatalogEntries).values({ companyId, applicationId, connectionId, name: "read", toolName: "read", versionHash: "fixture", riskLevel: "read", isReadOnly: true });
+      outer = client => connectionIntentService(client).complete(pending.interactionId!, connectionId, userId, { canManageOrganizationGrant: true });
+    }
+    const [profile] = await db.insert(toolProfiles).values({ companyId, profileKey: `app:${connectionId}`, name: "Shared fixture profile", defaultAction: "allow" }).returning();
+    await db.insert(toolProfileBindings).values({ companyId, profileId: profile.id, targetType: "company", targetId: companyId });
+    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(toolApplications).where(eq(toolApplications.id, applicationId)).for("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 [row] = await tx.execute(sql`SELECT pg_backend_pid() AS pid`);
+      pids.push(Number(row.pid));
+      return work(tx as unknown as Db);
+    });
+    const waitForLock = (index: number) => vi.waitFor(async () => {
+      expect(pids.length).toBeGreaterThan(index);
+      const [row] = await db.execute(sql`SELECT wait_event_type FROM pg_stat_activity WHERE pid = ${pids[index]}`);
+      expect(row.wait_event_type).toBe("Lock");
+    }, { timeout: 5000 });
+    const direct = run(client => toolAccessService(client).putConnectionInstalls(connectionId, { installs: [{ targetType: "agent", targetId: peerId }] }));
+    let nested: Promise<unknown> | undefined;
+    try {
+      await waitForLock(0);
+      nested = run(outer);
+      await waitForLock(1);
+    } finally { release(); await blocker; }
+    const results = await Promise.allSettled([direct, nested]);
+    for (const result of results) if (result.status === "rejected") throw result.reason;
+    expect(results.map(result => result.status)).toEqual(["fulfilled", "fulfilled"]);
+    expect((await toolAccessService(db).listConnectionInstalls(connectionId, companyId)).map(install => install.targetId).sort()).toEqual([agentId, peerId].sort());
+    if (caller === "connection intent") expect((results[1] as PromiseFulfilledResult<{ status: string }>).value.status).toBe("accepted");
+  }, 15000);
+  it("orders Slack search disconnect with concurrent personal delegation creation", async () => {
+    await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "alice", status: "active" });
+    await db.update(toolConnections).set({ connectionPurpose: "channel", transport: "chat_sdk", config: { provider: "slack" } }).where(eq(toolConnections.id, connectionId));
+    const [endpoint] = await db.insert(chatEndpoints).values({ companyId, connectionId, provider: "slack", publicId: randomUUID(), assignedAgentId: agentId, status: "active", providerAccountId: "T-FIXTURE" }).returning();
+    const [grant] = await db.insert(connectionGrants).values({ companyId, connectionId, kind: "user", subjectUserId: "alice", providerTenant: { slackSearch: { endpointId: endpoint.id, workspaceId: "T-FIXTURE", slackUserId: "U-FIXTURE", clientRevision: "fixture" } } }).returning();
+    let locked!: () => void; let resume!: () => void; let started!: (pid: number) => void;
+    const held = new Promise<void>(resolve => { locked = resolve; });
+    const mayDelegate = new Promise<void>(resolve => { resume = resolve; });
+    const disconnectStarted = new Promise<number>(resolve => { started = resolve; });
+    const creation = db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      await tx.select().from(connectionGrants).where(eq(connectionGrants.id, grant.id)).for("update");
+      locked(); await mayDelegate;
+      return toolAccessService(tx as unknown as Db).createConnectionGrantDelegation(connectionId, grant.id, agentId, "alice");
+    });
+    await held;
+    const disconnect = db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      const [row] = await tx.execute(sql`SELECT pg_backend_pid() AS pid`);
+      started(Number(row.pid));
+      return slackSearchOAuthService(tx as unknown as Db, "https://paperclip.example").disconnect(companyId, endpoint.id, "alice");
+    });
+    const completed = Promise.allSettled([creation, disconnect]);
+    try {
+      const pid = await disconnectStarted;
+      await vi.waitFor(async () => {
+        const [row] = await db.execute(sql`SELECT wait_event_type FROM pg_stat_activity WHERE pid = ${pid}`);
+        expect(row.wait_event_type).toBe("Lock");
+      }, { timeout: 5000 });
+    } finally { resume(); await completed; }
+    for (const result of await completed) if (result.status === "rejected") throw result.reason;
+    expect((await db.select().from(connectionGrants).where(eq(connectionGrants.id, grant.id)))[0].status).toBe("revoked");
+  }, 15000);
+  it("retries Slack disconnect when a grant appears while waiting for the connection", async () => {
+    await db.update(toolConnections).set({ connectionPurpose: "channel", transport: "chat_sdk", config: { provider: "slack" } }).where(eq(toolConnections.id, connectionId));
+    const [endpoint] = await db.insert(chatEndpoints).values({ companyId, connectionId, provider: "slack", publicId: randomUUID(), assignedAgentId: agentId, status: "active" }).returning();
+    let connectionLocked!: () => void; let releaseConnection!: () => void;
+    const held = new Promise<void>(resolve => { connectionLocked = resolve; });
+    const mayUnlock = new Promise<void>(resolve => { releaseConnection = resolve; });
+    const blocker = db.transaction(async tx => {
+      await tx.select().from(toolConnections).where(eq(toolConnections.id, connectionId)).for("no key update");
+      connectionLocked(); await mayUnlock;
+    });
+    await held;
+    let retries = 0; let disconnectPid: number | undefined; let writer: Promise<unknown> | undefined;
+    let writerLocked!: () => void; let resumeWriter!: () => void;
+    const grantHeld = new Promise<void>(resolve => { writerLocked = resolve; });
+    const mayWrite = new Promise<void>(resolve => { resumeWriter = resolve; });
+    const transaction = db.transaction.bind(db);
+    const spy = vi.spyOn(db, "transaction").mockImplementation((async (operation, config) => {
+      try { return await transaction(async tx => {
+        await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+        const [row] = await tx.execute(sql`SELECT pg_backend_pid() AS pid`);
+        disconnectPid ??= Number(row.pid);
+        return operation(tx);
+      }, config); }
+      catch (error) {
+        const failure = error as { code?: string; cause?: { code?: string } };
+        if ((failure.cause?.code ?? failure.code) === "55P03") { retries++; resumeWriter(); await writer; }
+        throw error;
+      }
+    }) as typeof db.transaction);
+    const disconnect = slackSearchOAuthService(db, "https://paperclip.example").disconnect(companyId, endpoint.id, "alice");
+    const disconnectResult = Promise.allSettled([disconnect]);
+    try {
+      await vi.waitFor(async () => {
+        expect(disconnectPid).toBeDefined();
+        const [row] = await db.execute(sql`SELECT wait_event_type FROM pg_stat_activity WHERE pid = ${disconnectPid!}`);
+        expect(row.wait_event_type).toBe("Lock");
+      }, { timeout: 5000 });
+      const [grant] = await db.insert(connectionGrants).values({ companyId, connectionId, kind: "user", subjectUserId: "alice" }).returning();
+      writer = db.transaction(async tx => {
+        await tx.select().from(connectionGrants).where(eq(connectionGrants.id, grant.id)).for("update");
+        writerLocked(); await mayWrite;
+        await tx.update(toolConnections).set({ name: "Saved during grant discovery" }).where(eq(toolConnections.id, connectionId));
+      });
+      await grantHeld;
+      releaseConnection(); await blocker;
+      await vi.waitFor(() => expect(retries).toBe(1), { timeout: 5000 });
+      await writer;
+      const [result] = await disconnectResult;
+      if (result.status === "rejected") throw result.reason;
+      expect((await db.select().from(connectionGrants).where(eq(connectionGrants.id, grant.id)))[0].status).toBe("revoked");
+    } finally { releaseConnection(); resumeWriter(); await blocker; await writer; await disconnectResult; spy.mockRestore(); }
+  }, 15000);
+  it("journals a personal grant delegation only for its target agent", async () => {
+    await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "alice", status: "active" });
+    const [grant] = await db.insert(connectionGrants).values({ companyId, connectionId, kind: "user", subjectUserId: "alice" }).returning();
+    const before = await events();
+    await toolAccessService(db).createConnectionGrantDelegation(connectionId, grant.id, agentId, "alice");
+    expect((await events()).slice(before.length).map(event => event.resourceId)).toEqual([agentId]);
+  });
+  it("keeps repeated delegation creation and revocation independent of an unrelated agent lock", async () => {
+    await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "alice", status: "active" });
+    const [grant] = await db.insert(connectionGrants).values({ companyId, connectionId, kind: "user", subjectUserId: "alice" }).returning();
+    const before = await events();
+    let locked!: () => void; let resume!: () => void;
+    const held = new Promise<void>(resolve => { locked = resolve; });
+    const mayFinish = new Promise<void>(resolve => { resume = resolve; });
+    const blocker = db.transaction(async tx => {
+      await tx.select().from(agents).where(eq(agents.id, peerId)).for("no key update");
+      locked(); await mayFinish;
+    });
+    await held;
+    try {
+      for (let index = 0; index < 2; index++) {
+        const delegation = await toolAccessService(db).createConnectionGrantDelegation(connectionId, grant.id, agentId, "alice");
+        await db.transaction(async tx => {
+          await tx.execute(sql`SET LOCAL lock_timeout = '2s'`);
+          await toolAccessService(tx as unknown as Db).revokeConnectionGrantDelegation(connectionId, grant.id, delegation.id);
+        });
+      }
+      expect((await events()).slice(before.length).map(row => row.resourceId)).toEqual([agentId, agentId, agentId, agentId]);
+      await db.transaction(async tx => {
+        await tx.execute(sql`SET LOCAL lock_timeout = '2s'`);
+        const service = toolAccessService(tx as unknown as Db);
+        const delegation = await service.createConnectionGrantDelegation(connectionId, grant.id, agentId, "alice");
+        await service.revokeConnectionGrantDelegation(connectionId, grant.id, delegation.id);
+      });
+      expect((await events()).slice(before.length).map(row => row.resourceId)).toEqual(Array(5).fill(agentId));
+    } finally { resume(); await blocker; }
+    expect(await db.select().from(connectionGrantDelegations).where(eq(connectionGrantDelegations.grantId, grant.id))).toEqual([]);
+  }, 15000);
+  it("invalidates both delegation targets on reassignment and ignores unchanged updates or rollback", async () => {
+    await db.insert(agents).values({ companyId, name: "Unrelated delegation peer", adapterType: "paperclip_runner" });
+    const [grant] = await db.insert(connectionGrants).values({ companyId, connectionId, kind: "user", subjectUserId: "alice" }).returning();
+    const [delegation] = await db.insert(connectionGrantDelegations).values({ companyId, grantId: grant.id, agentId, createdByUserId: "alice" }).returning();
+    const before = await events();
+    await db.update(connectionGrantDelegations).set({ agentId: peerId }).where(eq(connectionGrantDelegations.id, delegation.id));
+    const reassigned = await events();
+    expect(reassigned.slice(before.length).map(row => row.resourceId).sort()).toEqual([agentId, peerId].sort());
+    await db.update(connectionGrantDelegations).set({ agentId: peerId }).where(eq(connectionGrantDelegations.id, delegation.id));
+    expect(await events()).toEqual(reassigned);
+    await expect(db.transaction(async tx => {
+      await tx.delete(connectionGrantDelegations).where(eq(connectionGrantDelegations.id, delegation.id));
+      throw new Error("rollback delegation revocation");
+    })).rejects.toThrow("rollback delegation revocation");
+    expect(await events()).toEqual(reassigned);
+    expect(await db.select().from(connectionGrantDelegations).where(eq(connectionGrantDelegations.id, delegation.id))).toEqual([expect.objectContaining({ agentId: peerId })]);
+  });
+  it("limits delegation creation to its target without waiting on unrelated agents", async () => {
+    await db.insert(companyMemberships).values(["alice", "bob"].map(principalId => ({ companyId, principalType: "user", principalId, status: "active" })));
+    const grants = await db.insert(connectionGrants).values(["alice", "bob"].map(subjectUserId => ({ companyId, connectionId, kind: "user", subjectUserId }))).returning();
+    const baseline = await events();
+    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 pids: number[] = [];
+    const saves = grants.map((grant, index) => db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      const [row] = await tx.execute(sql`SELECT pg_backend_pid() AS pid`);
+      pids[index] = Number(row.pid);
+      return toolAccessService(tx as unknown as Db).createConnectionGrantDelegation(connectionId, grant.id, index === 0 ? higher : lower, grant.subjectUserId!);
+    }));
+    const completed = Promise.allSettled(saves);
+    try {
+      await expect(saves[0]).resolves.toMatchObject({ agentId: higher });
+      expect((await events()).slice(baseline.length).map(event => event.resourceId)).toEqual([higher]);
+      await vi.waitFor(async () => {
+        expect(pids).toHaveLength(2);
+        const rows = await db.execute(sql`SELECT pid FROM pg_stat_activity WHERE pid = ${pids[1]} AND wait_event_type = 'Lock'`);
+        expect(rows).toHaveLength(1);
+      }, { timeout: 5000 });
+    } finally { release(); await blocker; }
+    expect((await completed).map(result => result.status)).toEqual(["fulfilled", "fulfilled"]);
+    expect(await db.select().from(connectionGrantDelegations).where(eq(connectionGrantDelegations.companyId, companyId))).toHaveLength(2);
+    const before = await events();
+    expect(before.slice(baseline.length).map(event => event.resourceId).sort()).toEqual([lower, higher]);
+    await toolAccessService(db).createConnectionGrantDelegation(connectionId, grants[0].id, higher, grants[0].subjectUserId!);
+    expect(await events()).toEqual(before);
+  }, 15000);
+  it.each(["grant revocation", "membership suspension"] as const)("orders company locks before delegation cleanup during %s", async change => {
+    const [lower, higher] = [agentId, peerId].sort();
+    await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "owner", membershipRole: "owner", status: "active" });
+    const [member] = await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "alice", membershipRole: "member", status: "active" }).returning();
+    const [grant] = await db.insert(connectionGrants).values({ companyId, connectionId, kind: "user", subjectUserId: "alice" }).returning();
+    await toolAccessService(db).createConnectionGrantDelegation(connectionId, grant.id, higher, "alice");
+    let locked!: () => void; let resume!: () => void; let started!: (pid: number) => void;
+    const held = new Promise<void>(resolve => { locked = resolve; });
+    const maySave = new Promise<void>(resolve => { resume = resolve; });
+    const cleanupStarted = new Promise<number>(resolve => { started = resolve; });
+    const settings = db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      await tx.select().from(agents).where(eq(agents.id, lower)).for("no key update");
+      locked(); await maySave;
+      return toolAccessService(tx as unknown as Db).updateApplication(applicationId, { name: "Concurrent delegation settings" });
+    });
+    await held;
+    const cleanup = db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      const [row] = await tx.execute(sql`SELECT pg_backend_pid() AS pid`);
+      started(Number(row.pid));
+      return change === "grant revocation" ? toolAccessService(tx as unknown as Db).revokeConnectionGrant(connectionId, grant.id)
+        : accessService(tx as unknown as Db).updateMember(companyId, member.id, { status: "suspended" });
+    });
+    const completed = Promise.allSettled([settings, cleanup]);
+    try {
+      const pid = await cleanupStarted;
+      await vi.waitFor(async () => {
+        const [row] = await db.execute(sql`SELECT wait_event_type FROM pg_stat_activity WHERE pid = ${pid}`);
+        expect(row.wait_event_type).toBe("Lock");
+      }, { timeout: 5000 });
+    } finally { resume(); }
+    const results = await completed;
+    for (const result of results) if (result.status === "rejected") throw result.reason;
+    expect((await db.select().from(connectionGrants).where(eq(connectionGrants.id, grant.id)))[0].status).toBe("revoked");
+    expect(await db.select().from(connectionGrantDelegations).where(eq(connectionGrantDelegations.grantId, grant.id))).toEqual([]);
+  }, 15000);
+  it("orders a personal delegation revocation behind whole-grant cleanup", async () => {
+    await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "alice", status: "active" });
+    const [grant] = await db.insert(connectionGrants).values({ companyId, connectionId, kind: "user", subjectUserId: "alice" }).returning();
+    const delegation = await toolAccessService(db).createConnectionGrantDelegation(connectionId, grant.id, agentId, "alice");
+    let locked!: () => void; let resume!: () => void; let started!: (pid: number) => void;
+    const held = new Promise<void>(resolve => { locked = resolve; });
+    const mayRevoke = new Promise<void>(resolve => { resume = resolve; });
+    const personalStarted = new Promise<number>(resolve => { started = resolve; });
+    const whole = db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      await tx.select().from(connectionGrants).where(eq(connectionGrants.id, grant.id)).for("update");
+      await tx.select().from(agents).where(eq(agents.companyId, companyId)).orderBy(agents.id).for("no key update");
+      locked(); await mayRevoke;
+      return toolAccessService(tx as unknown as Db).revokeConnectionGrant(connectionId, grant.id);
+    });
+    await held;
+    const personal = db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      const [row] = await tx.execute(sql`SELECT pg_backend_pid() AS pid`);
+      started(Number(row.pid));
+      return toolAccessService(tx as unknown as Db).revokeConnectionGrantDelegation(connectionId, grant.id, delegation.id);
+    });
+    const completed = Promise.allSettled([whole, personal]);
+    try {
+      const pid = await personalStarted;
+      await vi.waitFor(async () => {
+        const [row] = await db.execute(sql`SELECT wait_event_type FROM pg_stat_activity WHERE pid = ${pid}`);
+        expect(row.wait_event_type).toBe("Lock");
+      }, { timeout: 5000 });
+    } finally { resume(); }
+    const [wholeResult, personalResult] = await completed;
+    if (wholeResult.status === "rejected") throw wholeResult.reason;
+    expect(personalResult).toMatchObject({ status: "rejected", reason: { status: 404 } });
+    expect(await db.select().from(connectionGrantDelegations).where(eq(connectionGrantDelegations.grantId, grant.id))).toEqual([]);
+  }, 15000);
+  it("orders grant revocation with a concurrent delegation creation", async () => {
+    await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "alice", status: "active" });
+    const [grant] = await db.insert(connectionGrants).values({ companyId, connectionId, kind: "user", subjectUserId: "alice" }).returning();
+    await toolAccessService(db).createConnectionGrantDelegation(connectionId, grant.id, agentId, "alice");
+    let locked!: () => void; let resume!: () => void; let started!: (pid: number) => void;
+    const held = new Promise<void>(resolve => { locked = resolve; });
+    const mayCreate = new Promise<void>(resolve => { resume = resolve; });
+    const revokerStarted = new Promise<number>(resolve => { started = resolve; });
+    const creation = db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      await tx.select().from(connectionGrants).where(eq(connectionGrants.id, grant.id)).for("update");
+      locked(); await mayCreate;
+      return toolAccessService(tx as unknown as Db).createConnectionGrantDelegation(connectionId, grant.id, peerId, "alice");
+    });
+    await held;
+    const revocation = db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      const [row] = await tx.execute(sql`SELECT pg_backend_pid() AS pid`);
+      started(Number(row.pid));
+      return toolAccessService(tx as unknown as Db).revokeConnectionGrant(connectionId, grant.id);
+    });
+    const completed = Promise.allSettled([creation, revocation]);
+    try {
+      const pid = await revokerStarted;
+      await vi.waitFor(async () => {
+        const [row] = await db.execute(sql`SELECT wait_event_type FROM pg_stat_activity WHERE pid = ${pid}`);
+        expect(row.wait_event_type).toBe("Lock");
+      }, { timeout: 5000 });
+    } finally { resume(); }
+    expect((await completed).map(result => result.status)).toEqual(["fulfilled", "fulfilled"]);
+    expect((await db.select().from(connectionGrants).where(eq(connectionGrants.id, grant.id)))[0].status).toBe("revoked");
+    expect(await db.select().from(connectionGrantDelegations).where(eq(connectionGrantDelegations.grantId, grant.id))).toEqual([]);
+    await expect(toolAccessService(db).createConnectionGrantDelegation(connectionId, grant.id, peerId, "alice")).rejects.toThrow("Only the active personal grant owner");
+  }, 15000);
+  it.each(["updateMember", "updateMemberAndPermissions", "archiveMember", "setUserCompanyAccess"] as const)("retries %s secret cleanup as a complete transaction", async command => {
+    const userId = `cleanup-${randomUUID()}`;
+    await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "owner", membershipRole: "owner", status: "active" });
+    const [member] = await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, membershipRole: "member", status: "active" }).returning();
+    const [definition] = await db.insert(userSecretDefinitions).values({ companyId, key: "legacy-cleanup", name: "Legacy cleanup fixture" }).returning();
+    const [secret] = await db.insert(companySecrets).values({ companyId, key: "legacy-cleanup", name: "Legacy cleanup fixture", scope: "user", ownerUserId: userId, userSecretDefinitionId: definition.id }).returning();
+    const [grant] = await db.insert(connectionGrants).values({ companyId, connectionId, kind: "user", subjectUserId: userId, credentialSecretRefs: [{ secretId: secret.id, configPath: "oauth.access_token" }] }).returning();
+    const before = await events();
+    let locked!: () => void; let resume!: () => void;
+    const held = new Promise<void>(resolve => { locked = resolve; });
+    const mayDisable = new Promise<void>(resolve => { resume = resolve; });
+    const writer = db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      await tx.select().from(companySecrets).where(eq(companySecrets.id, secret.id)).for("no key update");
+      locked(); await mayDisable;
+      return secretService(tx as unknown as Db).update(secret.id, { status: "disabled" });
+    });
+    await held;
+    let retries = 0;
+    let cleanupPid: number | undefined;
+    const transaction = db.transaction.bind(db);
+    const spy = vi.spyOn(db, "transaction").mockImplementation((async (operation, config) => {
+      try { return await transaction(async tx => {
+        const [row] = await tx.execute(sql`SELECT pg_backend_pid() AS pid`);
+        cleanupPid = Number(row.pid);
+        return operation(tx);
+      }, config); }
+      catch (error) {
+        const failure = error as { code?: string; cause?: { code?: string } };
+        if ((failure.cause?.code ?? failure.code) === "55P03") { retries++; resume(); await writer; }
+        throw error;
+      }
+    }) as typeof db.transaction);
+    const service = accessService(db);
+    const cleanup = command === "updateMember" ? service.updateMember(companyId, member.id, { status: "suspended" })
+      : command === "updateMemberAndPermissions" ? service.updateMemberAndPermissions(companyId, member.id, { status: "suspended", grants: [] }, "owner")
+        : command === "archiveMember" ? service.archiveMember(companyId, member.id)
+          : service.setUserCompanyAccess(userId, [], { actorUserId: "owner" });
+    const completed = Promise.allSettled([writer, cleanup]);
+    try {
+      await vi.waitFor(async () => {
+        if (retries === 1) return;
+        expect(cleanupPid).toBeDefined();
+        const [row] = await db.execute(sql`SELECT wait_event_type FROM pg_stat_activity WHERE pid = ${cleanupPid!}`);
+        expect(row.wait_event_type).toBe("Lock");
+      }, { timeout: 5000 });
+    } finally { resume(); await completed; spy.mockRestore(); }
+    for (const result of await completed) if (result.status === "rejected") throw result.reason;
+    expect(retries).toBe(1);
+    expect((await db.select().from(companyMemberships).where(eq(companyMemberships.id, member.id)))[0].status).toBe(command === "archiveMember" || command === "setUserCompanyAccess" ? "archived" : "suspended");
+    expect((await db.select().from(connectionGrants).where(eq(connectionGrants.id, grant.id)))[0].status).toBe("revoked");
+    expect(await db.select().from(companySecrets).where(eq(companySecrets.id, secret.id))).toEqual([]);
+    expect(await events()).toHaveLength(before.length + 4);
+  }, 15000);
+  it.each((["suspended", "archived"] as const).flatMap(status => (["revocation", "settings"] as const).map(writer => ({ status, writer }))))("orders membership $status cleanup with concurrent $writer", async ({ status, writer }) => {
+    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: "owner", membershipRole: "owner", status: "active" });
+    const [member] = await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "alice", membershipRole: "member", status: "active" }).returning();
+    const [definition] = await db.insert(userSecretDefinitions).values({ companyId, key: "cleanup", name: "Cleanup fixture" }).returning();
+    const [secret] = await db.insert(companySecrets).values({ companyId, key: "cleanup", name: "Cleanup fixture", scope: "user", ownerUserId: "alice", userSecretDefinitionId: definition.id }).returning();
+    const [grant] = await db.insert(connectionGrants).values({ companyId, connectionId, kind: "user", subjectUserId: "alice", credentialSecretRefs: [{ secretId: secret.id, configPath: "oauth.access_token" }] }).returning();
+    await toolAccessService(db).createConnectionGrantDelegation(connectionId, grant.id, agentId, "alice");
+    let locked!: () => void; let resume!: () => void; let started!: (pid: number) => void;
+    const held = new Promise<void>(resolve => { locked = resolve; });
+    const mayRevoke = new Promise<void>(resolve => { resume = resolve; });
+    const cleanupStarted = new Promise<number>(resolve => { started = resolve; });
+    const revocation = db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      if (writer === "revocation") await tx.select().from(connectionGrants).where(eq(connectionGrants.id, grant.id)).for("update");
+      else await tx.select().from(toolConnections).where(eq(toolConnections.id, connectionId)).for("update");
+      locked(); await mayRevoke;
+      const service = toolAccessService(tx as unknown as Db, { remoteHttpEndpointLookup: async () => [{ address: "8.8.8.8", family: 4 }] });
+      return writer === "revocation" ? service.revokeConnectionGrant(connectionId, grant.id) : service.updateConnection(connectionId, { name: "Concurrent settings" });
+    });
+    await held;
+    const cleanup = db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      const [row] = await tx.execute(sql`SELECT pg_backend_pid() AS pid`);
+      started(Number(row.pid));
+      const service = accessService(tx as unknown as Db);
+      return status === "suspended" ? service.updateMember(companyId, member.id, { status }) : service.archiveMember(companyId, member.id);
+    });
+    const completed = Promise.allSettled([revocation, cleanup]);
+    try {
+      const pid = await cleanupStarted;
+      await vi.waitFor(async () => {
+        const [row] = await db.execute(sql`SELECT wait_event_type FROM pg_stat_activity WHERE pid = ${pid}`);
+        expect(row.wait_event_type).toBe("Lock");
+      }, { timeout: 5000 });
+    } finally { resume(); }
+    expect((await completed).map(result => result.status)).toEqual(["fulfilled", "fulfilled"]);
+    expect((await db.select().from(companyMemberships).where(eq(companyMemberships.id, member.id)))[0].status).toBe(status);
+    expect((await db.select().from(connectionGrants).where(eq(connectionGrants.id, grant.id)))[0].status).toBe("revoked");
+    expect(await db.select().from(connectionGrantDelegations).where(eq(connectionGrantDelegations.grantId, grant.id))).toEqual([]);
+    expect(await db.select().from(companySecrets).where(eq(companySecrets.id, secret.id))).toEqual([]);
+    if (writer === "settings") expect((await db.select().from(toolConnections).where(eq(toolConnections.id, connectionId)))[0].name).toBe("Concurrent settings");
+  }, 15000);
+  it("orders audience replacement after membership cleanup without reversing grant locks", async () => {
+    await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: "owner", membershipRole: "owner", status: "active" });
+    const members = await db.insert(companyMemberships).values(["alice", "bob"].map(principalId => ({ companyId, principalType: "user", principalId, membershipRole: "member", status: "active" }))).returning();
+    const member = members.find(row => row.principalId === "alice")!;
+    const [definition] = await db.insert(userSecretDefinitions).values({ companyId, key: "audience", name: "Audience fixture" }).returning();
+    const [secret] = await db.insert(companySecrets).values({ companyId, key: "audience", name: "Audience fixture", scope: "user", userSecretDefinitionId: definition.id, ownerUserId: "alice" }).returning();
+    await db.insert(connectionGrants).values({ companyId, connectionId, kind: "user", subjectUserId: "alice", credentialSecretRefs: [{ secretId: secret.id, configPath: "oauth.access_token" }] });
+    const [organization] = await db.insert(connectionGrants).values({ companyId, connectionId, kind: "organization", isDefault: true }).returning();
+    await db.insert(connectionGrantMembers).values({ companyId, grantId: organization.id, subjectType: "user", subjectId: "alice" });
+    let locked!: () => void; let resume!: () => void; let started!: (pid: number) => void;
+    const held = new Promise<void>(resolve => { locked = resolve; });
+    const maySuspend = new Promise<void>(resolve => { resume = resolve; });
+    const audienceStarted = new Promise<number>(resolve => { started = resolve; });
+    const cleanup = db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      await tx.select().from(companyMemberships).where(eq(companyMemberships.id, member.id)).for("update");
+      locked(); await maySuspend;
+      return accessService(tx as unknown as Db).updateMember(companyId, member.id, { status: "suspended" });
+    });
+    await held;
+    const replacement = db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      const [row] = await tx.execute(sql`SELECT pg_backend_pid() AS pid`);
+      started(Number(row.pid));
+      return toolAccessService(tx as unknown as Db).replaceConnectionGrantMembers(connectionId, organization.id, ["bob"]);
+    });
+    const completed = Promise.allSettled([cleanup, replacement]);
+    try {
+      const pid = await audienceStarted;
+      await vi.waitFor(async () => {
+        const [row] = await db.execute(sql`SELECT wait_event_type FROM pg_stat_activity WHERE pid = ${pid}`);
+        expect(row.wait_event_type).toBe("Lock");
+      }, { timeout: 5000 });
+    } finally { resume(); }
+    expect((await completed).map(result => result.status)).toEqual(["fulfilled", "fulfilled"]);
+    expect((await db.select().from(companyMemberships).where(eq(companyMemberships.id, member.id)))[0].status).toBe("suspended");
+    expect((await db.select().from(connectionGrantMembers).where(eq(connectionGrantMembers.grantId, organization.id))).map(row => row.subjectId)).toEqual(["bob"]);
+  }, 15000);
+  it("invalidates only the owner of an agent secret binding", async () => {
+    const [secret] = await db.insert(companySecrets).values({ companyId, key: "binding", name: "Binding fixture" }).returning();
+    const before = (await events()).length;
+    const [binding] = await db.insert(companySecretBindings).values({ companyId, secretId: secret.id, targetType: "agent", targetId: agentId, configPath: "env.FIXTURE" }).returning();
+    await db.update(companySecretBindings).set({ versionSelector: "2" }).where(eq(companySecretBindings.id, binding.id));
+    await db.delete(companySecretBindings).where(eq(companySecretBindings.id, binding.id));
+    expect((await events()).slice(before).map(event => event.resourceId)).toEqual([agentId, agentId, agentId]);
+  });
+  it("allows an agent configuration save to finish while a connector refresh waits on that agent", async () => {
+    const higher = [agentId, peerId].sort().at(-1)!;
+    const [secret] = await db.insert(companySecrets).values({ companyId, key: "concurrent", name: "Concurrent fixture" }).returning();
+    await db.insert(companySecretBindings).values({ companyId, secretId: secret.id, targetType: "agent", targetId: higher, configPath: "env.FIXTURE" });
+    let locked!: () => void; let resume!: () => void; let started!: (pid: number) => void;
+    const agentLocked = new Promise<void>(resolve => { locked = resolve; });
+    const maySave = new Promise<void>(resolve => { resume = resolve; });
+    const connectorStarted = new Promise<number>(resolve => { started = resolve; });
+    const save = db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      await tx.select().from(agents).where(eq(agents.id, higher)).for("update");
+      locked(); await maySave;
+      await updateAgentConfigurationInTransaction(tx as unknown as Db, higher, { adapterConfig: { env: {} } });
+    });
+    await agentLocked;
+    const connector = db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      const [row] = await tx.execute(sql`SELECT pg_backend_pid() AS pid`);
+      started(Number(row.pid));
+      await tx.update(toolConnections).set({ enabled: false }).where(eq(toolConnections.id, connectionId));
+    });
+    try {
+      const pid = await connectorStarted;
+      await vi.waitFor(async () => {
+        const [row] = await db.execute(sql`SELECT wait_event_type FROM pg_stat_activity WHERE pid = ${pid}`);
+        expect(row.wait_event_type).toBe("Lock");
+      }, { timeout: 5000 });
+    } finally { resume(); await Promise.all([save, connector]); }
+    expect(await db.select().from(companySecretBindings).where(eq(companySecretBindings.targetId, higher))).toEqual([]);
+    expect((await db.select().from(toolConnections).where(eq(toolConnections.id, connectionId)))[0].enabled).toBe(false);
+  }, 15000);
+  it("ignores Vercel verification timestamps while retaining credential changes", async () => {
+    const credential = { provider: "vercel_connect" as const, subjectType: "user" as const, subjectId: "alice", tokenId: "fixture-token", expiresAt: "2099-01-01T00:00:00.000Z", lastVerifiedAt: "2000-01-01T00:00:00.000Z" };
+    const [grant] = await db.insert(connectionGrants).values({ companyId, connectionId, kind: "user", subjectUserId: "alice", externalCredential: credential }).returning();
+    const before = await events();
+    await db.update(connectionGrants).set({ externalCredential: { ...credential, lastVerifiedAt: "2000-01-02T00:00:00.000Z" } }).where(eq(connectionGrants.id, grant.id));
+    expect(await events()).toEqual(before);
+    await db.update(connectionGrants).set({ externalCredential: { ...credential, tokenId: "rotated-fixture-token" } }).where(eq(connectionGrants.id, grant.id));
+    expect(await events()).toHaveLength(before.length + 2);
+  });
+  it("ignores GitHub webhook health while retaining repository access changes", async () => {
+    const github = { userId: "fixture-user", login: "fixture", installationCount: 1, repositoryCount: 1, repositorySelection: "selected" as const, installationIds: ["1"], installationOwnerLogins: ["fixture"], accessRevision: "initial", lastWebhookAt: "2000-01-01T00:00:00.000Z", webhookHealth: "pending" as const };
+    const [grant] = await db.insert(connectionGrants).values({ companyId, connectionId, kind: "user", subjectUserId: "alice", providerTenant: { github } }).returning();
+    const before = await events();
+    await db.update(connectionGrants).set({ providerTenant: { github: { ...github, lastWebhookAt: "2000-01-02T00:00:00.000Z", webhookHealth: "healthy" } } }).where(eq(connectionGrants.id, grant.id));
+    expect(await events()).toEqual(before);
+    await db.update(connectionGrants).set({ providerTenant: { github: { ...github, accessRevision: "changed", repositoryCount: 0, repositorySelection: "none", installationIds: [] } } }).where(eq(connectionGrants.id, grant.id));
+    expect(await events()).toHaveLength(before.length + 2);
+  });
+  it("ignores OAuth grant lease acquisition and release while retaining access changes", async () => {
+    const oauth = { scopes: ["read"], tokenEndpoint: "https://oauth.example.test/token" };
+    const [grant] = await db.insert(connectionGrants).values({ companyId, connectionId, kind: "organization", providerTenant: { oauth } }).returning();
+    const before = await events();
+    await db.update(connectionGrants).set({ providerTenant: { oauth: { ...oauth, refreshLease: { id: "fixture-lease", expiresAt: "2099-01-01T00:00:00.000Z" } } } }).where(eq(connectionGrants.id, grant.id));
+    expect(await events()).toEqual(before);
+    await db.update(connectionGrants).set({ providerTenant: { oauth } }).where(eq(connectionGrants.id, grant.id));
+    expect(await events()).toEqual(before);
+    await db.update(connectionGrants).set({ providerTenant: { oauth: { ...oauth, scopes: ["read", "write"] } } }).where(eq(connectionGrants.id, grant.id));
+    expect(await events()).toHaveLength(before.length + 2);
+  });
+  it("releases removal agent locks when a concurrent credential writer holds the secret", async () => {
+    const [secret] = await db.insert(companySecrets).values({ companyId, key: "busy-delete", name: "Busy deletion fixture", status: "deleted" }).returning();
+    await db.update(toolConnections).set({ credentialSecretRefs: [{ secretId: secret.id, configPath: "token" }] }).where(eq(toolConnections.id, connectionId));
+    let locked!: () => void; let resume!: () => void;
+    const held = new Promise<void>(resolve => { locked = resolve; });
+    const mayWrite = new Promise<void>(resolve => { resume = resolve; });
+    const writer = db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      await tx.select().from(companySecrets).where(eq(companySecrets.id, secret.id)).for("no key update");
+      locked(); await mayWrite;
+      await tx.update(companySecrets).set({ latestVersion: 2 }).where(eq(companySecrets.id, secret.id));
+    });
+    await held;
+    let result: PromiseSettledResult<unknown> | undefined;
+    const removal = Promise.allSettled([secretService(db).remove(secret.id)]).then(rows => { result = rows[0]; });
+    try {
+      await vi.waitFor(() => expect(result).toBeDefined(), { timeout: 5000 });
+      expect(result).toMatchObject({ status: "rejected", reason: { status: 409, details: { retryable: true } } });
+    } finally { resume(); await writer; await removal; }
+    const [remaining] = await db.select().from(companySecrets).where(eq(companySecrets.id, secret.id));
+    expect(remaining.latestVersion).toBe(2);
+    await expect(secretService(db).remove(secret.id)).resolves.toMatchObject({ id: secret.id });
+  }, 15000);
+  it("orders secret deletion before cascading bindings against agent configuration saves", async () => {
+    const [secret] = await db.insert(companySecrets).values({ companyId, key: "delete-bound", name: "Deletion fixture" }).returning();
+    await db.insert(companySecretBindings).values({ companyId, secretId: secret.id, targetType: "agent", targetId: agentId, configPath: "env.FIXTURE" });
+    let locked!: () => void; let resume!: () => void; let started!: (pid: number) => void;
+    const held = new Promise<void>(resolve => { locked = resolve; });
+    const maySave = new Promise<void>(resolve => { resume = resolve; });
+    const deletionStarted = new Promise<number>(resolve => { started = resolve; });
+    const save = db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      await tx.select().from(agents).where(eq(agents.id, agentId)).for("no key update");
+      locked(); await maySave;
+      return updateAgentConfigurationInTransaction(tx as unknown as Db, agentId, { adapterConfig: { env: {} } });
+    });
+    await held;
+    const deletion = db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      const [row] = await tx.execute(sql`SELECT pg_backend_pid() AS pid`);
+      started(Number(row.pid));
+      return secretService(tx as unknown as Db).remove(secret.id);
+    });
+    const completed = Promise.allSettled([save, deletion]);
+    try {
+      const pid = await deletionStarted;
+      await vi.waitFor(async () => {
+        const [row] = await db.execute(sql`SELECT wait_event_type FROM pg_stat_activity WHERE pid = ${pid}`);
+        expect(row.wait_event_type).toBe("Lock");
+      }, { timeout: 5000 });
+    } finally { resume(); }
+    const results = await completed;
+    for (const result of results) if (result.status === "rejected") throw result.reason;
+    expect(await db.select().from(companySecrets).where(eq(companySecrets.id, secret.id))).toEqual([]);
+    expect(await db.select().from(companySecretBindings).where(eq(companySecretBindings.secretId, secret.id))).toEqual([]);
+  }, 15000);
+  it("captures grant revocation and referenced credential rotation without storing credentials in events", async () => {
+    const [secret] = await db.insert(companySecrets).values({ companyId, key: "fixture", name: "Fixture secret" }).returning();
+    const [grant] = await db.insert(connectionGrants).values({ companyId, connectionId, kind: "organization", credentialSecretRefs: [{ secretId: secret.id, configPath: "token" }] }).returning();
+    const before = (await events()).length;
+    await db.update(companySecrets).set({ latestVersion: 2 }).where(eq(companySecrets.id, secret.id));
+    expect(await events()).toHaveLength(before + 2);
+    await db.update(connectionGrants).set({ status: "revoked", revokedAt: new Date() }).where(eq(connectionGrants.id, grant.id));
+    expect(await events()).toHaveLength(before + 4);
+    expect(Object.keys((await events()).at(-1)!).sort()).toEqual(["action", "companyId", "createdAt", "id", "resourceId", "resourceType"]);
+    const unrelated = await db.insert(companySecrets).values({ companyId, key: "unrelated", name: "Unrelated" }).returning();
+    const unchanged = (await events()).length;
+    await db.update(companySecrets).set({ latestVersion: 2 }).where(eq(companySecrets.id, unrelated[0].id));
+    expect(await events()).toHaveLength(unchanged);
+  });
+
+});
@@ -1,5 +1,5 @@
 import { randomUUID } from "node:crypto";
-import { and, eq } from "drizzle-orm";
+import { and, eq, sql } from "drizzle-orm";
 import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
 import { agents, approvals, companies, companyMemberships, companySecrets, connectionGrants, createDb, heartbeatRuns, issues, toolCatalogEntries, toolActionRequests, toolConnections, toolPolicies, toolProfileBindings, toolProfiles, toolAccessAuditEvents } from "@paperclipai/db";
 import { toolAccessService } from "../services/tool-access.js";
@@ -132,6 +132,44 @@ const initialTools = [
    expect(retired.every((entry) => entry.status === "disabled")).toBe(true);
  });

+  it("orders concurrent Railway reconnect and SSH setup without any company agents", async () => {
+    const f = await fixture();
+    const [grant] = await db.select().from(connectionGrants).where(eq(connectionGrants.connectionId, f.connectionId));
+    const start = await f.service.startOAuth(f.company.id, f.connectionId, { redirectUri, actor });
+    let locked!: () => void; let resume!: () => void; let started!: (pid: number) => void;
+    const held = new Promise<void>(resolve => { locked = resolve; });
+    const mayReconnect = new Promise<void>(resolve => { resume = resolve; });
+    const sshStarted = new Promise<number>(resolve => { started = resolve; });
+    const service = (database: typeof db) => toolAccessService(database, { remoteHttpRequest: f.request, remoteHttpEndpointLookup: async () => [{ address: "8.8.8.8", family: 4 }] });
+    const callback = db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      await tx.select().from(connectionGrants).where(eq(connectionGrants.id, grant.id)).for("update");
+      locked(); await mayReconnect;
+      return service(tx as unknown as typeof db).completeOAuthCallback({ state: new URL(start.authorizationUrl).searchParams.get("state")!, code: "concurrent-reconnect", iss: railwayAuthorizationMetadata.issuer, redirectUri, actor });
+    });
+    await held;
+    const ssh = db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      const [row] = await tx.execute(sql`SELECT pg_backend_pid() AS pid`);
+      started(Number(row.pid));
+      return service(tx as unknown as typeof db).configureRailwaySsh(f.connectionId, f.company.id, { action: "prepare", grantId: grant.id }, actor);
+    });
+    const completed = Promise.allSettled([callback, ssh]);
+    try {
+      const pid = await sshStarted;
+      await vi.waitFor(async () => {
+        const [row] = await db.execute(sql`SELECT wait_event_type FROM pg_stat_activity WHERE pid = ${pid}`);
+        expect(row.wait_event_type).toBe("Lock");
+      }, { timeout: 5000 });
+    } finally { resume(); }
+    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 f.service.getConnection(f.connectionId)).config.railwaySsh).toMatchObject({ grantId: grant.id, publicKey: expect.stringMatching(/^ssh-ed25519 /), enabled: false });
+    const [updated] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, grant.id));
+    expect(updated.credentialSecretRefs.map(ref => ref.configPath)).toEqual(expect.arrayContaining(["oauth.access_token", "oauth.refresh_token", "railway.ssh_private_key"]));
+  }, 20000);
+
  it("preserves the dedicated SSH grant key on reconnect and removes it while disconnected", async () => {
    const f = await fixture();
    const [grant] = await db.select().from(connectionGrants).where(eq(connectionGrants.connectionId, f.connectionId));
@@ -9376,6 +9376,69 @@ describeEmbeddedPostgres("tool access service", () => {
    }
  });

+  it("orders managed reconnect with refresh and rejects superseded secret writes", async () => {
+    const company = await createCompany(db);
+    await createAgent(db, company.id);
+    const userId = `managed-refresh-${randomUUID()}`;
+    await grantBoardUser(db, company.id, userId, [], "owner");
+    const actor = { actorType: "user" as const, actorId: userId };
+    const connector = fakeGmailConnector(company.id, userId);
+    const refreshDb = createDb(tempDb!.connectionString, { maxConnections: 1 });
+    const service = createTestToolAccessService(db, { paperclipCloudConnector: connector });
+    const definition = getConnectableAppDefinition("gmail")!;
+    const availability = definition.ownershipAvailability;
+    definition.ownershipAvailability = { ...availability, platform_shared: true };
+    mockToolsList([]);
+    let allowRefresh!: () => void; let allowReconnect!: () => void;
+    try {
+      const connected = await service.connectGalleryApp(company.id, { galleryKey: "gmail", connectionMethodKey: "paperclip-draft", grantKind: "user", name: "Managed concurrent refresh" }, actor);
+      const start = () => service.startOAuth(company.id, connected.connectionId, { redirectUri: "https://paperclip.example/api/tools/oauth/cloud-connector/callback", actor });
+      const initial = await start();
+      await service.completePaperclipCloudConnectorCallback({ state: new URL(initial.authorizationUrl).searchParams.get("state")!, claimId: "initial", actor });
+      const [grant] = await db.select().from(connectionGrants).where(and(eq(connectionGrants.connectionId, connected.connectionId), eq(connectionGrants.subjectUserId, userId)));
+      const accessId = grant.credentialSecretRefs.find(ref => ref.configPath === "oauth.access_token")!.secretId;
+      const [before] = await db.select().from(companySecrets).where(eq(companySecrets.id, accessId));
+      const reconnect = await start();
+      const credentials = await connector.refresh({ subject: userId, companyId: company.id, profile: "gmail.draft", refreshToken: "fixture" });
+      let refreshing!: () => void;
+      const providerCalled = new Promise<void>(resolve => { refreshing = resolve; });
+      const mayRefresh = new Promise<void>(resolve => { allowRefresh = resolve; });
+      vi.mocked(connector.refresh).mockImplementation(async () => { refreshing(); await mayRefresh; return credentials; });
+      const [backend] = await refreshDb.execute(sql`SELECT pg_backend_pid() AS pid`);
+      const refresh = createTestToolAccessService(refreshDb, { paperclipCloudConnector: connector }).refreshOAuthGrantCredentials({ companyId: company.id, connectionId: connected.connectionId, grantId: grant.id, forceRefresh: true, actor });
+      const refreshResult = Promise.allSettled([refresh]);
+      await providerCalled;
+      let locked!: () => void;
+      const held = new Promise<void>(resolve => { locked = resolve; });
+      const mayReconnect = new Promise<void>(resolve => { allowReconnect = resolve; });
+      const callback = db.transaction(async tx => {
+        await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+        await tx.select().from(connectionGrants).where(eq(connectionGrants.id, grant.id)).for("update");
+        locked(); await mayReconnect;
+        return createTestToolAccessService(tx as unknown as Db, { paperclipCloudConnector: connector }).completePaperclipCloudConnectorCallback({ state: new URL(reconnect.authorizationUrl).searchParams.get("state")!, claimId: "reconnect", actor });
+      });
+      const callbackResult = Promise.allSettled([callback]);
+      await held;
+      allowRefresh();
+      try {
+        await vi.waitFor(async () => {
+          const [row] = await db.execute(sql`SELECT wait_event_type FROM pg_stat_activity WHERE pid = ${Number(backend.pid)}`);
+          expect(row.wait_event_type).toBe("Lock");
+        }, { timeout: 5000 });
+      } finally { allowReconnect(); }
+      const [completion] = await callbackResult;
+      if (completion.status === "rejected") throw completion.reason;
+      const [refreshCompletion] = await refreshResult;
+      expect(refreshCompletion).toMatchObject({ status: "rejected", reason: { status: 409, details: { code: "oauth_refresh_superseded", retryable: true } } });
+      expect((await db.select().from(companySecrets).where(eq(companySecrets.id, accessId)))[0].latestVersion).toBe(before.latestVersion + 1);
+      expect((await db.select().from(connectionGrants).where(eq(connectionGrants.id, grant.id)))[0].status).toBe("active");
+    } finally {
+      allowRefresh?.(); allowReconnect?.();
+      definition.ownershipAvailability = availability;
+      await refreshDb.$client.end({ timeout: 0 }).catch(() => undefined);
+    }
+  }, 20000);
+
  it("keeps brokered OAuth state retryable until credentials are durably stored", async () => {
    const company = await createCompany(db);
    const userId = `gmail-retry-${randomUUID()}`;
@@ -10617,6 +10680,133 @@ describeEmbeddedPostgres("tool access service", () => {
    }
  }, 20000);

+  it.each(["sharing", "settings", "application"] as const)("orders personal OAuth reconnect credential rotation with concurrent %s", async change => {
+    vi.stubEnv("PAPERCLIP_TOOL_OAUTH_SLACK_CLIENT_ID", "slack-client-id");
+    vi.stubEnv("PAPERCLIP_TOOL_OAUTH_SLACK_CLIENT_SECRET", "slack-client-secret");
+    const company = await createCompany(db);
+    await createAgent(db, company.id);
+    const userId = `oauth-sharer-${randomUUID()}`;
+    await grantBoardUser(db, company.id, userId, [], "owner");
+    const service = createTestToolAccessService(db);
+    const actor = { actorType: "user" as const, actorId: userId };
+    const connected = await service.connectGalleryApp(company.id, { galleryKey: "slack", name: "Concurrent sharing", grantKind: "user" }, actor);
+    const connection = await service.getConnection(connected.connectionId, company.id);
+    const redirectUri = "https://paperclip.example/api/tools/oauth/callback";
+    vi.spyOn(globalThis, "fetch").mockImplementation(async url => {
+      const href = String(url);
+      if (href === "https://slack.com/api/oauth.v2.access") return mcpHttpResponse(oauthToken("fixture-access-token", "fixture-refresh-token"));
+      if (href === "https://mcp.slack.com/mcp") return mcpHttpResponse(catalogResponse([{ name: "search_messages", annotations: { readOnlyHint: true } }]));
+      throw new Error(`unexpected fetch ${href}`);
+    });
+    const start = () => service.startOAuth(company.id, connected.connectionId, { redirectUri, actor, subjectUserId: userId });
+    const initial = await start();
+    await service.completeOAuthCallback({ state: new URL(initial.authorizationUrl).searchParams.get("state")!, code: "initial", redirectUri, actor });
+    const [grant] = await db.select().from(connectionGrants).where(and(eq(connectionGrants.connectionId, connected.connectionId), eq(connectionGrants.subjectUserId, userId)));
+    const reconnect = await start();
+    let locked!: () => void; let resume!: () => void; let started!: (pid: number) => void;
+    const held = new Promise<void>(resolve => { locked = resolve; });
+    const mayShare = new Promise<void>(resolve => { resume = resolve; });
+    const reconnectStarted = new Promise<number>(resolve => { started = resolve; });
+    const sharing = db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      if (change === "sharing") await tx.select().from(connectionGrants).where(eq(connectionGrants.id, grant.id)).for("update");
+      else if (change === "application") await tx.select().from(toolApplications).where(eq(toolApplications.id, connection.applicationId)).for("update");
+      else await tx.select().from(toolConnections).where(eq(toolConnections.id, connected.connectionId)).for("update");
+      locked(); await mayShare;
+      const writer = createTestToolAccessService(tx as unknown as Db);
+      return change === "sharing"
+        ? writer.finalizeOAuthAccess(company.id, connected.connectionId, { grantKind: "organization" }, actor)
+        : change === "application"
+          ? writer.updateApplication(connection.applicationId, { name: "Application saved concurrently" })
+          : writer.updateConnection(connected.connectionId, { name: "Settings saved concurrently" });
+    });
+    await held;
+    const callback = db.transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      const [row] = await tx.execute(sql`SELECT pg_backend_pid() AS pid`);
+      started(Number(row.pid));
+      return createTestToolAccessService(tx as unknown as Db).completeOAuthCallback({ state: new URL(reconnect.authorizationUrl).searchParams.get("state")!, code: "reconnect", redirectUri, actor });
+    });
+    const completed = Promise.allSettled([sharing, callback]);
+    try {
+      const pid = await reconnectStarted;
+      await vi.waitFor(async () => {
+        const [row] = await db.execute(sql`SELECT wait_event_type FROM pg_stat_activity WHERE pid = ${pid}`);
+        expect(row.wait_event_type).toBe("Lock");
+      }, { timeout: 5000 });
+    } finally { resume(); }
+    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 service.getConnection(connected.connectionId, company.id)).toMatchObject({ credentialPolicy: "per_user", status: "active" });
+    if (change === "sharing") expect((await service.listConnectionGrants(connected.connectionId, company.id)).grants).toEqual(expect.arrayContaining([expect.objectContaining({ kind: "organization", status: "active" })]));
+    else if (change === "application") {
+      const [application] = await db.select().from(toolApplications).where(eq(toolApplications.id, connection.applicationId));
+      expect(application.name).toBe("Application saved concurrently");
+    } else expect((await service.getConnection(connected.connectionId, company.id)).name).toBe("Settings saved concurrently");
+  }, 15000);
+
+  it.each(["app finishing", "new tool discovery"] as const)("orders %s with OAuth reconnect before either locks the connection", async change => {
+    vi.stubEnv("PAPERCLIP_TOOL_OAUTH_SLACK_CLIENT_ID", "slack-client-id");
+    vi.stubEnv("PAPERCLIP_TOOL_OAUTH_SLACK_CLIENT_SECRET", "slack-client-secret");
+    const company = await createCompany(db);
+    const userId = `oauth-sharer-${randomUUID()}`;
+    await grantBoardUser(db, company.id, userId, [], "owner");
+    const service = createTestToolAccessService(db);
+    const actor = { actorType: "user" as const, actorId: userId };
+    const connected = await service.connectGalleryApp(company.id, { galleryKey: "slack", name: "Concurrent sharing", grantKind: "user" }, actor);
+    const connection = await service.getConnection(connected.connectionId, company.id);
+    const redirectUri = "https://paperclip.example/api/tools/oauth/callback";
+    let discoverNewTool = false;
+    vi.spyOn(globalThis, "fetch").mockImplementation(async url => {
+      const href = String(url);
+      if (href === "https://slack.com/api/oauth.v2.access") return mcpHttpResponse(oauthToken("fixture-access-token", "fixture-refresh-token"));
+      if (href === "https://mcp.slack.com/mcp") return mcpHttpResponse(catalogResponse([{ name: "search_messages", annotations: { readOnlyHint: true } }, ...(discoverNewTool ? [{ name: "new_search", annotations: { readOnlyHint: true } }] : [])]));
+      throw new Error(`unexpected fetch ${href}`);
+    });
+    const start = () => service.startOAuth(company.id, connected.connectionId, { redirectUri, actor, subjectUserId: userId });
+    const initial = await start();
+    await service.completeOAuthCallback({ state: new URL(initial.authorizationUrl).searchParams.get("state")!, code: "initial", redirectUri, actor });
+    const reconnect = await start();
+    discoverNewTool = change === "new tool discovery";
+    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(toolApplications).where(eq(toolApplications.id, connection.applicationId)).for("update");
+      locked(); await resume;
+    });
+    await held;
+    const pids: number[] = [];
+    const transaction = db.transaction.bind(db);
+    const transactionSpy = vi.spyOn(db, "transaction").mockImplementation((async (operation, config) => transaction(async tx => {
+      await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+      const [row] = await tx.execute(sql`SELECT pg_backend_pid() AS pid`);
+      pids.push(Number(row.pid));
+      return operation(tx);
+    }, config)) as typeof db.transaction);
+    const run = (work: (service: ReturnType<typeof createTestToolAccessService>) => Promise<unknown>) => work(createTestToolAccessService(db));
+    const waitForLock = (index: number) => vi.waitFor(async () => {
+      expect(pids.length).toBeGreaterThan(index);
+      const [row] = await db.execute(sql`SELECT wait_event_type FROM pg_stat_activity WHERE pid = ${pids[index]}`);
+      expect(row.wait_event_type).toBe("Lock");
+    }, { timeout: 5000 });
+    const callback = run(writer => writer.completeOAuthCallback({ state: new URL(reconnect.authorizationUrl).searchParams.get("state")!, code: "reconnect", redirectUri, actor }));
+    let finishing: Promise<unknown> | undefined;
+    try {
+      await waitForLock(0);
+      finishing = run(writer => change === "new tool discovery" ? writer.refreshCatalog(connected.connectionId) : writer.finishGalleryAppConnection(company.id, connected.connectionId, { enabledCatalogEntryIds: [], askFirstCatalogEntryIds: [], access: "all_agents" }, actor));
+      await waitForLock(1);
+    } finally { release(); await blocker; }
+    const results = await Promise.allSettled([callback, finishing]);
+    for (const result of results) if (result.status === "rejected") throw result.reason;
+    transactionSpy.mockRestore();
+    expect(results.map(result => result.status)).toEqual(["fulfilled", "fulfilled"]);
+    expect(await service.getConnection(connected.connectionId, company.id)).toMatchObject({ credentialPolicy: "per_user", status: "active" });
+    if (change === "new tool discovery") expect(await db.select().from(toolCatalogEntries).where(eq(toolCatalogEntries.connectionId, connected.connectionId))).toEqual(expect.arrayContaining([expect.objectContaining({ toolName: "new_search" })]));
+    else expect(await db.select().from(toolProfiles).where(and(eq(toolProfiles.companyId, company.id), eq(toolProfiles.profileKey, `app:${connected.connectionId}`)))).toHaveLength(1);
+  }, 15000);
+
  it("promotes a personal OAuth identity only after Everyone in the company is chosen", async () => {
    vi.stubEnv("PAPERCLIP_TOOL_OAUTH_SLACK_CLIENT_ID", "slack-client-id");
    vi.stubEnv(
@@ -17583,6 +17773,51 @@ describeEmbeddedPostgres("tool access service", () => {
    ]);
  });

+  it("orders static-key reconnect with concurrent personal grant revocation", async () => {
+    const company = await createCompany(db);
+    await createAgent(db, company.id);
+    const userId = `key-reconnect-${randomUUID()}`;
+    const actor = { actorType: "user" as const, actorId: userId };
+    const service = createTestToolAccessService(db);
+    mockToolsList([readTool("list_zaps", "List")]);
+    await withGalleryServerUrl("github", PUBLIC_MCP_FIXTURE_URL, async () => {
+      const connected = await service.connectGalleryApp(company.id, { galleryKey: "github", connectionMethodKey: "mcp-key", name: "Key reconnect fixture", grantKind: "user", credentialValues: { "credentials.authorization": "fixture-old-key" } }, actor);
+      const { grants } = await service.listConnectionGrants(connected.connectionId, company.id);
+      const grant = grants.find(row => row.kind === "user")!;
+      let locked!: () => void; let resume!: () => void; let started!: (pid: number) => void;
+      const held = new Promise<void>(resolve => { locked = resolve; });
+      const mayRevoke = new Promise<void>(resolve => { resume = resolve; });
+      const reconnectStarted = new Promise<number>(resolve => { started = resolve; });
+      const revocation = db.transaction(async tx => {
+        await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+        await tx.select().from(connectionGrants).where(eq(connectionGrants.id, grant.id)).for("update");
+        locked(); await mayRevoke;
+        return createTestToolAccessService(tx as unknown as Db).revokeConnectionGrant(connected.connectionId, grant.id, actor);
+      });
+      await held;
+      const reconnect = db.transaction(async tx => {
+        await tx.execute(sql`SET LOCAL lock_timeout = '8s'`);
+        const [row] = await tx.execute(sql`SELECT pg_backend_pid() AS pid`);
+        started(Number(row.pid));
+        return createTestToolAccessService(tx as unknown as Db).reconnectGalleryApp(connected.connectionId, company.id, { credentialValues: { "credentials.authorization": "fixture-new-key" } }, actor);
+      });
+      const completed = Promise.allSettled([revocation, reconnect]);
+      try {
+        const pid = await reconnectStarted;
+        await vi.waitFor(async () => {
+          const [row] = await db.execute(sql`SELECT wait_event_type FROM pg_stat_activity WHERE pid = ${pid}`);
+          expect(row.wait_event_type).toBe("Lock");
+        }, { timeout: 5000 });
+      } finally { resume(); }
+      const results = await completed;
+      for (const result of results) if (result.status === "rejected") throw result.reason;
+      const [updatedGrant] = await db.select().from(connectionGrants).where(eq(connectionGrants.id, grant.id));
+      expect(updatedGrant.status).toBe("active");
+      const [secret] = await db.select().from(companySecrets).where(eq(companySecrets.id, grant.credentialSecretRefs[0]!.secretId));
+      expect(secret.latestVersion).toBe(2);
+    });
+  }, 15000);
+
  it("reconnects a personal key on the existing user grant without creating an organization credential", async () => {
    const company = await createCompany(db);
    const service = createTestToolAccessService(db);
@@ -1,6 +1,7 @@
 import { and, eq, inArray, ne, notInArray, sql } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
 import {
+  agents,
  companyMemberships,
  companySecretBindings,
  companySecrets,
@@ -36,6 +37,17 @@ type MemberArchiveInput = {
 export function accessService(db: Db) {
  const authorization = authorizationService(db);

+  async function memberAccessTransaction<T>(operation: (tx: Parameters<Parameters<Db["transaction"]>[0]>[0]) => Promise<T>): Promise<T> {
+    for (let attempt = 0; ; attempt++) {
+      try { return await db.transaction(operation); }
+      catch (error) {
+        const failure = error as { code?: string; cause?: { code?: string } };
+        if ((failure?.cause?.code ?? failure?.code) !== "55P03") throw error;
+        if (attempt >= 2) throw conflict("Member access changed during cleanup. Try again.", { retryable: true });
+      }
+    }
+  }
+
  async function sweepMemberConnectionAccess(
    tx: Parameters<Parameters<Db["transaction"]>[0]>[0],
    companyId: string,
@@ -229,6 +241,27 @@ export function accessService(db: Db) {
    const connectionSecretIdsToRemove = new Set(
      ownedSecretIds.filter((secretId) => !sharedConnectionSecretIds.has(secretId)),
    );
+    const affectedGrantIds = [...new Set([...grantIds, ...grantRefs
+      .filter(grant => affectedConnections.has(grant.connectionId))
+      .map(grant => grant.id)])];
+    if (affectedGrantIds.length > 0) {
+      await tx.select({ id: connectionGrants.id }).from(connectionGrants)
+        .where(and(eq(connectionGrants.companyId, companyId), inArray(connectionGrants.id, affectedGrantIds)))
+        .orderBy(connectionGrants.id).for("update");
+    }
+    if (affectedConnectionIds.length > 0) {
+      await tx.select({ id: toolConnections.id }).from(toolConnections)
+        .where(and(eq(toolConnections.companyId, companyId), inArray(toolConnections.id, affectedConnectionIds)))
+        .orderBy(toolConnections.id).for("update");
+    }
+    if (secretIdsToDelete.length > 0) {
+      await tx.select({ id: companySecrets.id }).from(companySecrets)
+        .where(and(eq(companySecrets.companyId, companyId), inArray(companySecrets.id, secretIdsToDelete)))
+        .orderBy(companySecrets.id).for("update", { noWait: true });
+    }
+    await tx.select({ id: agents.id }).from(agents)
+      .where(and(eq(agents.companyId, companyId), notInArray(agents.status, ["pending_approval", "terminated"])))
+      .orderBy(agents.id).for("no key update");
    const removedDelegations = grantIds.length === 0 ? [] : await tx
      .delete(connectionGrantDelegations)
      .where(and(
@@ -448,10 +481,11 @@ export function accessService(db: Db) {
    grants: GrantInput[],
    grantedByUserId: string | null,
  ) {
-    const member = await getMemberById(companyId, memberId);
-    if (!member) return null;
-
-    await db.transaction(async (tx) => {
+    return db.transaction(async (tx) => {
+      const member = await tx.select().from(companyMemberships)
+        .where(and(eq(companyMemberships.companyId, companyId), eq(companyMemberships.id, memberId)))
+        .for("update").then(rows => rows[0] ?? null);
+      if (!member) return null;
      await tx
        .delete(principalPermissionGrants)
        .where(
@@ -475,9 +509,8 @@ export function accessService(db: Db) {
          })),
        );
      }
+      return member;
    });
-
-    return member;
  }

  async function updateMemberAndPermissions(
@@ -490,7 +523,7 @@ export function accessService(db: Db) {
    },
    grantedByUserId: string | null,
  ) {
-    return db.transaction(async (tx) => {
+    return memberAccessTransaction(async (tx) => {
      await tx.execute(sql`
        select ${companyMemberships.id}
        from ${companyMemberships}
@@ -643,7 +676,7 @@ export function accessService(db: Db) {
  }

  async function archiveMember(companyId: string, memberId: string, input: MemberArchiveInput = {}) {
-    return db.transaction(async (tx) => {
+    return memberAccessTransaction(async (tx) => {
      await tx.execute(sql`
        select ${companyMemberships.id}
        from ${companyMemberships}
@@ -777,13 +810,14 @@ export function accessService(db: Db) {
  ) {
    const target = new Set(companyIds);

-    await db.transaction(async (tx) => {
+    await memberAccessTransaction(async (tx) => {
      // Serialize every company-access removal/reactivation with personal OAuth
      // completion, which locks the same membership row before writing secrets.
      const existing = await tx
        .select()
        .from(companyMemberships)
        .where(and(eq(companyMemberships.principalType, "user"), eq(companyMemberships.principalId, userId)))
+        .orderBy(companyMemberships.companyId, companyMemberships.id)
        .for("update");
      const existingByCompany = new Map(existing.map((row) => [row.companyId, row]));
      const toArchive = existing.filter((row) => !target.has(row.companyId) && row.status !== "archived");
@@ -1057,7 +1091,7 @@ export function accessService(db: Db) {
      status?: "pending" | "active" | "suspended";
    },
  ) {
-    return db.transaction(async (tx) => {
+    return memberAccessTransaction(async (tx) => {
      await tx.execute(sql`
        select ${companyMemberships.id}
        from ${companyMemberships}
@@ -1,6 +1,6 @@
 import { randomUUID } from "node:crypto";
 import { afterAll, beforeAll, describe, expect, it } from "vitest";
-import { eq } from "drizzle-orm";
+import { and, eq } from "drizzle-orm";
 import {
  agents,
  assets,
@@ -17,6 +17,7 @@ import {
  issueAttachments,
  issueComments,
  issues,
+  resourceLifecycleEvents,
  startEmbeddedPostgresTestDatabase,
  toolApplications,
  toolConnections,
@@ -289,6 +290,18 @@ suite("Teams same-transaction publication projection (real PostgreSQL)", () => {
    };
  }

+  it("ignores file-publication bookkeeping and journals endpoint instruction changes", async () => {
+    const f = await fixture("delivered");
+    const events = () => db.select().from(resourceLifecycleEvents).where(and(eq(resourceLifecycleEvents.companyId, f.companyId), eq(resourceLifecycleEvents.resourceType, "agent_runtime"))).orderBy(resourceLifecycleEvents.id);
+    const before = await events();
+    await f.project();
+    expect((await f.endpoint()).lastPublicationAt).not.toBeNull();
+    expect(await events()).toEqual(before);
+    await db.update(chatEndpoints).set({ communicationInstructions: "Use fixture replies" })
+      .where(eq(chatEndpoints.id, (await f.endpoint()).id));
+    expect(await events()).toHaveLength(before.length + 1);
+  });
+
  it.each([
    ["consent_pending", "pending"],
    ["consent_sending", "streaming"],
@@ -15,6 +15,7 @@ import {
  aiConnectionTaskPins,
  companies,
  toolConnections,
+  connectionGrants,
  toolCatalogEntries,
  toolProfiles,
  toolProfileEntries,
@@ -994,6 +995,9 @@ 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: connectionGrants.id }).from(connectionGrants)
+        .where(and(eq(connectionGrants.connectionId, connectionId), eq(connectionGrants.companyId, loaded.issue.companyId)))
+        .orderBy(connectionGrants.id).for("update");
      await lockToolConnectionRuntimeInputs(tx, loaded.issue.companyId, connectionId);
      let selectedConnection = await txAccess.getConnection(connectionId, loaded.issue.companyId);
      const selectedApplication = await txAccess.getApplication(
@@ -16,6 +16,7 @@ import {
 import { z } from "zod";
 import {
  badRequest,
+  conflict,
  forbidden,
  notFound,
  unprocessable,
@@ -43,6 +44,28 @@ export function slackSearchOAuthService(
 ) {
  const secrets = secretService(db);
  const vault = toolAccessService(db);
+  async function searchConnectionTransaction<T>(companyId: string, connectionId: string, userId: string,
+    operation: (tx: Parameters<Parameters<Db["transaction"]>[0]>[0]) => Promise<T>): Promise<T> {
+    for (let attempt = 0; ; attempt++) {
+      let guarded = false;
+      try {
+        return await db.transaction(async tx => {
+          const grantScope = and(eq(connectionGrants.companyId, companyId), eq(connectionGrants.connectionId, connectionId),
+            eq(connectionGrants.kind, "user"), eq(connectionGrants.subjectUserId, userId));
+          await tx.select({ id: connectionGrants.id }).from(connectionGrants).where(grantScope).orderBy(connectionGrants.id).for("update");
+          await tx.select({ id: toolConnections.id }).from(toolConnections)
+            .where(and(eq(toolConnections.id, connectionId), eq(toolConnections.companyId, companyId))).for("update");
+          await tx.select({ id: connectionGrants.id }).from(connectionGrants).where(grantScope).orderBy(connectionGrants.id).for("update", { noWait: true });
+          guarded = true;
+          return operation(tx);
+        });
+      } catch (error) {
+        const failure = error as { code?: string; cause?: { code?: string } };
+        if (guarded || (failure?.cause?.code ?? failure?.code) !== "55P03") throw error;
+        if (attempt >= 2) throw conflict("Slack search authorization changed. Try again.", { retryable: true });
+      }
+    }
+  }
  const redirectUri = () => {
    if (!publicBaseUrl)
      throw badRequest(
@@ -199,17 +222,7 @@ export function slackSearchOAuthService(
    /** Search-only credential acquisition; no caller may use this token for writes. */
    async accessToken(companyId: string, endpointId: string, userId: string) {
      const initial = await linked(companyId, endpointId, userId);
-      return db.transaction(async (tx) => {
-        await tx
-          .select({ id: toolConnections.id })
-          .from(toolConnections)
-          .where(
-            and(
-              eq(toolConnections.id, initial.connection.id),
-              eq(toolConnections.companyId, companyId),
-            ),
-          )
-          .for("update");
+      return searchConnectionTransaction(companyId, initial.connection.id, userId, async (tx) => {
        const [connection] = await tx
          .select()
          .from(toolConnections)
@@ -559,19 +572,9 @@ export function slackSearchOAuthService(
        throw forbidden(
          "Slack token identity does not match the linked account",
        );
-      return db.transaction(async (tx) => {
+      return searchConnectionTransaction(pending.companyId, row.connection.id, userId, async (tx) => {
        // Disconnect deletes pending and in-flight states. Consume under the same
        // connection lock as configuration/refresh so a late callback cannot revive access.
-        await tx
-          .select({ id: toolConnections.id })
-          .from(toolConnections)
-          .where(
-            and(
-              eq(toolConnections.id, row.connection.id),
-              eq(toolConnections.companyId, pending.companyId),
-            ),
-          )
-          .for("update");
        const [liveConnection] = await tx
          .select()
          .from(toolConnections)
@@ -699,17 +702,7 @@ export function slackSearchOAuthService(
    },
    async disconnect(companyId: string, endpointId: string, userId: string) {
      const row = await endpoint(companyId, endpointId);
-      await db.transaction(async (tx) => {
-        await tx
-          .select({ id: toolConnections.id })
-          .from(toolConnections)
-          .where(
-            and(
-              eq(toolConnections.id, row.connection.id),
-              eq(toolConnections.companyId, companyId),
-            ),
-          )
-          .for("update");
+      await searchConnectionTransaction(companyId, row.connection.id, userId, async (tx) => {
        await tx
          .update(connectionGrants)
          .set({
@@ -3432,7 +3432,24 @@ export function secretService(db: Db | DbTransaction) {
        }
      }
    }
-    await db.delete(companySecrets).where(eq(companySecrets.id, secretId));
+    for (let attempt = 0; ; attempt++) {
+      try {
+        await db.transaction(async tx => {
+          if (secret.scope !== "instance") {
+            await tx.select({ id: agents.id }).from(agents)
+              .where(eq(agents.companyId, secret.companyId)).orderBy(agents.id).for("no key update");
+          }
+          await tx.select({ id: companySecrets.id }).from(companySecrets)
+            .where(eq(companySecrets.id, secretId)).for("update", { noWait: true });
+          await tx.delete(companySecrets).where(eq(companySecrets.id, secretId));
+        });
+        break;
+      } catch (error) {
+        const failure = error as { code?: string; cause?: { code?: string } };
+        if ((failure?.cause?.code ?? failure?.code) !== "55P03") throw error;
+        if (attempt >= 2) throw conflict("Secret changed during removal. Retry removal.", { retryable: true });
+      }
+    }
    return secret;
  }

@@ -46,6 +46,7 @@ import {
  lte,
  max,
  ne,
+  notInArray,
  or,
  sql,
 } from "drizzle-orm";
@@ -11637,6 +11638,14 @@ export function toolAccessService(
              },
            };
            const updated = await db.transaction(async (tx) => {
+              const [current] = await tx.select({ id: connectionGrants.id }).from(connectionGrants)
+                .where(and(eq(connectionGrants.id, grant.id), eq(connectionGrants.companyId, grant.companyId),
+                  eq(connectionGrants.status, "active"),
+                  sql`${connectionGrants.providerTenant} -> 'oauth' -> 'refreshLease' ->> 'id' = ${lease.leaseId}`))
+                .for("update");
+              if (!current) throw conflict("OAuth credentials changed while refresh was in progress", {
+                code: "oauth_refresh_superseded", retryable: true,
+              });
              const txSecrets = secretService(tx);
              await txSecrets.rotate(
                accessRef.secretId,
@@ -13724,6 +13733,9 @@ export function toolAccessService(
        const attemptedConnection = connectionRow;
        try {
          await db.transaction(async (tx) => {
+            if (revivedGrantMutation) await tx.select({ id: connectionGrants.id }).from(connectionGrants)
+              .where(and(eq(connectionGrants.id, revivedGrantMutation.current.id), eq(connectionGrants.companyId, companyId)))
+              .for("update");
            const [latestConnection] = await tx
              .select()
              .from(toolConnections)
@@ -14611,6 +14623,15 @@ export function toolAccessService(
    ];

    const updated = await db.transaction(async (tx) => {
+      await tx.select({ id: connectionGrants.id }).from(connectionGrants)
+        .where(and(eq(connectionGrants.companyId, companyId), eq(connectionGrants.connectionId, connection.id)))
+        .orderBy(connectionGrants.id).for("update");
+      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 }).from(toolConnections)
+        .where(and(eq(toolConnections.id, connection.id), eq(toolConnections.companyId, companyId)))
+        .for("update");
      for (const field of providedFields) {
        const value = input.credentialValues[field.configPath]!.trim();
        if (field.placement === "url" && !remoteUrlCredentialMatchesPublicUrl(String(connection.config.url ?? ""), value)) {
@@ -15801,7 +15822,8 @@ export function toolAccessService(
                : eq(connectionGrants.isDefault, true),
          ),
        )
-        .limit(1);
+        .limit(1)
+        .for("update");
      await tx.select({ id: toolApplications.id }).from(toolApplications)
        .where(and(eq(toolApplications.id, connection.applicationId), eq(toolApplications.companyId, connection.companyId)))
        .for("update");
@@ -16435,7 +16457,8 @@ export function toolAccessService(
              eq(connectionGrants.subjectUserId, stateRow.subjectUserId!),
            ),
          )
-          .limit(1);
+          .limit(1)
+          .for("update");
        await tx.select({ id: toolApplications.id }).from(toolApplications)
          .where(and(eq(toolApplications.id, connection.applicationId), eq(toolApplications.companyId, connection.companyId)))
          .for("update");
@@ -16755,6 +16778,10 @@ export function toolAccessService(
          "Only a company owner, admin, or member with connection-manager permission can share credentials with the organization.",
        );
      }
+      await tx.select({ id: connectionGrants.id }).from(connectionGrants)
+        .where(and(eq(connectionGrants.companyId, connection.companyId), eq(connectionGrants.connectionId, connection.id),
+          eq(connectionGrants.kind, "organization"), eq(connectionGrants.isDefault, true)))
+        .for("update");
      await tx.select({ id: toolApplications.id }).from(toolApplications)
        .where(and(eq(toolApplications.id, connection.applicationId), eq(toolApplications.companyId, connection.companyId)))
        .for("update");
@@ -17148,6 +17175,16 @@ export function toolAccessService(
        ];
        const nowAt = new Date();
        await db.transaction(async (tx) => {
+          await tx.select({ id: connectionGrants.id }).from(connectionGrants)
+            .where(and(eq(connectionGrants.companyId, companyId), eq(connectionGrants.connectionId, connection.id),
+              or(eq(connectionGrants.id, personalGrant.id), and(eq(connectionGrants.kind, "organization"), eq(connectionGrants.isDefault, true)))))
+            .orderBy(connectionGrants.id).for("update");
+          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 }).from(toolConnections)
+            .where(and(eq(toolConnections.id, connection.id), eq(toolConnections.companyId, companyId)))
+            .for("update");
          const [existingOrganizationGrant] = await tx
            .select()
            .from(connectionGrants)
@@ -18489,19 +18526,9 @@ export function toolAccessService(
        // Serialize replacements before taking the current audience snapshot.
        // Otherwise a second replacement could read the old rows, wait behind
        // an in-flight replacement, and then miss locking a newly-added member.
-        await tx
-          .select({ id: connectionGrants.id })
-          .from(connectionGrants)
-          .where(
-            and(
-              eq(connectionGrants.id, grant.id),
-              eq(connectionGrants.companyId, connection.companyId),
-              eq(connectionGrants.connectionId, connection.id),
-            ),
-          )
-          .for("update");
+        await tx.execute(sql`SELECT pg_advisory_xact_lock(hashtextextended(${"paperclip.connection_grant_audience:" + grant.id}, 0))`);
        const existingAudience = await tx
-          .select({ subjectId: connectionGrantMembers.subjectId })
+          .select()
          .from(connectionGrantMembers)
          .where(
            and(
@@ -18564,6 +18591,19 @@ export function toolAccessService(
            );
          }
        }
+        await tx
+          .select({ id: connectionGrants.id })
+          .from(connectionGrants)
+          .where(
+            and(
+              eq(connectionGrants.id, grant.id),
+              eq(connectionGrants.companyId, connection.companyId),
+              eq(connectionGrants.connectionId, connection.id),
+            ),
+          )
+          .for("update");
+        const existingUserSet = new Set(existingUserIds);
+        if (requested.length === existingUserSet.size && requested.every(id => existingUserSet.has(id))) return existingAudience;
        await tx
          .delete(connectionGrantMembers)
          .where(
@@ -18656,11 +18696,14 @@ export function toolAccessService(
              eq(connectionGrants.status, "active"),
            ),
          )
+          .for("update")
          .limit(1);
        if (!grant)
          throw forbidden(
            "Only the active personal grant owner can create a delegation",
          );
+        await tx.select({ id: toolConnections.id }).from(toolConnections)
+          .where(eq(toolConnections.id, connection.id)).for("key share");
        const [targetAgent] = await tx
          .select({ id: agents.id })
          .from(agents)
@@ -18670,6 +18713,7 @@ export function toolAccessService(
              eq(agents.companyId, connection.companyId),
            ),
          )
+          .for("no key update")
          .limit(1);
        if (!targetAgent) throw notFound("Agent not found");
        const [existing] = await tx
@@ -18713,16 +18757,28 @@ export function toolAccessService(
      actor?: ActorInfo,
    ) => {
      const connection = await getConnectionRow(idOrUid);
-      const [delegation] = await db
-        .delete(connectionGrantDelegations)
-        .where(
-          and(
-            eq(connectionGrantDelegations.id, delegationId),
-            eq(connectionGrantDelegations.companyId, connection.companyId),
-            eq(connectionGrantDelegations.grantId, grantId),
-          ),
-        )
-        .returning();
+      const delegation = await db.transaction(async tx => {
+        const [grant] = await tx.select({ id: connectionGrants.id }).from(connectionGrants)
+          .where(and(eq(connectionGrants.id, grantId), eq(connectionGrants.companyId, connection.companyId), eq(connectionGrants.connectionId, connection.id)))
+          .for("update");
+        if (!grant) return undefined;
+        const [target] = await tx.select({ agentId: connectionGrantDelegations.agentId }).from(connectionGrantDelegations)
+          .where(and(eq(connectionGrantDelegations.id, delegationId), eq(connectionGrantDelegations.companyId, connection.companyId), eq(connectionGrantDelegations.grantId, grantId)));
+        if (!target) return undefined;
+        await tx.select({ id: agents.id }).from(agents)
+          .where(and(eq(agents.companyId, connection.companyId), eq(agents.id, target.agentId))).for("no key update");
+        const [removed] = await tx
+          .delete(connectionGrantDelegations)
+          .where(
+            and(
+              eq(connectionGrantDelegations.id, delegationId),
+              eq(connectionGrantDelegations.companyId, connection.companyId),
+              eq(connectionGrantDelegations.grantId, grantId),
+            ),
+          )
+          .returning();
+        return removed;
+      });
      if (!delegation) throw notFound("Connection grant delegation not found");
      const binding = actorBinding(actor);
      await db.insert(toolAccessAuditEvents).values({
@@ -18852,6 +18908,13 @@ export function toolAccessService(
        providerRevocation = "local_only_shared_client";
      }
      const grant = await db.transaction(async (tx) => {
+        const [lockedGrant] = await tx.select({ id: connectionGrants.id }).from(connectionGrants)
+          .where(and(eq(connectionGrants.id, grantId), eq(connectionGrants.companyId, connection.companyId), eq(connectionGrants.connectionId, connection.id)))
+          .for("update");
+        if (!lockedGrant) throw notFound("Connection grant not found");
+        await tx.select({ id: agents.id }).from(agents)
+          .where(and(eq(agents.companyId, connection.companyId), notInArray(agents.status, ["pending_approval", "terminated"])))
+          .orderBy(agents.id).for("no key update");
        const removedDelegations = await tx
          .delete(connectionGrantDelegations)
          .where(
@@ -19274,9 +19337,9 @@ export function toolAccessService(
    configureRailwaySsh: async (connectionId: string, companyId: string, input: ConfigureRailwaySsh, actor: ActorInfo): Promise<RailwaySshSetup | null> => {
      const knownHosts = input.action === "enable" ? validateRailwayKnownHosts(input.knownHosts) : "";
      const setup = await db.transaction(async (tx) => {
+        const [grant] = await tx.select().from(connectionGrants).where(and(eq(connectionGrants.id, input.grantId), eq(connectionGrants.connectionId, connectionId), eq(connectionGrants.companyId, companyId))).for("update");
        const [connection] = await tx.select().from(toolConnections).where(and(eq(toolConnections.id, connectionId), eq(toolConnections.companyId, companyId))).for("update");
        if (!connection || !isRailwayConnection(connection) || (connection.status !== "active" && input.action !== "remove")) throw unprocessable("Connect Railway before configuring container access.");
-        const [grant] = await tx.select().from(connectionGrants).where(and(eq(connectionGrants.id, input.grantId), eq(connectionGrants.connectionId, connectionId), eq(connectionGrants.companyId, companyId))).for("update");
        if (!grant || (grant.status !== "active" && input.action !== "remove")) throw unprocessable("Select an active Railway authorization.");
        if (grant.kind === "user" && (actor.actorType !== "user" || actor.actorId !== grant.subjectUserId)) throw forbidden("Only this authorization's owner can configure its SSH key.");
        const existing = asRecord(connection.config.railwaySsh);