feat: add opt-in admission hooks for shared provider HTTP (#643)

Add opt-in host admission, pre-dispatch commitment, and nonblocking outcome feedback at the shared guarded provider HTTP seam. Preserve trusted runtime identity, terminal denial, redirect/retry coverage, and egress revalidation while retaining existing credential and connection-store behavior.

Fix cancellation/permit accounting, immediate terminal fencing on dispatch failure, and feedback delaying response consumption. Scope excludes credential revision snapshots, body-completion observation, and SDK transports that bypass shared fetch.

Validation: catalog generation, fix-check, full local test suite, GitHub CI, and package installation/startup/Bun verification passed.

docs/configuration.md

@@ -4,6 +4,9 @@ Managed provider actions can be configured through the [Connector Marketplace](m
 
 OpenConnector is configured with environment variables.
 
+Embedding hosts can also configure opt-in [provider HTTP dispatch hooks](provider-http-dispatch.md)
+through runtime options to admit, delay or deny each upstream HTTP attempt.
+
 | Variable                                    | Default                   | Purpose                                                                                                                                                                     |
 | ------------------------------------------- | ------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
 | `PORT`                                      | `3000`                    | Local HTTP server port.                                                                                                                                                     |

docs/provider-http-dispatch.md

@@ -0,0 +1,129 @@
+# Provider HTTP dispatch hooks
+
+Embedding hosts can opt into admission for individual provider HTTP attempts with
+`createConnectorRuntime({ ..., providerHttpDispatch })` or
+`createConnectApp({ ..., providerHttpDispatch })`. The hook is disabled when the
+option is absent. It applies to shared provider fetchers, including module-level
+fetchers, and survives rewrapping a fetcher with a different egress policy.
+
+```typescript
+const providerHttpDispatch = {
+  async bindAuthority(context) {
+    return { workspaceId: hostWorkspaceId, connectionLineageId: await lineageFor(context.connectionId) };
+  },
+  async beforeAttempt(attempt, signal) {
+    const permit = await arbiter.admit(attempt, signal);
+    if (!permit.allowed) return { allow: false, retryAfterSeconds: permit.retryAfterSeconds };
+    return {
+      allow: true,
+      onDispatch: () => arbiter.commitDispatch(attempt.attemptId),
+      onResult: (result) => arbiter.recordResult(attempt.attemptId, result),
+    };
+  },
+};
+```
+
+The arbiter and identity resolver belong to the host. Open Connector does not
+provide a scheduler, budget store, default rate policy, or durable settlement and
+recovery protocol. Workspace/account mapping, priorities and fairness also remain
+host responsibilities.
+
+## Admission and denial
+
+`beforeAttempt` may wait to delay a request or return `allow: false` to deny it.
+A denial becomes `rate_limited` / HTTP 429, with `Retry-After` when the hook
+provides a nonnegative safe integer `retryAfterSeconds`. Authority-binding,
+admission and dispatch-commit failures fail closed with the same sanitized denial.
+Callback error strings are never exposed to clients.
+
+The runtime retains an invocation-local denial marker, so provider-specific error
+mapping cannot turn admission denial into invalid credentials or an upstream
+failure. Denial is terminal for that invocation: retries, fallbacks and concurrent
+siblings that have not yet dispatched cannot send subsequent requests.
+Independent explicit invocations have independent markers.
+
+**A rejected HTTP attempt does not establish that an entire action is safe to
+retry.** Earlier requests in a multi-request action may already have produced side
+effects. A later denial cannot undo them; the caller must choose a replay policy
+based on the operation's idempotency.
+
+Each attempt has a unique `attemptId`. Each fetch invocation has a `requestId`,
+shared by its redirect hops, with `redirectHop` starting at zero. A provider's
+internal HTTP retry is another fetch invocation and requires another admission.
+Followed redirects are admitted individually with the actual origin and rewritten
+method. A fetch using `redirect: "manual"` admits only its initial HTTP attempt.
+
+Admission runs after the initial URL/DNS guard. After admission and dispatch
+commitment, the guard runs again before egress because queued work may wait long
+enough for DNS answers to change. Existing private-network policy, redirect checks
+and cross-origin credential stripping remain in force.
+
+## Identity
+
+The frozen context contains allowlisted runtime-owned operation, catalog service
+and action, execution ID, and resolved connection ID and stored connection name
+when available. Connection identity is not copied from action input or incoming
+request headers. Credentials, profile metadata, URL paths/queries, request
+headers/bodies and response bodies are absent.
+
+`bindAuthority` receives that context and may resolve host-owned authority. Its
+frozen output copies only `workspaceId`, `connectionLineageId` and `workClass`.
+New-connection validation and initial OAuth exchange may have no established
+connection identity. The request's outer fallback scope has operation `runtime`.
+A host requiring stricter identity must deny unknown contexts.
+
+These IDs identify the resolved connection; they do not attest a credential
+revision or fence authorization changes. Existing credential mutability, OAuth
+refresh/single-flight, connection-store APIs and Trigger ownership behavior are
+preserved when admission is enabled.
+
+## Result feedback
+
+`onDispatch` is awaited before transport starts. A host can persist its dispatch
+commitment there; failure prevents egress. Admission and commitment should honor
+the cancellation signal and use bounded callbacks.
+
+`onResult` starts exactly once for each returned permit, including a late permit
+after queued cancellation. It receives a frozen result:
+
+- `response`: status and the response's `Retry-After` header. This observes response
+  headers, not body completion or completion of an upstream business operation.
+- `transport_error`: transport failed after dispatch started; the request may have
+  reached the provider.
+- `not_dispatched`: cancellation or failed dispatch commitment/revalidation
+  prevented transport from starting.
+
+Feedback is not awaited before returning the response or propagating a transport
+error or cancellation. The exact original Response and body behavior are retained.
+Feedback failure optionally invokes `onFeedbackError(attempt)` and cannot replace
+the transport outcome. The host is responsible for the lifetime and durability of
+its asynchronous feedback, and must not treat a returned Response as confirmation
+that its bookkeeping has settled.
+
+Queued cancellation returns promptly even if admission ignores the signal. A late
+permit receives `not_dispatched`; cancellation after dispatch starts does not
+retract the request. `not_dispatched` does not roll back an already-persisted host
+commitment. Open Connector does not refund budgets or establish completion,
+release a concurrency slot, or select a recovery policy.
+
+## Coverage and standalone use
+
+Coverage is HTTP passing through the shared guarded provider fetch: provider
+requests, content downloads, credential validation, OAuth requests and Trigger
+proxies. SDK transports that bypass shared fetch, including Alibaba OSS SDK
+actions, are outside this hook. Home Assistant WebSockets, MQTT over WebSockets,
+and IMAP/SMTP TCP/TLS also remain outside it. Existing transport guards are
+unchanged. Platform transit storage, remote SaaS/Marketplace execution and host
+control-plane traffic do not acquire local-provider HTTP permits.
+
+Library callers can use `withProviderHttpDispatch(context, run, options)` from
+`src/core/provider-http-dispatch.ts` around standalone provider fetches. Nested
+scopes inherit the hook and replace identity; concurrent scopes remain isolated.
+Use `runWithProviderHttpDispatch` at an asynchronous invocation boundary when the
+called library may catch and remap transport errors: it retains the original
+sanitized admission denial even when the library returns a converted error result.
+
+Hosts should use their own control-plane transport for arbiter RPCs rather than
+recursively calling provider fetchers from a hook. The scope uses
+`AsyncLocalStorage.run/getStore`, available on Node, Bun and Workers with the
+repository's `nodejs_compat` configuration.

src/connection-service.test.ts

@@ -1,7 +1,9 @@
 import type { IConnectionStore, StoredConnection } from "./connection-service.ts";
+import type { ProviderHttpAttempt, ProviderHttpDispatchOptions } from "./core/provider-http-dispatch.ts";
 import type { ActionExecutor, CredentialValidators, ProviderDefinition, ResolvedCredential } from "./core/types.ts";
 import type { MarketplaceService } from "./marketplace/marketplace-service.ts";
 import type { OAuthClientConfig } from "./oauth/oauth-client-config-service.ts";
+import type { IOAuthCredentialRefresher } from "./oauth/oauth-credential-refresh-service.ts";
 import type { IProviderLoader } from "./providers/provider-loader.ts";
 
 import { afterEach, describe, expect, it, vi } from "vitest";
@@ -824,6 +826,87 @@ describe("ConnectionService", () => {
     expect(fetcher).toHaveBeenCalledOnce();
   });
 
+  it("binds OAuth refresh to the resolved stored connection identity", async () => {
+    const store = new MemoryConnectionStore();
+    const oauthClientConfigs = createOAuthClientConfigs([oauthProvider]);
+    const attempts: ProviderHttpAttempt[] = [];
+    const service = createService([oauthProvider], {
+      oauthCredentials: new OAuthCredentialRefreshService(oauthClientConfigs),
+      store,
+      providerHttpDispatch: {
+        beforeAttempt: (attempt) => {
+          attempts.push(attempt);
+          return { allow: true };
+        },
+      },
+    });
+    await oauthClientConfigs.upsertConfig({
+      service: "example",
+      clientId: "client-id",
+      clientSecret: "client-secret",
+    });
+    const original = await store.set("example", "binding-fixture", {
+      authType: "oauth2",
+      accessToken: "expired-token",
+      tokenType: "Bearer",
+      refreshToken: "refresh-token",
+      expiresAt: "2026-01-01T00:00:00.000Z",
+      profile: testProfile,
+      metadata: {},
+    });
+    vi.stubGlobal(
+      "fetch",
+      vi.fn(async () => Response.json({ access_token: "fresh-token", expires_in: 3600, token_type: "Bearer" })),
+    );
+    const target = await service.resolveForExecution("example", undefined, original.id);
+    expect(attempts).toHaveLength(1);
+    expect(attempts[0]?.context).toMatchObject({
+      operation: "oauth",
+      service: "example",
+      connectionId: original.id,
+      connectionName: "binding-fixture",
+    });
+    if (target.kind !== "local") throw new Error("Expected local connection");
+    await expect(target.getCredential("example")).resolves.toMatchObject({ accessToken: "fresh-token" });
+    expect(JSON.stringify(attempts)).not.toMatch(/expired-token|refresh-token|client-secret/);
+  });
+
+  it("keeps boolean-only adapter OAuth refresh working with admission configured", async () => {
+    const memory = new MemoryConnectionStore();
+    const store: IConnectionStore = {
+      get: memory.get.bind(memory),
+      set: memory.set.bind(memory),
+      updateCredential: memory.updateCredential.bind(memory),
+      delete: memory.delete.bind(memory),
+      list: memory.list.bind(memory),
+    };
+    const expired = {
+      authType: "oauth2" as const,
+      accessToken: "expired-token",
+      tokenType: "Bearer",
+      refreshToken: "refresh-token",
+      expiresAt: "2026-01-01T00:00:00.000Z",
+      profile: testProfile,
+      metadata: {},
+    };
+    const refresh = vi.fn(async () => ({ ...expired, accessToken: "fresh-token" }));
+    await store.set("example", "default", expired);
+    await expect(
+      createService([oauthProvider], { store, oauthCredentials: { refresh } }).getCredential("example"),
+    ).resolves.toMatchObject({ accessToken: "fresh-token" });
+    expect(refresh).toHaveBeenCalledOnce();
+    await store.set("example", "default", expired);
+    const guarded = createService([oauthProvider], {
+      store,
+      oauthCredentials: { refresh },
+      providerHttpDispatch: { beforeAttempt: () => ({ allow: true }) },
+    });
+    const target = await guarded.resolveForExecution("example");
+    if (target.kind !== "local") throw new Error("Expected local connection");
+    await expect(target.getCredential("example")).resolves.toMatchObject({ accessToken: "fresh-token" });
+    expect(refresh).toHaveBeenCalledTimes(2);
+  });
+
   it("does not overwrite a connection recreated during OAuth refresh", async () => {
     const store = new MemoryConnectionStore();
     const oauthClientConfigs = createOAuthClientConfigs([oauthProvider]);
@@ -1070,6 +1153,26 @@ describe("ConnectionService", () => {
     });
   });
 
+  it("preserves mutable execution credentials when admission is configured", async () => {
+    const store = new MemoryConnectionStore();
+    const credential = {
+      authType: "api_key" as const,
+      apiKey: "original-key",
+      values: { apiKey: "original-key" },
+      profile: testProfile,
+      metadata: {},
+    };
+    await store.set("uptimerobot", "default", credential);
+    const target = await createService([apiKeyProvider], {
+      store,
+      providerHttpDispatch: { beforeAttempt: () => ({ allow: true }) },
+    }).resolveForExecution("uptimerobot");
+    if (target.kind !== "local") throw new Error("Expected local connection");
+    expect(await target.getCredential("uptimerobot")).toBe(credential);
+    credential.apiKey = "updated-key";
+    await expect(target.getCredential("uptimerobot")).resolves.toMatchObject({ apiKey: "updated-key" });
+  });
+
   it("resolves each service credential once per forConnection scope", async () => {
     const store = new MemoryConnectionStore();
     const service = createService([apiKeyProvider, customCredentialProvider], { store });
@@ -1355,16 +1458,18 @@ describe("ConnectionService disconnect revocation", () => {
 });
 
 interface CreateServiceOptions {
+  providerHttpDispatch?: ProviderHttpDispatchOptions;
   logger?: ReturnType<typeof createTestLogger>;
-  oauthCredentials?: OAuthCredentialRefreshService;
+  oauthCredentials?: IOAuthCredentialRefresher;
   providerLoader?: IProviderLoader;
-  store?: MemoryConnectionStore;
+  store?: IConnectionStore;
 }
 
 function createService(providers: ProviderDefinition[], options: CreateServiceOptions = {}): ConnectionService {
   const catalog = createCatalogStore(providers);
 
   return new ConnectionService({
+    providerHttpDispatch: options.providerHttpDispatch,
     catalog,
     logger: options.logger,
     oauthCredentials: options.oauthCredentials,

src/connection-service.ts

@@ -1,4 +1,5 @@
 import type { CatalogStore, RuntimeProviderDefinition } from "./catalog-store.ts";
+import type { ProviderHttpDispatchOptions } from "./core/provider-http-dispatch.ts";
 import type {
   ApiKeyAuthDefinition,
   AuthType,
@@ -17,7 +18,11 @@ import type { IProviderLoader } from "./providers/provider-loader.ts";
 
 import { normalizeCredentialValues } from "./core/credential-fields.ts";
 import { apiKeyCredentialFields } from "./core/provider-setup.ts";
-import { providerFetch } from "./providers/provider-runtime.ts";
+import {
+  providerFetch,
+  ProviderDispatchRequestError,
+  withProviderHttpDispatchResult,
+} from "./providers/provider-runtime.ts";
 
 export const defaultConnectionName = "default";
 const connectionNamePattern = /^[a-zA-Z0-9][a-zA-Z0-9_-]{0,63}$/;
@@ -59,6 +64,7 @@ export interface ConnectWithoutAuthInput {
 }
 
 export interface ConnectionServiceOptions {
+  providerHttpDispatch?: ProviderHttpDispatchOptions;
   catalog: CatalogStore;
   oauthCredentials?: IOAuthCredentialRefresher;
   providerLoader: IProviderLoader;
@@ -177,6 +183,7 @@ type OAuthCredential = Extract<ResolvedCredential, { authType: "oauth2" }>;
  * run public actions without configuration.
  */
 export class ConnectionService {
+  private readonly providerHttpDispatch?: ProviderHttpDispatchOptions;
   private readonly catalog: CatalogStore;
   private readonly oauthCredentialRefreshes = new Map<string, Promise<OAuthCredential>>();
   private readonly oauthCredentials?: IOAuthCredentialRefresher;
@@ -186,6 +193,7 @@ export class ConnectionService {
   private readonly marketplace?: MarketplaceService;
 
   constructor(input: ConnectionServiceOptions) {
+    this.providerHttpDispatch = input.providerHttpDispatch;
     this.catalog = input.catalog;
     this.oauthCredentials = input.oauthCredentials;
     this.providerLoader = input.providerLoader;
@@ -872,7 +880,11 @@ export class ConnectionService {
     refresher: IOAuthCredentialRefresher,
   ): Promise<OAuthCredential> {
     const { id, revision, service, connectionName } = connection;
-    const nextCredential = await refresher.refresh(service, credential);
+    const nextCredential = await withProviderHttpDispatchResult(
+      { operation: "oauth", service, connectionId: id, connectionName },
+      () => refresher.refresh(service, credential),
+      this.providerHttpDispatch,
+    );
     const updated = await this.store.updateCredential(
       {
         id,
@@ -899,13 +911,19 @@ export class ConnectionService {
   ): Promise<CredentialValidationResult> {
     this.assertNotCancelled(signal);
     try {
-      const result = (await validate()) ?? {};
+      const result =
+        (await withProviderHttpDispatchResult(
+          { operation: "credential_validation", service },
+          validate,
+          this.providerHttpDispatch,
+        )) ?? {};
       this.assertNotCancelled(signal);
       return result;
     } catch (error) {
       if (signal?.aborted) {
         throw cancelledConnectionError();
       }
+      if (error instanceof ProviderDispatchRequestError) throw error;
       throw new ConnectionError(
         "credential_verification_failed",
         error instanceof Error ? error.message : `${service} credential verification failed.`,

src/core/guarded-fetch.ts

@@ -1,3 +1,5 @@
+import type { GuardedHttpDispatcher } from "./provider-http-dispatch.ts";
+
 import { assertPublicHttpUrl, classifyIpAddress, isEgressTrustedHost, isIpAddress, isIpv4Address } from "./request.ts";
 
 /**
@@ -16,6 +18,8 @@ export interface ResolvedAddress {
 export type GuardedFetchDnsLookup = (hostname: string) => Promise<ResolvedAddress[]>;
 
 export interface GuardedFetchOptions {
+  /** Optional dispatcher at the screened raw-transport seam; every redirect hop passes through it. */
+  dispatchAttempt?: GuardedHttpDispatcher;
   /**
    * Base transport issuing the actual requests. Defaults to the global fetch,
    * resolved per call so test stubs installed later still apply.
@@ -138,6 +142,8 @@ const bodyHeaders = ["content-encoding", "content-language", "content-length", "
 
 /** Base transport behind each guarded fetch (undefined = global fetch) so re-wrapping never stacks guards. */
 const guardedFetchBases = new WeakMap<typeof fetch, typeof fetch | undefined>();
+/** Retain admission when a caller replaces the egress policy without stacking guards. */
+const guardedFetchDispatchers = new WeakMap<typeof fetch, GuardedHttpDispatcher | undefined>();
 
 let defaultLookupOverridden = false;
 let defaultLookupOverride: GuardedFetchDnsLookup | null = null;
@@ -199,19 +205,40 @@ export function unwrapGuardedFetch(fetcher: typeof fetch | undefined): typeof fe
  */
 export function createGuardedFetch(options: GuardedFetchOptions = {}): typeof fetch {
   const baseFetch = unwrapGuardedFetch(options.fetch);
+  const dispatchAttempt = options.dispatchAttempt ?? (options.fetch && guardedFetchDispatchers.get(options.fetch));
   const createError = options.createError ?? ((message: string) => new TypeError(message));
   const maxRedirects = options.maxRedirects ?? defaultMaxRedirects;
   const guardedFetch = (async (input: RequestInfo | URL, init?: RequestInit): Promise<Response> => {
     const transport = baseFetch ?? globalThis.fetch;
+    const requestId = dispatchAttempt ? crypto.randomUUID() : "";
+    let redirectHop = 0;
     const fetchTransport = async (
       transportInput: RequestInfo | URL,
       transportInit?: RequestInit,
     ): Promise<Response> => {
-      try {
-        return await transport(transportInput, transportInit);
-      } catch (error) {
-        throw options.mapTransportError?.(error) ?? error;
-      }
+      const send = async (): Promise<Response> => {
+        try {
+          return await transport(transportInput, transportInit);
+        } catch (error) {
+          throw options.mapTransportError?.(error) ?? error;
+        }
+      };
+      const transportRequest = transportInput instanceof Request ? transportInput : undefined;
+      return dispatchAttempt
+        ? dispatchAttempt(
+            Object.freeze({
+              requestId,
+              redirectHop,
+              origin: url.origin,
+              method: (transportInit?.method ?? transportRequest?.method ?? "GET").toUpperCase(),
+            }),
+            transportInit?.signal ?? transportRequest?.signal ?? undefined,
+            send,
+            async () => {
+              await guardHop(url.toString(), "request URL");
+            },
+          )
+        : send();
     };
     const allowPrivateNetwork =
       typeof options.allowPrivateNetwork === "function"
@@ -237,7 +264,11 @@ export function createGuardedFetch(options: GuardedFetchOptions = {}): typeof fe
 
     const redirectMode = init?.redirect ?? request?.redirect ?? "follow";
     if (redirectMode !== "follow") {
-      return fetchTransport(input, init);
+      // Keep the screened URL and method stable while admission is queued. A
+      // caller may otherwise mutate a URL or RequestInit before transport runs.
+      return dispatchAttempt
+        ? fetchTransport(request ?? (input instanceof URL ? new URL(url) : url.toString()), init ? { ...init } : init)
+        : fetchTransport(input, init);
     }
 
     let method = (init?.method ?? request?.method ?? "GET").toUpperCase();
@@ -245,6 +276,7 @@ export function createGuardedFetch(options: GuardedFetchOptions = {}): typeof fe
     let body: BodyInit | null | undefined = init?.body !== undefined ? init.body : request?.body;
 
     for (let redirects = 0; ; redirects++) {
+      redirectHop = redirects;
       const response =
         redirects === 0
           ? request
@@ -305,6 +337,7 @@ export function createGuardedFetch(options: GuardedFetchOptions = {}): typeof fe
   }) as typeof fetch;
 
   guardedFetchBases.set(guardedFetch, baseFetch);
+  guardedFetchDispatchers.set(guardedFetch, dispatchAttempt);
   return guardedFetch;
 }
 

src/core/provider-http-dispatch.test.ts

@@ -0,0 +1,671 @@
+import type { ProviderHttpAttempt, ProviderHttpPermit } from "./provider-http-dispatch.ts";
+
+import { once } from "node:events";
+import { createServer } from "node:http";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { lacunaActionHandlers, skipRetryDelay } from "../providers/lacuna/runtime.ts";
+import {
+  createProviderFetch,
+  providerFetch,
+  toProviderExecutionError,
+  withProviderHttpDispatchResult,
+} from "../providers/provider-runtime.ts";
+import { createGuardedFetch } from "./guarded-fetch.ts";
+import { dispatchProviderHttpAttempt, withProviderHttpDispatch } from "./provider-http-dispatch.ts";
+
+afterEach(() => vi.unstubAllGlobals());
+
+function deferred<T>(): { promise: Promise<T>; resolve(value: T): void } {
+  let resolve: (value: T) => void = () => {};
+  const promise = new Promise<T>((done) => {
+    resolve = done;
+  });
+  return { promise, resolve };
+}
+
+const context = {
+  operation: "action",
+  service: "example",
+  actionId: "example.read",
+  connectionId: "connection-1",
+  connectionName: "binding-1",
+} as const;
+
+describe("provider HTTP dispatch", () => {
+  it("returns the exact response without reading or wrapping its body", async () => {
+    const original = Response.json({ fixture: true });
+    const transport = vi.fn<typeof fetch>().mockResolvedValue(original);
+    const onResult = vi.fn();
+    const response = await withProviderHttpDispatch(
+      context,
+      () => createProviderFetch({ fetch: transport })("https://example.com"),
+      { beforeAttempt: () => ({ allow: true, onResult }) },
+    );
+    expect(response).toBe(original);
+    expect(response.bodyUsed).toBe(false);
+    expect(onResult).toHaveBeenCalledExactlyOnceWith({ kind: "response", status: 200, retryAfter: undefined });
+    expect(await response.json()).toEqual({ fixture: true });
+  });
+
+  it("delivers a native response body before pending feedback can consume its deadline", async () => {
+    const server = createServer((_request, response) => response.end("completed provider result"));
+    server.listen(0, "127.0.0.1");
+    await once(server, "listening");
+    const controller = new AbortController();
+    const feedback = deferred<void>();
+    let deadline: ReturnType<typeof setTimeout> | undefined;
+    let completeFeedback: ReturnType<typeof setTimeout> | undefined;
+    try {
+      const address = server.address();
+      if (!address || typeof address === "string") throw new Error("Missing test listener");
+      const fetcher = createProviderFetch({
+        fetch: async (_url, init) => fetch(`http://127.0.0.1:${address.port}`, init),
+      });
+      const response = await withProviderHttpDispatch(
+        context,
+        () => fetcher("https://example.com", { method: "POST", signal: controller.signal }),
+        {
+          beforeAttempt: () => ({
+            allow: true,
+            onResult: () => {
+              deadline = setTimeout(() => controller.abort(), 500);
+              completeFeedback = setTimeout(() => feedback.resolve(), 600);
+              return feedback.promise;
+            },
+          }),
+        },
+      );
+      expect(await response.text()).toBe("completed provider result");
+    } finally {
+      clearTimeout(deadline);
+      clearTimeout(completeFeedback);
+      feedback.resolve();
+      controller.abort();
+      server.close();
+      server.closeAllConnections();
+    }
+  });
+
+  it("accounts for a permit when admission resolution and cancellation share a microtask turn", async () => {
+    const controller = new AbortController();
+    const ready = deferred<void>();
+    const gate = deferred<ProviderHttpPermit>();
+    const transport = vi.fn(async () => new Response("ok"));
+    const onResult = vi.fn();
+    const target = { requestId: "queued", redirectHop: 0, method: "POST", origin: "https://example.com" };
+    const pending = withProviderHttpDispatch(
+      context,
+      () => dispatchProviderHttpAttempt(target, controller.signal, transport, async () => {}),
+      {
+        beforeAttempt: () => {
+          ready.resolve();
+          return gate.promise;
+        },
+      },
+    );
+    await ready.promise;
+    gate.resolve({ allow: true, onResult });
+    queueMicrotask(() => controller.abort());
+    await expect(pending).rejects.toMatchObject({ name: "AbortError" });
+    expect(transport).not.toHaveBeenCalled();
+    expect(onResult).toHaveBeenCalledExactlyOnceWith({ kind: "not_dispatched", reason: "cancelled" });
+  });
+
+  it.each(["dispatch", "revalidation"])("makes a %s failure terminal before its feedback settles", async (phase) => {
+    const enteredFeedback = deferred<void>();
+    const feedback = deferred<void>();
+    const transport = vi.fn(async () => new Response("side effect completed"));
+    const target = { requestId: "failed", redirectHop: 0, method: "POST", origin: "https://example.com" };
+    const fail = (): never => {
+      throw new Error("Commit or guard failure");
+    };
+    let admissions = 0;
+    const pending = withProviderHttpDispatchResult(
+      context,
+      async () => {
+        const first = dispatchProviderHttpAttempt(target, undefined, transport, async () => {
+          if (phase === "revalidation") fail();
+        }).catch(() => undefined);
+        await enteredFeedback.promise;
+        await dispatchProviderHttpAttempt(
+          { ...target, requestId: "sibling" },
+          undefined,
+          transport,
+          async () => {},
+        ).catch(() => undefined);
+        feedback.resolve();
+        await first;
+      },
+      {
+        beforeAttempt: () => {
+          admissions++;
+          return {
+            allow: true,
+            onDispatch: admissions === 1 && phase === "dispatch" ? fail : undefined,
+            onResult: () => {
+              enteredFeedback.resolve();
+              return feedback.promise;
+            },
+          };
+        },
+      },
+    );
+    await expect(pending).rejects.toMatchObject({ status: 429, code: "rate_limited" });
+    expect(admissions).toBe(1);
+    expect(transport).not.toHaveBeenCalled();
+  });
+
+  it("does not wait for transport-error feedback or its failure observer", async () => {
+    const original = new Error("Original transport failure");
+    const feedback = deferred<void>();
+    const observer = deferred<void>();
+    const onFeedbackError = vi.fn(() => observer.promise);
+    const pending = withProviderHttpDispatch(
+      context,
+      () =>
+        createProviderFetch({
+          fetch: async () => {
+            throw original;
+          },
+        })("https://example.com"),
+      {
+        beforeAttempt: () => ({
+          allow: true,
+          onResult: async () => {
+            await feedback.promise;
+            throw new Error("Bookkeeping failure");
+          },
+        }),
+        onFeedbackError,
+      },
+    );
+    await expect(pending).rejects.toBe(original);
+    expect(onFeedbackError).not.toHaveBeenCalled();
+    feedback.resolve();
+    await vi.waitFor(() => expect(onFeedbackError).toHaveBeenCalledOnce());
+    observer.resolve();
+  });
+
+  it("returns cancellation before feedback for an admitted request has settled", async () => {
+    const controller = new AbortController();
+    const commit = deferred<void>();
+    const enteredCommit = deferred<void>();
+    const feedback = deferred<void>();
+    const onResult = vi.fn(() => feedback.promise);
+    const transport = vi.fn<typeof fetch>();
+    const pending = withProviderHttpDispatch(
+      context,
+      () => createProviderFetch({ fetch: transport })("https://example.com", { signal: controller.signal }),
+      {
+        beforeAttempt: () => ({
+          allow: true,
+          onDispatch: () => {
+            enteredCommit.resolve();
+            return commit.promise;
+          },
+          onResult,
+        }),
+      },
+    );
+    await enteredCommit.promise;
+    controller.abort();
+    await expect(pending).rejects.toMatchObject({ name: "AbortError" });
+    expect(onResult).toHaveBeenCalledExactlyOnceWith({ kind: "not_dispatched", reason: "cancelled" });
+    expect(transport).not.toHaveBeenCalled();
+    feedback.resolve();
+    commit.resolve();
+  });
+
+  it("retains sanitized denial when provider code returns a remapped result", async () => {
+    const transport = vi.fn<typeof fetch>();
+    const pending = withProviderHttpDispatchResult(
+      context,
+      async () => {
+        try {
+          await createProviderFetch({ fetch: transport })("https://example.com");
+          return { ok: true };
+        } catch {
+          return { ok: false, error: { code: "credential_verification_failed", status: 400 } };
+        }
+      },
+      { beforeAttempt: () => ({ allow: false, retryAfterSeconds: 28 }) },
+    );
+    await expect(pending).rejects.toMatchObject({
+      status: 429,
+      code: "rate_limited",
+      details: { retryAfterSeconds: 28 },
+    });
+    expect(transport).not.toHaveBeenCalled();
+  });
+
+  it("does not taint an explicit successful retry in an independent nested invocation", async () => {
+    const response = new Response("ok");
+    const transport = vi.fn<typeof fetch>().mockResolvedValue(response);
+    const fetcher = createProviderFetch({ fetch: transport });
+    let denied = false;
+    const result = await withProviderHttpDispatchResult(
+      context,
+      async () => {
+        await withProviderHttpDispatchResult(context, () => fetcher("https://example.com")).catch(() => undefined);
+        return withProviderHttpDispatchResult(context, () => fetcher("https://example.com"));
+      },
+      {
+        beforeAttempt: () => {
+          if (denied) return { allow: true };
+          denied = true;
+          return { allow: false, retryAfterSeconds: 28 };
+        },
+      },
+    );
+    expect(result).toBe(response);
+    expect(transport).toHaveBeenCalledOnce();
+  });
+
+  it("does not permit later side effects after a caught terminal denial in the same invocation", async () => {
+    const transport = vi.fn<typeof fetch>().mockResolvedValue(new Response("side effect completed"));
+    const fetcher = createProviderFetch({ fetch: transport });
+    const beforeAttempt = vi
+      .fn()
+      .mockReturnValueOnce({ allow: false, retryAfterSeconds: 28 })
+      .mockReturnValue({ allow: true });
+    const pending = withProviderHttpDispatchResult(
+      context,
+      async () => {
+        await fetcher("https://example.com/first").catch(() => undefined);
+        return fetcher("https://example.com/fallback", { method: "POST" }).catch(() => ({ ok: true }));
+      },
+      { beforeAttempt },
+    );
+    await expect(pending).rejects.toMatchObject({ status: 429, details: { retryAfterSeconds: 28 } });
+    expect(beforeAttempt).toHaveBeenCalledOnce();
+    expect(transport).not.toHaveBeenCalled();
+  });
+
+  it.each(["admission", "dispatch", "revalidation"])(
+    "fences a concurrent sibling denied while another attempt waits for %s",
+    async (phase) => {
+      const waiting = deferred<void>();
+      const gate = deferred<void>();
+      const transport = vi.fn(async () => new Response("side effect completed"));
+      const onResult = vi.fn();
+      const onDispatch = vi.fn(async () => {
+        if (phase === "dispatch") {
+          waiting.resolve();
+          await gate.promise;
+        }
+      });
+      const target = { requestId: "waiting", redirectHop: 0, method: "POST", origin: "https://example.com" };
+      const pending = withProviderHttpDispatchResult(
+        context,
+        async () => {
+          const first = dispatchProviderHttpAttempt(target, undefined, transport, async () => {
+            if (phase === "revalidation") {
+              waiting.resolve();
+              await gate.promise;
+            }
+          }).catch(() => undefined);
+          await waiting.promise;
+          await dispatchProviderHttpAttempt(
+            { ...target, requestId: "denied" },
+            undefined,
+            transport,
+            async () => {},
+          ).catch(() => undefined);
+          gate.resolve();
+          await first;
+          return { ok: true };
+        },
+        {
+          beforeAttempt: async (attempt) => {
+            if (attempt.requestId === "denied") return { allow: false, retryAfterSeconds: 28 };
+            if (phase === "admission") {
+              waiting.resolve();
+              await gate.promise;
+            }
+            return { allow: true, onDispatch, onResult };
+          },
+        },
+      );
+      await expect(pending).rejects.toMatchObject({ status: 429, details: { retryAfterSeconds: 28 } });
+      expect(transport).not.toHaveBeenCalled();
+      expect(onDispatch).toHaveBeenCalledTimes(phase === "admission" ? 0 : 1);
+      expect(onResult).toHaveBeenCalledExactlyOnceWith({ kind: "not_dispatched", reason: "dispatch_failed" });
+    },
+  );
+
+  it("preserves unset behavior and passes the original transport response through", async () => {
+    const response = new Response("ok");
+    const transport = vi.fn<typeof fetch>().mockResolvedValue(response);
+    const fetcher = createProviderFetch({ fetch: transport });
+    expect(await fetcher("https://example.com")).toBe(response);
+    expect(transport).toHaveBeenCalledOnce();
+  });
+
+  it("awaits admission before sending and reports the response without consuming it", async () => {
+    const gate = deferred<ProviderHttpPermit>();
+    const response = new Response("private response", {
+      status: 429,
+      headers: { "Retry-After": "37", "Set-Cookie": "secret" },
+    });
+    const transport = vi.fn<typeof fetch>().mockResolvedValue(response);
+    const onDispatch = vi.fn();
+    const onResult = vi.fn();
+    const beforeAttempt = vi.fn(() => gate.promise);
+    const pending = withProviderHttpDispatch(
+      context,
+      () => createProviderFetch({ fetch: transport })("https://example.com"),
+      { beforeAttempt },
+    );
+    await vi.waitFor(() => expect(beforeAttempt).toHaveBeenCalledOnce());
+    expect(transport).not.toHaveBeenCalled();
+    gate.resolve({ allow: true, onDispatch, onResult });
+    expect(await pending).toBe(response);
+    expect(onDispatch).toHaveBeenCalledOnce();
+    expect(onResult).toHaveBeenCalledExactlyOnceWith({ kind: "response", status: 429, retryAfter: "37" });
+    expect(response.bodyUsed).toBe(false);
+  });
+
+  it("maps denial to rate_limited with a Retry-After hint and sends no request", async () => {
+    const transport = vi.fn<typeof fetch>();
+    const failure = await withProviderHttpDispatch(
+      context,
+      () => createProviderFetch({ fetch: transport })("https://example.com"),
+      {
+        beforeAttempt: () => ({ allow: false, retryAfterSeconds: 91 }),
+      },
+    ).catch((error: unknown) => toProviderExecutionError(error, "failed"));
+    expect(failure).toMatchObject({
+      ok: false,
+      error: { code: "rate_limited", details: { status: 429, details: { retryAfterSeconds: 91 } } },
+    });
+    expect(transport).not.toHaveBeenCalled();
+  });
+
+  it.each(["bind", "admission", "dispatch"])("fails closed when the %s hook throws", async (phase) => {
+    const transport = vi.fn<typeof fetch>();
+    const onResult = vi.fn();
+    const fail = (): never => {
+      throw new Error("credential-secret");
+    };
+    const pending = withProviderHttpDispatch(
+      context,
+      () => createProviderFetch({ fetch: transport })("https://example.com"),
+      {
+        bindAuthority: phase === "bind" ? fail : undefined,
+        beforeAttempt:
+          phase === "admission"
+            ? fail
+            : () => ({ allow: true, onDispatch: phase === "dispatch" ? fail : undefined, onResult }),
+      },
+    );
+    await expect(pending).rejects.toMatchObject({ status: 429, code: "rate_limited" });
+    await expect(pending).rejects.not.toThrow("credential-secret");
+    expect(transport).not.toHaveBeenCalled();
+    if (phase === "dispatch")
+      expect(onResult).toHaveBeenCalledExactlyOnceWith({ kind: "not_dispatched", reason: "dispatch_failed" });
+  });
+
+  it("admits every redirect hop with its actual origin and rewritten method", async () => {
+    const transport = vi
+      .fn<typeof fetch>()
+      .mockResolvedValueOnce(
+        new Response(null, { status: 303, headers: { location: "https://other.example/new?token=secret" } }),
+      )
+      .mockResolvedValueOnce(new Response("ok"));
+    const attempts: ProviderHttpAttempt[] = [];
+    const onResult = vi.fn();
+    await withProviderHttpDispatch(
+      context,
+      () =>
+        createProviderFetch({ fetch: transport })("https://example.com/private", { method: "POST", body: "secret" }),
+      {
+        beforeAttempt: (attempt) => {
+          attempts.push(attempt);
+          return { allow: true, onResult };
+        },
+      },
+    );
+    expect(attempts.map(({ origin, method, redirectHop }) => ({ origin, method, redirectHop }))).toEqual([
+      { origin: "https://example.com", method: "POST", redirectHop: 0 },
+      { origin: "https://other.example", method: "GET", redirectHop: 1 },
+    ]);
+    expect(attempts[0]?.requestId).toBe(attempts[1]?.requestId);
+    expect(attempts[0]?.attemptId).not.toBe(attempts[1]?.attemptId);
+    expect(onResult).toHaveBeenCalledTimes(2);
+  });
+
+  it("can deny a redirect hop after the first response", async () => {
+    const transport = vi
+      .fn<typeof fetch>()
+      .mockResolvedValue(new Response(null, { status: 302, headers: { location: "https://other.example" } }));
+    await expect(
+      withProviderHttpDispatch(context, () => createProviderFetch({ fetch: transport })("https://example.com"), {
+        beforeAttempt: (attempt) => ({ allow: attempt.redirectHop === 0 }),
+      }),
+    ).rejects.toMatchObject({ status: 429 });
+    expect(transport).toHaveBeenCalledOnce();
+  });
+
+  it("snapshots a manual request target and method before delayed admission", async () => {
+    const url = new URL("https://example.com/private");
+    const init: RequestInit = { method: "GET", redirect: "manual" };
+    const transport = vi.fn<typeof fetch>().mockResolvedValue(new Response("ok"));
+    await withProviderHttpDispatch(context, () => createProviderFetch({ fetch: transport })(url, init), {
+      beforeAttempt: async () => {
+        url.href = "http://127.0.0.1/private";
+        init.method = "DELETE";
+        return { allow: true };
+      },
+    });
+    expect(transport).toHaveBeenCalledExactlyOnceWith(
+      new URL("https://example.com/private"),
+      expect.objectContaining({ method: "GET" }),
+    );
+  });
+
+  it("admits every real provider-internal retry", async () => {
+    const transport = vi
+      .fn<typeof fetch>()
+      .mockResolvedValueOnce(new Response(null, { status: 429, headers: { "Retry-After": "0" } }))
+      .mockResolvedValueOnce(new Response(null, { status: 503 }))
+      .mockResolvedValueOnce(Response.json({ results: [] }));
+    const beforeAttempt = vi.fn((_attempt: ProviderHttpAttempt) => ({ allow: true as const }));
+    await withProviderHttpDispatch(
+      { ...context, service: "lacuna", actionId: "lacuna.search" },
+      () =>
+        lacunaActionHandlers.search(
+          { query: "test" },
+          { fetcher: createProviderFetch({ fetch: transport }), sleep: skipRetryDelay },
+        ),
+      { beforeAttempt },
+    );
+    expect(beforeAttempt).toHaveBeenCalledTimes(3);
+    expect(transport).toHaveBeenCalledTimes(3);
+    expect(new Set(beforeAttempt.mock.calls.map(([attempt]) => attempt.requestId)).size).toBe(3);
+  });
+
+  it("retains scope through module-level and rewrapped fetchers", async () => {
+    const transport = vi.fn<typeof fetch>().mockResolvedValue(new Response("ok"));
+    vi.stubGlobal("fetch", transport);
+    const rewrapped = createProviderFetch({ fetch: providerFetch, skipDnsValidation: true });
+    const customPolicy = createGuardedFetch({ fetch: providerFetch, skipDnsValidation: true });
+    const beforeAttempt = vi.fn((_attempt: ProviderHttpAttempt) => ({ allow: true as const }));
+    await withProviderHttpDispatch(
+      context,
+      async () => {
+        await providerFetch("https://example.com");
+        await rewrapped("https://example.com");
+        await customPolicy("https://example.com");
+      },
+      { beforeAttempt },
+    );
+    expect(beforeAttempt).toHaveBeenCalledTimes(3);
+    expect(transport).toHaveBeenCalledTimes(3);
+    expect(beforeAttempt.mock.calls.map(([attempt]) => attempt.context.connectionName)).toEqual([
+      "binding-1",
+      "binding-1",
+      "binding-1",
+    ]);
+  });
+
+  it("rejects blocked initial and redirect URLs before admission", async () => {
+    const beforeAttempt = vi.fn(() => ({ allow: true as const }));
+    const transport = vi
+      .fn<typeof fetch>()
+      .mockResolvedValue(new Response(null, { status: 302, headers: { location: "http://169.254.169.254/latest" } }));
+    await withProviderHttpDispatch(
+      context,
+      async () => {
+        await expect(createProviderFetch({ fetch: transport })("http://127.0.0.1")).rejects.toThrow();
+        expect(beforeAttempt).not.toHaveBeenCalled();
+        await expect(createProviderFetch({ fetch: transport })("https://example.com")).rejects.toThrow();
+      },
+      { beforeAttempt },
+    );
+    expect(beforeAttempt).toHaveBeenCalledOnce();
+    expect(transport).toHaveBeenCalledOnce();
+  });
+
+  it("revalidates DNS after delayed admission", async () => {
+    let address = "93.184.216.34";
+    const transport = vi.fn<typeof fetch>();
+    const onResult = vi.fn();
+    const fetcher = createGuardedFetch({
+      fetch: transport,
+      lookup: async () => [{ address, family: 4 }],
+      dispatchAttempt: async (...args) => {
+        const { dispatchProviderHttpAttempt } = await import("./provider-http-dispatch.ts");
+        return dispatchProviderHttpAttempt(...args);
+      },
+    });
+    await expect(
+      withProviderHttpDispatch(context, () => fetcher("https://example.com"), {
+        beforeAttempt: () => {
+          address = "127.0.0.1";
+          return { allow: true, onResult };
+        },
+      }),
+    ).rejects.toThrow();
+    expect(transport).not.toHaveBeenCalled();
+    expect(onResult).toHaveBeenCalledExactlyOnceWith({ kind: "not_dispatched", reason: "dispatch_failed" });
+  });
+
+  it("cancels a queued attempt promptly and accounts for a late permit", async () => {
+    const controller = new AbortController();
+    const gate = deferred<ProviderHttpPermit>();
+    const transport = vi.fn<typeof fetch>();
+    const onResult = vi.fn();
+    const beforeAttempt = vi.fn(() => gate.promise);
+    const pending = withProviderHttpDispatch(
+      context,
+      () => createProviderFetch({ fetch: transport })("https://example.com", { signal: controller.signal }),
+      { beforeAttempt },
+    );
+    await vi.waitFor(() => expect(beforeAttempt).toHaveBeenCalledOnce());
+    controller.abort();
+    await expect(pending).rejects.toMatchObject({ name: "AbortError" });
+    gate.resolve({ allow: true, onResult });
+    await vi.waitFor(() =>
+      expect(onResult).toHaveBeenCalledExactlyOnceWith({ kind: "not_dispatched", reason: "cancelled" }),
+    );
+    expect(transport).not.toHaveBeenCalled();
+  });
+
+  it("reports an uncertain transport failure without leaking its raw error", async () => {
+    const transport = vi.fn<typeof fetch>().mockRejectedValue(new Error("secret-url-and-headers"));
+    const onResult = vi.fn();
+    await expect(
+      withProviderHttpDispatch(context, () => createProviderFetch({ fetch: transport })("https://example.com"), {
+        beforeAttempt: () => ({ allow: true, onResult }),
+      }),
+    ).rejects.toThrow();
+    expect(onResult).toHaveBeenCalledExactlyOnceWith({ kind: "transport_error" });
+  });
+
+  it("preserves a successful response when feedback fails", async () => {
+    const response = new Response("ok");
+    const transport = vi.fn<typeof fetch>().mockResolvedValue(response);
+    const onFeedbackError = vi.fn(() => {
+      throw new Error("observer failed");
+    });
+    expect(
+      await withProviderHttpDispatch(context, () => createProviderFetch({ fetch: transport })("https://example.com"), {
+        beforeAttempt: () => ({
+          allow: true,
+          onResult: () => {
+            throw new Error("durability failed");
+          },
+        }),
+        onFeedbackError,
+      }),
+    ).toBe(response);
+    expect(onFeedbackError).toHaveBeenCalledOnce();
+    expect(transport).toHaveBeenCalledOnce();
+  });
+
+  it("freezes allowlisted context and authority without exposing credential-bearing request data", async () => {
+    const attempts: ProviderHttpAttempt[] = [];
+    const inputContext = { ...context, connectionName: "binding-1", secret: "context-secret" };
+    const authority = { workspaceId: "workspace-1", connectionLineageId: "lineage-1", token: "authority-secret" };
+    const transport = vi.fn<typeof fetch>().mockResolvedValue(new Response("ok"));
+    await withProviderHttpDispatch(
+      inputContext,
+      () =>
+        createProviderFetch({ fetch: transport })(
+          "https://example.com/private-secret?token=query-secret#fragment-secret",
+          { headers: { authorization: "header-secret" }, method: "POST", body: "body-secret" },
+        ),
+      {
+        bindAuthority: () => authority,
+        beforeAttempt: (attempt) => {
+          attempts.push(attempt);
+          return { allow: true };
+        },
+      },
+    );
+    const attempt = attempts[0]!;
+    expect(JSON.stringify(attempt)).not.toContain("secret");
+    expect(Object.isFrozen(attempt)).toBe(true);
+    expect(Object.isFrozen(attempt.context)).toBe(true);
+    expect(Object.isFrozen(attempt.authority)).toBe(true);
+    inputContext.connectionName = "replaced-binding";
+    expect(attempt.context.connectionName).toBe("binding-1");
+    expect(Reflect.set(attempt.context, "connectionName", "forged-binding")).toBe(false);
+    authority.workspaceId = "changed";
+    expect(attempt.authority.workspaceId).toBe("workspace-1");
+  });
+
+  it("isolates concurrent work and restores its parent scope", async () => {
+    const gate = deferred<void>();
+    const attempts: ProviderHttpAttempt[] = [];
+    const fetcher = createProviderFetch({
+      fetch: vi.fn<typeof fetch>().mockImplementation(async () => new Response("ok")),
+    });
+    const options = {
+      beforeAttempt: (attempt: ProviderHttpAttempt) => {
+        attempts.push(attempt);
+        return { allow: true as const };
+      },
+    };
+    await withProviderHttpDispatch(
+      context,
+      async () => {
+        await Promise.all([
+          withProviderHttpDispatch({ ...context, connectionId: "first" }, async () => {
+            await gate.promise;
+            await fetcher("https://example.com");
+          }),
+          withProviderHttpDispatch({ ...context, connectionId: "second" }, async () => {
+            await fetcher("https://example.com");
+            gate.resolve();
+          }),
+        ]);
+        await fetcher("https://example.com");
+      },
+      options,
+    );
+    expect(attempts.map((attempt) => attempt.context.connectionId)).toEqual(["second", "first", "connection-1"]);
+    await fetcher("https://example.com");
+    expect(attempts).toHaveLength(3);
+  });
+});

src/core/provider-http-dispatch.ts

@@ -0,0 +1,276 @@
+import { AsyncLocalStorage } from "node:async_hooks";
+
+/** Runtime-owned identity. Never populate these fields from action input or HTTP headers. */
+export interface ProviderDispatchContext {
+  readonly operation: "runtime" | "action" | "proxy" | "trigger" | "credential_validation" | "oauth";
+  readonly service?: string;
+  readonly actionId?: string;
+  readonly executionId?: string;
+  readonly connectionId?: string;
+  /** Alias from the resolved stored connection, not the caller's requested alias. */
+  readonly connectionName?: string;
+}
+
+/** Opaque authority supplied by the embedding host, independently of provider credentials. */
+export interface ProviderDispatchAuthority {
+  readonly workspaceId?: string;
+  readonly connectionLineageId?: string;
+  readonly workClass?: string;
+}
+
+/** A screened HTTP hop. URLs, paths, queries, headers and bodies are deliberately absent. */
+export interface ProviderHttpAttempt {
+  readonly attemptId: string;
+  readonly requestId: string;
+  readonly redirectHop: number;
+  readonly method: string;
+  readonly origin: string;
+  readonly context: ProviderDispatchContext;
+  readonly authority: ProviderDispatchAuthority;
+}
+
+/** Transport failures may have reached the provider; they are never evidence for a credit refund. */
+export type ProviderHttpAttemptResult =
+  | { readonly kind: "response"; readonly status: number; readonly retryAfter?: string }
+  | { readonly kind: "transport_error" }
+  | { readonly kind: "not_dispatched"; readonly reason: "cancelled" | "dispatch_failed" };
+
+export interface ProviderHttpPermit {
+  readonly allow: true;
+  /** Awaited immediately before transport. Persist dispatch commitment here; failures deny egress. */
+  readonly onDispatch?: () => void | Promise<void>;
+  /** Started exactly once after a permit is returned; never awaited before returning the transport outcome. */
+  readonly onResult?: (result: ProviderHttpAttemptResult) => void | Promise<void>;
+}
+
+export interface ProviderHttpDenial {
+  readonly allow: false;
+  /** Retry-After delay in whole seconds. */
+  readonly retryAfterSeconds?: number;
+}
+
+export interface ProviderHttpDispatchOptions {
+  /** Await admission to delay a request, or return a denial. Honour the signal while queued. */
+  readonly beforeAttempt: (
+    attempt: ProviderHttpAttempt,
+    signal: AbortSignal | undefined,
+  ) => ProviderHttpPermit | ProviderHttpDenial | Promise<ProviderHttpPermit | ProviderHttpDenial>;
+  /** Resolve workspace/lineage authority from runtime identity, never request credentials. */
+  readonly bindAuthority?: (
+    context: ProviderDispatchContext,
+  ) => ProviderDispatchAuthority | Promise<ProviderDispatchAuthority>;
+  /** Feedback failure leaves the attempt spent/unknown. This observer cannot change the transport result. */
+  readonly onFeedbackError?: (attempt: ProviderHttpAttempt) => void | Promise<void>;
+}
+
+interface DispatchScope {
+  options: ProviderHttpDispatchOptions;
+  context: ProviderDispatchContext;
+  receipt: { denial?: ProviderHttpDispatchError };
+}
+
+export interface GuardedHttpAttempt {
+  readonly requestId: string;
+  readonly redirectHop: number;
+  readonly method: string;
+  readonly origin: string;
+}
+
+export type GuardedHttpDispatcher = (
+  attempt: GuardedHttpAttempt,
+  signal: AbortSignal | undefined,
+  transport: () => Promise<Response>,
+  revalidate: () => Promise<void>,
+) => Promise<Response>;
+
+const scopes = new AsyncLocalStorage<DispatchScope>();
+
+/** Safe retryable denial, including hook/readiness failures; it contains no callback error details. */
+export class ProviderHttpDispatchError extends Error {
+  readonly retryAfterSeconds?: number;
+
+  constructor(retryAfterSeconds?: number) {
+    super("Provider HTTP dispatch is temporarily unavailable.");
+    this.retryAfterSeconds = retryAfterSeconds;
+  }
+}
+
+/**
+ * Bind a hook to asynchronous provider work. Nested scopes inherit the configured hook;
+ * concurrent requests stay isolated. Standalone providerFetch users can opt in here too.
+ */
+export function withProviderHttpDispatch<T>(
+  context: ProviderDispatchContext,
+  run: () => T,
+  options?: ProviderHttpDispatchOptions,
+): T {
+  const parent = scopes.getStore();
+  const configured = options ?? parent?.options;
+  if (!configured) return run();
+  // Copy an explicit allowlist so accidental credential-bearing extra fields cannot escape.
+  const snapshot: ProviderDispatchContext = Object.freeze({
+    operation: context.operation,
+    service: context.service,
+    actionId: context.actionId,
+    executionId: context.executionId,
+    connectionId: context.connectionId,
+    connectionName: context.connectionName,
+  });
+  return scopes.run(
+    { options: configured, context: snapshot, receipt: parent?.options === configured ? parent.receipt : {} },
+    run,
+  );
+}
+
+/**
+ * A provider invocation boundary: retain the original denial even if provider code catches,
+ * replaces, or converts it to a result. Each invocation owns its marker, so a sibling's
+ * denial cannot override another invocation's successful retry or fallback.
+ */
+export async function runWithProviderHttpDispatch<T>(
+  context: ProviderDispatchContext,
+  run: () => T | Promise<T>,
+  options?: ProviderHttpDispatchOptions,
+): Promise<T> {
+  return withProviderHttpDispatch(
+    context,
+    async () => {
+      const inherited = scopes.getStore();
+      if (!inherited) return run();
+      const scope: DispatchScope = { ...inherited, receipt: {} };
+      return scopes.run(scope, async () => {
+        try {
+          const result = await run();
+          if (scope.receipt.denial) throw scope.receipt.denial;
+          return result;
+        } catch (error) {
+          throw scope.receipt.denial ?? error;
+        }
+      });
+    },
+    options,
+  );
+}
+
+function deny(scope: DispatchScope, retryAfterSeconds?: number): ProviderHttpDispatchError {
+  return (scope.receipt.denial ??= Object.freeze(new ProviderHttpDispatchError(retryAfterSeconds)));
+}
+
+/** Called only at the screened raw-transport seam, once for every HTTP attempt and redirect hop. */
+export const dispatchProviderHttpAttempt: GuardedHttpDispatcher = async (target, signal, transport, revalidate) => {
+  const scope = scopes.getStore();
+  if (!scope) return transport();
+  signal?.throwIfAborted();
+  // Denial is terminal for this invocation. A provider may catch it, but must not complete
+  // later side effects that the boundary would then misreport as a retryable denial.
+  if (scope.receipt.denial) throw scope.receipt.denial;
+  let permit: ProviderHttpPermit | ProviderHttpDenial;
+  let attempt: ProviderHttpAttempt;
+  try {
+    const authority = await waitForSignal(Promise.resolve(scope.options.bindAuthority?.(scope.context)), signal);
+    signal?.throwIfAborted();
+    if (scope.receipt.denial) throw scope.receipt.denial;
+    attempt = Object.freeze({
+      attemptId: crypto.randomUUID(),
+      requestId: target.requestId,
+      redirectHop: target.redirectHop,
+      method: target.method,
+      origin: target.origin,
+      context: scope.context,
+      authority: Object.freeze({
+        workspaceId: authority?.workspaceId,
+        connectionLineageId: authority?.connectionLineageId,
+        workClass: authority?.workClass,
+      }),
+    });
+    permit = await waitForSignal(Promise.resolve(scope.options.beforeAttempt(attempt, signal)), signal, (late) => {
+      if (late?.allow)
+        void reportFeedback(scope.options, attempt, late, { kind: "not_dispatched", reason: "cancelled" });
+    });
+    if (!permit || typeof permit.allow !== "boolean") throw new ProviderHttpDispatchError();
+  } catch {
+    signal?.throwIfAborted();
+    throw deny(scope);
+  }
+  if (!permit.allow) {
+    const seconds = permit.retryAfterSeconds;
+    throw deny(scope, Number.isSafeInteger(seconds) && seconds! >= 0 ? seconds : undefined);
+  }
+  const admitted = permit;
+  const feedback = (result: ProviderHttpAttemptResult): Promise<void> =>
+    reportFeedback(scope.options, attempt, admitted, result);
+  try {
+    signal?.throwIfAborted();
+    if (scope.receipt.denial) throw scope.receipt.denial;
+    await waitForSignal(Promise.resolve(admitted.onDispatch?.()), signal);
+    if (scope.receipt.denial) throw scope.receipt.denial;
+    // Admission can wait arbitrarily long. Re-screen DNS/URL before issuing the admitted hop.
+    await waitForSignal(revalidate(), signal);
+    signal?.throwIfAborted();
+    if (scope.receipt.denial) throw scope.receipt.denial;
+  } catch {
+    const denial = signal?.aborted ? undefined : deny(scope);
+    void feedback({ kind: "not_dispatched", reason: signal?.aborted ? "cancelled" : "dispatch_failed" });
+    signal?.throwIfAborted();
+    throw denial;
+  }
+  let response: Response;
+  try {
+    response = await transport();
+  } catch (error) {
+    void feedback({ kind: "transport_error" });
+    throw error;
+  }
+  // Report every HTTP status, including redirect responses. No response body is read or exposed.
+  void feedback({
+    kind: "response",
+    status: response.status,
+    retryAfter: response.headers.get("retry-after") ?? undefined,
+  });
+  return response;
+};
+
+async function reportFeedback(
+  options: ProviderHttpDispatchOptions,
+  attempt: ProviderHttpAttempt,
+  permit: ProviderHttpPermit,
+  result: ProviderHttpAttemptResult,
+): Promise<void> {
+  try {
+    await permit.onResult?.(Object.freeze(result));
+  } catch {
+    try {
+      await options.onFeedbackError?.(attempt);
+    } catch {
+      // Feedback cannot turn an already-issued request into a failure that a caller might replay.
+    }
+  }
+}
+
+function waitForSignal<T>(
+  pending: Promise<T>,
+  signal: AbortSignal | undefined,
+  onLate?: (value: T) => void,
+): Promise<T> {
+  if (!signal) return pending;
+  return new Promise<T>((resolve, reject) => {
+    let cancelled = false;
+    const abort = (): void => {
+      cancelled = true;
+      reject(signal.reason);
+    };
+    if (signal.aborted) abort();
+    else signal.addEventListener("abort", abort, { once: true });
+    void pending.then(
+      (value) => {
+        signal.removeEventListener("abort", abort);
+        if (cancelled) onLate?.(value);
+        else resolve(value);
+      },
+      (error: unknown) => {
+        signal.removeEventListener("abort", abort);
+        reject(error);
+      },
+    );
+  });
+}

src/oauth/oauth-flow-service.ts

@@ -17,7 +17,11 @@ import type { OAuthTokenResult } from "./oauth-token.ts";
 
 import { createHash, randomBytes } from "node:crypto";
 import { ConnectionError } from "../connection-service.ts";
-import { providerFetch } from "../providers/provider-runtime.ts";
+import {
+  providerFetch,
+  ProviderDispatchRequestError,
+  withProviderHttpDispatchResult,
+} from "../providers/provider-runtime.ts";
 import { requestAuthorizationCodeToken } from "./oauth-token.ts";
 
 /**
@@ -251,7 +255,9 @@ export class OAuthFlowService {
 
     const providerOAuth = await this.providerLoader.loadProviderOAuthRuntime?.(service);
     const resolvedAuthorizationUrl = providerOAuth?.buildAuthorizationUrl
-      ? await providerOAuth.buildAuthorizationUrl({ authorizationUrl, clientConfig: config, now })
+      ? await withProviderHttpDispatchResult({ operation: "oauth", service }, () =>
+          providerOAuth.buildAuthorizationUrl!({ authorizationUrl, clientConfig: config, now }),
+        )
       : authorizationUrl.toString();
 
     return {
@@ -288,34 +294,47 @@ export class OAuthFlowService {
         new OAuthFlowError("oauth_token_exchange_failed", message);
       const providerOAuth = await this.providerLoader.loadProviderOAuthRuntime?.(pending.service);
       let tokenResponse: OAuthTokenResult;
-      if (providerOAuth?.exchangeCode) {
-        tokenResponse = await providerOAuth.exchangeCode({
-          code: input.code,
-          callbackParameters: input.callbackParameters,
-          clientConfig: config,
-          redirectUri,
-          tokenUrl,
-          fetcher: providerFetch,
-          signal: input.signal,
-          createError,
-        });
-      } else {
-        tokenResponse = await requestAuthorizationCodeToken({
-          code: input.code,
-          state: pending.state,
-          clientId: config.clientId,
-          clientSecret: config.clientSecret,
-          redirectUri,
-          responseEnvelope: auth.tokenResponseEnvelope,
-          tokenRequestFields: auth.tokenRequestFields,
-          tokenEndpointAuthMethod: auth.tokenEndpointAuthMethod,
-          tokenRequestFormat: auth.tokenRequestFormat,
-          tokenUrl,
-          extraFields: createTokenExtraFields(pending, auth.tokenRequestCallbackParameters, input.callbackParameters),
-          signal: input.signal,
-          createError,
-        });
-      }
+      tokenResponse = await withProviderHttpDispatchResult(
+        {
+          operation: "oauth",
+          service: pending.service,
+          connectionId: request?.target?.id,
+        },
+        async () => {
+          if (providerOAuth?.exchangeCode) {
+            return providerOAuth.exchangeCode({
+              code: input.code,
+              callbackParameters: input.callbackParameters,
+              clientConfig: config,
+              redirectUri,
+              tokenUrl,
+              fetcher: providerFetch,
+              signal: input.signal,
+              createError,
+            });
+          } else {
+            return requestAuthorizationCodeToken({
+              code: input.code,
+              state: pending.state,
+              clientId: config.clientId,
+              clientSecret: config.clientSecret,
+              redirectUri,
+              responseEnvelope: auth.tokenResponseEnvelope,
+              tokenRequestFields: auth.tokenRequestFields,
+              tokenEndpointAuthMethod: auth.tokenEndpointAuthMethod,
+              tokenRequestFormat: auth.tokenRequestFormat,
+              tokenUrl,
+              extraFields: createTokenExtraFields(
+                pending,
+                auth.tokenRequestCallbackParameters,
+                input.callbackParameters,
+              ),
+              signal: input.signal,
+              createError,
+            });
+          }
+        },
+      );
       const refreshParameters = readCallbackParameters(auth.tokenRequestCallbackParameters, input.callbackParameters);
       const providerSecret = mergeOAuthProviderSecret(tokenResponse.providerSecret, refreshParameters);
       const oauthCredential = {
@@ -376,6 +395,10 @@ export class OAuthFlowService {
         ...(request?.returnUri ? { returnUri: callbackReturnUri(request, "success") } : {}),
       };
     } catch (error) {
+      if (error instanceof ProviderDispatchRequestError) {
+        if (request) await this.requests.fail(request.connectionRequestId, "rate_limited", error.message);
+        throw error;
+      }
       if (request) {
         const code =
           error instanceof OAuthFlowError &&

src/oauth/oauth-token.ts

@@ -8,6 +8,7 @@ import {
   createProviderTimeout,
   isAbortLikeError,
   providerFetch,
+  ProviderDispatchRequestError,
   providerUserAgent,
 } from "../providers/provider-runtime.ts";
 
@@ -258,6 +259,7 @@ async function requestToken(input: TokenRequest): Promise<OAuthTokenResult> {
     });
   } catch (error) {
     timeout.cleanup();
+    if (error instanceof ProviderDispatchRequestError) throw error;
     if (input.signal?.aborted) {
       throw input.createError("OAuth token request was cancelled.");
     }

src/providers/provider-runtime.ts

@@ -1,3 +1,4 @@
+import type { ProviderDispatchContext, ProviderHttpDispatchOptions } from "../core/provider-http-dispatch.ts";
 import type {
   ActionExecutor,
   ExecutionContext,
@@ -23,6 +24,11 @@ import {
   requiredString,
 } from "../core/cast.ts";
 import { createGuardedFetch } from "../core/guarded-fetch.ts";
+import {
+  dispatchProviderHttpAttempt,
+  ProviderHttpDispatchError,
+  runWithProviderHttpDispatch,
+} from "../core/provider-http-dispatch.ts";
 import { readBoundedResponseBytes } from "../core/request.ts";
 
 /**
@@ -53,6 +59,14 @@ export interface ProviderFetchOptions {
  */
 export function createProviderFetch(options: ProviderFetchOptions = {}): ProviderFetch {
   return createGuardedFetch({
+    dispatchAttempt: async (attempt, signal, transport, revalidate) => {
+      try {
+        return await dispatchProviderHttpAttempt(attempt, signal, transport, revalidate);
+      } catch (error) {
+        if (error instanceof ProviderHttpDispatchError) throw new ProviderDispatchRequestError(error.retryAfterSeconds);
+        throw error;
+      }
+    },
     fetch: options.fetch,
     allowPrivateNetwork: options.allowPrivateNetwork,
     skipDnsValidation: options.skipDnsValidation,
@@ -78,6 +92,20 @@ export function createProviderFetch(options: ProviderFetchOptions = {}): Provide
  */
 export const providerFetch: ProviderFetch = createProviderFetch();
 
+/** Preserve admission denials at the shared runtime boundary despite provider-specific error mapping. */
+export async function withProviderHttpDispatchResult<T>(
+  context: ProviderDispatchContext,
+  run: () => T | Promise<T>,
+  options?: ProviderHttpDispatchOptions,
+): Promise<T> {
+  try {
+    return await runWithProviderHttpDispatch(context, run, options);
+  } catch (error) {
+    if (error instanceof ProviderHttpDispatchError) throw new ProviderDispatchRequestError(error.retryAfterSeconds);
+    throw error;
+  }
+}
+
 /**
  * Default User-Agent sent by local provider executors.
  */
@@ -274,6 +302,13 @@ export class ProviderRequestError extends Error {
   }
 }
 
+/** A dispatch denial is retryable and must not be wrapped as a bad provider credential. */
+export class ProviderDispatchRequestError extends ProviderRequestError {
+  constructor(retryAfterSeconds?: number) {
+    super(429, "Provider HTTP dispatch is temporarily unavailable.", { retryAfterSeconds }, "rate_limited");
+  }
+}
+
 /**
  * Return the 400 error providers throw for invalid action input or credentials.
  * The message is surfaced verbatim as the `invalid_input` execution error.

src/server/actions/action-runner.ts

@@ -1,6 +1,7 @@
 import type { CatalogStore } from "../../catalog-store.ts";
 import type { ConnectionService, ConnectionSummary, ExecutionConnection } from "../../connection-service.ts";
 import type { ActionPolicyDecision, ActionPolicySnapshot } from "../../core/action-policy.ts";
+import type { ProviderHttpDispatchOptions } from "../../core/provider-http-dispatch.ts";
 import type { RuntimeLogger, ExecutionContext, ExecutionResult, TransitFileWriter } from "../../core/types.ts";
 import type { MarketplaceService } from "../../marketplace/marketplace-service.ts";
 import type { IProviderLoader } from "../../providers/provider-loader.ts";
@@ -9,10 +10,17 @@ import type { IRunLogStore, RunLog, RunLogCaller, RunLogListInput, RunLogPage }
 
 import { ConnectionError } from "../../connection-service.ts";
 import { executeAction as executeProviderAction } from "../../core/execution.ts";
+import { withProviderHttpDispatch } from "../../core/provider-http-dispatch.ts";
+import {
+  ProviderDispatchRequestError,
+  toProviderExecutionError,
+  withProviderHttpDispatchResult,
+} from "../../providers/provider-runtime.ts";
 import { SaasError } from "../../saas/saas-client.ts";
 import { safeRunLogError, summarizeForRunLog } from "./run-log-summary.ts";
 
 export interface ActionRunnerOptions {
+  providerHttpDispatch?: ProviderHttpDispatchOptions;
   catalog: CatalogStore;
   providerLoader: IProviderLoader;
   connections: ConnectionService;
@@ -55,6 +63,15 @@ export class ActionRunner {
   }
 
   async run(input: RunActionInput): Promise<ActionRunResult | undefined> {
+    const action = this.options.catalog.actionsById.get(input.actionId);
+    return withProviderHttpDispatch(
+      { operation: "action", service: action?.service, actionId: action?.id },
+      () => this.runAction(input),
+      this.options.providerHttpDispatch,
+    );
+  }
+
+  private async runAction(input: RunActionInput): Promise<ActionRunResult | undefined> {
     const action = this.options.catalog.actionsById.get(input.actionId);
     if (!action) {
       this.options.logger?.warn(
@@ -134,30 +151,43 @@ export class ActionRunner {
               : undefined;
           input.signal?.throwIfAborted();
           const saasReference = connection.kind === "saas" ? connection.reference : undefined;
-          result = await executeProviderAction(
-            action,
-            saasReference
-              ? async (actionInput) => {
-                  if (!this.options.saas)
-                    throw new SaasError("oauth_source_unavailable", "SaaS execution is unavailable.", 503);
-                  const remote = await this.options.saas.executeAction(
-                    saasReference,
-                    action.service,
-                    action.id,
-                    actionInput,
-                    input.signal,
-                  );
-                  remoteExecutionId = remote.executionId;
-                  return { ok: true, output: remote.output };
-                }
-              : connection.kind === "marketplace"
-                ? (actionInput) => this.options.marketplace!.execute(action.id, actionInput, input.signal)
-                : executor,
-            input.input,
-            this.createExecutionContext(
-              connection.kind === "local" ? connection.getCredential : async () => undefined,
-              input.signal,
-            ),
+          const resolvedConnection = connection;
+          result = await withProviderHttpDispatchResult(
+            {
+              operation: "action",
+              service: action.service,
+              actionId: action.id,
+              executionId,
+              connectionId: connection.summary?.id,
+              connectionName: connection.summary?.connectionName,
+            },
+            () =>
+              executeProviderAction(
+                action,
+                saasReference
+                  ? async (actionInput) => {
+                      if (!this.options.saas)
+                        throw new SaasError("oauth_source_unavailable", "SaaS execution is unavailable.", 503);
+                      const remote = await this.options.saas.executeAction(
+                        saasReference,
+                        action.service,
+                        action.id,
+                        actionInput,
+                        input.signal,
+                      );
+                      remoteExecutionId = remote.executionId;
+                      return { ok: true, output: remote.output };
+                    }
+                  : resolvedConnection.kind === "marketplace"
+                    ? (actionInput) => this.options.marketplace!.execute(action.id, actionInput, input.signal)
+                    : executor,
+                input.input,
+                this.createExecutionContext(
+                  resolvedConnection.kind === "local" ? resolvedConnection.getCredential : async () => undefined,
+                  input.signal,
+                ),
+              ),
+            this.options.providerHttpDispatch,
           );
           if (input.signal?.aborted) {
             result = cancelledExecutionResult();
@@ -178,6 +208,8 @@ export class ActionRunner {
           failureStatus = error.status;
           retryAfter = error.retryAfter;
           result = { ok: false, error: { code: error.code, message: error.message } };
+        } else if (error instanceof ProviderDispatchRequestError) {
+          result = toProviderExecutionError(error, error.message);
         } else {
           result =
             error instanceof ConnectionError

src/server/connect-app.ts

@@ -1,5 +1,6 @@
 import type { CatalogStore } from "../catalog-store.ts";
 import type { ActionPolicyService } from "../core/action-policy.ts";
+import type { ProviderHttpDispatchOptions } from "../core/provider-http-dispatch.ts";
 import type { RuntimeLogger, TransitFileUpload } from "../core/types.ts";
 import type { IProviderLoader } from "../providers/provider-loader.ts";
 import type { RuntimeJwtVerifier } from "./api/runtime-jwt.ts";
@@ -28,6 +29,7 @@ import { RuntimeTokenService } from "./storage/runtime-token-service.ts";
 export interface ConnectAppOptions {
   catalog: CatalogStore;
   providerLoader: IProviderLoader;
+  providerHttpDispatch?: ProviderHttpDispatchOptions;
   runtimeDatabase: RuntimeDatabase;
   transitFiles: ITransitFileService;
   uploadTransitFile?: (request: Request) => Promise<TransitFileUpload>;
@@ -86,6 +88,7 @@ export async function createConnectApp(options: ConnectAppOptions): Promise<Conn
     isCustomClientConfigAvailable: (service) => options.secretCodec.encrypted && isCustomClientConfigAllowed(service),
   });
   const connections = new ConnectionService({
+    providerHttpDispatch: options.providerHttpDispatch,
     catalog: options.catalog,
     oauthCredentials: new OAuthCredentialRefreshService(oauthClientConfigs, options.providerLoader),
     providerLoader: options.providerLoader,
@@ -94,6 +97,7 @@ export async function createConnectApp(options: ConnectAppOptions): Promise<Conn
     marketplace,
   });
   const actions = new ActionRunner({
+    providerHttpDispatch: options.providerHttpDispatch,
     catalog: options.catalog,
     providerLoader: options.providerLoader,
     connections,
@@ -105,6 +109,7 @@ export async function createConnectApp(options: ConnectAppOptions): Promise<Conn
   });
 
   const triggers = new TriggerRunner({
+    providerHttpDispatch: options.providerHttpDispatch,
     catalog: options.catalog,
     providerLoader: options.providerLoader,
     connections,
@@ -127,6 +132,7 @@ export async function createConnectApp(options: ConnectAppOptions): Promise<Conn
       logger: options.logger,
     }),
     app: new ConnectServer({
+      providerHttpDispatch: options.providerHttpDispatch,
       catalog: options.catalog,
       publicOrigin: options.publicOrigin,
       providerLoader: options.providerLoader,

src/server/connect-server.ts

@@ -2,6 +2,7 @@ import type { CatalogStore, RuntimeActionDefinition } from "../catalog-store.ts"
 import type { ConnectionService, ConnectionSummary } from "../connection-service.ts";
 import type { ActionPolicySnapshot } from "../core/action-policy.ts";
 import type { ActionSearchDocument, ActionSearchIndexProvider } from "../core/action-search.ts";
+import type { ProviderHttpDispatchOptions } from "../core/provider-http-dispatch.ts";
 import type { RuntimeLogger, TransitFileUpload } from "../core/types.ts";
 import type { MarketplaceConfigInput, MarketplaceService } from "../marketplace/marketplace-service.ts";
 import type { OAuthClientConfigInput } from "../oauth/oauth-client-config-service.ts";
@@ -34,9 +35,11 @@ import {
   requiredStringArray,
 } from "../core/cast.ts";
 import { PromiseCache } from "../core/promise-cache.ts";
+import { withProviderHttpDispatch } from "../core/provider-http-dispatch.ts";
 import { MarketplaceError } from "../marketplace/marketplace-service.ts";
 import { OAuthClientConfigError, OAuthClientConfigService } from "../oauth/oauth-client-config-service.ts";
 import { OAuthCallbackError, OAuthFlowError, OAuthFlowService } from "../oauth/oauth-flow-service.ts";
+import { ProviderDispatchRequestError, toProviderExecutionError } from "../providers/provider-runtime.ts";
 import { SaasError } from "../saas/saas-client.ts";
 import {
   ActionInputDepthError,
@@ -123,6 +126,7 @@ export async function preloadOptionalServerModules(): Promise<void> {
  * Dependencies required to construct the local connector server.
  */
 export interface IConnectServerOptions {
+  providerHttpDispatch?: ProviderHttpDispatchOptions;
   catalog: CatalogStore;
   /** Public origin of this runtime, used for the HTTP request examples in Action guides. */
   publicOrigin: string;
@@ -167,6 +171,7 @@ export class ConnectServer {
     this.actionSearch = options.actionSearch ?? createActionSearchIndexProvider(options.catalog.actions);
     this.actionPolicy = options.actionPolicy ?? new ActionPolicyService();
     this.proxyRunner = new ProxyRunner({
+      providerHttpDispatch: options.providerHttpDispatch,
       catalog: options.catalog,
       providerLoader: options.providerLoader,
       connections: options.connections,
@@ -179,6 +184,10 @@ export class ConnectServer {
     const app = new Hono();
     const auth = this.options.auth ?? {};
 
+    app.use("*", async (_context, next) => {
+      await withProviderHttpDispatch({ operation: "runtime" }, next, this.options.providerHttpDispatch);
+    });
+
     app.use("*", async (context, next) => {
       await next();
       const cachePolicy = getResponseCachePolicy(context.req.method, context.req.path, context.res.status);
@@ -410,6 +419,14 @@ export class ConnectServer {
     if (this.options.registerStaticRoutes) this.options.registerStaticRoutes(app);
     else app.notFound(notFound);
     app.onError((error, context) => {
+      if (error instanceof ProviderDispatchRequestError) {
+        return writeRuntimeFailure(context, {
+          status: 429,
+          errorCode: "rate_limited",
+          message: error.message,
+          data: toProviderExecutionError(error, error.message).error?.details,
+        });
+      }
       if (error instanceof SaasError) {
         this.options.logger?.warn(
           {

src/server/connector-runtime.test.ts

@@ -1,3 +1,4 @@
+import type { ProviderHttpAttempt } from "../core/provider-http-dispatch.ts";
 import type { ConnectorRuntime, ConnectorRuntimeOptions } from "./connector-runtime.ts";
 
 import { cp, mkdtemp, mkdir, readFile, rm } from "node:fs/promises";
@@ -20,6 +21,93 @@ afterEach(async () => {
 });
 
 describe("headless runtime", () => {
+  it("preserves admission denial after a real provider remaps it into validation and transport errors", async () => {
+    let deny = false;
+    const fetcher = vi.fn<typeof fetch>(async () => Response.json({ credit_balance: 100 }));
+    vi.stubGlobal("fetch", fetcher);
+    runtime = await createConnectorRuntime({
+      ...(await fixture(["ninjapear"])),
+      providerHttpDispatch: {
+        beforeAttempt: () => (deny ? { allow: false, retryAfterSeconds: 62 } : { allow: true }),
+      },
+    });
+    expect((await request("/v1/connections/ninjapear/connect/api-key", { apiKey: "fixture-secret" })).status).toBe(200);
+    const accounts = (await (await request("/v1/apps", undefined, "runtime-token")).json()).data;
+    deny = true;
+    const executed = await runtime.fetch(
+      new Request(`${publicOrigin}/v1/actions/ninjapear.get_credit_balance`, {
+        method: "POST",
+        headers: {
+          authorization: "Bearer runtime-token",
+          "content-type": "application/json",
+          "x-oo-connector-alias": accounts[0].alias,
+        },
+        body: JSON.stringify({ input: {} }),
+      }),
+    );
+    const validated = await request("/v1/connections/ninjapear/connect/api-key", { apiKey: "another-secret" });
+    for (const response of [executed, validated]) {
+      expect(response.status).toBe(429);
+      expect(response.headers.get("Retry-After")).toBe("62");
+      expect(await response.json()).toMatchObject({ errorCode: "rate_limited" });
+    }
+    expect(fetcher).toHaveBeenCalledOnce();
+  });
+
+  it("binds resolved connection authority and exposes dispatch denial as 429 on action, proxy and validation routes", async () => {
+    const attempts: ProviderHttpAttempt[] = [];
+    let deny = false;
+    const fetcher = vi.fn<typeof fetch>(async () => Response.json({ id: 1, login: "fixture-account" }));
+    vi.stubGlobal("fetch", fetcher);
+    runtime = await createConnectorRuntime({
+      ...(await fixture()),
+      providerHttpDispatch: {
+        bindAuthority: (context) => ({ workspaceId: "host-workspace", connectionLineageId: context.connectionId }),
+        beforeAttempt: (attempt) => {
+          attempts.push(attempt);
+          return deny ? { allow: false, retryAfterSeconds: 45 } : { allow: true };
+        },
+      },
+    });
+    const saved = await request("/v1/connections/github/connect/api-key", { apiKey: "fixture-secret" });
+    expect(saved.status).toBe(200);
+    const connection = (await saved.json()).data;
+    expect(attempts[0]?.context).toMatchObject({ operation: "credential_validation", service: "github" });
+    const accounts = (await (await request("/v1/apps", undefined, "runtime-token")).json()).data;
+    deny = true;
+    for (const [route, body] of [
+      ["/v1/actions/github.get_current_user", { input: {}, connectionId: "untrusted" }],
+      ["/v1/proxy/github", { method: "GET", endpoint: "/user", connectionId: "untrusted" }],
+    ] as const) {
+      const response = await runtime.fetch(
+        new Request(`${publicOrigin}${route}`, {
+          method: "POST",
+          headers: {
+            authorization: "Bearer runtime-token",
+            "content-type": "application/json",
+            "x-oo-connector-alias": accounts[0].alias,
+          },
+          body: JSON.stringify(body),
+        }),
+      );
+      expect(response.status).toBe(429);
+      expect(response.headers.get("Retry-After")).toBe("45");
+      expect(await response.json()).toMatchObject({ errorCode: "rate_limited" });
+      expect(attempts.at(-1)?.context.connectionId).toBe(connection.id);
+      expect(attempts.at(-1)?.context.connectionName).toBe(accounts[0].alias);
+      expect(attempts.at(-1)?.authority).toMatchObject({
+        workspaceId: "host-workspace",
+        connectionLineageId: connection.id,
+      });
+    }
+    const refused = await request("/v1/connections/github/connect/api-key", {
+      apiKey: "another-secret",
+    });
+    expect(refused.status).toBe(429);
+    expect(refused.headers.get("Retry-After")).toBe("45");
+    expect(fetcher).toHaveBeenCalledOnce();
+    expect(JSON.stringify(attempts)).not.toMatch(/fixture-secret|another-secret|untrusted/);
+  });
   it("serves the existing connection and action contracts under a host mount without a dashboard", async () => {
     const options = await fixture();
     vi.stubGlobal(
@@ -280,12 +368,12 @@ it("starts persisted cleanup without HTTP traffic and waits for its abort on clo
   }
 });
 
-async function fixture(): Promise<ConnectorRuntimeOptions> {
+async function fixture(services = ["github"]): Promise<ConnectorRuntimeOptions> {
   const root = await mkdtemp(join(tmpdir(), "open-connector-runtime-"));
   directories.push(root);
   const catalogDir = join(root, "catalog");
   await mkdir(catalogDir);
-  for (const service of ["github"])
+  for (const service of services)
     await cp(resolve(import.meta.dirname, `../../catalog/apps/${service}.json`), join(catalogDir, `${service}.json`));
   return {
     dataDir: join(root, "state"),

src/server/connector-runtime.ts

@@ -1,4 +1,5 @@
 import type { ActionPolicyConfig } from "../core/action-policy.ts";
+import type { ProviderHttpDispatchOptions } from "../core/provider-http-dispatch.ts";
 import type { RuntimeLogger } from "../core/types.ts";
 import type { RuntimeJwtConfig } from "./api/runtime-jwt.ts";
 import type { S3TransitClientOptions } from "./files/s3-transit-files.ts";
@@ -80,6 +81,8 @@ export interface ConnectorRuntimeOptions {
   jwt?: RuntimeJwtConfig;
   postgres?: ConnectorPostgresOptions;
   network?: ConnectorNetworkOptions;
+  /** Opt-in admission and result feedback for every provider HTTP transport attempt. */
+  providerHttpDispatch?: ProviderHttpDispatchOptions;
   /** Allow or block actions, proxies and Triggers by name. */
   actionPolicy?: ActionPolicyConfig;
   /** Services, or `*`, whose connections may carry their own OAuth client instead of the configured one. */
@@ -209,6 +212,7 @@ async function openRuntime(options: ConnectorRuntimeOptions): Promise<ConnectorR
     const { app, runtimeAuthConfigured, saasCleanup, triggerMaintenance } = await createConnectApp({
       catalog,
       providerLoader: new ProviderLoader(executorModules),
+      providerHttpDispatch: options.providerHttpDispatch,
       runtimeDatabase: database,
       transitFiles,
       uploadTransitFile: createNodeTransitFileUpload({ transitFiles, tempDir }),

src/server/proxy/proxy-runner.ts

@@ -1,18 +1,26 @@
 import type { CatalogStore } from "../../catalog-store.ts";
 import type { ConnectionService } from "../../connection-service.ts";
 import type { ActionPolicySnapshot } from "../../core/action-policy.ts";
+import type { ProviderHttpDispatchOptions } from "../../core/provider-http-dispatch.ts";
 import type { RuntimeLogger, ProxyRequestInput, ProxyResponse } from "../../core/types.ts";
 import type { IProviderLoader } from "../../providers/provider-loader.ts";
 import type { SaasExecutionService } from "../../saas/saas-execution-service.ts";
 
 import { ConnectionError } from "../../connection-service.ts";
 import { optionalInteger, optionalRecord, requiredRecord, requiredString } from "../../core/cast.ts";
+import { withProviderHttpDispatch } from "../../core/provider-http-dispatch.ts";
+import {
+  ProviderDispatchRequestError,
+  toProviderExecutionError,
+  withProviderHttpDispatchResult,
+} from "../../providers/provider-runtime.ts";
 import { SaasError } from "../../saas/saas-client.ts";
 import { mapConnectionErrorStatus } from "../api/runtime-api.ts";
 
 export type ProxyFailureStatus = 400 | 402 | 403 | 404 | 409 | 413 | 429 | 500 | 501 | 502 | 503 | 504;
 
 export interface ProxyRunnerOptions {
+  providerHttpDispatch?: ProviderHttpDispatchOptions;
   catalog: CatalogStore;
   providerLoader: IProviderLoader;
   connections: ConnectionService;
@@ -61,6 +69,15 @@ export class ProxyRunner {
   }
 
   async run(input: RunProxyInput): Promise<ProxyRunResult> {
+    const provider = this.options.catalog.providers.find((candidate) => candidate.service === input.service);
+    return withProviderHttpDispatch(
+      { operation: "proxy", service: provider?.service },
+      () => this.runProxy(input),
+      this.options.providerHttpDispatch,
+    );
+  }
+
+  private async runProxy(input: RunProxyInput): Promise<ProxyRunResult> {
     const provider = this.options.catalog.providers.find((candidate) => candidate.service === input.service);
     if (!provider) {
       return {
@@ -174,10 +191,20 @@ export class ProxyRunner {
           message: `Proxy execution is not supported for ${provider.service}.`,
           meta: { service: provider.service },
         };
-      const result = await executor(request.input, {
-        getCredential: target.getCredential,
-        signal: input.signal,
-      });
+      const result = await withProviderHttpDispatchResult(
+        {
+          operation: "proxy",
+          service: provider.service,
+          connectionId: target.summary?.id,
+          connectionName: target.summary?.connectionName,
+        },
+        () =>
+          executor(request.input, {
+            getCredential: target.getCredential,
+            signal: input.signal,
+          }),
+        this.options.providerHttpDispatch,
+      );
       const durationMs = Date.now() - startedAtMs;
       if (result.ok) {
         this.options.logger?.info(
@@ -224,6 +251,16 @@ export class ProxyRunner {
           meta: { service: provider.service, executionId, remoteExecutionId: error.remoteExecutionId },
         };
       }
+      if (error instanceof ProviderDispatchRequestError) {
+        return {
+          ok: false,
+          status: 429,
+          errorCode: "rate_limited",
+          message: error.message,
+          data: toProviderExecutionError(error, error.message).error?.details,
+          meta: { service: provider.service },
+        };
+      }
       if (error instanceof ConnectionError) {
         const missingConnectionDecision =
           error.code === "connection_not_found" ? input.policy.evaluateConnection() : undefined;

src/server/trigger-runtime.test.ts

@@ -1,3 +1,5 @@
+import type { IConnectionStore } from "../connection-service.ts";
+import type { ProviderHttpAttempt } from "../core/provider-http-dispatch.ts";
 import type { ProviderDefinition, ProviderProxyExecutor } from "../core/types.ts";
 import type { ConnectorProxyRequest } from "../triggers/common/proxy.ts";
 import type { ConnectApp } from "./connect-app.ts";
@@ -5,12 +7,14 @@ import type { ConnectApp } from "./connect-app.ts";
 import { createHmac } from "node:crypto";
 import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { createCatalogStore } from "../catalog-store.ts";
+import { ConnectionService } from "../connection-service.ts";
 import { ActionPolicyService } from "../core/action-policy.ts";
 import { provider as feishu } from "../providers/feishu_app_bot/definition.ts";
 import { provider as github } from "../providers/github/definition.ts";
 import { provider as gmail } from "../providers/gmail/definition.ts";
 import { provider as linear } from "../providers/linear/definition.ts";
 import { ProviderLoader } from "../providers/provider-loader.ts";
+import { TriggerRunner } from "../triggers/trigger-runner.ts";
 import { createConnectApp } from "./connect-app.ts";
 import { TransitFileService } from "./files/transit-files.ts";
 import { PlainTextSecretCodec } from "./secrets/secret-codec-core.ts";
@@ -29,6 +33,10 @@ let hooks: Record<string, unknown>[];
 let nextId: number;
 let failDelete: boolean;
 let loseCreate: boolean;
+let useNativeProxy: boolean;
+let denyProviderDispatch: boolean;
+let dispatchAttempts: ProviderHttpAttempt[];
+let fixtureProviderLoader: ProviderLoader;
 
 beforeEach(async () => {
   database = new SqliteRuntimeDatabase(":memory:");
@@ -37,6 +45,9 @@ beforeEach(async () => {
   nextId = 1;
   failDelete = false;
   loseCreate = false;
+  useNativeProxy = false;
+  denyProviderDispatch = false;
+  dispatchAttempts = [];
   const sources: ProviderDefinition[] = [github, gmail, linear, feishu];
   const modulePaths: Record<string, string> = {
     github: "../providers/github/executors.ts",
@@ -49,10 +60,16 @@ beforeEach(async () => {
       provider.service,
       async () => {
         const native = await import(modulePaths[provider.service]!);
+        if (useNativeProxy) return native;
         const proxy: ProviderProxyExecutor = async (request, context) => {
           context.signal?.throwIfAborted();
           const credential = await context.getCredential(provider.service);
-          const key = credential?.authType === "api_key" ? credential.apiKey : "unknown";
+          const key =
+            credential?.authType === "api_key"
+              ? credential.apiKey
+              : credential?.authType === "oauth2"
+                ? credential.accessToken
+                : "unknown";
           calls.push({ service: provider.service, key, request: request as ConnectorProxyRequest });
           let data: unknown = {};
           let status = 200;
@@ -122,9 +139,16 @@ beforeEach(async () => {
   });
   token = created.token;
   tokenId = created.record.id;
+  fixtureProviderLoader = new ProviderLoader(modules);
   connector = await createConnectApp({
+    providerHttpDispatch: {
+      beforeAttempt: (attempt) => {
+        dispatchAttempts.push(attempt);
+        return denyProviderDispatch ? { allow: false, retryAfterSeconds: 46 } : { allow: true };
+      },
+    },
     catalog: createCatalogStore(sources),
-    providerLoader: new ProviderLoader(modules),
+    providerLoader: fixtureProviderLoader,
     runtimeDatabase: database,
     transitFiles: new TransitFileService({
       rootDir: "/unused-trigger-test",
@@ -146,6 +170,7 @@ afterEach(async () => {
   await connector.saasCleanup.close();
   database.close();
   vi.restoreAllMocks();
+  vi.unstubAllGlobals();
 });
 
 function credential(key: string, accountId: string) {
@@ -180,6 +205,82 @@ async function subscription() {
 }
 
 describe("Trigger runtime HTTP boundary", () => {
+  it("retains Trigger OAuth refresh with a boolean-only custom connection store and admission configured", async () => {
+    const sql = database.connectionStore;
+    const store: IConnectionStore = {
+      get: sql.get.bind(sql),
+      set: sql.set.bind(sql),
+      updateCredential: sql.updateCredential.bind(sql),
+      delete: sql.delete.bind(sql),
+      list: sql.list.bind(sql),
+    };
+    const oauth = {
+      authType: "oauth2" as const,
+      accessToken: "expired-token",
+      refreshToken: "refresh-token",
+      tokenType: "Bearer",
+      expiresAt: "2026-01-01T00:00:00.000Z",
+      profile: credential("unused", "work-account").profile,
+      metadata: { providerAccountVerified: true },
+    };
+    await store.set("github", "work", oauth);
+    const catalog = createCatalogStore([github]);
+    const providerHttpDispatch = { beforeAttempt: () => ({ allow: true as const }) };
+    const connections = new ConnectionService({
+      providerHttpDispatch,
+      catalog,
+      store,
+      providerLoader: fixtureProviderLoader,
+      oauthCredentials: {
+        refresh: async () => ({ ...oauth, accessToken: "fresh-token", expiresAt: "2099-01-01T00:00:00.000Z" }),
+      },
+    });
+    const runner = new TriggerRunner({
+      providerHttpDispatch,
+      catalog,
+      connections,
+      providerLoader: fixtureProviderLoader,
+      store: database.triggerStore,
+    });
+    await expect(
+      runner.run({
+        service: "github",
+        triggerId: "github.on_repo_event",
+        connectionId,
+        policy: new ActionPolicyService().createSnapshot(),
+        grant: {
+          tokenId,
+          allowedActions: [],
+          blockedActions: [],
+          allowedProxies: [],
+          allowedTriggers: ["github.on_repo_event"],
+        },
+        request: { operation: "reconcile", config, endpointUrl: callback, active: true, requestKey: "legacy-refresh" },
+        signal: new AbortController().signal,
+      }),
+    ).resolves.toMatchObject({ subscription: { id: expect.any(String) } });
+    expect(calls.length).toBeGreaterThan(0);
+    expect(calls.every((call) => call.key === "fresh-token")).toBe(true);
+  });
+
+  it("preserves admission denial through native Trigger proxy and integration error mapping", async () => {
+    useNativeProxy = true;
+    denyProviderDispatch = true;
+    const fetcher = vi.fn<typeof fetch>();
+    vi.stubGlobal("fetch", fetcher);
+    const response = await request(reconcile());
+    expect(response.status).toBe(429);
+    expect(response.headers.get("Retry-After")).toBe("46");
+    expect(await response.json()).toMatchObject({ errorCode: "rate_limited" });
+    expect(fetcher).not.toHaveBeenCalled();
+    expect(dispatchAttempts).toHaveLength(1);
+    expect(dispatchAttempts[0]?.context).toMatchObject({
+      operation: "trigger",
+      connectionId,
+      connectionName: "work",
+    });
+  });
+
   it("runs Trigger-only grants against the stable non-default connection and denies Action and public proxy", async () => {
     await subscription();
     expect(calls.every((call) => call.key === "work-key")).toBe(true);

src/triggers/trigger-runner.ts

@@ -1,6 +1,7 @@
 import type { CatalogStore } from "../catalog-store.ts";
 import type { ConnectionService } from "../connection-service.ts";
 import type { ActionPolicySnapshot } from "../core/action-policy.ts";
+import type { ProviderHttpDispatchOptions } from "../core/provider-http-dispatch.ts";
 import type { IProviderLoader } from "../providers/provider-loader.ts";
 import type { RuntimeGrant } from "../server/storage/runtime-token-service.ts";
 import type { IntegrationDefinition } from "./common/integration.ts";
@@ -11,6 +12,8 @@ import type { TriggerSubscription, TriggerStore } from "./store.ts";
 
 import { ConnectionError } from "../connection-service.ts";
 import { optionalInteger, optionalRecord } from "../core/cast.ts";
+import { withProviderHttpDispatch } from "../core/provider-http-dispatch.ts";
+import { withProviderHttpDispatchResult } from "../providers/provider-runtime.ts";
 import { HttpRequestError } from "../server/api/http-utils.ts";
 import { mapConnectionErrorStatus } from "../server/api/runtime-api.ts";
 import { resolveTriggerConfig } from "./common/config.ts";
@@ -23,6 +26,7 @@ import {
 import { executeSubscription, executeSubscriptionOperation } from "./subscriptions.ts";
 
 interface TriggerRunnerOptions {
+  providerHttpDispatch?: ProviderHttpDispatchOptions;
   catalog: CatalogStore;
   providerLoader: IProviderLoader;
   connections: ConnectionService;
@@ -46,6 +50,15 @@ export class TriggerRunner {
   }
 
   async run(input: RunTriggerInput): Promise<unknown> {
+    const provider = this.options.catalog.providers.find((candidate) => candidate.service === input.service);
+    return withProviderHttpDispatch(
+      { operation: "trigger", service: provider?.service },
+      () => this.runTrigger(input),
+      this.options.providerHttpDispatch,
+    );
+  }
+
+  private async runTrigger(input: RunTriggerInput): Promise<unknown> {
     const decision = input.policy.evaluateTrigger(input.triggerId);
     if (!decision.allowed) throw new HttpRequestError(decision.code, decision.message, 403);
     const stateful = input.request.operation !== "read" && input.request.operation !== "options";
@@ -67,7 +80,7 @@ export class TriggerRunner {
     if (request.operation === "receive") {
       if (definition.snapshot.type !== "integration" || ("eventSource" in definition && definition.eventSource))
         throw new HttpRequestError("invalid_input", "This Trigger has no remote webhook.");
-      return this.providerOperation(() =>
+      return this.providerOperation(input.service, () =>
         executeSubscription(
           this.options.store,
           definition as IntegrationDefinition,
@@ -84,7 +97,7 @@ export class TriggerRunner {
       request.operation === "options",
     );
     const context = { config, connector: target.proxy, now: new Date(), signal: input.signal };
-    return this.providerOperation(async () => {
+    return this.providerOperation(input.service, async () => {
       switch (request.operation) {
         case "options": {
           if (
@@ -142,7 +155,7 @@ export class TriggerRunner {
     if (record.mode === "resource") {
       if (!(definition as IntegrationDefinition).resources)
         throw new HttpRequestError("trigger_not_supported", "Resource subscriptions are unavailable.", 501);
-      await this.providerOperation(() =>
+      await this.providerOperation(record.service, () =>
         executeResourceSubscriptionOperation(
           this.options.store,
           (definition as IntegrationDefinition).resources!,
@@ -156,7 +169,7 @@ export class TriggerRunner {
     } else {
       if (definition.snapshot.type !== "integration")
         throw new HttpRequestError("trigger_not_found", "The remote Trigger definition is unavailable.", 404);
-      await this.providerOperation(() =>
+      await this.providerOperation(record.service, () =>
         executeSubscriptionOperation(
           record,
           save,
@@ -234,10 +247,20 @@ export class TriggerRunner {
       const proxy: ConnectorProxy = {
         execute: async (request, requestSignal) => {
           signal.throwIfAborted();
-          const result = await executor(request, {
-            getCredential: target.getCredential,
-            signal: requestSignal ?? signal,
-          });
+          const result = await withProviderHttpDispatch(
+            {
+              operation: "trigger",
+              service,
+              connectionId: stored.id,
+              connectionName: stored.connectionName,
+            },
+            () =>
+              executor(request, {
+                getCredential: target.getCredential,
+                signal: requestSignal ?? signal,
+              }),
+            this.options.providerHttpDispatch,
+          );
           if (result.ok) return result.response;
           const status = optionalInteger(optionalRecord(result.error.details)?.status) ?? 502;
           return { status, data: { error: result.error.message } };
@@ -259,7 +282,15 @@ export class TriggerRunner {
     }
   }
 
-  private async providerOperation<T>(run: () => Promise<T>): Promise<T> {
+  private async providerOperation<T>(service: string, run: () => Promise<T>): Promise<T> {
+    return withProviderHttpDispatchResult(
+      { operation: "trigger", service },
+      () => this.runProviderOperation(run),
+      this.options.providerHttpDispatch,
+    );
+  }
+
+  private async runProviderOperation<T>(run: () => Promise<T>): Promise<T> {
     try {
       return await run();
     } catch (error) {