diff --git a/.changeset/tool-sync-lifecycle.md b/.changeset/tool-sync-lifecycle.md new file mode 100644 index 0000000000..c67b69b568 --- /dev/null +++ b/.changeset/tool-sync-lifecycle.md @@ -0,0 +1,15 @@ +--- +"@executor-js/sdk": patch +"@executor-js/plugin-mcp": patch +"@executor-js/plugin-openapi": patch +--- + +**Fix: a broken integration is no longer re-dialed on every freshness window, and concurrent reads no longer duplicate the same refresh** + +A tools read refreshes stale tool catalogs before it answers, and the executor stamped the connection's last-synced time even when the listing failed. A server that had been unreachable for a month therefore reported as "synced 30 seconds ago", earned a fresh handshake every freshness window, failed again, and re-stamped — indefinitely. Handshakes that could only ever be refused, because the credential itself had been revoked, were the single largest share of that traffic. Separately, two reads arriving at once had no way to see each other, so both dialed the same server for the same catalog. + +Connections now carry a real sync lifecycle. The last-synced stamp is written only by a listing that actually succeeded, so freshness is honest; a drift signal is recorded alongside it rather than erasing it, so "the catalog changed" and "this has never synced" are finally different states with different diagnoses. The drift signal is an opaque token rather than a timestamp, because the question it answers is "was this catalog invalidated since the listing now finishing began" — a version question, which a wall clock can only answer as precisely as it ticks. A listing captures the token it observed at its start and clears it only if that same token is still there, so a signal arriving mid-listing carries a different token, survives, and re-lists on the next read, while a resolved one is genuinely gone and stops being scanned. + +Failed listings walk a jittered retry ladder that doubles up to a six-hour ceiling. A listing refused on authentication parks the connection, which suppresses only the clock asking a working catalog to re-verify itself: an explicit drift signal, a config revision, and a connection that has never synced at all all still go through, the last because parking a connection with no catalog serves nothing at all until a human notices, and nothing here observes a credential repaired upstream. The park is cleared outright by anything that could plausibly have fixed it — an explicit refresh, a connection edit, an OAuth reconnect, a healthy check, a config revision, or a fresh drift signal. Before dialing, a refresh claims the connection with a leased write token, so exactly one of any number of concurrent readers does the work and the rest answer from the existing catalog; a claimant that loses its lease discards its result instead of overwriting a newer one. + +Plugins can now classify an incomplete listing as `auth`, `unreachable`, `protocol`, or `config` via `ResolveToolsResult.incompleteKind`. The MCP plugin reports the first three from the handshake (including reauthorization demands and 401/403 responses, which it previously discarded), and the OpenAPI plugin reports `config`. The tools read's spans report each refresh as one of `synced`, `incomplete`, `lost_claim` or `fail`, with a counter each, so a plugin reporting bad news correctly is no longer indistinguishable from a refresh that crashed or from one that succeeded — alongside the claim outcome and per-reason skip counts. diff --git a/.claude/skills/prod-telemetry/SKILL.md b/.claude/skills/prod-telemetry/SKILL.md index 8437b5f568..2a34e140c9 100644 --- a/.claude/skills/prod-telemetry/SKILL.md +++ b/.claude/skills/prod-telemetry/SKILL.md @@ -41,18 +41,58 @@ join the same traces via traceparent). `.connection` (present only when the read was filtered), `executor.tools.result_count`, and the catalog-refresh counters `executor.tools.sync.candidates` (rows the stale scan returned) / - `executor.tools.sync.synced` / `executor.tools.sync.failed`. A slow tools + `executor.tools.sync.synced` / `.incomplete` / `.failed` / `.lost_claim` / + `.skipped_claimed` / `.skipped_backoff` / `.skipped_parked`. A slow tools read is almost always `candidates` > 0: subtract the child span durations to - confirm. A connection stuck permanently stale shows up as `failed` > 0 on - every read for its scope — the refresh is best-effort and never fails the - read, so this counter and the `executor tool catalog refresh failed` warning - are the only signals it emits. + confirm. + + `candidates` is the one to alert on for scan cost: in steady state it is + ZERO. Every trigger is cleared by the listing that answers it, drift marks + included (`connection.tools_stale_token` is compare-and-set to NULL), so a + scope whose `candidates` never returns to 0 has connections that are due and + never resolving, not a busy fleet. + + The four terminal counters partition every attempt, and the distinction that + matters when triaging is `incomplete` vs `failed`. `incomplete` is the PLUGIN + reporting bad news correctly — a refused credential, an unreachable server, + an unreadable spec — with the existing catalog kept and the retry ladder + advanced. `failed` is the refresh raising or dying before reaching any + verdict: the executor's own error channel, swallowed to keep the read + answerable, and the thing to page on. A connection stuck permanently stale + shows up as one or the other on every read for its scope; `failed` > 0 also + emits the `executor tool catalog refresh failed` warning, which is its only + other signal. `lost_claim` counts listings that completed and were then + discarded because another attempt had re-claimed the connection — wasted + upstream work that looks like success from every angle except the write. + + The three `skipped_*` counters are how the sync lifecycle reports its work + avoidance, and they are the ones to watch after a rollout: + `skipped_parked` counts connections whose credential was refused (an `auth` + verdict) AND whose catalog is merely past its freshness window — the park + gates re-verification only, so a never-synced connection with the same + verdict reports `skipped_backoff` instead and keeps walking the ladder. + `skipped_backoff` counts ones inside their failure ladder, and + `skipped_claimed` counts refreshes another concurrent read had already taken. + `synced` collapsing toward the skip counters is the intended shape — it means + dead upstreams stopped costing handshakes. `skipped_claimed` rising with + request concurrency for one integration is cross-isolate deduplication + working. + - `executor.tools.sync` (child of the above, one per refreshed connection; older spans carry the previous name `executor.tools.sync_stale` and no trigger/outcome attrs) — `executor.integration`, `executor.connection`, `executor.tools.sync.trigger` - (`stale_marked`/`config_revised`/`expired`) and - `executor.tools.sync.outcome` (`ok`/`fail`). The matching warning log carries + (`cold`/`stale_marked`/`config_revised`/`expired`), + `executor.tools.sync.claimed` (bool — false means another reader owned the + refresh and this one did nothing) and `executor.tools.sync.outcome` + (`synced`/`incomplete`/`lost_claim`/`fail`). Older spans report `ok` where a + current one reports `synced` or `incomplete`; that rename is the same + distinction as the counters above, so any query matching `outcome == "ok"` + needs both spellings to span the change. `cold` split out of `stale_marked` + when the drift mark became its own column: before that, a never-synced + connection and one invalidated mid-invocation were the same row, so spans + predating it report every never-synced connection as `stale_marked`. The + matching warning log carries `integration`, `connection`, `trigger` and `errorTags` only: a refresh runs on a live credential, so no cause or upstream message is logged and Axiom will not have one to search. diff --git a/apps/cloud/drizzle/0016_icy_jocasta.sql b/apps/cloud/drizzle/0016_icy_jocasta.sql new file mode 100644 index 0000000000..d6d4e77c56 --- /dev/null +++ b/apps/cloud/drizzle/0016_icy_jocasta.sql @@ -0,0 +1,6 @@ +ALTER TABLE "connection" ADD COLUMN "tools_stale_token" text;--> statement-breakpoint +ALTER TABLE "connection" ADD COLUMN "tools_sync_claim_id" text;--> statement-breakpoint +ALTER TABLE "connection" ADD COLUMN "tools_sync_claim_at" bigint;--> statement-breakpoint +ALTER TABLE "connection" ADD COLUMN "tools_sync_failures" bigint;--> statement-breakpoint +ALTER TABLE "connection" ADD COLUMN "tools_sync_retry_at" bigint;--> statement-breakpoint +ALTER TABLE "connection" ADD COLUMN "tools_sync_error_kind" text; \ No newline at end of file diff --git a/apps/cloud/drizzle/meta/0016_snapshot.json b/apps/cloud/drizzle/meta/0016_snapshot.json new file mode 100644 index 0000000000..b1214e0cd0 --- /dev/null +++ b/apps/cloud/drizzle/meta/0016_snapshot.json @@ -0,0 +1,1522 @@ +{ + "id": "2d6f4645-3838-4d48-ae79-15d2b5ad9149", + "prevId": "d666b31a-c3d1-4bd7-9bd6-85f2abc4fb55", + "version": "7", + "dialect": "postgresql", + "tables": { + "public.accounts": { + "name": "accounts", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.memberships": { + "name": "memberships", + "schema": "", + "columns": { + "account_id": { + "name": "account_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "organization_id": { + "name": "organization_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "memberships_account_id_accounts_id_fk": { + "name": "memberships_account_id_accounts_id_fk", + "tableFrom": "memberships", + "tableTo": "accounts", + "columnsFrom": ["account_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + }, + "memberships_organization_id_organizations_id_fk": { + "name": "memberships_organization_id_organizations_id_fk", + "tableFrom": "memberships", + "tableTo": "organizations", + "columnsFrom": ["organization_id"], + "columnsTo": ["id"], + "onDelete": "cascade", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "memberships_account_id_organization_id_pk": { + "name": "memberships_account_id_organization_id_pk", + "columns": ["account_id", "organization_id"] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.organizations": { + "name": "organizations", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "slug": { + "name": "slug", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "organizations_slug_unique": { + "name": "organizations_slug_unique", + "columns": [ + { + "expression": "slug", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.artifact": { + "name": "artifact", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "title": { + "name": "title", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "description": { + "name": "description", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "code": { + "name": "code", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "bindings": { + "name": "bindings", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "preview": { + "name": "preview", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "owner": { + "name": "owner", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "subject": { + "name": "subject", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "artifact_uidx": { + "name": "artifact_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "owner", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "subject", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.blob": { + "name": "blob", + "schema": "", + "columns": { + "namespace": { + "name": "namespace", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "key": { + "name": "key", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "value": { + "name": "value", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "id": { + "name": "id", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "blob_id_uidx": { + "name": "blob_id_uidx", + "columns": [ + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.connection": { + "name": "connection", + "schema": "", + "columns": { + "integration": { + "name": "integration", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "template": { + "name": "template", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "provider": { + "name": "provider", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "item_ids": { + "name": "item_ids", + "type": "json", + "primaryKey": false, + "notNull": true + }, + "identity_label": { + "name": "identity_label", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "description": { + "name": "description", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "last_health": { + "name": "last_health", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "tools_synced_at": { + "name": "tools_synced_at", + "type": "bigint", + "primaryKey": false, + "notNull": false + }, + "tools_stale_token": { + "name": "tools_stale_token", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "tools_sync_claim_id": { + "name": "tools_sync_claim_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "tools_sync_claim_at": { + "name": "tools_sync_claim_at", + "type": "bigint", + "primaryKey": false, + "notNull": false + }, + "tools_sync_failures": { + "name": "tools_sync_failures", + "type": "bigint", + "primaryKey": false, + "notNull": false + }, + "tools_sync_retry_at": { + "name": "tools_sync_retry_at", + "type": "bigint", + "primaryKey": false, + "notNull": false + }, + "tools_sync_error_kind": { + "name": "tools_sync_error_kind", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "oauth_client": { + "name": "oauth_client", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "oauth_client_owner": { + "name": "oauth_client_owner", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "refresh_item_id": { + "name": "refresh_item_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "expires_at": { + "name": "expires_at", + "type": "bigint", + "primaryKey": false, + "notNull": false + }, + "oauth_scope": { + "name": "oauth_scope", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "oauth_token_url": { + "name": "oauth_token_url", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "provider_state": { + "name": "provider_state", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "owner": { + "name": "owner", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "subject": { + "name": "subject", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "connection_uidx": { + "name": "connection_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "owner", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "subject", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "integration", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "name", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.definition": { + "name": "definition", + "schema": "", + "columns": { + "integration": { + "name": "integration", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "connection": { + "name": "connection", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "plugin_id": { + "name": "plugin_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "schema": { + "name": "schema", + "type": "json", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "owner": { + "name": "owner", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "subject": { + "name": "subject", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "definition_uidx": { + "name": "definition_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "owner", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "subject", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "integration", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "connection", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "name", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.integration": { + "name": "integration", + "schema": "", + "columns": { + "slug": { + "name": "slug", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "plugin_id": { + "name": "plugin_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "description": { + "name": "description", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "config": { + "name": "config", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "health_check": { + "name": "health_check", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "config_revised_at": { + "name": "config_revised_at", + "type": "bigint", + "primaryKey": false, + "notNull": false + }, + "can_remove": { + "name": "can_remove", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": true + }, + "can_refresh": { + "name": "can_refresh", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "integration_uidx": { + "name": "integration_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "slug", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.oauth_client": { + "name": "oauth_client", + "schema": "", + "columns": { + "slug": { + "name": "slug", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "authorization_url": { + "name": "authorization_url", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "token_url": { + "name": "token_url", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "grant": { + "name": "grant", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "client_id": { + "name": "client_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "client_secret_item_id": { + "name": "client_secret_item_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "resource": { + "name": "resource", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "origin_kind": { + "name": "origin_kind", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "origin_integration": { + "name": "origin_integration", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "origin_issuer": { + "name": "origin_issuer", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "origin_redirect_uri": { + "name": "origin_redirect_uri", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "owner": { + "name": "owner", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "subject": { + "name": "subject", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "oauth_client_uidx": { + "name": "oauth_client_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "owner", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "subject", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "slug", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.oauth_session": { + "name": "oauth_session", + "schema": "", + "columns": { + "state": { + "name": "state", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "client_slug": { + "name": "client_slug", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "integration": { + "name": "integration", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "template": { + "name": "template", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "redirect_url": { + "name": "redirect_url", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "pkce_verifier": { + "name": "pkce_verifier", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "identity_label": { + "name": "identity_label", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "payload": { + "name": "payload", + "type": "json", + "primaryKey": false, + "notNull": true + }, + "expires_at": { + "name": "expires_at", + "type": "bigint", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "owner": { + "name": "owner", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "subject": { + "name": "subject", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "oauth_session_uidx": { + "name": "oauth_session_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "state", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.plugin_storage": { + "name": "plugin_storage", + "schema": "", + "columns": { + "plugin_id": { + "name": "plugin_id", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "collection": { + "name": "collection", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "key": { + "name": "key", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "data": { + "name": "data", + "type": "json", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "owner": { + "name": "owner", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "subject": { + "name": "subject", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "plugin_storage_uidx": { + "name": "plugin_storage_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "owner", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "subject", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "plugin_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "collection", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.private_executor_cloud_settings": { + "name": "private_executor_cloud_settings", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "version": { + "name": "version", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true, + "default": "'1.0.0'" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.subject": { + "name": "subject", + "schema": "", + "columns": { + "external_id": { + "name": "external_id", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "last_seen_at": { + "name": "last_seen_at", + "type": "bigint", + "primaryKey": false, + "notNull": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "subject_uidx": { + "name": "subject_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "external_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.tool": { + "name": "tool", + "schema": "", + "columns": { + "integration": { + "name": "integration", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "connection": { + "name": "connection", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "plugin_id": { + "name": "plugin_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "name": { + "name": "name", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "description": { + "name": "description", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "input_schema": { + "name": "input_schema", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "output_schema": { + "name": "output_schema", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "annotations": { + "name": "annotations", + "type": "json", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "owner": { + "name": "owner", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "subject": { + "name": "subject", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "tool_uidx": { + "name": "tool_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "owner", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "subject", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "integration", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "connection", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "name", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.tool_policy": { + "name": "tool_policy", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "pattern": { + "name": "pattern", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "action": { + "name": "action", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "position": { + "name": "position", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "updated_at": { + "name": "updated_at", + "type": "timestamp", + "primaryKey": false, + "notNull": true + }, + "row_id": { + "name": "row_id", + "type": "varchar(255)", + "primaryKey": true, + "notNull": true + }, + "tenant": { + "name": "tenant", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "owner": { + "name": "owner", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + }, + "subject": { + "name": "subject", + "type": "varchar(255)", + "primaryKey": false, + "notNull": true + } + }, + "indexes": { + "tool_policy_uidx": { + "name": "tool_policy_uidx", + "columns": [ + { + "expression": "tenant", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "owner", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "subject", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + } + }, + "enums": {}, + "schemas": {}, + "sequences": {}, + "roles": {}, + "policies": {}, + "views": {}, + "_meta": { + "columns": {}, + "schemas": {}, + "tables": {} + } +} diff --git a/apps/cloud/drizzle/meta/_journal.json b/apps/cloud/drizzle/meta/_journal.json index fa90570831..e62deab0af 100644 --- a/apps/cloud/drizzle/meta/_journal.json +++ b/apps/cloud/drizzle/meta/_journal.json @@ -113,6 +113,13 @@ "when": 1785355354955, "tag": "0015_equal_the_leader", "breakpoints": true + }, + { + "idx": 16, + "version": "7", + "when": 1786889156029, + "tag": "0016_icy_jocasta", + "breakpoints": true } ] } diff --git a/apps/cloud/src/db/executor-schema.ts b/apps/cloud/src/db/executor-schema.ts index e23d30d077..3c318f54da 100644 --- a/apps/cloud/src/db/executor-schema.ts +++ b/apps/cloud/src/db/executor-schema.ts @@ -61,6 +61,12 @@ export const connection = pgTable( description: text("description"), last_health: json("last_health"), tools_synced_at: bigint("tools_synced_at", { mode: "bigint" }), + tools_stale_token: text("tools_stale_token"), + tools_sync_claim_id: text("tools_sync_claim_id"), + tools_sync_claim_at: bigint("tools_sync_claim_at", { mode: "bigint" }), + tools_sync_failures: bigint("tools_sync_failures", { mode: "bigint" }), + tools_sync_retry_at: bigint("tools_sync_retry_at", { mode: "bigint" }), + tools_sync_error_kind: text("tools_sync_error_kind"), oauth_client: text("oauth_client"), oauth_client_owner: text("oauth_client_owner"), refresh_item_id: text("refresh_item_id"), diff --git a/packages/core/sdk/src/connections.test.ts b/packages/core/sdk/src/connections.test.ts index 7ed046f65b..266796c28c 100644 --- a/packages/core/sdk/src/connections.test.ts +++ b/packages/core/sdk/src/connections.test.ts @@ -605,7 +605,12 @@ describe("tool catalog sync safety", () => { yield* Effect.promise(() => config.db.updateMany("connection", { where: (b) => b.and(b("integration", "=", String(INTEG)), b("name", "=", "main")), - set: { tools_synced_at: null }, + // The failed sync above put this connection on the retry ladder, so + // the next read is held off until its backoff window elapses. Move + // the window into the past rather than sleeping through it; the + // failure count stays, so the successful sync below has to clear + // the whole ledger and not just the health record. + set: { tools_synced_at: null, tools_sync_retry_at: Date.now() - 1 }, }), ); yield* executor.tools.list({ integration: INTEG }); diff --git a/packages/core/sdk/src/core-schema.ts b/packages/core/sdk/src/core-schema.ts index baa71ee003..9960a55e3b 100644 --- a/packages/core/sdk/src/core-schema.ts +++ b/packages/core/sdk/src/core-schema.ts @@ -219,9 +219,47 @@ export const coreTables = defineTables({ // accounts list shows alive/expired AT A GLANCE (the customer ask) // instead of only after a per-row manual probe. Null = never checked. last_health: nullableJsonColumn("last_health"), - // Epoch ms of the last tool (re)production for this connection. Stale - // vs the integration's `config_revised_at` → re-produced on next read. + // Epoch ms of the last AUTHORITATIVE tool (re)production for this + // connection. Stale vs the integration's `config_revised_at` → + // re-produced on next read. Never stamped by an incomplete or failed + // listing: a month-dead MCP server that reads as "synced 30s ago" is + // re-dialed once per freshness window forever. tools_synced_at: nullableBigintColumn("tools_synced_at"), + // Set when an event declares this catalog drifted (an MCP + // `tools/list_changed`, an unknown-tool rejection); null when there is no + // outstanding drift. Separate from `tools_synced_at` so marking a catalog + // stale no longer destroys the last-verified timestamp — "drifted" and + // "never synced" are different states with different diagnoses. + // + // An OPAQUE token, not a timestamp: it answers "was this catalog + // invalidated since the listing now finishing began", which is a version + // question, and a wall clock answers it only as precisely as it ticks. A + // listing captures the token it observed at its start and clears the + // column only if that same token is still there, so a mark that landed + // mid-listing carries a different token, survives the clear, and re-lists. + // Never compared for order. + tools_stale_token: nullableTextColumn("tools_stale_token"), + // The refresh lease. Concurrent reads land in different Workers isolates + // with no shared memory, so the row IS the coordination medium: a + // refresh compare-and-sets its own nonce here before dialing, and the + // losers serve the existing catalog instead of duplicating the + // handshake. `tools_sync_claim_at` bounds the lease + // (`TOOL_SYNC_CLAIM_LEASE_MS`), so an isolate evicted mid-listing frees + // the connection rather than wedging it. + tools_sync_claim_id: nullableTextColumn("tools_sync_claim_id"), + tools_sync_claim_at: nullableBigintColumn("tools_sync_claim_at"), + // The failure ladder: CONSECUTIVE failed listings, and the earliest next + // attempt derived from them (see `tool-sync-schedule`). Zeroed by an + // authoritative listing. `tools_sync_retry_at` is also written on + // success, where it carries the freshness horizon rather than a gate. + tools_sync_failures: nullableBigintColumn("tools_sync_failures"), + tools_sync_retry_at: nullableBigintColumn("tools_sync_retry_at"), + // Why the last listing failed, as the closed `ToolSyncErrorKind` set + // (`auth` | `unreachable` | `protocol` | `config`). `auth` parks the + // connection: retrying a rejected credential cannot change the verdict, + // and 401 handshakes were the single largest source of wasted upstream + // dials. + tools_sync_error_kind: nullableTextColumn("tools_sync_error_kind"), oauth_client: nullableTextColumn("oauth_client"), // The OWNER of `oauth_client` (a Personal connection may be minted through // a shared Workspace app), set together with `oauth_client`; null for @@ -455,14 +493,21 @@ export const TOOL_INVOCATION_COLUMNS = [ "updated_at", ] as const satisfies readonly (keyof ToolRow)[]; /** The connection columns the tools read's catalog-refresh scan projects: the - * address `produceConnectionTools` needs plus the stamp its trigger check - * reads. Credentials, OAuth state and health JSON stay unread — the scan runs - * on every `tools.list`, and in steady state it must match no rows at all. */ + * address `produceConnectionTools` needs plus the whole sync lifecycle its + * eligibility check reads (`ToolSyncCandidate`). Credentials, OAuth state and + * health JSON stay unread — the scan runs on every `tools.list`, and in steady + * state it must match no rows at all. */ export const CONNECTION_CATALOG_SCAN_COLUMNS = [ "owner", "integration", "name", "tools_synced_at", + "tools_stale_token", + "tools_sync_claim_id", + "tools_sync_claim_at", + "tools_sync_failures", + "tools_sync_retry_at", + "tools_sync_error_kind", ] as const satisfies readonly (keyof ConnectionRow)[]; export type DefinitionRow = FumaRow; export type ToolPolicyRow = FumaRow; diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index e587f8c38e..514365801b 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -43,6 +43,18 @@ import { type ToolRow, type ToolPolicyRow, } from "./core-schema"; +import { + TOOL_SYNC_CLAIM_LEASE_MS, + TOOL_SYNC_JITTER_MAX, + TOOL_SYNC_JITTER_MIN, + decideToolSync, + isToolSyncErrorKind, + scheduleAfterFailure, + scheduleAfterSuccess, + type ToolSyncCandidate, + type ToolSyncErrorKind, + type ToolSyncState, +} from "./tool-sync-schedule"; import { ElicitationDeclinedError, ElicitationResponse, @@ -678,25 +690,133 @@ export interface ExecutorConfig; + +/** One attempt at producing a connection's tools: why it is happening, and the + * refresh claim it holds while it does. `claim` is null for `explicit` + * production, which is authoritative by construction and does not coordinate + * with anyone. */ +interface ToolSyncAttempt { + readonly trigger: ToolSyncTrigger; + readonly claim: string | null; +} + +/** Production asked for by a caller who is waiting for the answer. */ +const EXPLICIT_TOOL_SYNC: ToolSyncAttempt = { trigger: "explicit", claim: null }; + +/** + * What one attempt at producing a connection's tools ended up with. Three + * outcomes rather than a boolean, because they are three genuinely different + * things to be told and only one of them is a sync: + * + * - `synced` an AUTHORITATIVE listing landed: the catalog persisted is the + * upstream's tool set, and the freshness stamp is honest. A + * legitimate empty catalog (no credential bound, or a plugin + * with no `resolveTools`) is one of these — "no tools" verified + * is still verified. + * - `incomplete` the plugin reported it could not produce an authoritative + * listing. The existing catalog was KEPT, the failure ladder + * advanced, and nothing claimed freshness. + * - `lost_claim` the listing completed and then found another attempt had + * re-claimed the connection and answered first, so nothing of it + * was written at all. + * + * Collapsing `incomplete` into success is how a fleet of month-dead upstreams + * reports as fully synced; collapsing it into `lost_claim` would blame + * concurrency for a broken server. + * + * `tools` is what the caller should answer with in every case — the fresh + * listing when it was persisted, the persisted catalog otherwise. + */ +type ToolProductionOutcome = "synced" | "incomplete" | "lost_claim"; + +interface ToolProduction { + readonly tools: readonly Tool[]; + readonly outcome: ToolProductionOutcome; +} + +/** + * How much work a read's catalog refresh found and did, for the enclosing + * `executor.tools.list` span. `candidates` is what the stale scan returned — + * the number the scan's filter scoping exists to keep at zero. + * + * The four terminal counters partition every listing that actually ran — a + * candidate that never won its claim is `skippedClaimed`, not a listing — and + * each says something the others cannot: + * + * - `synced` an authoritative listing was persisted (see + * {@link ToolProduction}). + * - `incomplete` the PLUGIN reported it could not produce a listing — a refused + * credential, an unreachable server, an unreadable spec. The + * catalog was kept and the ladder advanced. + * - `failed` the refresh raised or died before reaching a verdict, and was + * swallowed to keep the read answerable. This is the executor's + * own error channel, not the upstream's — an `incomplete` is a + * connection reporting bad news correctly, a `failed` is one + * nobody has a verdict for. + * - `lostClaim` the listing completed and was then discarded because another + * attempt had re-claimed the connection — wasted upstream work, + * and the one outcome that looks like a success from every angle + * except the write. + * + * The three `skipped*` counts are the whole point of the lifecycle columns: a + * fleet whose reads are dominated by `skipped_parked` is one where dead + * credentials stopped costing handshakes. + */ interface CatalogRefreshSummary { readonly candidates: number; readonly synced: number; + readonly incomplete: number; readonly failed: number; + readonly lostClaim: number; + readonly skippedClaimed: number; + readonly skippedBackoff: number; + readonly skippedParked: number; } -const NO_CATALOG_REFRESH: CatalogRefreshSummary = { candidates: 0, synced: 0, failed: 0 }; +const NO_CATALOG_REFRESH: CatalogRefreshSummary = { + candidates: 0, + synced: 0, + incomplete: 0, + failed: 0, + lostClaim: 0, + skippedClaimed: 0, + skippedBackoff: 0, + skippedParked: 0, +}; + +/** A nullable bigint column as epoch milliseconds. The drivers hand these back + * as `bigint` (Postgres) or `number` (sqlite/D1), and every comparison in the + * sync lifecycle is arithmetic. Narrow rather than `unknown`: `Number()` turns + * anything else into `NaN`, and a `NaN` stamp compares false against every + * bound, which would silently classify a connection as fresh forever. */ +const asEpochMs = (value: bigint | number | null | undefined): number | null => + value == null ? null : Number(value); + +/** A backoff jitter factor in [`TOOL_SYNC_JITTER_MIN`, `TOOL_SYNC_JITTER_MAX`]. + * The randomness lives here rather than in `tool-sync-schedule`, which stays a + * pure module; `Math.random` is how this package sources it everywhere else. */ +const toolSyncJitter = (): number => + TOOL_SYNC_JITTER_MIN + Math.random() * (TOOL_SYNC_JITTER_MAX - TOOL_SYNC_JITTER_MIN); + +/** A refresh claim nonce. Only has to be unique among the attempts racing for + * one connection inside one lease window, and it is never an address. */ +const toolSyncClaimNonce = (): string => + `sync_${Date.now().toString(36)}${Math.random().toString(36).slice(2, 10)}`; + +/** A drift mark, as an opaque token. Generated exactly like the claim nonce and + * for the same reason: it only has to be distinguishable from the value a + * concurrent listing observed, which is what makes the clear a compare-and-set + * rather than a timestamp comparison that ties inside one millisecond. Never + * parsed, never ordered, never an address. */ +const toolsStaleToken = (): string => + `stale_${Date.now().toString(36)}${Math.random().toString(36).slice(2, 10)}`; /** The error tags in a swallowed refresh cause, as a stable comma-joined set. * Tags only: this cause can carry an upstream HTTP request/response, so no @@ -2563,23 +2683,163 @@ export const createExecutor = ({ status: "degraded", checkedAt: Date.now(), - detail: `${toolSyncHealthDetailPrefix}: ${reason}`, + detail: `Tool sync failing: ${reason}`, }); const syncHealthReason = (result: ResolveToolsResult): string => result.incompleteReason ?? "plugin returned an incomplete tool catalog"; + // The terminal outcome of a listing that did NOT produce an authoritative + // catalog. `tools_synced_at` is deliberately untouched: stamping it here is + // what made a month-dead server report as "synced 30s ago" and earn a fresh + // handshake every freshness window. Any drift signal (`tools_stale_token`) + // also stands, because nothing resolved it — the retry ladder is the only + // thing holding the next attempt off, and an `auth` verdict parks a + // connection the clock is merely asking to re-verify. + // + // `priorFailures` is the CONSECUTIVE count already on the row, so the + // `failures` derived here is `>= 1` — the contract `scheduleAfterFailure` + // states and deliberately does not clamp. + const toolSyncFailureSet = ( + priorFailures: number, + kind: ToolSyncErrorKind | null, + reason: string, + ): Record => { + const now = Date.now(); + const failures = priorFailures + 1; + return { + tools_sync_failures: failures, + tools_sync_retry_at: scheduleAfterFailure( + now, + toolSyncBackoffBaseMs, + failures, + toolSyncJitter(), + ), + tools_sync_error_kind: kind, + tools_sync_claim_id: null, + tools_sync_claim_at: null, + last_health: toolSyncHealth(reason), + updated_at: new Date(), + }; + }; + + /** The column resets that put a connection back at the start of the retry + * ladder — failure count, next-attempt instant, and the error kind that + * may have parked it. Applied by every path that can plausibly have fixed + * the cause: an explicit refresh, a connection edit, an OAuth re-mint, a + * healthy probe, and a fresh drift signal. Folded into those writes' own + * `set` wherever they already issue one. */ + const TOOL_SYNC_LADDER_RESET = { + tools_sync_failures: null, + tools_sync_retry_at: null, + tools_sync_error_kind: null, + } as const; + + /** The standalone form, for the one caller with no write of its own to fold + * into. Locked: it can run concurrently with a read's refresh fan-out, and + * a bare UPDATE issued while another fiber holds an open BEGIN on the + * shared SQLite connection is enrolled in that fiber's transaction. */ + const resetToolSyncLadder = (ref: ConnectionRef): Effect.Effect => + withCatalogPersistLock( + core.updateMany("connection", { + where: (b: AnyCb) => + b.and( + byOwner(ref.owner)(b), + b("integration", "=", String(ref.integration)), + b("name", "=", String(ref.name)), + ), + set: { ...TOOL_SYNC_LADDER_RESET }, + }), + ); + + /** + * Take the refresh lease for a connection, or report that someone else + * holds it. Returns the claim nonce on success, `null` when the connection + * is already claimed. + * + * A compare-and-set, because there is no shared memory to coordinate + * through: concurrent `tools.list` calls land in different Workers isolates + * and the ROW is the only medium they share. The UPDATE's WHERE is the + * compare (the claim is free, or its lease has run out); the read-back is + * the verdict, because fumadb's `updateMany` returns void and reports no + * row count. That read-back is sound rather than a heuristic: under READ + * COMMITTED Postgres re-evaluates the predicate against the row version the + * concurrent writer committed, so exactly one of two racing UPDATEs can see + * a free claim, and sqlite/D1 serialize writes outright. + * + * Locked and outside any transaction: this write runs during the read's + * concurrent fan-out, where another fiber may hold an open BEGIN on the + * shared SQLite connection. The permit is not reentrant, so callers must + * hold nothing when they get here. + */ + const claimConnectionForSync = ( + ref: ConnectionRef, + ): Effect.Effect => { + const connectionWhere = (b: AnyCb) => + b.and( + byOwner(ref.owner)(b), + b("integration", "=", String(ref.integration)), + b("name", "=", String(ref.name)), + ); + return withCatalogPersistLock( + Effect.gen(function* () { + const nonce = toolSyncClaimNonce(); + const now = Date.now(); + yield* core.updateMany("connection", { + where: (b: AnyCb) => + b.and( + connectionWhere(b), + b.or( + b.isNull("tools_sync_claim_id"), + b.isNull("tools_sync_claim_at"), + b("tools_sync_claim_at", "<", now - TOOL_SYNC_CLAIM_LEASE_MS), + ), + ), + set: { tools_sync_claim_id: nonce, tools_sync_claim_at: now }, + }); + const readBack = yield* core.findFirst("connection", { + where: connectionWhere, + select: ["tools_sync_claim_id"], + }); + return readBack?.tools_sync_claim_id === nonce ? nonce : null; + }), + ); + }; + const produceConnectionTools = ( integrationRow: IntegrationRow, ref: ConnectionRef, - mode: "explicit" | "background" = "explicit", - ): Effect.Effect => + attempt: ToolSyncAttempt = EXPLICIT_TOOL_SYNC, + ): Effect.Effect => Effect.gen(function* () { + // Captured before anything is dialed, and it is what stamps + // `tools_synced_at`: the catalog this listing returns is the upstream's + // tool set as of NOW, not as of whenever the handshake happened to + // finish. Anything that invalidates the catalog after this instant + // must out-date the stamp rather than be swallowed by it. + const listingStartedAt = Date.now(); const runtime = runtimes.get(integrationRow.plugin_id); const keys = yield* Effect.try({ try: () => ownedKeys(ref.owner), @@ -2598,41 +2858,166 @@ export const createExecutor = + // Only ever read INSIDE the outcome's own transaction, where `core` + // resolves to the transaction handle: the tool rows it gates are + // deleted and reinserted, and no WHERE clause on the connection can + // condition those. + const holdsClaim = (): Effect.Effect => + attempt.claim === null + ? Effect.succeed(true) + : core + .findFirst("connection", { + where: connectionWhere, + select: ["tools_sync_claim_id"], + }) + .pipe(Effect.map((row) => row?.tools_sync_claim_id === attempt.claim)); + + // The terminal write carries the claim in its WHERE, which makes it a + // compare-and-set rather than a check-then-act. `withCatalogPersistLock` + // is an in-process semaphore and cannot see the other Workers isolate + // that re-claimed this connection while we were listing; the predicate + // can, and an update that no longer matches simply writes nothing. + const outcomeWhere = (b: AnyCb) => + attempt.claim === null + ? connectionWhere(b) + : b.and(connectionWhere(b), b("tools_sync_claim_id", "=", attempt.claim)); + + // The terminal outcome of an authoritative listing, in ONE write: the + // catalog is verified as of `listingStartedAt`, the failure ledger that + // sent us here is settled, and the lease is released. `last_health` is + // cleared only when the previous outcome was a sync failure, so a + // genuine health-check verdict survives. The failure COUNT is what says + // so — `tools_sync_error_kind` is optional, since a plugin is allowed to + // report an incomplete listing without classifying it. + // + // The stamp is the instant the listing STARTED, not the instant it + // landed. It records freshness and NOTHING else: resolving the drift + // mark is a separate compare-and-set inside `replaceCatalog`, so a + // signal that arrived mid-listing is not swallowed by a stamp that + // happens to sort after it. + const successSet = (row: ConnectionRow | null) => { + const recovering = Number(row?.tools_sync_failures ?? 0) > 0; + return { + tools_synced_at: listingStartedAt, + tools_sync_failures: null, + tools_sync_retry_at: scheduleAfterSuccess(listingStartedAt, toolsSyncTtlMs), + tools_sync_error_kind: null, + tools_sync_claim_id: null, + tools_sync_claim_at: null, + ...(recovering ? { last_health: null, updated_at: new Date() } : {}), + }; + }; + + /** Record a NON-authoritative outcome, or report that the attempt lost + * its lease and wrote nothing. Returns false in the latter case. + * + * Locked: this runs inside a read's concurrent fan-out, and a bare + * UPDATE issued while another fiber holds an open BEGIN on the shared + * SQLite connection is enrolled in that fiber's transaction — and lost + * with it if it rolls back. The permit is NOT reentrant, so nothing + * reachable from here may take it again. Transactional for the same + * reason as {@link replaceCatalog}: `holdsClaim` is only sound read + * inside the transaction that acts on it. */ + const writeSyncOutcome = ( + set: Record, + ): Effect.Effect => withCatalogPersistLock( - core.updateMany("connection", { - where: connectionWhere, - set: { - tools_synced_at: Date.now(), - last_health: toolSyncHealth(reason), - updated_at: new Date(), - }, - }), + transaction( + Effect.gen(function* () { + if (!(yield* holdsClaim())) return false; + yield* core.updateMany("connection", { where: outcomeWhere, set }); + return true; + }), + ), + ); + + /** Replace the persisted catalog, record the outcome, and resolve the + * drift mark this listing answered — all three together, or none of + * them. Returns false when the attempt lost its lease. + * + * That atomicity is dialect-conditional: on D1 + * (`interactiveTransactions: false`) there is no transaction, the + * statements auto-commit one by one, and an interrupted run can leave a + * prefix of them applied. It still fails CLOSED, because the drift clear + * is the LAST statement: any prefix short of the whole leaves the token + * in place, so the connection stays drifted and re-lists. + * + * Already holds the persist permit, which is not reentrant: nothing in + * here may take it again. + * + * INVARIANT on the drift clear: it is a compare-and-set against the + * token observed when this listing STARTED, never an unconditional + * null. `connections.markToolsStale` writes a fresh token from another + * fiber (a `tools/list_changed` arriving mid-invocation), and a mark + * that landed while we were dialing describes a tool set this listing + * never saw — its token differs, the WHERE misses, and the drift + * survives to re-list on the next read. + * + * It fails CLOSED, and it converges: a clear that misses leaves the + * connection drifted, so it re-lists, observes the token that is now + * there, and clears it next time. */ + const replaceCatalog = ( + toolRows: readonly Record[], + definitionRows: readonly Record[], + set: Record, + ): Effect.Effect => + withCatalogPersistLock( + transaction( + Effect.gen(function* () { + if (!(yield* holdsClaim())) return false; + yield* core.deleteMany("tool", { where }); + yield* core.deleteMany("definition", { where }); + yield* core.createMany("tool", toolRows); + yield* core.createMany("definition", definitionRows); + yield* core.updateMany("connection", { where: outcomeWhere, set }); + if (observedStaleToken !== null) { + // Keyed on the connection and the token, deliberately NOT on + // `outcomeWhere`: the write above already released the lease + // (`successSet` nulls `tools_sync_claim_id`), so a predicate + // carrying this attempt's claim would match nothing from here + // on. The lease was checked by `holdsClaim` at the top of this + // transaction, which is where it belongs. + yield* core.updateMany("connection", { + where: (b: AnyCb) => + b.and(connectionWhere(b), b("tools_stale_token", "=", observedStaleToken)), + set: { tools_stale_token: null }, + }); + } + return true; + }), + ), + ); + + const persistedTools = ( + outcome: ToolProductionOutcome, + ): Effect.Effect => + core.findMany("tool", { where }).pipe( + Effect.map((rows) => ({ + tools: rows.map((row) => rowToTool(row as ConnectionToolRow)), + outcome, + })), ); // Defense in depth (and cleanup for rows created before the create-time @@ -2643,37 +3028,26 @@ export const createExecutor = rowToTool(row as ConnectionToolRow)); + return yield* persistedTools(recorded ? "incomplete" : "lost_claim"); } if ( - mode === "background" && + attempt.trigger !== "explicit" && runtime.plugin.remoteToolCatalog === true && result.tools.length === 0 ) { @@ -2720,14 +3104,28 @@ export const createExecutor = 0) { const reason = "background tool sync produced an authoritative empty catalog for a connection with existing tools"; - yield* stampSyncedWithHealth(reason); + // Treated as a failure, not a listing: an empty catalog where tools + // existed is far more often a broken server than a real change, and + // the plugin said nothing about why. Unclassified, so it backs off + // but never parks. + const recorded = yield* writeSyncOutcome(failureSet(null, reason)); + // `recorded` for the same reason as above: the warning fires even when + // the lease was lost and nothing was written. yield* Effect.logWarning("executor tool sync preserved nonzero catalog", { reason, integration: String(ref.integration), connection: String(ref.name), existingToolCount: keptRows.length, + recorded, }); - return keptRows.map((row) => rowToTool(row as ConnectionToolRow)); + // Re-read on a lost lease: `keptRows` predates the write, and the + // attempt that owns the connection may have replaced the catalog + // since. + if (!recorded) return yield* persistedTools("lost_claim"); + return { + tools: keptRows.map((row) => rowToTool(row as ConnectionToolRow)), + outcome: "incomplete", + }; } } @@ -2760,38 +3158,36 @@ export const createExecutor = - rowToTool( - { - tenant: keys.tenant, - owner: keys.owner, - subject: keys.subject, - integration: String(ref.integration), - connection: String(ref.name), - plugin_id: integrationRow.plugin_id, - name: String(tool.name), - description: tool.description ?? "", - input_schema: tool.inputSchema ?? null, - output_schema: tool.outputSchema ?? null, - annotations: tool.annotations ?? null, - created_at: now, - updated_at: now, - } as ConnectionToolRow, - tool.annotations, + return { + outcome: "synced", + tools: result.tools.map((tool: ToolDef) => + rowToTool( + { + tenant: keys.tenant, + owner: keys.owner, + subject: keys.subject, + integration: String(ref.integration), + connection: String(ref.name), + plugin_id: integrationRow.plugin_id, + name: String(tool.name), + description: tool.description ?? "", + input_schema: tool.inputSchema ?? null, + output_schema: tool.outputSchema ?? null, + annotations: tool.annotations ?? null, + created_at: now, + updated_at: now, + } as ConnectionToolRow, + tool.annotations, + ), ), - ); + }; }); // ------------------------------------------------------------------ @@ -2965,7 +3361,8 @@ export const createExecutor = Effect.succeed([] as readonly Tool[])), + Effect.asVoid, + Effect.catchTag("IntegrationNotFoundError", () => Effect.void), ); const row = yield* findConnectionRow(ref); @@ -3048,8 +3445,12 @@ export const createExecutor = Effect.succeed([] as readonly Tool[])), + Effect.asVoid, + Effect.catchTag("IntegrationNotFoundError", () => Effect.void), ); const row = yield* findConnectionRow(ref); @@ -3174,7 +3576,10 @@ export const createExecutor = = { updated_at: new Date() }; + // An edit is a caller acting on the connection, which is the same + // signal an explicit refresh carries: restart the ladder rather than + // leave a parked connection parked on evidence that predates the edit. + const set: Record = { updated_at: new Date(), ...TOOL_SYNC_LADDER_RESET }; if (input.description !== undefined) set.description = input.description; if (input.identityLabel !== undefined) set.identity_label = input.identityLabel; yield* core.updateMany("connection", { @@ -3256,13 +3661,22 @@ export const createExecutor = ({ status: "unknown", checkedAt: Date.now() }); + // Only `connectionCheckHealth` persists a verdict, and a HEALTHY one is + // direct evidence that whatever parked this connection's tool sync is gone: + // the credential resolves and the upstream answers. Un-park in the same + // write, so the next read re-lists instead of waiting for a human. const persistHealthResult = ( ref: ConnectionRef, result: HealthCheckResult, @@ -3275,7 +3689,11 @@ export const createExecutor = => - core.updateMany("connection", { - where: (b: AnyCb) => - b.and( - byOwner(ref.owner)(b), - b("integration", "=", String(ref.integration)), - b("name", "=", String(ref.name)), - ), - set: { tools_synced_at: null }, - }); + withCatalogPersistLock( + core.updateMany("connection", { + where: (b: AnyCb) => + b.and( + byOwner(ref.owner)(b), + b("integration", "=", String(ref.integration)), + b("name", "=", String(ref.name)), + ), + set: { tools_stale_token: toolsStaleToken(), ...TOOL_SYNC_LADDER_RESET }, + }), + ); // ------------------------------------------------------------------ // Active policy source. @@ -3653,17 +4093,16 @@ export const createExecutor = => @@ -3720,8 +4163,19 @@ export const createExecutor = b.and( @@ -3732,11 +4186,12 @@ export const createExecutor = ; + readonly priorFailures: number; }[] = []; + let skippedClaimed = 0; + let skippedBackoff = 0; + let skippedParked = 0; for (const connection of connections) { const integrationRow = integrationBySlug.get(connection.integration); if (!integrationRow) continue; @@ -3761,22 +4220,34 @@ export const createExecutor = - produceConnectionTools(integrationRow, ref, "background").pipe( - Effect.andThen(Effect.annotateCurrentSpan({ "executor.tools.sync.outcome": "ok" })), - Effect.as(true), - // catchCause, not catch: a plugin DEFECT is not in the error - // channel, and a tools READ must not be failable by one - // connection's misbehaving refresh. The stale-but-working - // catalog stays in place and the next read retries. - // - // Warning, not debug: swallowing this is the whole point, so the - // log line is the only signal that a connection is serving a - // stale catalog. Enumerable fields only — a refresh runs on a - // live credential and its cause can embed the upstream request - // and response. - Effect.catchCause((cause) => - Effect.annotateCurrentSpan({ "executor.tools.sync.outcome": "fail" }).pipe( - Effect.andThen( - Effect.logWarning("executor tool catalog refresh failed", { - integration: String(ref.integration), - connection: String(ref.name), - trigger, - errorTags: causeErrorTags(cause), - }), - ), - Effect.as(false), + ({ integrationRow, ref, state, priorFailures }) => + Effect.gen(function* () { + const claim = yield* claimConnectionForSync(ref); + yield* Effect.annotateCurrentSpan({ + "executor.tools.sync.claimed": claim !== null, + }); + if (claim === null) return "skipped_claimed" as const; + return yield* produceConnectionTools(integrationRow, ref, { + trigger: state, + claim, + }).pipe( + // Reported verbatim: `synced`, `incomplete` and `lost_claim` + // are three different things to be told, and only the first is + // a sync. Folding the other two into it is how a fleet that + // persists nothing reads as fully healthy. + Effect.flatMap((produced) => + Effect.annotateCurrentSpan({ + "executor.tools.sync.outcome": produced.outcome, + }).pipe(Effect.as(produced.outcome)), ), - ), + // catchCause, not catch: a plugin DEFECT is not in the error + // channel, and a tools READ must not be failable by one + // connection's misbehaving refresh. The stale-but-working + // catalog stays in place and the next read retries. + // + // Warning, not debug: swallowing this is the whole point, so + // the log line is the only signal that a connection is serving + // a stale catalog. Enumerable fields only — a refresh runs on a + // live credential and its cause can embed the upstream request + // and response. + Effect.catchCause((cause) => + // Cancellation first: a `tools.list` abandoned by its caller + // must not be recorded as an upstream failure, and writing a + // retry schedule from an interrupted fiber would persist a + // verdict nobody reached. Let it through untouched — the + // lease expires on its own. + Cause.hasInterrupts(cause) + ? Effect.interrupt + : Effect.gen(function* () { + yield* Effect.annotateCurrentSpan({ + "executor.tools.sync.outcome": "fail", + }); + // A genuine failure walks the retry ladder; a DEFECT + // only releases the lease. The four error kinds describe + // what an upstream did, and a bug in this process is + // none of them — it is loud, repeated and fixable, so it + // keeps retrying rather than quietly backing off under a + // wrong label. + yield* withCatalogPersistLock( + core.updateMany("connection", { + where: (b: AnyCb) => + b.and( + byOwner(ref.owner)(b), + b("integration", "=", String(ref.integration)), + b("name", "=", String(ref.name)), + b("tools_sync_claim_id", "=", claim), + ), + set: Cause.hasFails(cause) + ? toolSyncFailureSet( + priorFailures, + null, + "the tool catalog refresh failed", + ) + : { tools_sync_claim_id: null, tools_sync_claim_at: null }, + }), + ).pipe(Effect.ignore); + yield* Effect.logWarning("executor tool catalog refresh failed", { + integration: String(ref.integration), + connection: String(ref.name), + trigger: state, + errorTags: causeErrorTags(cause), + }); + return "fail" as const; + }), + ), + ); + }).pipe( Effect.withSpan("executor.tools.sync", { attributes: { "executor.integration": String(ref.integration), "executor.connection": String(ref.name), - "executor.tools.sync.trigger": trigger, + "executor.tools.sync.trigger": state, }, }), ), { concurrency: 4 }, ); - const synced = outcomes.filter((ok) => ok).length; return { candidates: connections.length, - synced, - failed: outcomes.length - synced, + synced: outcomes.filter((outcome) => outcome === "synced").length, + incomplete: outcomes.filter((outcome) => outcome === "incomplete").length, + failed: outcomes.filter((outcome) => outcome === "fail").length, + lostClaim: outcomes.filter((outcome) => outcome === "lost_claim").length, + skippedClaimed: + skippedClaimed + outcomes.filter((outcome) => outcome === "skipped_claimed").length, + skippedBackoff, + skippedParked, }; }); @@ -3892,7 +4419,12 @@ export const createExecutor = { readonly Tool[], ConnectionNotFoundError | IntegrationNotFoundError | StorageFailure >; - /** Mark a connection's persisted tool catalog stale (clears its sync - * stamp) without re-listing inline. The next tools read re-produces it. - * For signals that arrive mid-invocation — e.g. an MCP server sending + /** Record that a connection's persisted tool catalog has drifted, without + * re-listing inline. The next tools read re-produces it. For signals that + * arrive mid-invocation — e.g. an MCP server sending * `notifications/tools/list_changed` or rejecting a call as an unknown - * tool — where an inline `refresh` would block the caller. */ + * tool — where an inline `refresh` would block the caller. The last + * verified sync time is preserved; the drift is recorded alongside it. + * + * Safe to call at any point, including while a listing for the same + * connection is in flight: the mark is recorded as a fresh token, and a + * listing only resolves the token it observed when it started. A signal + * that arrives mid-listing therefore survives it and re-lists. + * + * NOT from inside `ctx.transaction(...)`, though: the write takes a + * non-reentrant executor permit, and on sqlite an open BEGIN blocking on + * that permit can silently drop a concurrent catalog rebuild. Call it + * before the transaction, or after it commits. */ readonly markToolsStale: (ref: ConnectionRef) => Effect.Effect; /** Resolve a connection's value through its provider (and OAuth refresh). * null if the provider can't produce one. */ @@ -335,6 +347,14 @@ export interface ResolveToolsResult { /** Human-readable reason for an incomplete listing. Persisted by core when it * preserves the prior catalog so operators can see why data is stale. */ readonly incompleteReason?: string; + /** What KIND of failure made the listing incomplete, as the structured + * counterpart to `incompleteReason`. Core persists it on the connection and + * schedules the next attempt from it: `auth` parks the connection until a + * human re-authorizes, because retrying a rejected credential cannot change + * the verdict, while the other kinds walk the retry ladder. Omit when the + * plugin genuinely cannot tell them apart — an unclassified failure still + * backs off, it just never parks. */ + readonly incompleteKind?: ToolSyncErrorKind; } export interface ProjectToolSchemaInput { diff --git a/packages/core/sdk/src/tool-sync-schedule.test.ts b/packages/core/sdk/src/tool-sync-schedule.test.ts new file mode 100644 index 0000000000..f57247215f --- /dev/null +++ b/packages/core/sdk/src/tool-sync-schedule.test.ts @@ -0,0 +1,493 @@ +import { describe, expect, it } from "@effect/vitest"; + +import { + TOOL_SYNC_BACKOFF_CEILING_MS, + TOOL_SYNC_CLAIM_LEASE_MS, + TOOL_SYNC_ERROR_KINDS, + TOOL_SYNC_JITTER_MAX, + TOOL_SYNC_JITTER_MIN, + classifyToolSync, + decideToolSync, + isSyncEligible, + isToolSyncBackedOff, + isToolSyncClaimLive, + isToolSyncErrorKind, + isToolSyncParked, + scheduleAfterFailure, + scheduleAfterSuccess, + type ToolSyncCandidate, +} from "./tool-sync-schedule"; + +// The catalog sync lifecycle is a pure decision function, and these are the +// decisions it exists to make: what a persisted catalog IS, whether anyone is +// allowed to re-list it right now, and how long a failing connection is left +// alone. Testing it here rather than only through the executor is what keeps +// the ladder, the cap and the park rule honest without a database and a live +// upstream in the loop. + +const NOW = 1_700_000_000_000; +const TTL = 15 * 60 * 1000; + +/** A drift mark, as `connections.markToolsStale` writes it. Opaque: nothing + * here may read structure out of it, and nothing may order two of them. */ +const STALE_TOKEN = "stale_m3k1x9q2"; + +/** A row every case overrides into the state it is about. Neutral on purpose: + * naming it for one of the states under test (it started life as `fresh`) + * makes `classifyToolSync(fresh({ toolsSyncedAt: null }))` read as a + * contradiction with the assertion beside it. */ +const candidate = (overrides: Partial = {}): ToolSyncCandidate => ({ + toolsSyncedAt: NOW - 1000, + toolsStaleToken: null, + toolsSyncClaimId: null, + toolsSyncClaimAt: null, + toolsSyncFailures: 0, + toolsSyncRetryAt: null, + toolsSyncErrorKind: null, + configRevisedAt: null, + remoteToolCatalog: true, + ttlMs: TTL, + ...overrides, +}); + +describe("isToolSyncErrorKind", () => { + it("narrows exactly the kinds the closed set declares", () => { + // Pinned as a literal, not derived: the set is closed because + // `isToolSyncParked`'s auth-only rule and the persisted-column parse depend + // on it, so a fifth kind has to be a deliberate edit here rather than a + // silent widening. + expect([...TOOL_SYNC_ERROR_KINDS]).toEqual(["auth", "unreachable", "protocol", "config"]); + for (const kind of TOOL_SYNC_ERROR_KINDS) { + expect(isToolSyncErrorKind(kind)).toBe(true); + } + }); + + it("rejects anything else a text column could hold", () => { + for (const value of ["", "AUTH", "authentication", null, undefined, 401, {}]) { + expect(isToolSyncErrorKind(value)).toBe(false); + } + }); +}); + +describe("classifyToolSync", () => { + it("calls a never-synced, never-marked connection cold", () => { + expect(classifyToolSync(candidate({ toolsSyncedAt: null }), NOW)).toBe("cold"); + }); + + it("calls a connection with a drift mark stale_marked, keeping its stamp", () => { + expect( + classifyToolSync( + candidate({ toolsSyncedAt: NOW - 60_000, toolsStaleToken: STALE_TOKEN }), + NOW, + ), + ).toBe("stale_marked"); + }); + + it("reads the drift mark as presence, never against the stamp", () => { + // The whole point of the token encoding. `tools_stale_token` answers "was + // this catalog invalidated since the listing now finishing began", which is + // a version question; the earlier encoding asked it of a wall clock, and + // `Date.now()` has millisecond granularity, so a mark and a stamp landing in + // one millisecond tied. No stamp, however far ahead of the mark, resolves + // it — only the compare-and-set that nulls the column does. + for (const toolsSyncedAt of [null, 0, NOW - 60_000, NOW, NOW + 60_000]) { + expect( + classifyToolSync(candidate({ toolsSyncedAt, toolsStaleToken: STALE_TOKEN }), NOW), + ).toBe("stale_marked"); + } + }); + + it("leaves a connection whose drift mark was cleared resolved", () => { + // An authoritative listing clears the token it observed, and a cleared + // column is the only thing that settles a drift. This is also what keeps + // the read's scan at zero rows in steady state: nothing ever cleared the + // old timestamp column, so every connection that had ever drifted was + // re-selected on every read, forever. + expect( + classifyToolSync(candidate({ toolsSyncedAt: NOW - 1000, toolsStaleToken: null }), NOW), + ).toBe("fresh"); + }); + + it("calls a never-synced connection carrying a drift mark stale_marked, not cold", () => { + expect( + classifyToolSync(candidate({ toolsSyncedAt: null, toolsStaleToken: STALE_TOKEN }), NOW), + ).toBe("stale_marked"); + }); + + it("calls a catalog older than its integration's config revision config_revised", () => { + expect( + classifyToolSync( + candidate({ toolsSyncedAt: NOW - 60_000, configRevisedAt: NOW - 30_000 }), + NOW, + ), + ).toBe("config_revised"); + }); + + it("still sees a config revision that landed in the same millisecond as the stamp", () => { + // Same rule as the drift mark, and for a stronger reason: `tools_synced_at` + // is stamped when the listing STARTED, so a revision inside that + // millisecond describes a config the listing cannot have read. A plugin + // without a remote catalog has no TTL backstop to catch it later. + expect(classifyToolSync(candidate({ toolsSyncedAt: NOW, configRevisedAt: NOW }), NOW)).toBe( + "config_revised", + ); + }); + + it("leaves a catalog newer than the config revision alone", () => { + expect( + classifyToolSync( + candidate({ toolsSyncedAt: NOW - 30_000, configRevisedAt: NOW - 60_000 }), + NOW, + ), + ).toBe("fresh"); + }); + + it("expires a remote catalog on the far side of the TTL only", () => { + // Both sides of the boundary: flipping the comparison, or shifting the + // subtraction, changes when every remote catalog re-dials its upstream. + expect(classifyToolSync(candidate({ toolsSyncedAt: NOW - TTL }), NOW)).toBe("fresh"); + expect(classifyToolSync(candidate({ toolsSyncedAt: NOW - TTL - 1 }), NOW)).toBe("expired"); + }); + + it("never expires a catalog the plugin does not list remotely", () => { + // An OpenAPI catalog is derived from a stored spec: time alone cannot make + // it wrong, and re-listing it on a timer is pure waste. + expect( + classifyToolSync(candidate({ toolsSyncedAt: NOW - TTL - 1, remoteToolCatalog: false }), NOW), + ).toBe("fresh"); + }); + + it("never expires anything when the TTL is disabled", () => { + expect(classifyToolSync(candidate({ toolsSyncedAt: 0, ttlMs: null }), NOW)).toBe("fresh"); + }); + + it("ranks an explicit drift signal above a config revision and the clock", () => { + expect( + classifyToolSync( + candidate({ + toolsSyncedAt: NOW - TTL - 1, + toolsStaleToken: STALE_TOKEN, + configRevisedAt: NOW - 1, + }), + NOW, + ), + ).toBe("stale_marked"); + }); + + it("ranks a config revision above the clock", () => { + expect( + classifyToolSync(candidate({ toolsSyncedAt: NOW - TTL - 1, configRevisedAt: NOW - 1 }), NOW), + ).toBe("config_revised"); + }); +}); + +describe("isToolSyncClaimLive", () => { + it("is false with no claim", () => { + expect(isToolSyncClaimLive(candidate(), NOW)).toBe(false); + }); + + it("is true inside the lease", () => { + expect( + isToolSyncClaimLive( + candidate({ + toolsSyncClaimId: "sync_abc", + toolsSyncClaimAt: NOW - TOOL_SYNC_CLAIM_LEASE_MS + 1, + }), + NOW, + ), + ).toBe(true); + }); + + it("is false once the lease has run out", () => { + expect( + isToolSyncClaimLive( + candidate({ + toolsSyncClaimId: "sync_abc", + toolsSyncClaimAt: NOW - TOOL_SYNC_CLAIM_LEASE_MS, + }), + NOW, + ), + ).toBe(false); + }); + + it("treats a claim with no timestamp as dead rather than eternal", () => { + // Otherwise a half-written row would lock a connection out of every future + // refresh, forever, with nothing to expire it. + expect( + isToolSyncClaimLive(candidate({ toolsSyncClaimId: "sync_abc", toolsSyncClaimAt: null }), NOW), + ).toBe(false); + }); +}); + +describe("isToolSyncBackedOff", () => { + it("holds a failing connection off until its retry instant", () => { + const backedOff = candidate({ toolsSyncFailures: 2, toolsSyncRetryAt: NOW + 1 }); + expect(isToolSyncBackedOff(backedOff, NOW)).toBe(true); + expect(isToolSyncBackedOff(backedOff, NOW + 1)).toBe(false); + }); + + it("ignores the retry instant a SUCCESS wrote", () => { + // After a success `tools_sync_retry_at` carries the freshness horizon, not + // a gate: a drift signal arriving inside the window must heal the catalog + // immediately instead of waiting the window out. + const drifted = candidate({ + toolsSyncFailures: 0, + toolsSyncRetryAt: NOW + TTL, + toolsStaleToken: STALE_TOKEN, + }); + expect(isToolSyncBackedOff(drifted, NOW)).toBe(false); + expect(isSyncEligible(drifted, NOW)).toBe(true); + }); +}); + +describe("isToolSyncParked", () => { + it("parks only on an auth verdict", () => { + expect(isToolSyncParked(candidate({ toolsSyncErrorKind: "auth" }))).toBe(true); + for (const kind of ["unreachable", "protocol", "config", null] as const) { + expect(isToolSyncParked(candidate({ toolsSyncErrorKind: kind }))).toBe(false); + } + }); +}); + +describe("decideToolSync", () => { + it("passes a due connection with a clear ledger, reporting what made it due", () => { + expect(decideToolSync(candidate({ toolsSyncedAt: null }), NOW)).toEqual({ + kind: "refresh", + trigger: "cold", + }); + }); + + it("skips a fresh connection", () => { + expect(decideToolSync(candidate(), NOW)).toEqual({ kind: "skip", reason: "fresh" }); + }); + + it("skips a parked connection however far past its freshness window it is", () => { + expect( + decideToolSync( + candidate({ + toolsSyncedAt: NOW - 30 * TTL, + toolsSyncFailures: 9, + toolsSyncErrorKind: "auth", + toolsSyncRetryAt: NOW - 1, + }), + NOW, + ), + ).toEqual({ kind: "skip", reason: "parked" }); + }); + + it("skips a connection inside its backoff window", () => { + expect( + decideToolSync( + candidate({ + toolsSyncedAt: null, + toolsSyncFailures: 1, + toolsSyncErrorKind: "unreachable", + toolsSyncRetryAt: NOW + 1, + }), + NOW, + ), + ).toEqual({ kind: "skip", reason: "backoff" }); + }); + + it("skips a connection another attempt is already holding", () => { + expect( + decideToolSync( + candidate({ toolsSyncedAt: null, toolsSyncClaimId: "sync_abc", toolsSyncClaimAt: NOW }), + NOW, + ), + ).toEqual({ kind: "skip", reason: "claimed" }); + }); + + it("re-admits a connection whose claimant died", () => { + expect( + decideToolSync( + candidate({ + toolsSyncedAt: null, + toolsSyncClaimId: "sync_abc", + toolsSyncClaimAt: NOW - TOOL_SYNC_CLAIM_LEASE_MS, + }), + NOW, + ), + ).toEqual({ kind: "refresh", trigger: "cold" }); + }); + + // The skip reason is not cosmetic: the read fans it into three separate + // operator counters, so a clause reordering silently misattributes every sync + // skip. Each case below satisfies SEVERAL clauses at once, which is the only + // way the order itself is what the assertion pins. + describe("precedence", () => { + it("reports fresh over an auth verdict left on the row", () => { + // A parked connection that has since been re-listed is FRESH. Ranking the + // park first would count every healthy connection carrying a stale error + // kind as parked, on every read. + expect(decideToolSync(candidate({ toolsSyncErrorKind: "auth" }), NOW)).toEqual({ + kind: "skip", + reason: "fresh", + }); + }); + + it("reports parked over a backoff window that is still open", () => { + expect( + decideToolSync( + candidate({ + toolsSyncedAt: NOW - TTL - 1, + toolsSyncErrorKind: "auth", + toolsSyncFailures: 3, + toolsSyncRetryAt: NOW + TTL, + }), + NOW, + ), + ).toEqual({ kind: "skip", reason: "parked" }); + }); + + it("reports backoff, not parked, for a COLD connection with an auth verdict", () => { + // The park does not apply to `cold`, so the ladder is what holds this + // connection off — and the reason an operator sees has to say so, because + // the read fans the two into different counters and they mean opposite + // things: `skipped_parked` is waste permanently avoided, `skipped_backoff` + // is a retry that will happen. + expect( + decideToolSync( + candidate({ + toolsSyncedAt: null, + toolsSyncErrorKind: "auth", + toolsSyncFailures: 3, + toolsSyncRetryAt: NOW + TTL, + }), + NOW, + ), + ).toEqual({ kind: "skip", reason: "backoff" }); + }); + + it("reports backoff over a live claim", () => { + expect( + decideToolSync( + candidate({ + toolsSyncedAt: null, + toolsSyncFailures: 2, + toolsSyncRetryAt: NOW + 1, + toolsSyncClaimId: "sync_abc", + toolsSyncClaimAt: NOW, + }), + NOW, + ), + ).toEqual({ kind: "skip", reason: "backoff" }); + }); + }); + + // The park answers exactly one question: "is re-VERIFYING a catalog we can + // still serve worth dialing a credential that was refused". Every other + // trigger outranks it. + describe("park gates expired and nothing else", () => { + it("refreshes a parked connection whose integration config was revised", () => { + // This is the repair path: an `auth` verdict caused by integration + // configuration is fixed by editing that configuration, and + // `integrations.update` is the one invalidation that does not clear the + // ladder. Swallowing it leaves the fixed connection parked forever. + expect( + decideToolSync( + candidate({ + toolsSyncedAt: NOW - 60_000, + configRevisedAt: NOW - 1, + toolsSyncErrorKind: "auth", + }), + NOW, + ), + ).toEqual({ kind: "refresh", trigger: "config_revised" }); + }); + + it("refreshes a parked connection that was marked stale", () => { + expect( + decideToolSync( + candidate({ + toolsSyncedAt: NOW - 60_000, + toolsStaleToken: STALE_TOKEN, + toolsSyncErrorKind: "auth", + }), + NOW, + ), + ).toEqual({ kind: "refresh", trigger: "stale_marked" }); + }); + + it("refreshes a parked connection that has never synced at all", () => { + // A parked `cold` connection has NO catalog to serve, so parking it is + // not "keep serving what we have", it is "serve nothing until a human + // notices". The ladder still caps the cost at a few dials a day, and it + // is what lets a credential repaired upstream — where nothing in this + // system observes the repair — recover on its own. + expect( + decideToolSync(candidate({ toolsSyncedAt: null, toolsSyncErrorKind: "auth" }), NOW), + ).toEqual({ kind: "refresh", trigger: "cold" }); + }); + + it("still parks when the clock is only asking for a re-verification", () => { + expect( + decideToolSync( + candidate({ toolsSyncedAt: NOW - TTL - 1, toolsSyncErrorKind: "auth" }), + NOW, + ), + ).toEqual({ kind: "skip", reason: "parked" }); + }); + }); +}); + +describe("isSyncEligible", () => { + it("projects the decision to a boolean", () => { + expect(isSyncEligible(candidate({ toolsSyncedAt: null }), NOW)).toBe(true); + expect(isSyncEligible(candidate(), NOW)).toBe(false); + }); +}); + +describe("scheduleAfterSuccess", () => { + it("puts the next attempt one freshness window out", () => { + expect(scheduleAfterSuccess(NOW, TTL)).toBe(NOW + TTL); + }); + + it("schedules nothing when time-based re-sync is disabled", () => { + expect(scheduleAfterSuccess(NOW, null)).toBe(null); + }); +}); + +describe("scheduleAfterFailure", () => { + const delay = (failures: number, jitter = 1, ttlMs = TTL) => + scheduleAfterFailure(NOW, ttlMs, failures, jitter) - NOW; + + it("doubles the wait with every consecutive failure", () => { + // `failures` is the count INCLUDING the failure being recorded, so 1 is the + // first rung. Out-of-contract counts are not clamped — a caller passing the + // prior count would run the whole ladder one step short, and that is a bug + // to surface rather than launder. + expect(delay(1)).toBe(TTL); + expect(delay(2)).toBe(2 * TTL); + expect(delay(3)).toBe(4 * TTL); + expect(delay(4)).toBe(8 * TTL); + }); + + it("caps the ladder rather than growing without bound", () => { + // A server dead for a week is still probed a few times a day, which is what + // makes recovery automatic. + expect(delay(20)).toBe(TOOL_SYNC_BACKOFF_CEILING_MS); + expect(delay(1000)).toBe(TOOL_SYNC_BACKOFF_CEILING_MS); + }); + + it("stays within the jitter bounds at every rung", () => { + for (let failures = 1; failures <= 12; failures += 1) { + const base = Math.min(TOOL_SYNC_BACKOFF_CEILING_MS, TTL * 2 ** (failures - 1)); + expect(delay(failures, TOOL_SYNC_JITTER_MIN)).toBe(Math.round(base * TOOL_SYNC_JITTER_MIN)); + expect(delay(failures, TOOL_SYNC_JITTER_MAX)).toBe(Math.round(base * TOOL_SYNC_JITTER_MAX)); + for (const jitter of [TOOL_SYNC_JITTER_MIN, 0.93, 1.07, TOOL_SYNC_JITTER_MAX]) { + const value = delay(failures, jitter); + expect(value).toBeGreaterThanOrEqual(Math.round(base * TOOL_SYNC_JITTER_MIN)); + expect(value).toBeLessThanOrEqual(Math.round(base * TOOL_SYNC_JITTER_MAX)); + } + } + }); + + it("returns a whole number of milliseconds for the bigint column", () => { + expect(Number.isInteger(scheduleAfterFailure(NOW, TTL, 3, 0.8137))).toBe(true); + }); + + it("scales the ladder to whatever base it is given", () => { + expect(delay(5, 1, 1000)).toBe(16_000); + }); +}); diff --git a/packages/core/sdk/src/tool-sync-schedule.ts b/packages/core/sdk/src/tool-sync-schedule.ts new file mode 100644 index 0000000000..f68122a661 --- /dev/null +++ b/packages/core/sdk/src/tool-sync-schedule.ts @@ -0,0 +1,263 @@ +// --------------------------------------------------------------------------- +// Tool-catalog sync lifecycle — the pure state model behind every decision the +// executor makes about re-listing a connection's tools. +// +// No storage, no clock, no randomness: the connection's columns arrive as a +// value, `now` and the backoff jitter arrive as arguments, and every function +// here is total. That is what makes the ladder, the cap and the eligibility +// rules testable directly rather than through a database and a live upstream, +// and it is what will let PR 3's background scheduler reuse the same rules +// without reimplementing them against a different clock. +// --------------------------------------------------------------------------- + +/** Why a catalog listing could not be completed, as a CLOSED set persisted on + * `connection.tools_sync_error_kind`. + * + * Structural on purpose: the previous mechanism was a `"Tool sync failing: "` + * prefix on the human-readable health detail, which meant every consumer that + * wanted to know "is this connection's catalog broken, and how" had to match a + * display string. These four values are the whole vocabulary: + * + * - `auth` the credential is rejected or absent. Retrying the same grant + * cannot change the verdict, so it PARKS the connection (see + * {@link isToolSyncParked}) until a human re-authorizes. + * - `unreachable` the upstream did not answer (DNS, connect, timeout). + * - `protocol` the upstream answered but the listing could not be read. + * - `config` the integration's own configuration cannot produce a listing + * (unparseable spec, disabled transport, missing endpoint). */ +export type ToolSyncErrorKind = "auth" | "unreachable" | "protocol" | "config"; + +export const TOOL_SYNC_ERROR_KINDS = [ + "auth", + "unreachable", + "protocol", + "config", +] as const satisfies readonly ToolSyncErrorKind[]; + +/** Narrow a persisted `tools_sync_error_kind` column back to the closed set. + * The column is plain text, so a value written by a newer build (or edited by + * hand) has to be parsed rather than trusted. */ +export const isToolSyncErrorKind = (value: unknown): value is ToolSyncErrorKind => + typeof value === "string" && (TOOL_SYNC_ERROR_KINDS as readonly string[]).includes(value); + +/** + * What a connection's persisted catalog is, relative to now. + * + * Every non-`fresh` value is exactly the reason a read has to re-list, so this + * doubles as the refresh trigger reported on the `executor.tools.sync` span. + * + * `cold` and `stale_marked` were one value before the `tools_stale_token` + * column existed: `connections.markToolsStale` cleared `tools_synced_at`, which + * made a connection that had never synced indistinguishable from one whose + * catalog was invalidated mid-invocation, and destroyed the last-verified + * timestamp in the process. + */ +export type ToolSyncState = "fresh" | "cold" | "stale_marked" | "config_revised" | "expired"; + +/** Why an otherwise-due connection is not being refreshed right now. Enumerable + * so the read's span can count each reason. */ +export type ToolSyncSkip = "fresh" | "parked" | "backoff" | "claimed"; + +/** + * A connection's sync lifecycle columns, plus the two facts that live on its + * integration and plugin (`config_revised_at`, whether the plugin lists a live + * remote catalog) and the executor's freshness window. + * + * Decoded from the row by the caller: bigint columns arrive as `bigint | null` + * from the driver, and the text `tools_sync_error_kind` has to pass + * {@link isToolSyncErrorKind} before it gets here. + */ +export interface ToolSyncCandidate { + /** Epoch ms of the last AUTHORITATIVE listing. Never stamped by a failed or + * incomplete one — a month-dead server must not read as "synced 30s ago". */ + readonly toolsSyncedAt: number | null; + /** The outstanding drift mark, or null when there is none. An opaque token + * (see `connection.tools_stale_token`) — its presence is the whole signal + * and its value is only ever compared for equality, by the listing that + * clears it. */ + readonly toolsStaleToken: string | null; + /** The nonce of the refresh attempt currently holding the write lease. */ + readonly toolsSyncClaimId: string | null; + /** Epoch ms the claim was taken, against which the lease expires. */ + readonly toolsSyncClaimAt: number | null; + /** CONSECUTIVE failed listings; zeroed by an authoritative one. */ + readonly toolsSyncFailures: number; + /** Epoch ms of the earliest next attempt. */ + readonly toolsSyncRetryAt: number | null; + readonly toolsSyncErrorKind: ToolSyncErrorKind | null; + /** The integration's `config_revised_at`: its last tool-affecting config + * change. */ + readonly configRevisedAt: number | null; + /** The plugin lists a live remote catalog (an MCP server, whose tool set can + * change with no executor-visible signal), so time alone can expire it. */ + readonly remoteToolCatalog: boolean; + /** The freshness window; `null` disables time-based expiry entirely. */ + readonly ttlMs: number | null; +} + +/** + * How long a refresh claim stays valid. + * + * A lease rather than a lock: the claimant can be a Workers isolate that is + * evicted mid-handshake, and nothing would ever release its claim. Long enough + * to cover a slow upstream listing (discovery is bounded well under this), + * short enough that a lost claimant costs one wasted freshness window. + */ +export const TOOL_SYNC_CLAIM_LEASE_MS = 60_000; + +/** The longest a failing connection is left alone. Past this the ladder stops + * doubling: a server that has been dead for a week still gets probed a few + * times a day, which is what makes recovery automatic. */ +export const TOOL_SYNC_BACKOFF_CEILING_MS = 6 * 60 * 60 * 1000; + +/** Bounds of the multiplicative jitter applied to every backoff delay, so a + * fleet that lost the same upstream at the same moment does not re-dial it in + * lockstep. The factor itself is supplied by the caller — this module owns no + * randomness. */ +export const TOOL_SYNC_JITTER_MIN = 0.8; +export const TOOL_SYNC_JITTER_MAX = 1.2; + +/** + * Classify a connection's catalog. Total, and ordered by authority: an explicit + * drift signal outranks a config revision, which outranks the clock. + * + * A drift mark is read as PRESENCE, never as an instant: `tools_stale_token` is + * an opaque token, and the listing that resolves it clears it by + * compare-and-set on the token it observed. Comparing a drift TIMESTAMP against + * `tools_synced_at` instead is what made the two stamps landing in one + * millisecond ambiguous, and left the column set forever on every connection + * that had ever drifted. + * + * The config-revision comparison stays inclusive of the stamp's own millisecond + * (`syncedAt <= configRevisedAt`). `Date.now()` has millisecond granularity and + * `tools_synced_at` records when the listing STARTED, so a revision landing in + * that millisecond describes a config the listing cannot have read. Swallowing + * it would leave the connection serving a catalog that predates the change with + * no TTL backstop — a plugin without a remote catalog never reaches the + * `expired` branch below. + */ +export const classifyToolSync = (candidate: ToolSyncCandidate, now: number): ToolSyncState => { + const syncedAt = candidate.toolsSyncedAt; + if (candidate.toolsStaleToken !== null) return "stale_marked"; + if (syncedAt === null) return "cold"; + if (candidate.configRevisedAt !== null && syncedAt <= candidate.configRevisedAt) { + return "config_revised"; + } + if (candidate.remoteToolCatalog && candidate.ttlMs !== null && syncedAt < now - candidate.ttlMs) { + return "expired"; + } + return "fresh"; +}; + +/** Whether another attempt currently holds the write lease. A claim with no + * `claim_at` is treated as dead rather than eternal, so a half-written row can + * never wedge a connection out of every future refresh. */ +export const isToolSyncClaimLive = (candidate: ToolSyncCandidate, now: number): boolean => + candidate.toolsSyncClaimId !== null && + candidate.toolsSyncClaimAt !== null && + now - candidate.toolsSyncClaimAt < TOOL_SYNC_CLAIM_LEASE_MS; + +/** + * Whether the failure ladder is still holding this connection off. + * + * Gated on `failures > 0` deliberately. `tools_sync_retry_at` is also written + * after a SUCCESS, where it records the TTL horizon so a scheduler can order + * work by one column; treating that as a gate would make a `list_changed` + * notification wait out the freshness window before it could heal the catalog. + * An event-driven drift re-lists immediately; only a failing one waits. + */ +export const isToolSyncBackedOff = (candidate: ToolSyncCandidate, now: number): boolean => + candidate.toolsSyncFailures > 0 && + candidate.toolsSyncRetryAt !== null && + now < candidate.toolsSyncRetryAt; + +/** Whether the connection is parked: its credential is rejected, and no amount + * of retrying the same grant will change that. Cleared by the human-initiated + * paths that can actually fix it (explicit refresh, connection update, an + * OAuth re-mint, a healthy probe, a fresh drift signal). {@link decideToolSync} + * applies it to `expired` alone — every other refresh trigger outranks it. */ +export const isToolSyncParked = (candidate: ToolSyncCandidate): boolean => + candidate.toolsSyncErrorKind === "auth"; + +/** + * Refresh this connection, or pass it over and say why. ONE decision rather + * than a predicate plus a re-classification, so a caller that needs both the + * verdict and the trigger cannot end up asking two questions with two answers. + * + * Every clause that can stop a refresh lives here, and that is the point: a + * read's inline refresh, an explicit signal and a background sweep all consult + * this, so "when do we dial an upstream" has one answer rather than one per + * caller. A `connection.status` lifecycle (disabled, revoked) belongs here too + * when it lands, as one more skip clause — putting it at a call site is what + * would reintroduce the divergence this exists to prevent. + * + * Ordered cheapest-and-most-final first, so the reason an operator sees on the + * span is the one they would name: fresh, then parked, then backoff, then + * claimed. A live claim observed on the row is only a pre-filter: the binding + * decision is the compare-and-set the caller runs against the database, and two + * readers that both see a free claim here still resolve to one refresh. + * + * The park gates `expired` and NOTHING else. `expired` is the only trigger that + * is purely the clock asking a connection which already has a working catalog + * to re-verify it, and re-dialing a rejected credential to re-verify a catalog + * we can still serve is exactly the waste the park exists to stop. + * + * `stale_marked` and `config_revised` are somebody telling us the world + * changed. A config revision in particular is how an `auth` verdict caused by + * integration configuration (the key in the wrong place, the wrong token + * endpoint) gets FIXED, and `integrations.update` is the one invalidation path + * that does not clear the ladder, so swallowing it would leave the repaired + * connection parked forever. + * + * `cold` is deliberately NOT parked, and that is a change from parking every + * clock-driven trigger. A parked connection that has never synced has no + * catalog to serve, so parking it is not "keep serving what we have", it is + * "serve nothing, forever, until a human notices". The retry ladder already + * caps a never-synced auth failure at roughly four dials a day, and letting the + * clock drive it back is what restores automatic recovery when the credential + * is repaired upstream — where nothing in this system observes the repair. + */ +export type ToolSyncDecision = + | { readonly kind: "skip"; readonly reason: ToolSyncSkip } + | { readonly kind: "refresh"; readonly trigger: Exclude }; + +export const decideToolSync = (candidate: ToolSyncCandidate, now: number): ToolSyncDecision => { + const state = classifyToolSync(candidate, now); + if (state === "fresh") return { kind: "skip", reason: "fresh" }; + if (state === "expired" && isToolSyncParked(candidate)) return { kind: "skip", reason: "parked" }; + if (isToolSyncBackedOff(candidate, now)) return { kind: "skip", reason: "backoff" }; + if (isToolSyncClaimLive(candidate, now)) return { kind: "skip", reason: "claimed" }; + return { kind: "refresh", trigger: state }; +}; + +/** THE eligibility predicate, for callers that only need the yes/no. */ +export const isSyncEligible = (candidate: ToolSyncCandidate, now: number): boolean => + decideToolSync(candidate, now).kind === "refresh"; + +/** The earliest next attempt after an AUTHORITATIVE listing: one freshness + * window out, or `null` when time-based re-sync is disabled. Not a gate (see + * {@link isToolSyncBackedOff}) — it is the horizon a scheduler sorts by. */ +export const scheduleAfterSuccess = (now: number, ttlMs: number | null): number | null => + ttlMs === null ? null : now + ttlMs; + +/** + * The earliest next attempt after a FAILED listing: `ttlMs × 2^(failures−1)`, + * capped at {@link TOOL_SYNC_BACKOFF_CEILING_MS}, scaled by `jitter`. + * + * `failures` is the count INCLUDING the failure being recorded, so it is `>= 1` + * and the first failure waits one plain TTL. Neither argument is clamped: a + * call site that passes the PRIOR count is indistinguishable from a correct one + * at the first rung and then runs the whole ladder one step short, permanently + * and invisibly, which is a bug worth seeing rather than laundering into a + * plausible delay. `jitter` belongs to the caller on the same terms and is + * expected in [{@link TOOL_SYNC_JITTER_MIN}, {@link TOOL_SYNC_JITTER_MAX}]. + * + * Rounded: the result lands in a bigint column. + */ +export const scheduleAfterFailure = ( + now: number, + ttlMs: number, + failures: number, + jitter: number, +): number => + Math.round(now + Math.min(TOOL_SYNC_BACKOFF_CEILING_MS, ttlMs * 2 ** (failures - 1)) * jitter); diff --git a/packages/core/sdk/src/tools-sync-scope.test.ts b/packages/core/sdk/src/tools-sync-scope.test.ts index 42715520f5..cd4b0e7a8e 100644 --- a/packages/core/sdk/src/tools-sync-scope.test.ts +++ b/packages/core/sdk/src/tools-sync-scope.test.ts @@ -1,12 +1,13 @@ import { describe, expect, it } from "@effect/vitest"; -import { Effect, Latch } from "effect"; +import { Duration, Effect, Fiber, Latch } from "effect"; import { CONNECTION_CATALOG_SCAN_COLUMNS } from "./core-schema"; -import { createExecutor } from "./executor"; +import { DEFAULT_TOOLS_SYNC_TTL_MS, createExecutor } from "./executor"; import type { FumaDb } from "./fuma-runtime"; import { AuthTemplateSlug, ConnectionName, IntegrationSlug, ToolName } from "./ids"; import { definePlugin } from "./plugin"; import { makeTestConfig, memoryCredentialsPlugin } from "./testing"; +import type { ToolSyncErrorKind } from "./tool-sync-schedule"; // A tools READ refreshes stale catalogs before it answers, and each refresh is // a live upstream handshake. These cases pin the three properties that keep @@ -18,6 +19,22 @@ const INTEG_A = IntegrationSlug.make("alpha"); const INTEG_B = IntegrationSlug.make("beta"); const TEMPLATE = AuthTemplateSlug.make("apiKey"); +/** How long any latch in this file is allowed to hold a fiber. Far longer than + * a passing run needs (the whole suite is milliseconds), and only ever reached + * when the overlap a case is establishing never happened — a sequential + * refresh fan-out, or a claim compare-and-set that stopped excluding. Reaching + * it lets the fiber through so the case fails on its own assertion instead of + * hanging the suite into an opaque timeout with no diff. */ +const LATCH_TIMEOUT_MS = 5000; + +const withLatchTimeout = (effect: Effect.Effect) => + effect.pipe( + Effect.timeoutOrElse({ + duration: Duration.millis(LATCH_TIMEOUT_MS), + orElse: () => Effect.succeed(undefined), + }), + ); + /** A plugin owning two integrations that records every `resolveTools` it is * asked for, as `/`. The recorded list IS the cost * a read pays upstream, so most assertions below are about its contents. @@ -34,21 +51,37 @@ const makeCountingPlugin = () => { let holdUntil: number | null = null; let inFlight = 0; let peakInFlight = 0; + let incompleteKind: ToolSyncErrorKind | null | undefined; const gate = Latch.makeUnsafe(true); + const entered = Latch.makeUnsafe(false); const plugin = definePlugin(() => ({ id: "counting" as const, storage: () => ({}), + // Stands in for an MCP server: a catalog only the upstream knows, so the + // freshness TTL can expire it. Without this the `expired` state is + // unreachable from these tests, and `expired` is the one state the park + // gates. + remoteToolCatalog: true, resolveTools: (input) => Effect.gen(function* () { const slug = String(input.connection.integration); resolved.push(`${slug}/${String(input.connection.name)}`); inFlight += 1; peakInFlight = Math.max(peakInFlight, inFlight); + entered.openUnsafe(); if (holdUntil !== null && inFlight >= holdUntil) gate.openUnsafe(); - yield* gate.await; + yield* withLatchTimeout(gate.await); inFlight -= 1; if (dying) return yield* Effect.die("resolveTools blew up"); + if (incompleteKind !== undefined) { + return { + tools: [], + incomplete: true, + incompleteReason: "the fixture upstream refused", + ...(incompleteKind === null ? {} : { incompleteKind }), + }; + } return { tools: [{ name: ToolName.make(`${slug}_${toolSuffix}`), description: toolSuffix }], }; @@ -65,6 +98,14 @@ const makeCountingPlugin = () => { * the `config_revised_at` stamp — is on the plugin-facing surface. */ revise: (slug: IntegrationSlug) => ctx.core.integrations.update(slug, { config: { revision: 2 } }), + /** The `stale_marked` trigger as a plugin actually fires it (an MCP + * `tools/list_changed`, an unknown-tool rejection). */ + markStale: (slug: IntegrationSlug, name: string) => + ctx.connections.markToolsStale({ + owner: "org", + integration: slug, + name: ConnectionName.make(name), + }), }), }))(); @@ -82,6 +123,22 @@ const makeCountingPlugin = () => { gate.closeUnsafe(); }, peakConcurrentResolves: () => peakInFlight, + /** Make every resolve report a non-authoritative listing. `null` leaves it + * unclassified (a plugin that knows it failed but not why); a kind is what + * core parks or backs off on. `undefined` restores real listings. */ + resolveIncompleteAs: (kind: ToolSyncErrorKind | null | undefined) => { + incompleteKind = kind; + }, + /** Block every resolve until {@link releaseResolves}, so a second read can + * be driven all the way through while the first is mid-handshake. */ + blockResolves: () => { + entered.closeUnsafe(); + gate.closeUnsafe(); + }, + releaseResolves: () => gate.openUnsafe(), + /** Resolves when a resolve has actually STARTED — the claim is taken by + * then, which is the state the contention cases need to observe. */ + awaitResolveStarted: withLatchTimeout(entered.await), }; }; @@ -111,13 +168,16 @@ const observeConnectionReads = (db: FumaDb) => { const value: unknown = Reflect.get(target, prop, receiver); if (typeof value !== "function") return value; + // Cast to the interface's OWN member types, never to a hand-written + // mirror: the recorder's contract is with `AbstractQuery`, and a + // re-declared signature is one that silently stops matching it. if (prop === "withContext") { - const withContext = value as (context: unknown) => FumaDb; + const withContext = value as NonNullable; return (context: unknown) => wrap(Reflect.apply(withContext, inner, [context])); } if (prop === "transaction") { - const transaction = value as (run: (orm: FumaDb) => Promise) => Promise; + const transaction = value as FumaDb["transaction"]; return (run: (orm: FumaDb) => Promise) => { openTransactions += 1; peakOpenTransactions = Math.max(peakOpenTransactions, openTransactions); @@ -130,10 +190,7 @@ const observeConnectionReads = (db: FumaDb) => { } if (prop !== "findMany") return value; - const findMany = value as ( - table: string, - options?: { readonly select?: unknown }, - ) => Promise; + const findMany = value as FumaDb["findMany"]; return async (table: string, options?: { readonly select?: unknown }) => { const rows = await Reflect.apply(findMany, inner, [table, options]); if (table === "connection") { @@ -151,9 +208,30 @@ const observeConnectionReads = (db: FumaDb) => { }; }; -/** The projection the catalog-refresh scan is expected to ask for, as a plain - * array so a structural compare against the recorded `select` reads cleanly. */ -const SCAN_PROJECTION = [...CONNECTION_CATALOG_SCAN_COLUMNS]; +/** The projection the catalog-refresh scan is expected to ask for, written out + * rather than re-exported from the production constant: comparing the + * implementation to itself would catch only the `select` option being deleted, + * never the failure the constant's own docstring names — someone adding + * `value`, `oauth_state` or `last_health` to it so the every-read scan starts + * pulling credential and health JSON. */ +const SCAN_PROJECTION = [ + "owner", + "integration", + "name", + "tools_synced_at", + "tools_stale_token", + "tools_sync_claim_id", + "tools_sync_claim_at", + "tools_sync_failures", + "tools_sync_retry_at", + "tools_sync_error_kind", +]; + +describe("connection catalog scan projection", () => { + it("is the address plus the sync lifecycle, and nothing else", () => { + expect([...CONNECTION_CATALOG_SCAN_COLUMNS]).toEqual(SCAN_PROJECTION); + }); +}); const makeHarness = () => Effect.acquireRelease( @@ -179,15 +257,30 @@ const makeHarness = () => template: TEMPLATE, value: "secret-token", }), - /** Clear every connection's catalog stamp — the `stale_marked` trigger, - * as `connections.markToolsStale` leaves it. */ - markEveryCatalogStale: () => + /** Clear every connection's catalog stamp, which is the `cold` trigger: + * a row with no stamp and no drift mark has never synced. NOT the + * drift trigger — `connections.markToolsStale` leaves `tools_synced_at` + * alone and writes `tools_stale_token`, which is what + * `executor.counting.markStale` drives. Used by the cases that only + * need a connection to be due, whatever made it due. */ + clearEveryCatalogStamp: () => Effect.promise(() => observed.db.updateMany("connection", { where: (b) => b.isNotNull("tools_synced_at"), set: { tools_synced_at: null }, }), ), + /** Push every catalog past the freshness window, which is the `expired` + * trigger: a connection that HAS a working catalog and is only being + * asked to re-verify it. Distinct from `cold` in exactly the way the + * park cares about, so the two cannot share a helper. */ + expireEveryCatalog: () => + Effect.promise(() => + observed.db.updateMany("connection", { + where: (b) => b.isNotNull("tools_synced_at"), + set: { tools_synced_at: Date.now() - DEFAULT_TOOLS_SYNC_TTL_MS - 1 }, + }), + ), /** Push every catalog stamp into the past. The `config_revised` trigger * compares `tools_synced_at` against a stamp taken later in the same * test; on an in-memory database both can land in one millisecond and @@ -200,6 +293,53 @@ const makeHarness = () => set: { tools_synced_at: Date.now() - 60_000 }, }), ), + /** The sync lifecycle columns, as persisted. Read straight from the + * database rather than through a projection, because "what did the + * failed refresh actually write" is the whole question. */ + syncStateOf: (name: string) => + Effect.promise(() => + observed.db.findFirst("connection", { where: (b) => b("name", "=", name) }), + ).pipe( + Effect.flatMap((row) => + // A missing row is a defect, not a state. Projecting it into a + // populated "clean" record is what would let `syncedAt` and + // `claimId` assert null against an implementation that DELETED + // the connection rather than one that correctly declined to + // stamp it. + row == null + ? Effect.die(`no connection row named ${name}`) + : Effect.succeed({ + syncedAt: row.tools_synced_at == null ? null : Number(row.tools_synced_at), + staleToken: + row.tools_stale_token == null ? null : String(row.tools_stale_token), + failures: row.tools_sync_failures == null ? 0 : Number(row.tools_sync_failures), + retryAt: + row.tools_sync_retry_at == null ? null : Number(row.tools_sync_retry_at), + errorKind: row.tools_sync_error_kind ?? null, + claimId: row.tools_sync_claim_id ?? null, + }), + ), + ), + /** Re-claim a connection out from under whoever holds it, as a second + * Workers isolate does when it observes an expired lease. There is no + * in-process seam for this — the row IS the coordination medium. */ + stealClaim: (name: string) => + Effect.promise(() => + observed.db.updateMany("connection", { + where: (b) => b("name", "=", name), + set: { tools_sync_claim_id: "another-isolate", tools_sync_claim_at: Date.now() }, + }), + ), + /** Move every backoff window into the past. The ladder's first rung is + * the freshness TTL (15 minutes by default), so the alternative is + * sleeping through it. */ + elapseEveryBackoff: () => + Effect.promise(() => + observed.db.updateMany("connection", { + where: (b) => b.isNotNull("tools_sync_retry_at"), + set: { tools_sync_retry_at: Date.now() - 1 }, + }), + ), }; }), ({ executor, testDb }) => @@ -218,7 +358,7 @@ describe("tools read catalog refresh scope", () => { const harness = yield* makeHarness(); yield* harness.connect(INTEG_A, "main"); yield* harness.connect(INTEG_B, "main"); - yield* harness.markEveryCatalogStale(); + yield* harness.clearEveryCatalogStamp(); harness.resolved.length = 0; const alpha = yield* harness.executor.tools.list({ integration: INTEG_A }); @@ -264,7 +404,7 @@ describe("tools read catalog refresh scope", () => { yield* harness.connect(INTEG_A, "main"); yield* harness.connect(INTEG_A, "second"); yield* harness.connect(INTEG_B, "main"); - yield* harness.markEveryCatalogStale(); + yield* harness.clearEveryCatalogStamp(); harness.resolved.length = 0; harness.connectionScans.length = 0; @@ -275,8 +415,9 @@ describe("tools read catalog refresh scope", () => { }); expect(harness.resolved).toEqual(["alpha/second"]); - // One row, and only the four columns the refresh actually reads: the - // scan runs on every read, so the projection is part of its cost. + // One row, and only the columns the refresh actually reads — the + // connection's address plus its sync lifecycle. The scan runs on every + // read, so the projection is part of its cost. expect(harness.connectionScans).toEqual([{ rows: 1, select: SCAN_PROJECTION }]); expect(tools.map((tool) => String(tool.name))).toEqual(["alpha_deploy"]); }), @@ -314,7 +455,7 @@ describe("tools read catalog refresh scope", () => { Effect.gen(function* () { const harness = yield* makeHarness(); yield* harness.connect(INTEG_A, "main"); - yield* harness.markEveryCatalogStale(); + yield* harness.clearEveryCatalogStamp(); harness.startDying(); harness.resolved.length = 0; @@ -345,7 +486,7 @@ describe("tools read catalog refresh scope", () => { // Make the rebuild observably different from what `connect` already // persisted, so a refresh whose write never landed is visible. harness.renameToolsTo("redeploy"); - yield* harness.markEveryCatalogStale(); + yield* harness.clearEveryCatalogStamp(); harness.resolved.length = 0; // Arm the overlap latch only now: `connect` resolves one connection at // a time, and a latch waiting for a second in-flight resolve would @@ -373,3 +514,372 @@ describe("tools read catalog refresh scope", () => { ), ); }); + +// --------------------------------------------------------------------------- +// Sync lifecycle. Concurrent reads have no shared memory (each is potentially +// its own Workers isolate), and a broken upstream has no way of telling us to +// stop calling it. Both are answered by state on the connection row: a refresh +// claim, a failure ladder, and a stamp that is only ever written by a listing +// that actually succeeded. +// --------------------------------------------------------------------------- + +describe("tools read catalog refresh lifecycle", () => { + it.effect("lets exactly one of two concurrent reads dial a drifted connection", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness(); + yield* harness.connect(INTEG_A, "main"); + // Genuinely drifted, through the path a plugin actually uses: the + // connection keeps its stamp and carries a `tools_stale_token`. + yield* harness.executor.counting.markStale(INTEG_A, "main"); + harness.renameToolsTo("redeploy"); + harness.resolved.length = 0; + harness.blockResolves(); + + // First reader: forked, and parked inside `resolveTools` — which means + // it has already taken the claim. + const first = yield* Effect.forkChild( + harness.executor.tools.list({ integration: INTEG_A }), + ); + yield* harness.awaitResolveStarted; + + // Second reader runs to completion against the same drifted connection + // while the first still holds the claim. + const second = yield* harness.executor.tools.list({ integration: INTEG_A }); + + // One handshake between them, not one each. + expect(harness.resolved).toEqual(["alpha/main"]); + // And the loser still ANSWERS — from the stale-but-working catalog. + expect(second.map((tool) => String(tool.name))).toEqual(["alpha_deploy"]); + + harness.releaseResolves(); + const firstTools = yield* Fiber.join(first); + expect(firstTools.map((tool) => String(tool.name))).toEqual(["alpha_redeploy"]); + expect(harness.resolved).toEqual(["alpha/main"]); + + // The winner released its lease on the way out. + expect((yield* harness.syncStateOf("main")).claimId).toBeNull(); + }), + ), + ); + + it.effect("discards a listing whose lease was taken while it was in flight", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness(); + yield* harness.connect(INTEG_A, "main"); + yield* harness.clearEveryCatalogStamp(); + harness.renameToolsTo("redeploy"); + harness.resolved.length = 0; + harness.blockResolves(); + + const slow = yield* Effect.forkChild(harness.executor.tools.list({ integration: INTEG_A })); + yield* harness.awaitResolveStarted; + + // Another isolate saw the lease expire and re-claimed the connection. + // The in-flight listing is now the OLDER one, whatever order it lands + // in, and must not overwrite what the new claimant will write. + yield* harness.stealClaim("main"); + harness.releaseResolves(); + + const tools = yield* Fiber.join(slow); + expect(tools.map((tool) => String(tool.name))).toEqual(["alpha_deploy"]); + + const state = yield* harness.syncStateOf("main"); + // Nothing of the discarded attempt survives: not the catalog, and not + // the freshness stamp that would have hidden the drift. + expect(state.syncedAt).toBeNull(); + expect(state.claimId).toBe("another-isolate"); + }), + ), + ); + + it.effect("holds a failing connection off until its backoff window elapses", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness(); + yield* harness.connect(INTEG_A, "main"); + yield* harness.clearEveryCatalogStamp(); + harness.resolveIncompleteAs(null); + harness.resolved.length = 0; + + yield* harness.executor.tools.list({ integration: INTEG_A }); + expect(harness.resolved).toEqual(["alpha/main"]); + + const afterFailure = yield* harness.syncStateOf("main"); + expect(afterFailure.failures).toBe(1); + expect(afterFailure.retryAt).not.toBeNull(); + + // Inside the window: no upstream call at all, however many reads land. + yield* harness.executor.tools.list({ integration: INTEG_A }); + yield* harness.executor.tools.list({ integration: INTEG_A }); + expect(harness.resolved).toEqual(["alpha/main"]); + + // Past it: one more attempt, and the ladder advances. + yield* harness.elapseEveryBackoff(); + yield* harness.executor.tools.list({ integration: INTEG_A }); + expect(harness.resolved).toEqual(["alpha/main", "alpha/main"]); + expect((yield* harness.syncStateOf("main")).failures).toBe(2); + }), + ), + ); + + it.effect("parks a connection whose credential was refused, and un-parks on refresh", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness(); + yield* harness.connect(INTEG_A, "main"); + // EXPIRED, not cold: the connection has a working catalog and the clock + // is only asking it to re-verify. That is the one thing the park gates, + // and the case below pins the other side of the rule. + yield* harness.expireEveryCatalog(); + harness.resolveIncompleteAs("auth"); + harness.resolved.length = 0; + + yield* harness.executor.tools.list({ integration: INTEG_A }); + expect(harness.resolved).toEqual(["alpha/main"]); + expect((yield* harness.syncStateOf("main")).errorKind).toBe("auth"); + + // Parked: not even elapsing the backoff window lets a read dial again. + // Retrying a rejected credential to re-verify a catalog we can still + // serve cannot change the verdict, and 401 handshakes were the largest + // single source of wasted upstream calls. + yield* harness.elapseEveryBackoff(); + yield* harness.executor.tools.list({ integration: INTEG_A }); + expect(harness.resolved).toEqual(["alpha/main"]); + + // An explicit refresh is a human saying "I fixed it": it bypasses the + // park, dials, and clears the ledger on success. + harness.resolveIncompleteAs(undefined); + yield* harness.executor.connections.refresh({ + owner: "org", + integration: INTEG_A, + name: ConnectionName.make("main"), + }); + expect(harness.resolved).toEqual(["alpha/main", "alpha/main"]); + + const recovered = yield* harness.syncStateOf("main"); + expect(recovered.errorKind).toBeNull(); + expect(recovered.failures).toBe(0); + expect(recovered.syncedAt).not.toBeNull(); + }), + ), + ); + + it.effect("walks the ladder for a never-synced connection whose credential was refused", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness(); + yield* harness.connect(INTEG_A, "main"); + // COLD: no catalog at all. Parking this serves nothing, forever, until + // a human notices — and nothing in this system observes a credential + // repaired upstream, so the clock has to be what brings it back. + yield* harness.clearEveryCatalogStamp(); + harness.resolveIncompleteAs("auth"); + harness.resolved.length = 0; + + yield* harness.executor.tools.list({ integration: INTEG_A }); + expect(harness.resolved).toEqual(["alpha/main"]); + expect((yield* harness.syncStateOf("main")).errorKind).toBe("auth"); + + // The LADDER holds it off, not the park — so reads inside the window + // still cost nothing. + yield* harness.executor.tools.list({ integration: INTEG_A }); + expect(harness.resolved).toEqual(["alpha/main"]); + + // Past the rung, a credential fixed upstream is picked up with no human + // in the loop. Under a park that gated `cold` this read never dials and + // the connection advertises no tools indefinitely. + harness.resolveIncompleteAs(undefined); + yield* harness.elapseEveryBackoff(); + + const tools = yield* harness.executor.tools.list({ integration: INTEG_A }); + expect(harness.resolved).toEqual(["alpha/main", "alpha/main"]); + expect(tools.map((tool) => String(tool.name))).toEqual(["alpha_deploy"]); + + const recovered = yield* harness.syncStateOf("main"); + expect(recovered.errorKind).toBeNull(); + expect(recovered.failures).toBe(0); + expect(recovered.syncedAt).not.toBeNull(); + }), + ), + ); + + it.effect("keeps retrying a parked connection whose integration config was revised", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness(); + yield* harness.connect(INTEG_A, "main"); + yield* harness.backdateEveryCatalog(); + // Editing the integration's config is how an `auth` verdict CAUSED by + // configuration is fixed (the key in the wrong place, the wrong token + // endpoint), and `integrations.update` is the one invalidation that + // does not clear the ladder. So this is the case where a park could + // swallow the repair outright. + yield* harness.executor.counting.revise(INTEG_A); + harness.resolveIncompleteAs("auth"); + harness.resolved.length = 0; + + yield* harness.executor.tools.list({ integration: INTEG_A }); + expect(harness.resolved).toEqual(["alpha/main"]); + expect((yield* harness.syncStateOf("main")).errorKind).toBe("auth"); + + // The park does NOT apply while the revision is outstanding — but the + // ladder still does, so a broken config costs one dial per rung rather + // than one per read. + yield* harness.executor.tools.list({ integration: INTEG_A }); + expect(harness.resolved).toEqual(["alpha/main"]); + + // Past the rung, the fixed config is actually tried. Under a park that + // ignored the revision this read would never dial and the connection + // would serve its old catalog forever. + harness.resolveIncompleteAs(undefined); + harness.renameToolsTo("redeploy"); + yield* harness.elapseEveryBackoff(); + + const tools = yield* harness.executor.tools.list({ integration: INTEG_A }); + expect(harness.resolved).toEqual(["alpha/main", "alpha/main"]); + expect(tools.map((tool) => String(tool.name))).toEqual(["alpha_redeploy"]); + + const recovered = yield* harness.syncStateOf("main"); + expect(recovered.errorKind).toBeNull(); + expect(recovered.failures).toBe(0); + }), + ), + ); + + it.effect("marking a catalog stale records the drift without erasing the stamp", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness(); + yield* harness.connect(INTEG_A, "main"); + + const synced = yield* harness.syncStateOf("main"); + expect(synced.syncedAt).not.toBeNull(); + expect(synced.staleToken).toBeNull(); + + yield* harness.executor.counting.markStale(INTEG_A, "main"); + + const drifted = yield* harness.syncStateOf("main"); + // Both facts survive: when it last verified, and that it has since + // drifted. Clearing the stamp used to conflate this with "never synced". + expect(drifted.syncedAt).toBe(synced.syncedAt); + expect(drifted.staleToken).not.toBeNull(); + + harness.renameToolsTo("redeploy"); + harness.resolved.length = 0; + const tools = yield* harness.executor.tools.list({ integration: INTEG_A }); + + expect(harness.resolved).toEqual(["alpha/main"]); + expect(tools.map((tool) => String(tool.name))).toEqual(["alpha_redeploy"]); + + // The re-list settled the drift by CLEARING the token it observed. It + // is a compare-and-set, not an unconditional null (the case below is + // the mark it must not clear), and the clear is what makes the + // resolution observable rather than inferred from two timestamps that + // can land in the same millisecond. + const settled = yield* harness.syncStateOf("main"); + expect(settled.staleToken).toBeNull(); + expect(settled.syncedAt).not.toBeNull(); + + // And the connection is genuinely done: the next read does not dial + // it — and, the property the token encoding exists for, does not even + // SCAN it. A drift mark nothing ever clears leaves the scan's + // `tools_stale_token IS NOT NULL` arm matching this row on every read + // for the rest of its life, which is the steady state PR 1 established + // and its `candidates` gauge measures. + harness.resolved.length = 0; + harness.connectionScans.length = 0; + yield* harness.executor.tools.list({ integration: INTEG_A }); + expect(harness.resolved).toEqual([]); + expect(harness.connectionScans).toEqual([{ rows: 0, select: SCAN_PROJECTION }]); + }), + ), + ); + + it.effect("keeps a drift signal that landed while the listing was in flight", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness(); + yield* harness.connect(INTEG_A, "main"); + yield* harness.executor.counting.markStale(INTEG_A, "main"); + const observed = yield* harness.syncStateOf("main"); + harness.renameToolsTo("redeploy"); + harness.resolved.length = 0; + harness.blockResolves(); + + // A read is dialing, and parked inside `resolveTools`. + const listing = yield* Effect.forkChild( + harness.executor.tools.list({ integration: INTEG_A }), + ); + yield* harness.awaitResolveStarted; + + // The server sends `tools/list_changed` WHILE we are listing: a + // different fiber, no coordination with the one in flight, and it + // describes a tool set this listing cannot have seen. + yield* harness.executor.counting.markStale(INTEG_A, "main"); + const marked = yield* harness.syncStateOf("main"); + // A FRESH token, which is the entire mechanism: the listing will clear + // only the token it observed at its start, and this is not that token. + expect(marked.staleToken).not.toBe(observed.staleToken); + + harness.releaseResolves(); + yield* Fiber.join(listing); + + // The mark survived the success write, verbatim. Nulling + // `tools_stale_token` unconditionally here is a lost update, and for a + // plugin with no remote catalog nothing else would ever re-list the + // connection. + const state = yield* harness.syncStateOf("main"); + expect(state.staleToken).toBe(marked.staleToken); + + // Which is the fact that matters: the next read dials again. + harness.renameToolsTo("rerelease"); + harness.resolved.length = 0; + const tools = yield* harness.executor.tools.list({ integration: INTEG_A }); + expect(harness.resolved).toEqual(["alpha/main"]); + expect(tools.map((tool) => String(tool.name))).toEqual(["alpha_rerelease"]); + + // And it CONVERGES. That read observed the surviving token and cleared + // it, so the drift is settled in one extra listing rather than leaving + // the connection permanently marked — a clear that misses costs one + // re-list, never a standing scan. + expect((yield* harness.syncStateOf("main")).staleToken).toBeNull(); + }), + ), + ); + + it.effect("never advances the sync stamp for a listing that did not succeed", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness(); + yield* harness.connect(INTEG_A, "main"); + const synced = yield* harness.syncStateOf("main"); + + harness.resolveIncompleteAs("unreachable"); + // One drift signal, then four attempts at it. A failed refresh leaves + // the drift mark standing (nothing resolved it), so the connection + // stays due and only the ladder holds it off — which is exactly what + // `elapseEveryBackoff` steps past. + yield* harness.executor.counting.markStale(INTEG_A, "main"); + harness.resolved.length = 0; + for (let attempt = 0; attempt < 4; attempt += 1) { + yield* harness.elapseEveryBackoff(); + yield* harness.executor.tools.list({ integration: INTEG_A }); + } + expect(harness.resolved).toHaveLength(4); + + const state = yield* harness.syncStateOf("main"); + // The anti-lying assertion. A month of failed refreshes must leave + // `tools_synced_at` where the last SUCCESSFUL listing put it, or every + // freshness decision downstream is made on a fabricated timestamp. + expect(state.syncedAt).toBe(synced.syncedAt); + expect(state.failures).toBe(4); + expect(state.errorKind).toBe("unreachable"); + // The drift mark also stands: only an authoritative listing clears it, + // and none of these four were one. + expect(state.staleToken).not.toBeNull(); + }), + ), + ); +}); diff --git a/packages/plugins/mcp/src/sdk/discover.ts b/packages/plugins/mcp/src/sdk/discover.ts index 2734e73f54..83cf8ba986 100644 --- a/packages/plugins/mcp/src/sdk/discover.ts +++ b/packages/plugins/mcp/src/sdk/discover.ts @@ -105,10 +105,18 @@ export const discoverTools = ( const httpStatus = Predicate.isTagged(failure, "McpConnectionError") ? failure.httpStatus : undefined; + // Same reason: a reauthorization demand is an auth verdict with no + // status code, and flattening it into a bare connect failure would make + // it indistinguishable from an unreachable server. + const reauthorizationRequired = Predicate.isTagged( + failure, + "McpOAuthReauthorizationRequired", + ); return new McpToolDiscoveryError({ stage: "connect", message: `Failed connecting to MCP server: ${failure.message}`, ...(httpStatus !== undefined ? { httpStatus } : {}), + ...(reauthorizationRequired ? { reauthorizationRequired } : {}), }); }), ); diff --git a/packages/plugins/mcp/src/sdk/errors.ts b/packages/plugins/mcp/src/sdk/errors.ts index e3bba2faf1..02b9c81962 100644 --- a/packages/plugins/mcp/src/sdk/errors.ts +++ b/packages/plugins/mcp/src/sdk/errors.ts @@ -27,6 +27,12 @@ export class McpToolDiscoveryError extends Schema.TaggedErrorClass { - if (failure.httpStatus !== undefined) { - // A failed connect can't be healthy; the shared classifier decides - // expired vs degraded so "which statuses mean expired" lives in one place. - const classified = classifyHttpStatus(failure.httpStatus); - return classified === "expired" ? "expired" : "degraded"; - } - const lower = failure.message.toLowerCase(); - const authWalled = - lower.includes("oauth re-authorization") || - lower.includes("unauthorized") || - lower.includes("forbidden"); - return authWalled ? "expired" : "degraded"; + if (failure.reauthorizationRequired === true) return "expired"; + if (failure.httpStatus === undefined) return "degraded"; + return classifyHttpStatus(failure.httpStatus) === "expired" ? "expired" : "degraded"; }; const legacyOAuthClientSlugCandidate = (value: string): string | null => { @@ -1213,18 +1212,38 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { // ----------------------------------------------------------------------- resolveTools: ({ config, connection, template, getValues, httpClientLayer }) => Effect.gen(function* () { + // `kind` is null for a failure the plugin cannot honestly place in the + // closed set: core backs it off like any other, but never parks it. + const incomplete = ( + kind: ToolSyncErrorKind | null, + reason: string, + ): ResolveToolsResult => ({ + tools: [], + incomplete: true, + incompleteReason: reason, + ...(kind === null ? {} : { incompleteKind: kind }), + }); + const parsed = parseMcpIntegrationConfig(config); - if (!parsed) return { tools: [] as readonly ToolDef[], incomplete: true }; + if (!parsed) { + return incomplete("config", "The MCP integration config could not be read."); + } - // Discovery tolerates unresolved credentials (an open server lists - // tools unauthenticated; a bad value just yields zero tools). - const values = yield* getValues().pipe( - Effect.orElseSucceed(() => ({}) as Record), - ); + // Discovery tolerates unresolved credentials (an open server lists tools + // unauthenticated; a bad value just yields zero tools) — but a secret + // STORE that cannot answer is a different fact. Dialing with a silently + // emptied value map produces a 401, which classifies as `auth`, which + // PARKS the connection until a human re-authorizes: a credential store + // blip would take working connections offline and blame the user's + // credential. Unclassified instead, so it retries. + const values = yield* getValues().pipe(Effect.result); + if (Result.isFailure(values)) { + return incomplete(null, "The connection's stored credential could not be read."); + } const built = yield* buildConnectorInput( parsed, - values, + values.success, template === null ? null : String(template), allowStdio, httpClientLayer, @@ -1233,28 +1252,44 @@ export const mcpPlugin = definePlugin((options?: McpPluginOptions) => { Effect.result, ); - const manifest = Result.isSuccess(built) - ? yield* discoverTools(built.success).pipe( - Effect.map((m) => ({ ok: true as const, manifest: m })), - Effect.catch(() => Effect.succeed({ ok: false as const, manifest: null })), - Effect.withSpan("mcp.plugin.discover_tools", { - attributes: { "mcp.connection.name": String(connection.name) }, - }), - ) - : { ok: false as const, manifest: null }; + // A connector that could not even be built is a configuration problem + // (stdio disabled, no usable endpoint), never an upstream one — nothing + // was dialed. + if (Result.isFailure(built)) { + return incomplete("config", "The MCP server connection could not be prepared."); + } + + // The discovery failure is CLASSIFIED rather than discarded: core parks + // an `auth` verdict instead of re-dialing a server that will keep + // refusing the same credential, and 401 handshakes were the single + // largest source of wasted upstream dials. The reason stays generic — + // it reaches a stored health record, and a discovery cause can embed + // the upstream request and response. + const discovered = yield* discoverTools(built.success).pipe( + Effect.result, + Effect.withSpan("mcp.plugin.discover_tools", { + attributes: { "mcp.connection.name": String(connection.name) }, + }), + ); - if (!manifest.ok || !manifest.manifest) { - return { tools: [] as readonly ToolDef[], incomplete: true }; + if (Result.isFailure(discovered)) { + const failure = discovered.failure; + // The same two structural signals `mcpLivenessFailureStatus` reads, so + // the health verdict and the sync verdict cannot drift apart. + if (mcpLivenessFailureStatus(failure) === "expired") { + return incomplete("auth", "The MCP server rejected the credential."); + } + return failure.stage === "connect" + ? incomplete("unreachable", "The MCP server could not be reached.") + : incomplete("protocol", "The MCP server's tool listing could not be read."); } - return { tools: manifest.manifest.tools.map(toToolDef) }; + + return { tools: discovered.success.tools.map(toToolDef) }; }).pipe( Effect.withSpan("mcp.plugin.resolve_tools", { attributes: { "mcp.connection.name": String(connection.name) }, }), - ) as Effect.Effect< - { readonly tools: readonly ToolDef[]; readonly incomplete?: boolean }, - StorageFailure - >, + ), invokeTool: ({ ctx, toolRow, credential, args, elicit }) => Effect.gen(function* () { diff --git a/packages/plugins/openapi/src/sdk/backing.ts b/packages/plugins/openapi/src/sdk/backing.ts index 4e6e5ba95a..ff4cef2e91 100644 --- a/packages/plugins/openapi/src/sdk/backing.ts +++ b/packages/plugins/openapi/src/sdk/backing.ts @@ -590,14 +590,25 @@ export const resolveOpenApiBackedTools = ({ readonly storage: OpenapiStore; }): Effect.Effect => Effect.gen(function* () { + // Always `config`: an OpenAPI catalog is derived entirely from the stored + // spec and its operation bindings, so nothing here can fail on a credential + // or an upstream. Every incomplete listing below means the integration's + // own persisted artifacts cannot produce a catalog, which is a fixable + // configuration state rather than something to keep re-dialing. const incomplete = (reason: string): ResolveToolsResult => ({ tools: [], definitions: {}, incomplete: true, incompleteReason: reason, + incompleteKind: "config", }); const openApiConfig = decodeOpenApiIntegrationConfig(config); - if (!openApiConfig) return { tools: [], definitions: {} }; + // Incomplete, never an authoritative empty catalog. Core reads an + // authoritative listing as the truth and DELETES every persisted tool row + // for the integration, and this plugin does not set `remoteToolCatalog`, so + // the nonzero-catalog guard that would otherwise catch it never runs — an + // unreadable config blob would wipe a working catalog. + if (!openApiConfig) return incomplete("The OpenAPI integration config could not be read."); if (openApiConfig.specHash != null) { const defsJson = yield* storage .getDefs(openApiConfig.specHash)