Encode the drift mark as a token, not a timestamp

Replace connection.tools_stale_at (bigint ms) with tools_stale_token (nullable text). The column answers "was this catalog invalidated since the listing now finishing began" — a version question, which a wall clock answers only as precisely as it ticks. markToolsStale writes a fresh token; an authoritative listing captures the token it observed at its start and clears it by compare-and-set inside replaceCatalog's existing transaction, so a mark that landed mid-listing carries a different token and survives. A missed clear fails closed and converges on the next re-list. This restores the read scan's zero-rows steady state: nothing ever cleared the timestamp column, so the prefilter's drift arm re-selected every once-drifted connection on every read forever. Contract change from a1fdc72d6: the same-millisecond stale/synced tie is gone rather than resolved either way, so the schedule tests that pinned inclusive comparison now pin token presence instead. Deliberate. Also: - produceConnectionTools returns outcome synced/incomplete/lost_claim instead of a boolean, with its own counter and span attribute. A plugin reporting bad news correctly is no longer counted as a sync, nor as a refresh that crashed. writeSyncOutcome reports whether it wrote, so an incomplete listing that had also lost its lease reports lost_claim. - The auth park gates expired only. A never-synced parked connection has no catalog to serve, so it walks the retry ladder instead, which caps the waste and restores recovery for credentials repaired upstream. - markToolsStale runs under the catalog persist lock; it was the one lifecycle write left exposed to enrollment in another fiber's open transaction. Migration 0016 and its snapshot are edited in place: unpushed, never run.

Rhys Sullivan committed 2 months ago

Commit e48fe131 parent a1fdc7211 files · +509 −219Browse files

Unified · Split

11 files

.changeset/tool-sync-lifecycle.md +4 −2
.claude/skills/prod-telemetry/SKILL.md +40 −17
apps/cloud/drizzle/0016_icy_jocasta.sql +1 −1
apps/cloud/drizzle/meta/0016_snapshot.json +3 −3
apps/cloud/src/db/executor-schema.ts +1 −1
packages/core/sdk/src/core-schema.ts +15 −10
packages/core/sdk/src/executor.ts +204 −88
packages/core/sdk/src/plugin.ts +6 −1
packages/core/sdk/src/tool-sync-schedule.test.ts +86 −35
packages/core/sdk/src/tool-sync-schedule.ts +47 −37
packages/core/sdk/src/tools-sync-scope.test.ts +102 −24

.changeset/tool-sync-lifecycle.md+4−2

@@ -8,6 +8,8 @@ · 7 unmodified lines
8		8
9	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.	9	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.
10		10
11	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. Failed listings walk a jittered retry ladder that doubles up to a six-hour ceiling, and a listing refused on authentication parks the connection until something that could plausibly fix it happens: an explicit refresh, a connection edit, an OAuth reconnect, a healthy check, an integration config revision, or a fresh drift signal. The park holds off the clock, never an explicit invalidation, so a credential problem caused by integration configuration is retried as soon as that configuration is edited. 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.	11	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.
12		12
13	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 gained the claim outcome and per-reason skip counts.	13	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.
		14
		15	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.

.claude/skills/prod-telemetry/SKILL.md+40−17

@@ -41,23 +41,42 @@ join the same traces via traceparent). · 40 unmodified lines
41	`.connection` (present only when the read was filtered),	41	`.connection` (present only when the read was filtered),
42	`executor.tools.result_count`, and the catalog-refresh counters	42	`executor.tools.result_count`, and the catalog-refresh counters
43	`executor.tools.sync.candidates` (rows the stale scan returned) /	43	`executor.tools.sync.candidates` (rows the stale scan returned) /
44	`executor.tools.sync.synced` / `executor.tools.sync.failed` /	44	`executor.tools.sync.synced` / `.incomplete` / `.failed` / `.lost_claim` /
45	`executor.tools.sync.skipped_claimed` / `.skipped_backoff` /	45	`.skipped_claimed` / `.skipped_backoff` / `.skipped_parked`. A slow tools
46	`.skipped_parked`. A slow tools read is almost always `candidates` > 0:	46	read is almost always `candidates` > 0: subtract the child span durations to
47	subtract the child span durations to confirm. A connection stuck permanently	47	confirm.
48	stale shows up as `failed` > 0 on every read for its scope — the refresh is	48
49	best-effort and never fails the read, so this counter and the	49	`candidates` is the one to alert on for scan cost: in steady state it is
50	`executor tool catalog refresh failed` warning are the only signals it emits.	50	ZERO. Every trigger is cleared by the listing that answers it, drift marks
		51	included (`connection.tools_stale_token` is compare-and-set to NULL), so a
		52	scope whose `candidates` never returns to 0 has connections that are due and
		53	never resolving, not a busy fleet.
		54
		55	The four terminal counters partition every attempt, and the distinction that
		56	matters when triaging is `incomplete` vs `failed`. `incomplete` is the PLUGIN
		57	reporting bad news correctly — a refused credential, an unreachable server,
		58	an unreadable spec — with the existing catalog kept and the retry ladder
		59	advanced. `failed` is the refresh raising or dying before reaching any
		60	verdict: the executor's own error channel, swallowed to keep the read
		61	answerable, and the thing to page on. A connection stuck permanently stale
		62	shows up as one or the other on every read for its scope; `failed` > 0 also
		63	emits the `executor tool catalog refresh failed` warning, which is its only
		64	other signal. `lost_claim` counts listings that completed and were then
		65	discarded because another attempt had re-claimed the connection — wasted
		66	upstream work that looks like success from every angle except the write.
51		67
52	The three `skipped_*` counters are how the sync lifecycle reports its work	68	The three `skipped_*` counters are how the sync lifecycle reports its work
53	avoidance, and they are the ones to watch after a rollout:	69	avoidance, and they are the ones to watch after a rollout:
54	`skipped_parked` counts connections whose credential was refused (an `auth`	70	`skipped_parked` counts connections whose credential was refused (an `auth`
55	verdict, which no retry can fix), `skipped_backoff` counts ones inside their	71	verdict) AND whose catalog is merely past its freshness window — the park
56	failure ladder, and `skipped_claimed` counts refreshes another concurrent	72	gates re-verification only, so a never-synced connection with the same
57	read had already taken. `candidates` staying high while `synced` collapses	73	verdict reports `skipped_backoff` instead and keeps walking the ladder.
58	toward the skip counters is the intended shape — it means dead upstreams	74	`skipped_backoff` counts ones inside their failure ladder, and
59	stopped costing handshakes. `skipped_claimed` rising with request concurrency	75	`skipped_claimed` counts refreshes another concurrent read had already taken.
60	for one integration is cross-isolate deduplication working.	76	`synced` collapsing toward the skip counters is the intended shape — it means
		77	dead upstreams stopped costing handshakes. `skipped_claimed` rising with
		78	request concurrency for one integration is cross-isolate deduplication
		79	working.
61		80
62	- `executor.tools.sync` (child of the above, one per refreshed connection;	81	- `executor.tools.sync` (child of the above, one per refreshed connection;
63	older spans carry the previous name `executor.tools.sync_stale` and no	82	older spans carry the previous name `executor.tools.sync_stale` and no
@@ -66,10 +85,14 @@ join the same traces via traceparent). · 2 unmodified lines
66	(`cold`/`stale_marked`/`config_revised`/`expired`),	85	(`cold`/`stale_marked`/`config_revised`/`expired`),
67	`executor.tools.sync.claimed` (bool — false means another reader owned the	86	`executor.tools.sync.claimed` (bool — false means another reader owned the
68	refresh and this one did nothing) and `executor.tools.sync.outcome`	87	refresh and this one did nothing) and `executor.tools.sync.outcome`
69	(`ok`/`fail`). `cold` split out of `stale_marked` when `tools_stale_at`	88	(`synced`/`incomplete`/`lost_claim`/`fail`). Older spans report `ok` where a
70	landed: before that, a never-synced connection and one invalidated	89	current one reports `synced` or `incomplete`; that rename is the same
71	mid-invocation were the same row, so spans predating it report every	90	distinction as the counters above, so any query matching `outcome == "ok"`
72	never-synced connection as `stale_marked`. The matching warning log carries	91	needs both spellings to span the change. `cold` split out of `stale_marked`
		92	when the drift mark became its own column: before that, a never-synced
		93	connection and one invalidated mid-invocation were the same row, so spans
		94	predating it report every never-synced connection as `stale_marked`. The
		95	matching warning log carries
73	`integration`, `connection`, `trigger` and `errorTags` only: a refresh runs	96	`integration`, `connection`, `trigger` and `errorTags` only: a refresh runs
74	on a live credential, so no cause or upstream message is logged and Axiom	97	on a live credential, so no cause or upstream message is logged and Axiom
75	will not have one to search.	98	will not have one to search.

apps/cloud/drizzle/0016_icy_jocasta.sql+1−1

@@ -1,4 +1,4 @@
1	ALTER TABLE "connection" ADD COLUMN "tools_stale_at" bigint;--> statement-breakpoint	1	ALTER TABLE "connection" ADD COLUMN "tools_stale_token" text;--> statement-breakpoint
2	ALTER TABLE "connection" ADD COLUMN "tools_sync_claim_id" text;--> statement-breakpoint	2	ALTER TABLE "connection" ADD COLUMN "tools_sync_claim_id" text;--> statement-breakpoint
3	ALTER TABLE "connection" ADD COLUMN "tools_sync_claim_at" bigint;--> statement-breakpoint	3	ALTER TABLE "connection" ADD COLUMN "tools_sync_claim_at" bigint;--> statement-breakpoint
4	ALTER TABLE "connection" ADD COLUMN "tools_sync_failures" bigint;--> statement-breakpoint	4	ALTER TABLE "connection" ADD COLUMN "tools_sync_failures" bigint;--> statement-breakpoint

apps/cloud/drizzle/meta/0016_snapshot.json+3−3

@@ -376,9 +376,9 @@ · 375 unmodified lines
376	"primaryKey": false,	376	"primaryKey": false,
377	"notNull": false	377	"notNull": false
378	},	378	},
379	"tools_stale_at": {	379	"tools_stale_token": {
380	"name": "tools_stale_at",	380	"name": "tools_stale_token",
381	"type": "bigint",	381	"type": "text",
382	"primaryKey": false,	382	"primaryKey": false,
383	"notNull": false	383	"notNull": false
384	},	384	},

apps/cloud/src/db/executor-schema.ts+1−1

@@ -61,7 +61,7 @@ export const connection = pgTable( · 60 unmodified lines
61	description: text("description"),	61	description: text("description"),
62	last_health: json("last_health"),	62	last_health: json("last_health"),
63	tools_synced_at: bigint("tools_synced_at", { mode: "bigint" }),	63	tools_synced_at: bigint("tools_synced_at", { mode: "bigint" }),
64	tools_stale_at: bigint("tools_stale_at", { mode: "bigint" }),	64	tools_stale_token: text("tools_stale_token"),
65	tools_sync_claim_id: text("tools_sync_claim_id"),	65	tools_sync_claim_id: text("tools_sync_claim_id"),
66	tools_sync_claim_at: bigint("tools_sync_claim_at", { mode: "bigint" }),	66	tools_sync_claim_at: bigint("tools_sync_claim_at", { mode: "bigint" }),
67	tools_sync_failures: bigint("tools_sync_failures", { mode: "bigint" }),	67	tools_sync_failures: bigint("tools_sync_failures", { mode: "bigint" }),

packages/core/sdk/src/core-schema.ts+15−10

       // 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"),
-      // Epoch ms an event declared this catalog drifted (an MCP
+      // Set when an event declares this catalog drifted (an MCP
-      // `tools/list_changed`, an unknown-tool rejection). Separate from
+      // `tools/list_changed`, an unknown-tool rejection); null when there is no
-      // `tools_synced_at` so marking a catalog stale no longer destroys the
+      // outstanding drift. Separate from `tools_synced_at` so marking a catalog
-      // last-verified timestamp — "drifted" and "never synced" are different
+      // stale no longer destroys the last-verified timestamp — "drifted" and
-      // states with different diagnoses. OUT-DATED by the next authoritative
+      // "never synced" are different states with different diagnoses.
-      // listing rather than cleared by it: the marking fiber and the listing
+      //
-      // fiber are different isolates, so a write that nulled this column would
+      // An OPAQUE token, not a timestamp: it answers "was this catalog
-      // erase a drift signal that landed mid-listing.
+      // invalidated since the listing now finishing began", which is a version
-      tools_stale_at: nullableBigintColumn("tools_stale_at"),
+      // 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
   "integration",
   "name",
   "tools_synced_at",
-  "tools_stale_at",
+  "tools_stale_token",
   "tools_sync_claim_id",
   "tools_sync_claim_at",
   "tools_sync_failures",

packages/core/sdk/src/executor.ts+204−88

 /** What sent a connection through tool production, for the per-connection span.
  *  The read path reports the classified {@link ToolSyncState} verbatim (`cold`
- *  and `stale_marked` are finally distinguishable, now that `tools_stale_at`
+ *  and `stale_marked` are finally distinguishable, now that `tools_stale_token`
  *  records a drift without erasing the last-verified stamp); `explicit` is
  *  every human-initiated production — create, re-mint, `connections.refresh` —
  *  which answers to a caller and so skips the read path's gates entirely. */
 /** 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.
+/**
+ * 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.
  *
- * `lostClaim` is a distinct outcome rather than a flavour of success: the
+ * Collapsing `incomplete` into success is how a fleet of month-dead upstreams
- * listing completed, and then found another attempt had re-claimed the
+ * reports as fully synced; collapsing it into `lost_claim` would blame
- * connection and answered first, so nothing of it was persisted. Counting that
+ * concurrency for a broken server.
- * as a sync is how a fleet whose refreshes all lose their lease reports as
+ *
- * fully synced. `tools` is what the caller should answer with either way — the
+ * `tools` is what the caller should answer with in every case — the fresh
- * persisted catalog when the claim was lost, the fresh listing when it held. */
+ * listing when it was persisted, the persisted catalog otherwise.
+ */
+type ToolProductionOutcome = "synced" | "incomplete" | "lost_claim";
 interface ToolProduction {
   readonly tools: readonly Tool[];
-  readonly lostClaim: boolean;
+  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 —
+ * How much work a read's catalog refresh found and did, for the enclosing
- *  the number the scan's filter scoping exists to keep at zero. `failed`
+ * `executor.tools.list` span. `candidates` is what the stale scan returned —
- *  counts refreshes that were swallowed to keep the read answerable, so a
+ * the number the scan's filter scoping exists to keep at zero.
- *  connection stuck permanently stale is visible as a rate rather than only
+ *
- *  as an absent `synced`. `lostClaim` counts listings that completed and were
+ * The four terminal counters partition every listing that actually ran — a
- *  then discarded because another attempt had re-claimed the connection —
+ * candidate that never won its claim is `skippedClaimed`, not a listing — and
- *  wasted upstream work, and the one outcome that looks like a success from
+ * each says something the others cannot:
- *  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`
+ * - `synced`     an authoritative listing was persisted (see
- *  is one where dead credentials stopped costing handshakes. */
+ *                {@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;
 const NO_CATALOG_REFRESH: CatalogRefreshSummary = {
   candidates: 0,
   synced: 0,
+  incomplete: 0,
   failed: 0,
   lostClaim: 0,
   skippedClaimed: 0,
 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
  *  message, field or rendered cause from it may reach a log line. A defect
     // 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_at`)
+    // 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 the
+    // thing holding the next attempt off, and an `auth` verdict parks a
-    // connection outright.
+    // 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`
             b("name", "=", String(ref.name)),
           );
+        // The row as it stood when this listing began, read before anything is
+        // dialed. Two things come out of it: the failure count the ladder
+        // advances from, and the drift token this listing is answering.
+        const existingRow = yield* findConnectionRow(ref);
+        const failureSet = (kind: ToolSyncErrorKind | null, reason: string) =>
+          toolSyncFailureSet(Number(existingRow?.tools_sync_failures ?? 0), kind, reason);
+        // THE drift mark this listing observed, or null if there was none. It is
+        // a version token, not an instant: an authoritative listing clears the
+        // column only if this exact value is still in it, so a mark written
+        // while we were dialing carries a different token, survives the clear,
+        // and re-lists the connection on the next read.
+        const observedStaleToken =
+          existingRow?.tools_stale_token == null ? null : String(existingRow.tools_stale_token);
         // Does this attempt still own the connection? A read-path refresh takes
         // a lease before dialing, and a slow one can lose it: another reader
         // sees the lease expire, re-claims, re-lists and finishes first. The
         // report an incomplete listing without classifying it.
         //
         // The stamp is the instant the listing STARTED, not the instant it
-        // landed, and `tools_stale_at` is deliberately NOT cleared. Clearing it
+        // landed. It records freshness and NOTHING else: resolving the drift
-        // was a lost update: `connections.markToolsStale` writes that column
+        // mark is a separate compare-and-set inside `replaceCatalog`, so a
-        // from another fiber (a `tools/list_changed` arriving mid-invocation),
+        // signal that arrived mid-listing is not swallowed by a stamp that
-        // and a drift signal landing while we were dialing describes a tool set
+        // happens to sort after it.
-        // this listing never saw. `classifyToolSync` compares the two stamps
-        // instead, so a mark that predates the listing is out-dated by it and
-        // one that postdates it survives to re-list.
         const successSet = (row: ConnectionRow | null) => {
           const recovering = Number(row?.tools_sync_failures ?? 0) > 0;
           return {
           };
         };
-        // Locked: these run inside a read's concurrent fan-out, and a bare
+        /** Record a NON-authoritative outcome, or report that the attempt lost
-        // UPDATE issued while another fiber holds an open BEGIN on the shared
+         *  its lease and wrote nothing. Returns false in the latter case.
-        // SQLite connection is enrolled in that fiber's transaction — and lost
+         *
-        // with it if it rolls back. The permit is NOT reentrant, so nothing
+         *  Locked: this runs inside a read's concurrent fan-out, and a bare
-        // reachable from here may take it again.
+         *  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<string, unknown>,
-        ): Effect.Effect<void, StorageFailure> =>
+        ): Effect.Effect<boolean, StorageFailure> =>
-          withCatalogPersistLock(core.updateMany("connection", { where: outcomeWhere, set }));
+          withCatalogPersistLock(
+            transaction(
+              Effect.gen(function* () {
+                if (!(yield* holdsClaim())) return false;
+                yield* core.updateMany("connection", { where: outcomeWhere, set });
+                return true;
+              }),
+            ),
+          );
-        /** Replace the persisted catalog and record the outcome together, or do
+        /** Replace the persisted catalog, record the outcome, and resolve the
-         *  neither. Returns false when the attempt lost its lease. */
+         *  drift mark this listing answered — all three together, or none of
+         *  them. Returns false when the attempt lost its lease.
+         *
+         *  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<string, unknown>[],
           definitionRows: readonly Record<string, unknown>[],
                 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 = (
-          lostClaim: boolean,
+          outcome: ToolProductionOutcome,
         ): Effect.Effect<ToolProduction, StorageFailure> =>
           core.findMany("tool", { where }).pipe(
             Effect.map((rows) => ({
               tools: rows.map((row) => rowToTool(row as ConnectionToolRow)),
-              lostClaim,
+              outcome,
             })),
           );
         // carry their token outside `item_ids`; no-auth (`"none"` template)
         // connections legitimately bind nothing (an empty `item_ids` is their
         // canonical shape) — both are exempt.
-        const existingRow = yield* findConnectionRow(ref);
-        const failureSet = (kind: ToolSyncErrorKind | null, reason: string) =>
-          toolSyncFailureSet(Number(existingRow?.tools_sync_failures ?? 0), kind, reason);
         if (
           existingRow &&
           existingRow.oauth_client == null &&
           // A legitimate clear IS an authoritative listing: the answer is "no
           // tools", verified as of now.
           if (yield* replaceCatalog([], [], successSet(existingRow))) {
-            return { tools: [], lostClaim: false };
+            return { tools: [], outcome: "synced" };
           }
-          return yield* persistedTools(true);
+          return yield* persistedTools("lost_claim");
         }
         if (!runtime?.plugin.resolveTools) {
           // No dynamic tools — clear any existing rows and return empty.
           if (yield* replaceCatalog([], [], successSet(existingRow))) {
-            return { tools: [], lostClaim: false };
+            return { tools: [], outcome: "synced" };
           }
-          return yield* persistedTools(true);
+          return yield* persistedTools("lost_claim");
         }
         const result: ResolveToolsResult = yield* runtime.plugin
           // verdict parks the connection outright.
           const reason = syncHealthReason(result);
           const kind = result.incompleteKind ?? null;
-          yield* writeSyncOutcome(failureSet(kind, reason));
+          // An attempt that had already lost its lease wrote nothing at all —
+          // not the ladder, not the health detail — so it reports `lost_claim`.
+          // Reporting `incomplete` there would attribute a verdict to a row the
+          // winning attempt owns.
+          const recorded = yield* writeSyncOutcome(failureSet(kind, reason));
           yield* Effect.logWarning("executor tool sync preserved catalog", {
             reason,
             errorKind: kind ?? "unclassified",
             integration: String(ref.integration),
             connection: String(ref.name),
           });
-          return yield* persistedTools(false);
+          return yield* persistedTools(recorded ? "incomplete" : "lost_claim");
         }
         if (
             // 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.
-            yield* writeSyncOutcome(failureSet(null, reason));
+            const recorded = yield* writeSyncOutcome(failureSet(null, reason));
             yield* Effect.logWarning("executor tool sync preserved nonzero catalog", {
               reason,
               integration: String(ref.integration),
               connection: String(ref.name),
               existingToolCount: keptRows.length,
             });
+            // 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)),
-              lostClaim: false,
+              outcome: "incomplete",
             };
           }
         }
         // this catalog with a listing at least as recent as ours. Answer from
         // what is persisted rather than overwriting it.
         if (!(yield* replaceCatalog(toolRows, definitionRows, successSet(existingRow)))) {
-          return yield* persistedTools(true);
+          return yield* persistedTools("lost_claim");
         }
         return {
-          lostClaim: false,
+          outcome: "synced",
           tools: result.tools.map((tool: ToolDef) =>
             rowToTool(
               {
     // catalog indistinguishable from one that had never synced. The drift also
     // un-parks — the server just told us its tool set moved, which is evidence
     // it is reachable and worth one more attempt.
+    //
+    // A FRESH token every time, unconditionally overwriting whatever was there.
+    // That is what makes a mark landing mid-listing survive: the listing clears
+    // the column only for the token it observed at its start, so overwriting
+    // with a new one is precisely how this signal outlives the listing that
+    // could not have seen it.
+    //
+    // Locked: this is a lifecycle write like every other, and it can run while
+    // a read's refresh fan-out holds an open BEGIN on the shared SQLite
+    // connection — where a bare UPDATE is enrolled in that transaction and lost
+    // with it if it rolls back. Losing THIS write loses a drift signal outright.
     const connectionsMarkToolsStale = (ref: ConnectionRef): Effect.Effect<void, StorageFailure> =>
-      core.updateMany("connection", {
+      withCatalogPersistLock(
-        where: (b: AnyCb) =>
+        core.updateMany("connection", {
-          b.and(
+          where: (b: AnyCb) =>
-            byOwner(ref.owner)(b),
+            b.and(
-            b("integration", "=", String(ref.integration)),
+              byOwner(ref.owner)(b),
-            b("name", "=", String(ref.name)),
+              b("integration", "=", String(ref.integration)),
-          ),
+              b("name", "=", String(ref.name)),
-        set: { tools_stale_at: Date.now(), ...TOOL_SYNC_LADDER_RESET },
+            ),
-      });
+          set: { tools_stale_token: toolsStaleToken(), ...TOOL_SYNC_LADDER_RESET },
+        }),
+      );
     // ------------------------------------------------------------------
     // Active policy source.
     // tool catalog is stale, scoped to that read's own filter — a read for one
     // integration never waits on another's upstream handshakes. Four triggers,
     // classified by `tool-sync-schedule`:
-    //  - cold: never synced at all (`tools_synced_at` and `tools_stale_at` both
+    //  - cold: never synced at all (`tools_synced_at` and `tools_stale_token`
-    //    NULL — a fresh row, or one a data migration reset);
+    //    both NULL — a fresh row, or one a data migration reset);
-    //  - stale-marked: `tools_stale_at` postdates the catalog
+    //  - stale-marked: `tools_stale_token` is set
     //    (`connections.markToolsStale` — an MCP `tools/list_changed`
     //    notification or unknown-tool rejection recorded the drift
     //    mid-invocation);
         // `blob.getMany`). The two NULL/NOT-NULL arms always stand, so the OR
         // can never degenerate to a bare `false`.
         //
-        // The drift arm is `tools_stale_at IS NOT NULL` rather than a
+        // The drift arm is exactly `tools_stale_token IS NOT NULL`, and it is
-        // column-to-column comparison against `tools_synced_at`, which the query
+        // exact rather than an over-selection because an authoritative listing
-        // builder does not offer. A listing OUT-DATES a drift mark rather than
+        // CLEARS the token by compare-and-set. That is what keeps the zero-rows
-        // clearing it (clearing it lost concurrent marks), so this arm
+        // steady state: encoding the mark as a timestamp nothing ever cleared
-        // over-selects every connection that has ever drifted and the per-row
+        // made this arm re-select every once-drifted connection on every read,
-        // classifier discards the resolved ones. Deliberate: the alternative is
+        // forever.
-        // the lost update, and the cost is one projected row per
-        // once-drifted connection.
         //
         // The revision arm is `<=`, matching `classifyToolSync`: `tools_synced_at`
         // is the instant the listing STARTED, so a revision landing in that same
               filter?.connection === undefined ? true : b("name", "=", String(filter.connection)),
               b.or(
                 b.isNull("tools_synced_at"),
-                b.isNotNull("tools_stale_at"),
+                b.isNotNull("tools_stale_token"),
                 revisedSlugs.length === 0 || latestRevision === null
                   ? false
                   : b.and(
           const candidate: ToolSyncCandidate = {
             toolsSyncedAt: asEpochMs(connection.tools_synced_at),
-            toolsStaleAt: asEpochMs(connection.tools_stale_at),
+            toolsStaleToken:
+              connection.tools_stale_token == null ? null : String(connection.tools_stale_token),
             toolsSyncClaimId:
               connection.tools_sync_claim_id == null
                 ? null
                 trigger: state,
                 claim,
               }).pipe(
-                // `lost_claim` is its own outcome, not a flavour of `ok`: the
+                // Reported verbatim: `synced`, `incomplete` and `lost_claim`
-                // listing ran, cost an upstream handshake, and was then thrown
+                // are three different things to be told, and only the first is
-                // away because another attempt had re-claimed the connection.
+                // a sync. Folding the other two into it is how a fleet that
-                // Reporting it as synced is how a fleet that persists nothing
+                // persists nothing reads as fully healthy.
-                // reads as fully healthy.
+                Effect.flatMap((produced) =>
-                Effect.flatMap((produced) => {
+                  Effect.annotateCurrentSpan({
-                  const outcome = produced.lostClaim ? ("lost_claim" as const) : ("ok" as const);
+                    "executor.tools.sync.outcome": produced.outcome,
-                  return Effect.annotateCurrentSpan({
+                  }).pipe(Effect.as(produced.outcome)),
-                    "executor.tools.sync.outcome": outcome,
+                ),
-                  }).pipe(Effect.as(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
         return {
           candidates: connections.length,
-          synced: outcomes.filter((outcome) => outcome === "ok").length,
+          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:
           "executor.tools.result_count": tools.length,
           "executor.tools.sync.candidates": refresh.candidates,
           "executor.tools.sync.synced": refresh.synced,
+          "executor.tools.sync.incomplete": refresh.incomplete,
           "executor.tools.sync.failed": refresh.failed,
           "executor.tools.sync.lost_claim": refresh.lostClaim,
           "executor.tools.sync.skipped_claimed": refresh.skippedClaimed,

packages/core/sdk/src/plugin.ts+6−1

      *  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. The last
-     *  verified sync time is preserved; the drift is recorded alongside it. */
+     *  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. */
     readonly markToolsStale: (ref: ConnectionRef) => Effect.Effect<void, StorageFailure>;
     /** Resolve a connection's value through its provider (and OAuth refresh).
      *  null if the provider can't produce one. */

packages/core/sdk/src/tool-sync-schedule.test.ts+86−35

 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> = {}): ToolSyncCandidate => ({
   toolsSyncedAt: NOW - 1000,
-  toolsStaleAt: null,
+  toolsStaleToken: null,
   toolsSyncClaimId: null,
   toolsSyncClaimAt: null,
   toolsSyncFailures: 0,
   it("calls a connection with a drift mark stale_marked, keeping its stamp", () => {
     expect(
-      classifyToolSync(candidate({ toolsSyncedAt: NOW - 60_000, toolsStaleAt: NOW - 1000 }), NOW),
+      classifyToolSync(
+        candidate({ toolsSyncedAt: NOW - 60_000, toolsStaleToken: STALE_TOKEN }),
+        NOW,
+      ),
     ).toBe("stale_marked");
   });
-  it("still sees a drift mark that landed in the same millisecond as the stamp", () => {
+  it("reads the drift mark as presence, never against the stamp", () => {
-    // Date.now() has millisecond granularity, and a `tools/list_changed`
+    // The whole point of the token encoding. `tools_stale_token` answers "was
-    // arriving inside the millisecond it invalidates must not be swallowed.
+    // this catalog invalidated since the listing now finishing began", which is
-    expect(classifyToolSync(candidate({ toolsSyncedAt: NOW, toolsStaleAt: NOW }), NOW)).toBe(
+    // a version question; the earlier encoding asked it of a wall clock, and
-      "stale_marked",
+    // `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 drift mark the listing out-dated resolved", () => {
+  it("leaves a connection whose drift mark was cleared resolved", () => {
-    // An authoritative listing does not clear `tools_stale_at`, it stamps
+    // An authoritative listing clears the token it observed, and a cleared
-    // `tools_synced_at` past it. A mark older than the stamp is a settled
+    // column is the only thing that settles a drift. This is also what keeps
-    // drift, and the connection must not re-list forever because the column is
+    // the read's scan at zero rows in steady state: nothing ever cleared the
-    // still set.
+    // old timestamp column, so every connection that had ever drifted was
+    // re-selected on every read, forever.
     expect(
-      classifyToolSync(candidate({ toolsSyncedAt: NOW - 1000, toolsStaleAt: NOW - 2000 }), NOW),
+      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, toolsStaleAt: NOW }), NOW)).toBe(
+    expect(
-      "stale_marked",
+      classifyToolSync(candidate({ toolsSyncedAt: null, toolsStaleToken: STALE_TOKEN }), NOW),
-    );
+    ).toBe("stale_marked");
   });
   it("calls a catalog older than its integration's config revision config_revised", () => {
   it("ranks an explicit drift signal above a config revision and the clock", () => {
     expect(
       classifyToolSync(
-        candidate({ toolsSyncedAt: NOW - TTL - 1, toolsStaleAt: NOW, configRevisedAt: NOW - 1 }),
+        candidate({
+          toolsSyncedAt: NOW - TTL - 1,
+          toolsStaleToken: STALE_TOKEN,
+          configRevisedAt: NOW - 1,
+        }),
         NOW,
       ),
     ).toBe("stale_marked");
     const drifted = candidate({
       toolsSyncFailures: 0,
       toolsSyncRetryAt: NOW + TTL,
-      toolsStaleAt: NOW,
+      toolsStaleToken: STALE_TOKEN,
     });
     expect(isToolSyncBackedOff(drifted, NOW)).toBe(false);
     expect(isSyncEligible(drifted, NOW)).toBe(true);
     expect(decideToolSync(candidate(), NOW)).toEqual({ kind: "skip", reason: "fresh" });
   });
-  it("skips a parked connection however overdue the clock says it is", () => {
+  it("skips a parked connection however far past its freshness window it is", () => {
     expect(
       decideToolSync(
         candidate({
-          toolsSyncedAt: null,
+          toolsSyncedAt: NOW - 30 * TTL,
           toolsSyncFailures: 9,
           toolsSyncErrorKind: "auth",
           toolsSyncRetryAt: NOW - 1,
       expect(
         decideToolSync(
           candidate({
-            toolsSyncedAt: null,
+            toolsSyncedAt: NOW - TTL - 1,
             toolsSyncErrorKind: "auth",
             toolsSyncFailures: 3,
             toolsSyncRetryAt: NOW + TTL,
       ).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(
     });
   });
-  // The park answers "the clock alone is not a reason to re-dial a rejected
+  // The park answers exactly one question: "is re-VERIFYING a catalog we can
-  // credential". It is not an answer to somebody telling us the world changed.
+  // still serve worth dialing a credential that was refused". Every other
-  describe("park yields to an explicit invalidation", () => {
+  // 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
     it("refreshes a parked connection that was marked stale", () => {
       expect(
         decideToolSync(
-          candidate({ toolsSyncedAt: NOW - 60_000, toolsStaleAt: NOW, toolsSyncErrorKind: "auth" }),
+          candidate({
+            toolsSyncedAt: NOW - 60_000,
+            toolsStaleToken: STALE_TOKEN,
+            toolsSyncErrorKind: "auth",
+          }),
           NOW,
         ),
       ).toEqual({ kind: "refresh", trigger: "stale_marked" });
     });
-    it("still parks when only the clock is asking", () => {
+    it("refreshes a parked connection that has never synced at all", () => {
-      for (const overrides of [
+      // A parked `cold` connection has NO catalog to serve, so parking it is
-        { toolsSyncedAt: null },
+      // not "keep serving what we have", it is "serve nothing until a human
-        { toolsSyncedAt: NOW - TTL - 1 },
+      // notices". The ladder still caps the cost at a few dials a day, and it
-      ] as const) {
+      // is what lets a credential repaired upstream — where nothing in this
-        expect(
+      // system observes the repair — recover on its own.
-          decideToolSync(candidate({ ...overrides, toolsSyncErrorKind: "auth" }), NOW),
+      expect(
-        ).toEqual({ kind: "skip", reason: "parked" });
+        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" });
     });
   });
 });

packages/core/sdk/src/tool-sync-schedule.ts+47−37

  * 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_at` column
+ * `cold` and `stale_marked` were one value before the `tools_stale_token`
- * existed: `connections.markToolsStale` cleared `tools_synced_at`, which made a
+ * column existed: `connections.markToolsStale` cleared `tools_synced_at`, which
- * connection that had never synced indistinguishable from one whose catalog was
+ * made a connection that had never synced indistinguishable from one whose
- * invalidated mid-invocation, and destroyed the last-verified timestamp in the
+ * catalog was invalidated mid-invocation, and destroyed the last-verified
- * process.
+ * timestamp in the process.
  */
 export type ToolSyncState = "fresh" | "cold" | "stale_marked" | "config_revised" | "expired";
   /** 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;
-  /** Epoch ms an event declared the persisted catalog drifted. */
+  /** The outstanding drift mark, or null when there is none. An opaque token
-  readonly toolsStaleAt: number | null;
+   *  (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. */
  * Classify a connection's catalog. Total, and ordered by authority: an explicit
  * drift signal outranks a config revision, which outranks the clock.
  *
- * BOTH invalidation comparisons are inclusive of the stamp's own millisecond
+ * A drift mark is read as PRESENCE, never as an instant: `tools_stale_token` is
- * (`toolsStaleAt >= syncedAt`, `syncedAt <= configRevisedAt`). `Date.now()` has
+ * an opaque token, and the listing that resolves it clears it by
- * millisecond granularity and `tools_synced_at` records when the listing
+ * compare-and-set on the token it observed. Comparing a drift TIMESTAMP against
- * STARTED, so a signal landing in that millisecond describes a change the
+ * `tools_synced_at` instead is what made the two stamps landing in one
- * listing cannot have seen. Swallowing it would leave the connection serving a
+ * millisecond ambiguous, and left the column set forever on every connection
- * catalog that predates the change with no TTL backstop — a plugin without a
+ * that had ever drifted.
- * remote catalog never reaches the `expired` branch below.
+ *
- *
+ * The config-revision comparison stays inclusive of the stamp's own millisecond
- * This is also what resolves a drift mark: an authoritative listing does not
+ * (`syncedAt <= configRevisedAt`). `Date.now()` has millisecond granularity and
- * clear `tools_stale_at`, it out-dates it. Clearing the column was a lost
+ * `tools_synced_at` records when the listing STARTED, so a revision landing in
- * update, because the fiber that marked the drift and the fiber finishing the
+ * that millisecond describes a config the listing cannot have read. Swallowing
- * listing are different isolates with no coordination between them.
+ * 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;
-  // A never-synced connection sorts before every stamp, which is why the
+  if (candidate.toolsStaleToken !== null) return "stale_marked";
-  // comparisons below read through `?? 0` rather than special-casing null.
-  if (candidate.toolsStaleAt !== null && candidate.toolsStaleAt >= (syncedAt ?? 0)) {
-    return "stale_marked";
-  }
   if (syncedAt === null) return "cold";
   if (candidate.configRevisedAt !== null && syncedAt <= candidate.configRevisedAt) {
     return "config_revised";
 /** 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), and yielded to by
+ *  OAuth re-mint, a healthy probe, a fresh drift signal). {@link decideToolSync}
- *  {@link decideToolSync} whenever an explicit invalidation is outstanding. */
+ *  applies it to `expired` alone — every other refresh trigger outranks it. */
 export const isToolSyncParked = (candidate: ToolSyncCandidate): boolean =>
   candidate.toolsSyncErrorKind === "auth";
  * 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 yields to an EXPLICIT invalidation. `cold` and `expired` are the
+ * The park gates `expired` and NOTHING else. `expired` is the only trigger that
- * clock talking, and re-dialing a rejected credential because time passed is
+ * is purely the clock asking a connection which already has a working catalog
- * exactly the waste the park exists to stop; `stale_marked` and
+ * to re-verify it, and re-dialing a rejected credential to re-verify a catalog
- * `config_revised` are somebody telling us the world changed. A config revision
+ * we can still serve is exactly the waste the park exists to stop.
- * in particular is how an `auth` verdict caused by integration configuration
+ *
- * (the key in the wrong place, the wrong token endpoint) gets FIXED, and
+ * `stale_marked` and `config_revised` are somebody telling us the world
- * `integrations.update` is the one invalidation path that does not clear the
+ * changed. A config revision in particular is how an `auth` verdict caused by
- * ladder, so swallowing it here would leave the repaired connection parked
+ * integration configuration (the key in the wrong place, the wrong token
- * forever.
+ * 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 }
 export const decideToolSync = (candidate: ToolSyncCandidate, now: number): ToolSyncDecision => {
   const state = classifyToolSync(candidate, now);
   if (state === "fresh") return { kind: "skip", reason: "fresh" };
-  const timeDriven = state === "cold" || state === "expired";
+  if (state === "expired" && isToolSyncParked(candidate)) return { kind: "skip", reason: "parked" };
-  if (timeDriven && 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 };

packages/core/sdk/src/tools-sync-scope.test.ts+102−24

 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";
   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);
   "integration",
   "name",
   "tools_synced_at",
-  "tools_stale_at",
+  "tools_stale_token",
   "tools_sync_claim_id",
   "tools_sync_claim_at",
   "tools_sync_failures",
         /** 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_at`, which is what
+         *  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: () =>
               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
                 ? Effect.die(`no connection row named ${name}`)
                 : Effect.succeed({
                     syncedAt: row.tools_synced_at == null ? null : Number(row.tools_synced_at),
-                    staleAt: row.tools_stale_at == null ? null : Number(row.tools_stale_at),
+                    staleToken: row.tools_stale_token ?? null,
                     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),
         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 newer `tools_stale_at`.
+        // 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;
       Effect.gen(function* () {
         const harness = yield* makeHarness();
         yield* harness.connect(INTEG_A, "main");
-        yield* harness.clearEveryCatalogStamp();
+        // 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;
         expect((yield* harness.syncStateOf("main")).errorKind).toBe("auth");
         // Parked: not even elapsing the backoff window lets a read dial again.
-        // Retrying a rejected credential cannot change the verdict, and 401
+        // Retrying a rejected credential to re-verify a catalog we can still
-        // handshakes were the largest single source of wasted upstream calls.
+        // 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"]);
     ),
   );
+  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 synced = yield* harness.syncStateOf("main");
         expect(synced.syncedAt).not.toBeNull();
-        expect(synced.staleAt).toBeNull();
+        expect(synced.staleToken).toBeNull();
         yield* harness.executor.counting.markStale(INTEG_A, "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.staleAt).not.toBeNull();
+        expect(drifted.staleToken).not.toBeNull();
         harness.renameToolsTo("redeploy");
         harness.resolved.length = 0;
         expect(harness.resolved).toEqual(["alpha/main"]);
         expect(tools.map((tool) => String(tool.name))).toEqual(["alpha_redeploy"]);
-        // The re-list settled the drift by OUT-DATING the mark, not by clearing
+        // The re-list settled the drift by CLEARING the token it observed. It
-        // it: the stamp is now at or past `tools_stale_at`, which is what
+        // is a compare-and-set, not an unconditional null (the case below is
-        // `classifyToolSync` reads. Clearing the column instead is the lost
+        // the mark it must not clear), and the clear is what makes the
-        // update the case below is about.
+        // resolution observable rather than inferred from two timestamps that
+        // can land in the same millisecond.
         const settled = yield* harness.syncStateOf("main");
-        expect(settled.staleAt).not.toBeNull();
+        expect(settled.staleToken).toBeNull();
         expect(settled.syncedAt).not.toBeNull();
-        expect(Number(settled.syncedAt)).toBeGreaterThanOrEqual(Number(settled.staleAt));
-        // And the connection is genuinely done: the next read neither scans it
+        // And the connection is genuinely done: the next read does not dial
-        // as due nor dials it again.
+        // 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 }]);
       }),
     ),
   );
         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();
         // 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. Nulling `tools_stale_at` here is
+        // The mark survived the success write, verbatim. Nulling
-        // a lost update, and for a plugin with no remote catalog nothing else
+        // `tools_stale_token` unconditionally here is a lost update, and for a
-        // would ever re-list the connection.
+        // plugin with no remote catalog nothing else would ever re-list the
+        // connection.
         const state = yield* harness.syncStateOf("main");
-        expect(state.staleAt).not.toBeNull();
+        expect(state.staleToken).toBe(marked.staleToken);
-        expect(Number(state.staleAt)).toBeGreaterThanOrEqual(Number(state.syncedAt));
         // Which is the fact that matters: the next read dials again.
         harness.renameToolsTo("rerelease");
         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();
       }),
     ),
   );
         expect(state.syncedAt).toBe(synced.syncedAt);
         expect(state.failures).toBe(4);
         expect(state.errorKind).toBe("unreachable");
-        expect(state.staleAt).not.toBeNull();
+        // The drift mark also stands: only an authoritative listing clears it,
+        // and none of these four were one.
+        expect(state.staleToken).not.toBeNull();
       }),
     ),
   );