From 955dd907e8c860f7279e281cd8b5f29283d41159 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Sun, 16 Aug 2026 14:54:54 -0600 Subject: [PATCH 1/8] Defer a tools read's expired catalog refresh to the host --- packages/core/sdk/src/executor.ts | 355 +++++++++++++----- packages/core/sdk/src/test-config.ts | 39 ++ .../core/sdk/src/tools-sync-scope.test.ts | 162 +++++++- 3 files changed, 462 insertions(+), 94 deletions(-) diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index 514365801b..a85af305be 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -665,6 +665,35 @@ export interface ExecutorConfig Effect.Effect; + /** + * Run `task` AFTER the operation that produced it has answered its caller. + * + * The read path hands the speculative half of its catalog refresh here: a TTL + * `expired` connection has a working catalog and no evidence at all that it + * drifted, so re-verifying it is worth doing soon and never worth making a + * read wait for. Invalidation triggers (`cold`, `stale_marked`, + * `config_revised`) do NOT come through here — the next read's correctness + * depends on them, so they stay inline. + * + * CONTRACT. `task` is already total (`Effect`: no failure channel, and + * its own defects are caught and logged inside), so a host only has to + * schedule it. Scheduling MUST NOT fail the caller — this is invoked mid-read + * and a background-work outage is not a read failure. Dropping a task is + * SAFE: nothing was written, the connection's row is still due, and the cost + * is one freshness window before the next read re-offers it. + * + * ABSENT means run inline, and that is the correct semantics for a host with + * no background capability rather than a compatibility shim: the work still + * has to happen, and a host that cannot outlive its own response can only do + * it before answering. + * + * On Workers the implementation must guarantee execution with `waitUntil` (or + * `ctx.waitUntil`) — an isolate that is evicted the instant its response is + * written runs nothing else. It must also keep whatever database handle the + * task closes over alive until the task finishes: a request-scoped pool that + * closes with the request takes the deferred batch down with it. + */ + readonly deferToolSync?: (task: Effect.Effect) => Effect.Effect; /** * Opt into the PLATFORM VIEW: a read-only, tenant-wide `executor.admin` * surface that reads across every subject in the tenant (see @@ -741,6 +770,23 @@ interface ToolProduction { readonly outcome: ToolProductionOutcome; } +/** A connection the read's scan found due, with everything one refresh attempt + * needs: which integration owns it, which connection it is, what made it due, + * and the failure count the retry ladder advances from. Named because the + * inline path and the deferred batch both consume it, and they must not drift + * into two shapes. */ +interface DueConnection { + readonly integrationRow: IntegrationRow; + readonly ref: ConnectionRef; + readonly state: Exclude; + readonly priorFailures: number; +} + +/** What one refresh attempt did, as reported to the read's summary. + * `skipped_claimed` is the compare-and-set losing, `fail` is the swallowed + * error/defect path; the rest are {@link ToolProductionOutcome} verbatim. */ +type CatalogRefreshOutcome = ToolProductionOutcome | "fail" | "skipped_claimed"; + /** * 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 — @@ -768,6 +814,15 @@ interface ToolProduction { * 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. + * + * `deferred` counts the candidates handed to `ExecutorConfig.deferToolSync` + * instead of being listed inline, and it is deliberately NOT folded into the + * four terminal counters: from the READ's point of view a deferred candidate + * has no outcome yet, and reporting one it did not observe is how a background + * batch that never ran would read as a fleet of healthy syncs. Each deferred + * listing still emits its own `executor.tools.sync` span with its own outcome + * when it eventually runs. `deferredOverflow` is what did not fit in the batch + * cap, so a read that keeps shedding work says so rather than capping quietly. */ interface CatalogRefreshSummary { readonly candidates: number; @@ -778,6 +833,8 @@ interface CatalogRefreshSummary { readonly skippedClaimed: number; readonly skippedBackoff: number; readonly skippedParked: number; + readonly deferred: number; + readonly deferredOverflow: number; } const NO_CATALOG_REFRESH: CatalogRefreshSummary = { @@ -789,8 +846,28 @@ const NO_CATALOG_REFRESH: CatalogRefreshSummary = { skippedClaimed: 0, skippedBackoff: 0, skippedParked: 0, + deferred: 0, + deferredOverflow: 0, }; +/** + * The most `expired` connections one read will hand to + * `ExecutorConfig.deferToolSync`. + * + * A cap rather than the whole due set, because the deferred batch is charged to + * the host's background budget, and on the tightest host that budget is a + * Workers `waitUntil` — bounded CPU, bounded wall clock, and shared with + * whatever else the request queued. Sixteen upstream listings at concurrency 4 + * is four sequential handshakes deep, which fits that budget with room to spare + * while still draining a realistically wide workspace in a couple of reads. + * + * Nothing is lost by capping: an `expired` connection that does not fit is + * still due, still serving its existing catalog, and re-offered on the very + * next read. The overflow is counted on the read's span (never logged) so a + * deployment that permanently sheds work is visible instead of quietly slow. + */ +const TOOL_SYNC_DEFERRED_BATCH_MAX = 16; + /** 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 @@ -2688,6 +2765,13 @@ export const createExecutor = ) => task); + // The retry ladder's first rung. It follows the freshness window when there // is a usable one, and falls back to the default otherwise: "don't re-dial // a server that just refused us" is a separate concern from "re-list a live @@ -4093,6 +4177,146 @@ export const createExecutor = => + 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": state, + }, + }), + ); + + /** + * The `expired` half of a read's due set, as one task for + * `ExecutorConfig.deferToolSync`. + * + * ONE task for the whole batch, not one per connection: the cap, the + * concurrency bound and the host's background budget are all properties of + * the batch, and a host handed sixteen independent tasks would have to + * rediscover every one of them. + * + * Total by construction. `refreshOneConnection` already swallows its own + * failures and defects per connection; the storage channel that survives it + * (the claim's own write) is caught here, because a deferred task runs with + * no caller to report to and a rejected background fiber is an unhandled + * error on some hosts. + * + * LOCKING. This runs the same `withCatalogPersistLock` the inline path + * does, and the permit belongs to THIS executor instance — so it excludes + * this read's own inline refreshes and nothing else. A later request in the + * same isolate builds its own executor with its own semaphore, and the two + * can interleave here. That is by design and already the production case: + * the lease compare-and-set on the connection row, not the in-process + * permit, is what makes exactly one attempt authoritative. The permit's job + * is narrower — keeping two persists on ONE shared sqlite connection from + * splitting each other's delete-then-insert — and it still covers every + * persist a single executor issues, deferred ones included. + */ + const deferredSyncTask = (batch: readonly DueConnection[]): Effect.Effect => + Effect.forEach(batch, refreshOneConnection, { concurrency: 4, discard: true }).pipe( + Effect.withSpan("executor.tools.sync.deferred", { + attributes: { "executor.tools.sync.deferred": batch.length }, + }), + Effect.catchCause((cause) => + Effect.logWarning("executor deferred tool catalog refresh failed", { + errorTags: causeErrorTags(cause), + }), + ), + ); + // Rebuild the connections a `tools.list` is about to read whose persisted // 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, @@ -4116,6 +4340,16 @@ export const createExecutor = => @@ -4201,12 +4435,7 @@ export const createExecutor = ; - readonly priorFailures: number; - }[] = []; + const due: DueConnection[] = []; let skippedClaimed = 0; let skippedBackoff = 0; let skippedParked = 0; @@ -4265,95 +4494,33 @@ export const createExecutor = - 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": state, - }, - }), - ), + due.filter((candidate) => candidate.state !== "expired"), + refreshOneConnection, { concurrency: 4 }, ); + // The speculative half, handed over AFTER the invalidation-driven + // listings have run: with no seam configured `deferToolSync` is the + // identity, so this is where the batch executes, and running it before + // the work a read's own correctness depends on would invert the + // priority for exactly the hosts that can least afford it. + const expired = due.filter((candidate) => candidate.state === "expired"); + const batch = expired.slice(0, TOOL_SYNC_DEFERRED_BATCH_MAX); + if (batch.length > 0) { + // Guarded like `onIntegrationChange`: this hook is a host's + // scheduling machinery, and a read must not become failable by it. + // Nothing is lost when it dies — the batch claimed nothing yet, so + // the connections are simply still due. + yield* deferToolSync(deferredSyncTask(batch)).pipe( + Effect.catchCause((cause) => + Effect.logWarning("executor deferred tool catalog refresh was not scheduled", { + deferred: batch.length, + errorTags: causeErrorTags(cause), + }), + ), + ); + } + return { candidates: connections.length, synced: outcomes.filter((outcome) => outcome === "synced").length, @@ -4364,6 +4531,8 @@ export const createExecutor = outcome === "skipped_claimed").length, skippedBackoff, skippedParked, + deferred: batch.length, + deferredOverflow: expired.length - batch.length, }; }); @@ -4425,6 +4594,8 @@ export const createExecutor = ["onIntegrationChange"]; + /** Install a `deferToolSync` collector, so background tool-catalog work is + * queued instead of run, and a test decides when it happens by yielding + * `drainBackgroundTasks`. + * + * OFF by default, which leaves `deferToolSync` absent and therefore every + * deferred batch inline — the behaviour every existing suite was written + * against, and the behaviour of a host with no background capability. */ + readonly collectBackgroundTasks?: boolean; +}; + +/** The `deferToolSync` collector behind `collectBackgroundTasks`, plus its + * drain. + * + * Deterministic on purpose: tasks run in enqueue order, one at a time, only + * when the drain is yielded. A background scheduler that ran them on a timer + * would make every assertion about "what the read did NOT do" a race. + * + * The drain takes the queue as it stands and runs that; work a drained task + * enqueues in turn belongs to the next drain, so one drain always terminates. */ +const makeBackgroundTaskCollector = () => { + const pending: Effect.Effect[] = []; + return { + deferToolSync: (task: Effect.Effect) => + Effect.sync(() => { + pending.push(task); + }), + drainBackgroundTasks: Effect.suspend(() => + Effect.forEach(pending.splice(0), (task) => task, { discard: true }), + ), + }; }; export const makeTestConfig = ( @@ -130,6 +160,9 @@ export const makeTestConfig = , "db"> & { readonly db: FumaDb; readonly testDb: TestFumaDb; + /** Run every background task queued so far. Always present, and empty unless + * `collectBackgroundTasks` installed the collector. */ + readonly drainBackgroundTasks: Effect.Effect; } => { const tenant = options?.tenant ?? "test-tenant"; const subject = options?.subject === undefined ? "test-subject" : options.subject; @@ -151,7 +184,13 @@ export const makeTestConfig = { }); }); -const makeHarness = () => +const makeHarness = (options?: { + /** Queue deferred tool-sync batches instead of running them, so a case can + * assert what the read did NOT do and then drive the batch itself. */ + readonly collectBackgroundTasks?: boolean; + /** Replace the seam outright, for the cases about the seam's own contract + * rather than about what gets deferred. */ + readonly deferToolSync?: (task: Effect.Effect) => Effect.Effect; +}) => Effect.acquireRelease( Effect.gen(function* () { const counting = makeCountingPlugin(); const config = makeTestConfig({ plugins: [memoryCredentialsPlugin(), counting.plugin] as const, + collectBackgroundTasks: options?.collectBackgroundTasks, }); const observed = observeConnectionReads(config.db); - const executor = yield* createExecutor({ ...config, db: observed.db }); + const executor = yield* createExecutor({ + ...config, + db: observed.db, + ...(options?.deferToolSync === undefined ? {} : { deferToolSync: options.deferToolSync }), + }); yield* executor.counting.seed(); return { ...counting, executor, testDb: config.testDb, + drainBackgroundTasks: config.drainBackgroundTasks, connectionScans: observed.connectionScans, peakOpenTransactions: observed.peakOpenTransactions, connect: (integration: IntegrationSlug, name: string) => @@ -883,3 +896,148 @@ describe("tools read catalog refresh lifecycle", () => { ), ); }); + +// --------------------------------------------------------------------------- +// Deferred refresh. A TTL `expired` catalog is one nothing has told us is +// wrong — it works, it is merely old — so re-verifying it is background work. +// Every other trigger is somebody reporting that the world changed, and the +// answer this read is about to give is wrong until it runs. These cases pin +// which half is which, and that the background half claims nothing until it +// actually runs. +// --------------------------------------------------------------------------- + +describe("tools read deferred catalog refresh", () => { + it.effect("serves the expired catalog without dialing, and refreshes on the drain", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness({ collectBackgroundTasks: true }); + yield* harness.connect(INTEG_A, "main"); + yield* harness.expireEveryCatalog(); + // Make the deferred listing observably different from what is already + // persisted, so "the drain refreshed it" is distinguishable from "the + // drain did nothing and the old rows were fine". + harness.renameToolsTo("redeploy"); + harness.resolved.length = 0; + + const served = yield* harness.executor.tools.list({ integration: INTEG_A }); + + // The read paid no upstream handshake and answered from the catalog it + // already had. This is the whole point of the layer. + expect(harness.resolved).toEqual([]); + expect(served.map((tool) => String(tool.name))).toEqual(["alpha_deploy"]); + // And it claimed nothing on the way past. The claim belongs to the + // attempt, so a batch that is enqueued and then never runs (an evicted + // isolate, a dropped `waitUntil`) strands no lease. + expect((yield* harness.syncStateOf("main")).claimId).toBeNull(); + + yield* harness.drainBackgroundTasks; + + expect(harness.resolved).toEqual(["alpha/main"]); + const refreshed = yield* harness.executor.tools.list({ integration: INTEG_A }); + expect(refreshed.map((tool) => String(tool.name))).toEqual(["alpha_redeploy"]); + }), + ), + ); + + it.effect("still dials a drifted connection inline, seam or no seam", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness({ collectBackgroundTasks: true }); + yield* harness.connect(INTEG_A, "main"); + // Drifted AND old: `classifyToolSync` ranks the drift signal above the + // clock, so this is an invalidation, not an expiry, and the read's own + // correctness depends on it. + yield* harness.expireEveryCatalog(); + yield* harness.executor.counting.markStale(INTEG_A, "main"); + 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"]); + }), + ), + ); + + it.effect("caps one read's deferred batch and leaves the rest for the next read", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness({ collectBackgroundTasks: true }); + // One more than the cap, so the boundary is exercised rather than + // approached. The overflow count itself rides the read's span + // (`executor.tools.sync.deferred_overflow`); what is asserted here is + // the behaviour behind it — work is shed, never dropped. + const names = Array.from({ length: 17 }, (_, index) => `conn${index}`); + for (const name of names) yield* harness.connect(INTEG_A, name); + yield* harness.expireEveryCatalog(); + harness.resolved.length = 0; + + yield* harness.executor.tools.list({ integration: INTEG_A }); + yield* harness.drainBackgroundTasks; + + expect(harness.resolved).toHaveLength(16); + + // The seventeenth is still due — still serving its catalog, and picked + // up by the very next read. + const deferredFirst = new Set(harness.resolved); + harness.resolved.length = 0; + yield* harness.executor.tools.list({ integration: INTEG_A }); + yield* harness.drainBackgroundTasks; + + const remaining = names + .map((name) => `alpha/${name}`) + .filter((address) => !deferredFirst.has(address)); + expect(remaining).toHaveLength(1); + expect(harness.resolved).toContain(remaining[0]); + }), + ), + ); + + it.effect("skips a deferred connection whose lease was taken before the drain ran", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness({ collectBackgroundTasks: true }); + yield* harness.connect(INTEG_A, "main"); + yield* harness.expireEveryCatalog(); + harness.renameToolsTo("redeploy"); + harness.resolved.length = 0; + + yield* harness.executor.tools.list({ integration: INTEG_A }); + + // Between the enqueue and the drain, another isolate claimed the + // connection and is listing it. The batch must lose the compare-and-set + // and stop there — not dial, not fail, not overwrite. + yield* harness.stealClaim("main"); + yield* harness.drainBackgroundTasks; + + expect(harness.resolved).toEqual([]); + const state = yield* harness.syncStateOf("main"); + expect(state.claimId).toBe("another-isolate"); + const tools = yield* harness.executor.tools.list({ integration: INTEG_A }); + expect(tools.map((tool) => String(tool.name))).toEqual(["alpha_deploy"]); + }), + ), + ); + + it.effect("answers the read even when the host's defer hook dies", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness({ + deferToolSync: () => Effect.die("the host's background queue exploded"), + }); + yield* harness.connect(INTEG_A, "main"); + yield* harness.expireEveryCatalog(); + harness.resolved.length = 0; + + const tools = yield* harness.executor.tools.list({ integration: INTEG_A }); + + // Scheduling is the host's machinery, and a read is not failable by it. + // Nothing was claimed, so the connection is simply still due. + expect(tools.map((tool) => String(tool.name))).toEqual(["alpha_deploy"]); + expect(harness.resolved).toEqual([]); + expect((yield* harness.syncStateOf("main")).claimId).toBeNull(); + }), + ), + ); +}); From d5459f417604d7292fcaa5c0171e36d89785246e Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Sun, 16 Aug 2026 14:54:54 -0600 Subject: [PATCH 2/8] Forward the deferred tool-sync seam through HostConfig --- packages/core/api/src/server/scoped-executor.ts | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/packages/core/api/src/server/scoped-executor.ts b/packages/core/api/src/server/scoped-executor.ts index ea0e33ce61..94a0f0c492 100644 --- a/packages/core/api/src/server/scoped-executor.ts +++ b/packages/core/api/src/server/scoped-executor.ts @@ -97,6 +97,15 @@ export interface HostConfigShape { * Hosts that record product analytics supply it; omitted -> no observation. */ readonly onIntegrationChange?: ExecutorConfig["onIntegrationChange"]; + /** + * Forwarded to `ExecutorConfig.deferToolSync`: how this host runs a tools + * read's speculative catalog refresh after the read has answered (see the sdk + * contract). A host with a background capability supplies it — a Workers + * `waitUntil`, a daemon fiber on a long-lived process. Omitted means the + * batch runs inline, which is the only thing a host with nowhere to put it + * can honestly do. + */ + readonly deferToolSync?: ExecutorConfig["deferToolSync"]; } export class HostConfig extends Context.Service()( @@ -284,6 +293,7 @@ export const makeScopedExecutor = < httpClientLayer, fetch: hostedFetch, onIntegrationChange: config.onIntegrationChange, + deferToolSync: config.deferToolSync, onElicitation: "accept-all", redirectUri, oauthCallbackStateOrgSlug: orgSlug, From 936f95e526fe662b1142fa1833cdce59aa7d6b53 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Sun, 16 Aug 2026 14:54:54 -0600 Subject: [PATCH 3/8] Run deferred tool syncs off the request on local, self-host and cloudflare --- apps/host-cloudflare/src/execution.ts | 8 +++++++- apps/host-selfhost/src/execution.ts | 9 ++++++++- apps/local/src/executor.ts | 5 +++++ 3 files changed, 20 insertions(+), 2 deletions(-) diff --git a/apps/host-cloudflare/src/execution.ts b/apps/host-cloudflare/src/execution.ts index 1ee0e4b1e9..0cdf81176d 100644 --- a/apps/host-cloudflare/src/execution.ts +++ b/apps/host-cloudflare/src/execution.ts @@ -12,7 +12,7 @@ import { } from "@executor-js/api/server"; import { makeDynamicWorkerExecutor } from "@executor-js/runtime-dynamic-worker"; import { makeQuickJsExecutor } from "@executor-js/runtime-quickjs"; -import { env } from "cloudflare:workers"; +import { env, waitUntil } from "cloudflare:workers"; import type { CloudflareConfig } from "./config"; import { makeCloudflarePlugins } from "./plugins"; @@ -56,6 +56,12 @@ export const makeCloudflareHostConfig = (config: CloudflareConfig): Layer.Layer< allowLocalNetwork: config.allowLocalNetwork, webBaseUrl: config.webBaseUrl, oauthCallbackPath: "/api/oauth/callback", + // A Worker, so a tools read's speculative catalog refresh needs `waitUntil` + // to survive the response at all. Unlike cloud's Hyperdrive plane there is + // nothing to drain it before: the storage here is a D1 binding read + // straight off `env`, with no pool and no close finalizer, so a detached + // batch still has everything it needs after the response is written. + deferToolSync: (task) => Effect.sync(() => waitUntil(Effect.runPromise(task))), }); /** diff --git a/apps/host-selfhost/src/execution.ts b/apps/host-selfhost/src/execution.ts index 270ffc4f8e..7763f2125a 100644 --- a/apps/host-selfhost/src/execution.ts +++ b/apps/host-selfhost/src/execution.ts @@ -1,4 +1,4 @@ -import { Layer } from "effect"; +import { Effect, Layer } from "effect"; import { CodeExecutorProvider, @@ -60,6 +60,13 @@ export const SelfHostHostConfig: Layer.Layer = Layer.sync(HostConfig event.kind === "added" ? "integration_added" : "integration_removed", { plugin_key: event.pluginKey }, ), + // Self-host is one long-lived process over one long-lived libSQL handle, so + // a speculative catalog refresh just goes to a detached fiber: the handle + // it needs outlives every request, and there is no isolate to keep alive. + // Detached, not `forkScoped` — the request scope this is called from closes + // as soon as the response is written, which is precisely what the work has + // to outlive. The task is total by contract, so nothing escapes the fork. + deferToolSync: (task) => Effect.asVoid(Effect.forkDetach(task)), }; }); diff --git a/apps/local/src/executor.ts b/apps/local/src/executor.ts index dd91838f35..782324f75d 100644 --- a/apps/local/src/executor.ts +++ b/apps/local/src/executor.ts @@ -201,6 +201,11 @@ const createLocalExecutorLayer = (options: LocalExecutorOptions = {}) => { event.kind === "added" ? "integration_added" : "integration_removed", { plugin_key: event.pluginKey }, ), + // The daemon owns its sqlite handle for its whole lifetime, so a tools + // read's speculative catalog refresh goes to a detached fiber and the + // read answers without waiting on an MCP server's handshake. The task + // is total by contract, so nothing escapes the fork. + deferToolSync: (task) => Effect.asVoid(Effect.forkDetach(task)), onElicitation: "accept-all", oauthEndpointUrlPolicy: { allowHttp: true }, // EXPLICIT OAuth callback — the daemon serves the v2 `/api/oauth/callback` From 2f2c2b0394c20c8a54e4e0df1da78be0d51c8526 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Sun, 16 Aug 2026 14:55:02 -0600 Subject: [PATCH 4/8] Defer MCP session tool syncs to waitUntil and pin the HTTP plane as inline --- apps/cloud/src/api.request-scope.node.test.ts | 54 +++++++++++ .../src/engine/execution-stack-metered.ts | 16 +++- apps/cloud/src/engine/execution-stack.ts | 91 +++++++++++++++---- apps/cloud/src/mcp/session-durable-object.ts | 52 ++++++++++- 4 files changed, 190 insertions(+), 23 deletions(-) diff --git a/apps/cloud/src/api.request-scope.node.test.ts b/apps/cloud/src/api.request-scope.node.test.ts index 1efc5bf864..617cb6bf41 100644 --- a/apps/cloud/src/api.request-scope.node.test.ts +++ b/apps/cloud/src/api.request-scope.node.test.ts @@ -154,6 +154,60 @@ describe("HttpRouter.toWebHandler request scoping", () => { }); }); +// --------------------------------------------------------------------------- +// WHEN the per-request resource dies, relative to the response bytes. +// +// `requestScopedMiddleware` wraps the ROUTE HANDLER in `Effect.scoped`, and the +// response is written later still, by `HttpEffect.toHandled`'s callback. So the +// postgres socket is already closed by the time the client has anything. That +// is why cloud's HTTP plane leaves `ExecutorConfig.deferToolSync` unset (see +// `engine/execution-stack.ts`): there is no window on this plane that is both +// after the response and before the socket closes, so `waitUntil` would only +// ever hand a dead pool to the deferred batch. +// +// Pinned rather than assumed. If the request lifecycle is ever restructured so +// the scope outlives the written response, this test fails and the deferral +// becomes available — which is a thing to be told, not to rediscover. +// --------------------------------------------------------------------------- + +describe("per-request resource lifetime vs the written response", () => { + it("releases the request-scoped resource before the response resolves", async () => { + const order: string[] = []; + const Tracked = Layer.effectDiscard( + Effect.acquireRelease( + Effect.sync(() => { + order.push("acquire"); + }), + () => + Effect.sync(() => { + order.push("release"); + }), + ), + ); + const Route = HttpRouter.add( + "GET", + "/", + Effect.sync(() => { + order.push("handler"); + return HttpServerResponse.jsonUnsafe({ ok: true }); + }), + ); + const handler = HttpRouter.toWebHandler( + Route.pipe( + Layer.provide(requestScopedMiddleware(Tracked).layer), + Layer.provideMerge(HttpServer.layerServices), + ), + { disableLogger: true }, + ).handler; + + const response = await handler(new Request("http://test.local/")); + order.push("response"); + + expect(response.status).toBe(200); + expect(order).toEqual(["acquire", "handler", "release", "response"]); + }); +}); + // --------------------------------------------------------------------------- // Regression test against the prod handler factory. If anyone reverts // `makeApiLive` back to wiring `RequestScopedServicesLive` via diff --git a/apps/cloud/src/engine/execution-stack-metered.ts b/apps/cloud/src/engine/execution-stack-metered.ts index 4ea70b6d43..56e3b6f68c 100644 --- a/apps/cloud/src/engine/execution-stack-metered.ts +++ b/apps/cloud/src/engine/execution-stack-metered.ts @@ -27,7 +27,7 @@ import { import { AutumnService } from "../extensions/billing/service"; import type { DbService } from "../db/db"; -import { CloudExecutionSeamsLayer } from "../engine/execution-stack"; +import { CloudExecutionSeamsLayer, makeCloudExecutionSeamsLayer } from "../engine/execution-stack"; import { makeExecutionLimitGate } from "./execution-gate"; import { makeCloudExecutionRateLimiter } from "./execution-rate-limit"; import { withExecutionUsageTracking } from "./execution-usage"; @@ -75,3 +75,17 @@ export const CloudMeteredExecutionStackLayer: Layer.Layer< never, AutumnService | DbService > = Layer.merge(CloudExecutionSeamsLayer, CloudMeteringEngineDecorator); + +/** + * The same stack over a caller-supplied `HostConfig`. The MCP session DO builds + * one so it can defer tool-catalog refreshes to its own `ctx.waitUntil` — its + * database handle outlives any single request, where the HTTP plane's does not. + * Everything else is identical, decorator included. + */ +export const makeCloudMeteredExecutionStackLayer = ( + hostConfig: Layer.Layer, +): Layer.Layer< + DbProvider | PluginsProvider | HostConfig | CodeExecutorProvider | EngineDecorator, + never, + AutumnService | DbService +> => Layer.merge(makeCloudExecutionSeamsLayer(hostConfig), CloudMeteringEngineDecorator); diff --git a/apps/cloud/src/engine/execution-stack.ts b/apps/cloud/src/engine/execution-stack.ts index 869bf58167..43a67f596f 100644 --- a/apps/cloud/src/engine/execution-stack.ts +++ b/apps/cloud/src/engine/execution-stack.ts @@ -41,6 +41,7 @@ import { HostConfig, PluginsProvider, collectTables, + type HostConfigShape, } from "@executor-js/api/server"; import { makeDynamicWorkerExecutor } from "@executor-js/runtime-dynamic-worker"; import type { AnyPlugin } from "@executor-js/sdk"; @@ -88,18 +89,69 @@ export const CloudPluginsProvider: Layer.Layer = Layer.succeed( */ export const CLOUD_MOUNT_PREFIX = "/api" as const; -export const CloudHostConfig: Layer.Layer = Layer.sync(HostConfig, () => ({ - // SSRF / private-network egress guard. Config-driven, NOT a test flag: - // production leaves `ALLOW_LOCAL_NETWORK` unset so the guard stays ON (`false`); - // the e2e dev-server env opts in with `"true"` so in-scenario fixture - // servers on localhost are reachable. See `hosted-http-client.ts`. - allowLocalNetwork: env.ALLOW_LOCAL_NETWORK === "true", - webBaseUrl: env.VITE_PUBLIC_SITE_URL ?? "https://executor.sh", - oauthCallbackPath: `${CLOUD_MOUNT_PREFIX}/oauth/callback`, - // WorkOS Vault is cloud's credential storage implementation detail, not a - // user-selectable provider surface. - exposeCredentialProviders: false, -})); +/** + * Cloud's host config, parameterized by the ONE seam its two execution planes + * genuinely disagree about: where a deferred tool-catalog refresh runs. + * + * The MCP session Durable Object holds its database handle for the life of the + * session, so it can hand the batch straight to `ctx.waitUntil` and forget it. + * The HTTP plane's handle dies with its request and gets no seam at all (see + * {@link CloudHostConfig}). Nothing else about the config differs, which is why + * this is a parameter rather than a second config module. + */ +export const makeCloudHostConfig = ( + deferToolSync: HostConfigShape["deferToolSync"], +): Layer.Layer => + Layer.sync(HostConfig, () => ({ + // SSRF / private-network egress guard. Config-driven, NOT a test flag: + // production leaves `ALLOW_LOCAL_NETWORK` unset so the guard stays ON (`false`); + // the e2e dev-server env opts in with `"true"` so in-scenario fixture + // servers on localhost are reachable. See `hosted-http-client.ts`. + allowLocalNetwork: env.ALLOW_LOCAL_NETWORK === "true", + webBaseUrl: env.VITE_PUBLIC_SITE_URL ?? "https://executor.sh", + oauthCallbackPath: `${CLOUD_MOUNT_PREFIX}/oauth/callback`, + // WorkOS Vault is cloud's credential storage implementation detail, not a + // user-selectable provider surface. + exposeCredentialProviders: false, + deferToolSync, + })); + +/** + * The HTTP plane's config, and the one seam it CANNOT fill: `deferToolSync` is + * absent, so a tools read over `/api/*` still refreshes an expired catalog + * inline. + * + * Not an omission. The postgres pool a request's executor closes over is + * acquired and released inside `requestScopedMiddleware`'s `Effect.scoped`, + * which wraps the ROUTE HANDLER — so the pool is torn down while the handler's + * `HttpServerResponse` is still a value, strictly before `HttpEffect.toHandled` + * writes it. There is therefore no window on this plane that is both after the + * response and before the socket closes: work scheduled with `waitUntil` finds + * a dead pool, and work drained inside the scope is simply inline work with + * extra steps. `api.request-scope.node.test.ts` pins that ordering, so a future + * change to the request lifecycle will say so rather than making this comment + * quietly wrong. + * + * Holding the pool open past the response instead — deferring the scope close + * into `waitUntil` — is the one thing that would make the window exist, and it + * is the thing `closePostgres` was written to prevent. That finalizer awaits + * the teardown precisely so live-plus-closing sockets stay bounded by what is + * in flight; unbounded, the backlog is the sustained-load cascade recorded in + * `db/db.ts`. Trading a socket leak for a background refresh is not a trade + * this plane should make, and it would make it on EVERY request, not just the + * ones that deferred. + * + * The alternative — giving the batch a short-lived pool of its own via + * `makeDbLayer()` — does not fit either: the task the sdk hands over is already + * bound to the executor that built it, so re-targeting it means rebuilding a + * whole second executor in the background. That is a background scheduler, a + * different feature with a different failure surface, not a wiring detail. + * + * Cloud's high-volume tools read is `tools/list` over MCP, and that plane DOES + * defer: the session Durable Object holds its database handle for the life of + * the session (see `mcp/session-durable-object.ts`). + */ +export const CloudHostConfig: Layer.Layer = makeCloudHostConfig(undefined); export const CloudCodeExecutorProvider: Layer.Layer = Layer.sync( CodeExecutorProvider, @@ -114,13 +166,16 @@ export const CloudCodeExecutorProvider: Layer.Layer = Laye * exported so that overlay builds over the SAME four seams. There is no neutral * no-op-decorator variant anymore: every cloud execution meters. */ +export const makeCloudExecutionSeamsLayer = ( + hostConfig: Layer.Layer, +): Layer.Layer< + DbProvider | PluginsProvider | HostConfig | CodeExecutorProvider, + never, + DbService +> => Layer.mergeAll(CloudDbProvider, CloudPluginsProvider, hostConfig, CloudCodeExecutorProvider); + export const CloudExecutionSeamsLayer: Layer.Layer< DbProvider | PluginsProvider | HostConfig | CodeExecutorProvider, never, DbService -> = Layer.mergeAll( - CloudDbProvider, - CloudPluginsProvider, - CloudHostConfig, - CloudCodeExecutorProvider, -); +> = makeCloudExecutionSeamsLayer(CloudHostConfig); diff --git a/apps/cloud/src/mcp/session-durable-object.ts b/apps/cloud/src/mcp/session-durable-object.ts index 31d6d4bf6b..f03a2ec1c7 100644 --- a/apps/cloud/src/mcp/session-durable-object.ts +++ b/apps/cloud/src/mcp/session-durable-object.ts @@ -46,7 +46,7 @@ import { mcpSessionStub } from "@executor-js/cloudflare/mcp/session-stub"; import { buildExecuteDescription, type ResumeResponse } from "@executor-js/execution"; // The DO meters executions just like the HTTP `/api/*` plane: it builds its -// engine with `CloudMeteredExecutionStackLayer`, so every MCP execution is +// engine with the metered execution stack, so every MCP execution is // tracked to Autumn (the MCP server is the primary execution surface, so leaving // it unmetered silently dropped the bulk of real usage). The billing service // (`AutumnService.Default`) is provided LOCALLY to the metered stack below, so @@ -66,9 +66,9 @@ import { type DrizzleDb, type DbServiceShape, } from "../db/db"; -import { makeExecutionStack } from "../engine/execution-stack"; +import { makeCloudHostConfig, makeExecutionStack } from "../engine/execution-stack"; import { preloadQuickJs } from "../quickjs"; -import { CloudMeteredExecutionStackLayer } from "../engine/execution-stack-metered"; +import { makeCloudMeteredExecutionStackLayer } from "../engine/execution-stack-metered"; import { AutumnService } from "../extensions/billing/service"; import { DoTelemetryLive, flushTracerProvider } from "../observability/telemetry"; import { @@ -236,6 +236,46 @@ export class McpSessionDOSqlite extends McpAgentSessionDOBase): Effect.Effect => + Effect.sync(() => { + this.ctx.waitUntil( + Effect.runPromise( + task.pipe( + Effect.provide(DoTelemetryLive), + Effect.andThen( + Effect.tryPromise({ + try: () => flushTracerProvider(), + catch: () => undefined, + }).pipe(Effect.ignore), + ), + ), + ), + ); + }); + protected override buildMcpServer( sessionMeta: SessionMeta, dbHandle: CloudSessionDbHandle, @@ -261,7 +301,11 @@ export class McpSessionDOSqlite extends McpAgentSessionDOBase Date: Sun, 16 Aug 2026 14:55:02 -0600 Subject: [PATCH 5/8] Cover the deferred refresh against a real MCP server --- .../plugins/mcp/src/sdk/catalog-sync.test.ts | 45 ++++++++++++++++++- 1 file changed, 43 insertions(+), 2 deletions(-) diff --git a/packages/plugins/mcp/src/sdk/catalog-sync.test.ts b/packages/plugins/mcp/src/sdk/catalog-sync.test.ts index 5948485714..227eed679f 100644 --- a/packages/plugins/mcp/src/sdk/catalog-sync.test.ts +++ b/packages/plugins/mcp/src/sdk/catalog-sync.test.ts @@ -39,12 +39,21 @@ const INTEG = IntegrationSlug.make("catalog_mcp"); const CONNECTION = ConnectionName.make("main"); const TEMPLATE = AuthTemplateSlug.make("none"); +const makeCatalogTestConfig = (options?: { readonly collectBackgroundTasks?: boolean }) => + makeTestConfig({ + plugins: [memoryCredentialsPlugin(), mcpPlugin()] as const, + collectBackgroundTasks: options?.collectBackgroundTasks, + }); + const makeCatalogTestExecutor = ( serverUrl: string, - options?: { readonly toolsSyncTtlMs?: number | null }, + options?: { + readonly toolsSyncTtlMs?: number | null; + readonly config?: ReturnType; + }, ) => createExecutor({ - ...makeTestConfig({ plugins: [memoryCredentialsPlugin(), mcpPlugin()] as const }), + ...(options?.config ?? makeCatalogTestConfig()), ...(options?.toolsSyncTtlMs === undefined ? {} : { toolsSyncTtlMs: options.toolsSyncTtlMs }), }).pipe( Effect.tap((executor) => @@ -144,6 +153,38 @@ describe("MCP tool-catalog sync (end-to-end)", () => { }), ); + it.effect("an expired catalog is re-listed in the background when a host can defer", () => + Effect.gen(function* () { + const mutable = makeMutableCatalogMcpServer(); + const server = yield* serveMcpServer(mutable.factory); + // The same instantly-stale executor as the case above, with the one + // difference that makes the read cheap: a host that can run work after + // it has answered. + const config = makeCatalogTestConfig({ collectBackgroundTasks: true }); + const executor = yield* makeCatalogTestExecutor(server.url, { + toolsSyncTtlMs: 0, + config, + }); + + expect(toolNames(yield* executor.tools.list())).toContain(mutable.initialToolName); + const sessionsAfterFirstList = server.sessionCount(); + + mutable.renameTool(); + + // The expired read serves what it has and dials nothing: an old catalog + // is not a wrong one, and nothing has said it drifted. + expect(toolNames(yield* executor.tools.list())).toContain(mutable.initialToolName); + expect(server.sessionCount()).toBe(sessionsAfterFirstList); + + // The work still happens — just not on the read's clock. + yield* config.drainBackgroundTasks; + + const refreshed = toolNames(yield* executor.tools.list()); + expect(refreshed).toContain(mutable.renamedToolName); + expect(refreshed).not.toContain(mutable.initialToolName); + }), + ); + it.effect("a fresh catalog inside the TTL is served from the persisted rows", () => Effect.gen(function* () { const mutable = makeMutableCatalogMcpServer(); From 09546375c0652d0955a0e6c1c7ed41e3c2503dcf Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Sun, 16 Aug 2026 14:55:02 -0600 Subject: [PATCH 6/8] Document the deferred sync counters and span --- .changeset/tool-sync-defer.md | 16 +++++++++ .claude/skills/prod-telemetry/SKILL.md | 46 ++++++++++++++++++++++---- 2 files changed, 56 insertions(+), 6 deletions(-) create mode 100644 .changeset/tool-sync-defer.md diff --git a/.changeset/tool-sync-defer.md b/.changeset/tool-sync-defer.md new file mode 100644 index 0000000000..d830ec9c6c --- /dev/null +++ b/.changeset/tool-sync-defer.md @@ -0,0 +1,16 @@ +--- +"@executor-js/sdk": patch +"@executor-js/api": patch +"@executor-js/local": patch +"@executor-js/host-selfhost": patch +"@executor-js/host-cloudflare": patch +"@executor-js/cloud": patch +--- + +**Fix: an old-but-working tool catalog no longer makes a tools read wait for an upstream handshake** + +A tools read re-lists any connection whose catalog is due, and until now it waited for every one of them before it answered. Most of that waiting was speculative. A catalog goes due for four reasons, and only three of them are somebody telling us it changed: the connection has never synced, a drift signal arrived mid-invocation, or the integration's configuration was revised. The fourth is only the clock — the freshness window elapsed on a connection whose catalog works and which nothing has reported as wrong. Re-verifying it is worth doing soon; it was never worth making a caller wait for, and on a workspace with several MCP servers it was the bulk of what a read paid for. + +Reads now split the two. The three invalidation triggers still re-list inline, because the answer the read is about to give is wrong until they run. An expired catalog is served as it stands and its listing is handed to the host to run once the read has answered, through a new optional `ExecutorConfig.deferToolSync`. A connection that both drifted and expired is an invalidation and stays inline; that falls out of the existing classification rather than needing a rule. The refresh claim is taken inside the deferred listing rather than when it is queued, so a batch that is enqueued and then never runs — an evicted isolate, a dropped background task — leaves no lease behind and the connection is simply offered again on the next read. Each read defers at most sixteen connections, because the background budget on the tightest host is a Cloudflare `waitUntil`; the rest stay due and are picked up by the following read, and both counts are reported on the read's span alongside the existing sync counters. + +A host with no way to outlive its own response leaves `deferToolSync` unset and the batch runs inline, which is the same work in the only order that host can do it in. The local daemon, self-host and the Cloudflare host run it detached; the cloud MCP session, whose database handle lives as long as the session, runs it under the session's `waitUntil`. Cloud's HTTP API plane deliberately does not: its postgres socket is released while the response is still a value, so there is no moment on that plane that is both after the response and before the connection closes. diff --git a/.claude/skills/prod-telemetry/SKILL.md b/.claude/skills/prod-telemetry/SKILL.md index 2a34e140c9..1d14746bfb 100644 --- a/.claude/skills/prod-telemetry/SKILL.md +++ b/.claude/skills/prod-telemetry/SKILL.md @@ -42,9 +42,9 @@ join the same traces via traceparent). `executor.tools.result_count`, and the catalog-refresh counters `executor.tools.sync.candidates` (rows the stale scan returned) / `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. + `.skipped_claimed` / `.skipped_backoff` / `.skipped_parked` / `.deferred` / + `.deferred_overflow`. A slow tools read is almost always `candidates` > 0: + subtract the child span durations to 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 @@ -78,9 +78,43 @@ join the same traces via traceparent). 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`, + `deferred` counts candidates handed to the host to list AFTER the read + answered — `expired` ones only, never an invalidation. It is deliberately not + part of the terminal partition: those listings have no outcome yet at the time + this span closes, and each reports its own on its own + `executor.tools.sync` span later. So on a host that defers, a read with + `deferred` > 0 and `synced` == 0 is the healthy shape, and `candidates` will + still be > 0 on the next read until the background batch lands. On a host that + does NOT defer (cloud's `/api/*` plane, which cannot — its postgres socket + closes before the response is written) the same work reports as `synced` + inline, so comparing the two planes' `synced` rates directly is a mistake. + + `deferred_overflow` is what did not fit in the per-read batch cap (16). A + steady non-zero value means a scope is shedding refresh work every read and + converging slowly — it is never dropped, but the freshness window it is + effectively running at is `ttl x ceil(due / 16)`. There is no log line for + this; the counter is the only signal. + +- `executor.tools.sync.deferred` — one per read that deferred anything, wrapping + the whole background batch, with `executor.tools.sync.deferred` (the batch + size) on it. Its parentage is the HOST's choice, so do not join on it. In + cloud (the MCP session DO) it is a ROOT span in its own trace: the DO + re-provides the tracer so the batch is not hung under a `executor.tools.list` + span that closed before it, which would re-inflate the very read latency the + deferral removes. On a long-lived host that forks a fiber instead, it stays a + child of that list span and simply outlives it. + + When triaging "the catalog is stale and nothing is re-listing", the absence of + this span on a host that reports `deferred` > 0 is the signal — the host + dropped the batch (an evicted isolate, a `waitUntil` that never ran). That + costs one freshness window and nothing else; it is not data loss. Search by + `executor.integration` on the child `executor.tools.sync` spans rather than by + trace, since cloud's batch has no trace in common with the read. + +- `executor.tools.sync` (child of `executor.tools.list` for an inline refresh + and of `executor.tools.sync.deferred` for a background one, 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` (`cold`/`stale_marked`/`config_revised`/`expired`), `executor.tools.sync.claimed` (bool — false means another reader owned the From ad0b3a537ab8b24aa2475672364b5eced100e6ac Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Sun, 16 Aug 2026 15:36:11 -0600 Subject: [PATCH 7/8] let interrupts through the deferred tool sync, and count an inline expired batch --- .claude/skills/prod-telemetry/SKILL.md | 5 +- packages/core/sdk/src/executor.ts | 155 ++++++++++++------ .../core/sdk/src/tools-sync-scope.test.ts | 54 +++++- 3 files changed, 165 insertions(+), 49 deletions(-) diff --git a/.claude/skills/prod-telemetry/SKILL.md b/.claude/skills/prod-telemetry/SKILL.md index 1d14746bfb..410722efcb 100644 --- a/.claude/skills/prod-telemetry/SKILL.md +++ b/.claude/skills/prod-telemetry/SKILL.md @@ -86,8 +86,9 @@ join the same traces via traceparent). `deferred` > 0 and `synced` == 0 is the healthy shape, and `candidates` will still be > 0 on the next read until the background batch lands. On a host that does NOT defer (cloud's `/api/*` plane, which cannot — its postgres socket - closes before the response is written) the same work reports as `synced` - inline, so comparing the two planes' `synced` rates directly is a mistake. + closes before the response is written) the same work runs inline and reports + as `synced`, with `deferred` flat at 0, so comparing the two planes' `synced` + rates directly is a mistake. `deferred_overflow` is what did not fit in the per-read batch cap (16). A steady non-zero value means a scope is shedding refresh work every read and diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index a85af305be..1e06ad6a3b 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -821,8 +821,11 @@ type CatalogRefreshOutcome = ToolProductionOutcome | "fail" | "skipped_claimed"; * has no outcome yet, and reporting one it did not observe is how a background * batch that never ran would read as a fleet of healthy syncs. Each deferred * listing still emits its own `executor.tools.sync` span with its own outcome - * when it eventually runs. `deferredOverflow` is what did not fit in the batch - * cap, so a read that keeps shedding work says so rather than capping quietly. + * when it eventually runs. A host with NO seam defers nothing — it lists the + * same batch inline, before answering — so there it is 0 and those listings sit + * in the terminal counters with every other one. `deferredOverflow` is what did + * not fit in the batch cap on either path, so a read that keeps shedding work + * says so rather than capping quietly. */ interface CatalogRefreshSummary { readonly candidates: number; @@ -2766,11 +2769,15 @@ export const createExecutor = ) => task); + // (`ExecutorConfig.deferToolSync`), or absent on a host with no way to + // outlive its own response — which then does the work before answering, the + // honest behaviour rather than a fallback (see the config doc). + // + // Deliberately NOT collapsed into an identity function: an identity makes + // the two cases indistinguishable to the caller, and the read reported a + // batch it had just run to completion itself as `deferred` work with no + // outcome. The read path branches on it once, in `refreshExpiredBatch`. + const deferToolSync = config.deferToolSync; // The retry ladder's first rung. It follows the freshness window when there // is a usable one, and falls back to the default otherwise: "don't re-dial @@ -4279,6 +4286,21 @@ export const createExecutor = => + Effect.forEach(candidates, refreshOneConnection, { concurrency: 4 }); + /** * The `expired` half of a read's due set, as one task for * `ExecutorConfig.deferToolSync`. @@ -4288,11 +4310,14 @@ export const createExecutor = => - Effect.forEach(batch, refreshOneConnection, { concurrency: 4, discard: true }).pipe( + const deferredSyncTask = (batch: readonly DueConnection[]): Effect.Effect => { + const logBatchFailure = (cause: Cause.Cause) => + Effect.logWarning("executor deferred tool catalog refresh failed", { + errorTags: causeErrorTags(cause), + }); + return refreshDueConnections(batch).pipe( + Effect.asVoid, Effect.withSpan("executor.tools.sync.deferred", { attributes: { "executor.tools.sync.deferred": batch.length }, }), - Effect.catchCause((cause) => - Effect.logWarning("executor deferred tool catalog refresh failed", { - errorTags: causeErrorTags(cause), - }), - ), + // The claim's own write is the one typed channel that survives + // `refreshOneConnection`, and it is caught FIRST so what reaches the + // cause handler below is defects and interrupts only. + Effect.catch((error) => logBatchFailure(Cause.fail(error))), + // catchCauseIf, not catchCause: v4's catchCause is an unfiltered + // handler over EVERY cause, interrupts included, and a caught + // self-interrupt is gone for good. It would swallow exactly the + // re-raise `refreshOneConnection` makes for an abandoned read — and + // because one child's interrupt takes its siblings down with it, the + // batch would report success having aborted fifteen live listings. + // Defects are still swallowed, for the reason above. + Effect.catchCauseIf((cause) => !Cause.hasInterrupts(cause), logBatchFailure), ); + }; + + /** + * The `expired` half of a read's due set, run wherever this host can run it. + * + * The two branches report DIFFERENT things, and that is the point. A seam + * takes the batch away unfinished, so the read has no outcomes to report + * and counts the handover instead (`deferred`). With no seam nothing is + * deferred at all: the batch runs here, on the read's own fiber, before the + * read answers — so its outcomes ARE the read's own and belong in the + * terminal partition like any other inline listing. Reporting those as + * `deferred` had an inline host reading as a fleet that syncs nothing. + * + * The no-seam branch is ordinary inline work in every respect, including a + * storage failure reaching the read: the invalidation half above already + * propagates one, and a claim this read could not write is the read's own + * failure, not a background outage. + */ + const refreshExpiredBatch = ( + batch: readonly DueConnection[], + ): Effect.Effect => + deferToolSync === undefined + ? refreshDueConnections(batch) + : deferToolSync(deferredSyncTask(batch)).pipe( + // Guarded like `onIntegrationChange`: this hook is a host's + // scheduling machinery, and a read must not become failable by it. + // Nothing is lost when it dies — the batch claimed nothing yet, so + // the connections are simply still due. Interrupts pass through for + // the same reason as in the task: a read abandoned mid-handover + // reached no verdict about the host's scheduler. + Effect.catchCauseIf( + (cause) => !Cause.hasInterrupts(cause), + (cause) => + Effect.logWarning("executor deferred tool catalog refresh was not scheduled", { + deferred: batch.length, + errorTags: causeErrorTags(cause), + }), + ), + Effect.as([]), + ); // Rebuild the connections a `tools.list` is about to read whose persisted // tool catalog is stale, scoped to that read's own filter — a read for one @@ -4489,37 +4566,19 @@ export const createExecutor = candidate.state !== "expired"), - refreshOneConnection, - { concurrency: 4 }, ); - // The speculative half, handed over AFTER the invalidation-driven - // listings have run: with no seam configured `deferToolSync` is the - // identity, so this is where the batch executes, and running it before - // the work a read's own correctness depends on would invert the - // priority for exactly the hosts that can least afford it. + // The speculative half, taken up AFTER the invalidation-driven listings + // have run: on a host with no seam this is where the batch executes, + // and running it before the work a read's own correctness depends on + // would invert the priority for exactly the hosts that can least afford + // it. const expired = due.filter((candidate) => candidate.state === "expired"); const batch = expired.slice(0, TOOL_SYNC_DEFERRED_BATCH_MAX); - if (batch.length > 0) { - // Guarded like `onIntegrationChange`: this hook is a host's - // scheduling machinery, and a read must not become failable by it. - // Nothing is lost when it dies — the batch claimed nothing yet, so - // the connections are simply still due. - yield* deferToolSync(deferredSyncTask(batch)).pipe( - Effect.catchCause((cause) => - Effect.logWarning("executor deferred tool catalog refresh was not scheduled", { - deferred: batch.length, - errorTags: causeErrorTags(cause), - }), - ), - ); - } + const expiredOutcomes = batch.length === 0 ? [] : yield* refreshExpiredBatch(batch); + const outcomes = [...inlineOutcomes, ...expiredOutcomes]; return { candidates: connections.length, @@ -4531,7 +4590,11 @@ export const createExecutor = outcome === "skipped_claimed").length, skippedBackoff, skippedParked, - deferred: batch.length, + // Only work this read handed away unfinished. A no-seam host ran the + // same batch inline and its outcomes are in the counters above. + deferred: deferToolSync === undefined ? 0 : batch.length, + // Unchanged by which branch ran it: the cap sheds work either way, + // and a read that keeps shedding says so. deferredOverflow: expired.length - batch.length, }; }); diff --git a/packages/core/sdk/src/tools-sync-scope.test.ts b/packages/core/sdk/src/tools-sync-scope.test.ts index d264d754c1..4d7f5ccf8d 100644 --- a/packages/core/sdk/src/tools-sync-scope.test.ts +++ b/packages/core/sdk/src/tools-sync-scope.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "@effect/vitest"; -import { Duration, Effect, Fiber, Latch } from "effect"; +import { Cause, Duration, Effect, Exit, Fiber, Latch } from "effect"; import { CONNECTION_CATALOG_SCAN_COLUMNS } from "./core-schema"; import { DEFAULT_TOOLS_SYNC_TTL_MS, createExecutor } from "./executor"; @@ -47,6 +47,7 @@ const withLatchTimeout = (effect: Effect.Effect) => const makeCountingPlugin = () => { const resolved: string[] = []; let dying = false; + let cancelling = false; let toolSuffix = "deploy"; let holdUntil: number | null = null; let inFlight = 0; @@ -73,6 +74,7 @@ const makeCountingPlugin = () => { if (holdUntil !== null && inFlight >= holdUntil) gate.openUnsafe(); yield* withLatchTimeout(gate.await); inFlight -= 1; + if (cancelling) return yield* Effect.interrupt; if (dying) return yield* Effect.die("resolveTools blew up"); if (incompleteKind !== undefined) { return { @@ -113,6 +115,11 @@ const makeCountingPlugin = () => { plugin, resolved, startDying: () => void (dying = true), + /** Make every resolve end in CANCELLATION rather than a failure or a + * defect: a listing whose caller hung up mid-handshake. The distinction is + * the whole point — an interrupt is the one cause the refresh must neither + * record a verdict for nor swallow. */ + startCancelling: () => void (cancelling = true), /** Change what the next resolve returns, so a rebuild that never landed is * distinguishable from one that did. Without this every refresh rewrites * the identical row and a persist that silently failed still looks @@ -1020,6 +1027,51 @@ describe("tools read deferred catalog refresh", () => { ), ); + it.effect("gives an interrupted batch back interrupted rather than as a clean drain", () => + Effect.scoped( + Effect.gen(function* () { + const harness = yield* makeHarness({ collectBackgroundTasks: true }); + yield* harness.connect(INTEG_A, "main"); + yield* harness.expireEveryCatalog(); + harness.resolved.length = 0; + + yield* harness.executor.tools.list({ integration: INTEG_A }); + + // The listing is cancelled once the batch is already in flight. + // `refreshOneConnection` re-raises that as an interrupt rather than + // writing a verdict nobody reached, and the batch must carry it out: + // one cancelled connection takes its concurrent siblings down with it, + // so a batch that swallowed the interrupt would report a clean drain + // over fifteen listings it had just aborted. + const before = yield* harness.syncStateOf("main"); + harness.startCancelling(); + const exit = yield* Effect.exit(harness.drainBackgroundTasks); + + const drain = Exit.isSuccess(exit) + ? "reported success" + : Cause.hasInterrupts(exit.cause) + ? "interrupted" + : "failed"; + expect(drain).toBe("interrupted"); + + // And it recorded no verdict on the way past: a cancelled refresh + // reached none, so the ladder stands exactly where the last real + // listing left it. Only the lease moved, and that expires on its own. + const after = yield* harness.syncStateOf("main"); + expect({ + failures: after.failures, + retryAt: after.retryAt, + errorKind: after.errorKind, + }).toEqual({ + failures: before.failures, + retryAt: before.retryAt, + errorKind: before.errorKind, + }); + expect(after.claimId).not.toBeNull(); + }), + ), + ); + it.effect("answers the read even when the host's defer hook dies", () => Effect.scoped( Effect.gen(function* () { From ad87861a16f3e8beb23b8209726729a9450c7cf3 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Sun, 16 Aug 2026 15:36:12 -0600 Subject: [PATCH 8/8] bound the deferred telemetry flush and correct the defer comments --- apps/cloud/src/mcp/session-durable-object.ts | 34 ++++++++++++++++---- apps/host-cloudflare/src/execution.ts | 7 ++++ 2 files changed, 34 insertions(+), 7 deletions(-) diff --git a/apps/cloud/src/mcp/session-durable-object.ts b/apps/cloud/src/mcp/session-durable-object.ts index f03a2ec1c7..c148cb2e31 100644 --- a/apps/cloud/src/mcp/session-durable-object.ts +++ b/apps/cloud/src/mcp/session-durable-object.ts @@ -243,10 +243,20 @@ export class McpSessionDOSqlite extends McpAgentSessionDOBase): Effect.Effect => Effect.sync(() => { @@ -265,11 +279,17 @@ export class McpSessionDOSqlite extends McpAgentSessionDOBase flushTracerProvider(), catch: () => undefined, - }).pipe(Effect.ignore), + }).pipe( + Effect.ignore, + Effect.timeoutOrElse({ + duration: `${TELEMETRY_FLUSH_TIMEOUT_MS} millis`, + orElse: () => Effect.void, + }), + ), ), ), ), diff --git a/apps/host-cloudflare/src/execution.ts b/apps/host-cloudflare/src/execution.ts index 0cdf81176d..c5a6b4b57b 100644 --- a/apps/host-cloudflare/src/execution.ts +++ b/apps/host-cloudflare/src/execution.ts @@ -61,6 +61,13 @@ export const makeCloudflareHostConfig = (config: CloudflareConfig): Layer.Layer< // nothing to drain it before: the storage here is a D1 binding read // straight off `env`, with no pool and no close finalizer, so a detached // batch still has everything it needs after the response is written. + // + // BARE `runPromise`, unlike cloud's DO, which re-provides its tracer here. + // This app has no OTel and no Sentry anywhere — `observability.ts` is a + // console error capture and the request path itself runs on the runtime's + // default no-op tracer — so there is nothing to provide, and the deferred + // batch's spans are dropped exactly like the read's own. Give it the tracer + // the way the DO does the day host-cf gains one. deferToolSync: (task) => Effect.sync(() => waitUntil(Effect.runPromise(task))), });