Expose trusted HTTP executor dispatch receipts

src/agent/hosted/executor-session.test.ts

@@ -639,7 +639,10 @@ describe("hosted executor session", () => {
   }
 
   it("observes preparation without attaching until the allocator supplies a ready endpoint", async () => {
-    const f = fixture();
+    const attachments: Parameters<
+      NonNullable<HostedExecutorSessionOptions["observeAttachment"]>
+    >[0][] = [];
+    const f = fixture({ observeAttachment: (receipt) => void attachments.push(receipt) });
     const endpoint = f.returned.endpoint!;
     f.returned.phase = "preparing";
     delete f.returned.endpoint;
@@ -650,10 +653,20 @@ describe("hosted executor session", () => {
     f.time.advanceBy(100);
     await tick();
     assertEquals(f.calls, ["allocate", "observe"]);
+    assertEquals(attachments, []);
     f.returned.phase = "ready";
     f.returned.endpoint = endpoint;
     f.time.advanceBy(100);
     await session.ready;
+    assertEquals(attachments, [{
+      allocationId: binding.allocationId,
+      generation: binding.generation,
+      invocationId: binding.invocationId,
+      podUid: endpoint.podUid,
+      nodeName: endpoint.nodeName,
+      declaredImage: image,
+    }]);
+    assert(Object.isFrozen(attachments[0]));
     assertEquals(f.calls.filter((call) => call === "allocate").length, 1);
     assertEquals(f.calls.filter((call) => call === "connect").length, 1);
     await session.close();
@@ -758,7 +771,12 @@ describe("hosted executor session", () => {
   });
 
   it("does not repeat a failed authenticated attachment or expose its error details", async () => {
-    const f = fixture();
+    let attachments = 0;
+    const f = fixture({
+      observeAttachment: () => {
+        attachments++;
+      },
+    });
     let attempts = 0;
     f.options.connectTransport = () => {
       attempts++;
@@ -770,6 +788,7 @@ describe("hosted executor session", () => {
     f.time.advanceBy(1000);
     await tick();
     assertEquals(attempts, 1);
+    assertEquals(attachments, 0);
     assertEquals(f.calls.includes("operations"), false);
     assertEquals(f.allocatedKey, new Uint8Array(32));
     assertEquals(f.time.pendingWaitCount, 0);

src/agent/hosted/executor-session.ts

@@ -59,6 +59,17 @@ export interface HostedExecutorSessionOptions {
   allocator: HostedExecutorAllocatorClient;
   /** Authenticate one connection. Snapshot the allocation key before awaiting. */
   connectTransport(options: ConnectExecutorTransportOptions): Promise<ExecutorNodeTransport>;
+  /** Trusted synchronous receipt sink, invoked after channel authentication and before readiness. */
+  observeAttachment?(
+    receipt: Readonly<{
+      allocationId: string;
+      generation: number;
+      invocationId: string;
+      podUid: string;
+      nodeName: string;
+      declaredImage: string;
+    }>,
+  ): void;
   /** Factory must release its own partial resources if construction throws. */
   createOperations(binding: Readonly<HostedExecutorBinding>, signal: AbortSignal): {
     operations: ReadonlyMap<string, ExecutorOperation>;
@@ -140,6 +151,7 @@ class Session implements HostedExecutorSession {
   readonly #expectedImage: string;
   readonly #allocator: HostedExecutorAllocatorClient;
   readonly #connect: HostedExecutorSessionOptions["connectTransport"];
+  readonly #observeAttachment: HostedExecutorSessionOptions["observeAttachment"];
   readonly #createOperations: HostedExecutorSessionOptions["createOperations"];
   readonly #clock: HostedExecutorSessionClock;
   readonly #pollInterval: number;
@@ -212,6 +224,7 @@ class Session implements HostedExecutorSession {
       release: options.allocator.release.bind(options.allocator),
     };
     this.#connect = options.connectTransport;
+    this.#observeAttachment = options.observeAttachment;
     this.#createOperations = options.createOperations;
     this.#clock = options.clock ?? createHostedExecutorSessionClock(Date.now());
     this.#pollInterval = positiveLimit(options.pollIntervalMs ?? 250, 30_000);
@@ -518,6 +531,14 @@ class Session implements HostedExecutorSession {
       void this.#channel.closed.then(() => this.#stop("channel-closed"));
       await awaitAbortable(this.#channel.ready, this.signal);
       this.#assertActive();
+      const endpoint = this.#endpoint!;
+      this.#observeAttachment?.(Object.freeze({
+        ...channelBinding,
+        podUid: endpoint.podUid,
+        nodeName: endpoint.nodeName,
+        declaredImage: endpoint.image,
+      }));
+      this.#assertActive();
       this.#cancelTimer("attachment");
       this.#isReady = true;
       this.#ready.resolve(this.#channel);

src/server/isolated-http/generation-executor.ts

@@ -3,7 +3,10 @@ import {
   getExecutorHttpInstallSchema,
   parseExecutorInstallation,
 } from "#veryfront/agent/hosted/executor-runtime-install-schema.ts";
-import type { HostedExecutorSession } from "#veryfront/agent/hosted/executor-session.ts";
+import type {
+  HostedExecutorSession,
+  HostedExecutorSessionOptions,
+} from "#veryfront/agent/hosted/executor-session.ts";
 import type { HostedExecutorSessionPool } from "#veryfront/agent/hosted/executor-session-pool.ts";
 import {
   getHostedExecutorAllocationRequestSchema,
@@ -13,6 +16,10 @@ import {
   sameHostedExecutorOwner,
 } from "#veryfront/agent/hosted/executor-session-schema.ts";
 import { verifyHostedRuntimeSourceBinding } from "#veryfront/agent/hosted/runtime-source-binding.ts";
+import {
+  formatTraceparent,
+  parseTraceparent,
+} from "#veryfront/observability/tracing/traceparent.ts";
 import { normalizeDynamicDenyHeaders } from "#veryfront/security/http/application-request.ts";
 import { completeOnResponseBodyConsumption } from "#veryfront/platform/compat/http/response-lifecycle.ts";
 import type { RenderGenerationExecutor } from "#veryfront/rendering/render-generation.ts";
@@ -112,6 +119,17 @@ export interface HostedHttpGenerationExecutorOptions {
   sessions: Pick<HostedExecutorSessionPool, "start">;
   /** Look up the broker's data for an admitted request. */
   context(request: Request): HostedHttpGenerationRequest | undefined;
+  /** Trusted synchronous observer of a request about to use an authenticated attachment. */
+  observeDispatch?(
+    receipt: Readonly<{
+      attachment: Parameters<NonNullable<HostedExecutorSessionOptions["observeAttachment"]>>[0];
+      brokerInstanceId: string;
+      projectId: string;
+      environmentId: string;
+      releaseId: string;
+      traceparent?: string;
+    }>,
+  ): void;
   /** Stop admitting requests after this long without one in flight. */
   idleTimeoutMs: number;
   /**
@@ -206,6 +224,10 @@ export function createHostedHttpGenerationExecutor(
   let released = false;
   let stopping: Promise<void> | undefined;
   let observation: Promise<unknown> | undefined;
+  const observeDispatch = options.observeDispatch;
+  let attachment:
+    | Parameters<NonNullable<HostedExecutorSessionOptions["observeAttachment"]>>[0]
+    | undefined;
 
   function retire(): void {
     if (retired) return;
@@ -224,6 +246,12 @@ export function createHostedHttpGenerationExecutor(
     }
     const started = sessions.start({
       ...input.session,
+      ...(observeDispatch === undefined ? {} : {
+        observeAttachment(receipt) {
+          attachment = receipt;
+          input.session.observeAttachment?.(receipt);
+        },
+      }),
       request: allocation,
       // The session is shared by every request of this generation. Only the generation
       // owns its lifetime; a caller's owner signal bounds that caller's request alone.
@@ -363,7 +391,26 @@ export function createHostedHttpGenerationExecutor(
         const current = session!;
         const signal = AbortSignal.any([request.signal, current.signal, lifetime.signal]);
         const owned = new Request(request, { signal });
-        const response = await current.runOwned(() => client!.fetch(owned, context.tracing));
+        const response = await current.runOwned(() => {
+          signal.throwIfAborted();
+          if (observeDispatch) {
+            if (!attachment) throw new Error("HTTP executor attachment receipt is unavailable");
+            const { owner, source, environmentId } = input.installation;
+            if (owner.scopeKind !== "project") throw new Error("HTTP executor owner is unverified");
+            observeDispatch(
+              Object.freeze({
+                attachment,
+                brokerInstanceId: input.session.expectedBrokerInstanceId,
+                projectId: owner.projectId,
+                environmentId,
+                releaseId: source.releaseId,
+                traceparent: formatTraceparent(parseTraceparent(context.tracing?.traceparent)),
+              }),
+            );
+            signal.throwIfAborted();
+          }
+          return client!.fetch(owned, context.tracing);
+        });
         return completeOnResponseBodyConsumption(
           response,
           finish,

src/server/isolated-http/hosted-http-broker.test.ts

@@ -53,6 +53,52 @@ describe("hosted HTTP executor broker", () => {
     }
   });
 
+  it("joins each warm dispatch to the same authenticated attachment and its own host trace", async () => {
+    const f = fixture(() => new Response("app"));
+    Object.assign(f.input.installation.source, { type: "environment", environmentName: "staging" });
+    const records: unknown[] = [];
+    const broker = createHostedHttpBroker({
+      maxActive: 1,
+      observeDispatch: (receipt) => void records.push(receipt),
+    });
+    const traces = [
+      "00-11111111111111111111111111111111-2222222222222222-01",
+      "00-33333333333333333333333333333333-4444444444444444-01",
+    ];
+    try {
+      for (const [index, traceparent] of traces.entries()) {
+        const response = await broker.fetch(
+          new Request("https://app.example/api", { headers: { traceparent: "untrusted" } }),
+          {
+            ...f.input,
+            tracing: {
+              traceparent: index === 0 ? traceparent : traceparent + "-private",
+              onRecords() {},
+            },
+          },
+        );
+        assertEquals(await response.text(), "app");
+      }
+      assertEquals(f.calls.filter((call) => call === "allocate").length, 1);
+      const attachment = (records[0] as { attachment: unknown }).attachment;
+      assertExists(attachment);
+      assertEquals(
+        records,
+        traces.map((traceparent) => ({
+          attachment,
+          brokerInstanceId: "broker-a",
+          projectId: "project-a",
+          environmentId: "environment-a",
+          releaseId: "release-a",
+          traceparent,
+        })),
+      );
+    } finally {
+      await broker.shutdown();
+      await broker.settled;
+    }
+  });
+
   it("delivers only the matched application snapshot through the allocation channel", async () => {
     let appValue: string | undefined;
     const setup = fixture(() => new Response(appValue), async (peer, installed) => {

src/server/isolated-http/hosted-http-broker.ts

@@ -64,6 +64,8 @@
 
 /** Admission and lifetime limits of the hosted HTTP broker's warm executors. */
 export interface HostedHttpBrokerOptions extends HostedExecutorSessionPoolOptions {
+  observeDispatch?:
+    import("./generation-executor.ts").HostedHttpGenerationExecutorOptions["observeDispatch"];
   /** Requests one executor serves at once. Default 8, maximum 16. */
   maxConcurrentRequestsPerExecutor?: number;
   /** Retire an executor after this long without a request in flight. Default 60 seconds, maximum 1 hour. */
@@ -176,6 +178,7 @@
  * executor with the same identity is refused, never rebound.
  */
 export function createHostedHttpBroker(options: HostedHttpBrokerOptions) {
+  const observeDispatch = options.observeDispatch;
   const sessions = createHostedExecutorSessionPool(options);
   const perExecutor = bounded(
     options.maxConcurrentRequestsPerExecutor ?? 8,
@@ -530,6 +533,7 @@
         input,
         sessions,
         context: (value) => contexts.get(value),
+        ...(observeDispatch === undefined ? {} : { observeDispatch }),
         idleTimeoutMs,
         retireBeforeDeadlineMs: drainTimeoutMs,
       });

src/server/isolated-http/installed-http-broker.test.ts

@@ -66,11 +66,16 @@ describe("installed HTTP broker", () => {
   });
   it("uses the configured bounded lifetime for long requests and streams", async () => {
     const f = createHostedHttpFixture(() => new Response("A"));
+    let attachments = 0;
+    const observeAttachment = () => {
+      attachments++;
+    };
     const options = {
       installations: [{ image: f.input.session.expectedImage, configuration }],
       expectedBrokerInstanceId: "broker-a",
       allocator: f.input.session.allocator,
       connectTransport: f.input.session.connectTransport,
+      observeAttachment,
       maxActive: 1,
       authorize: async () => {},
       requestLifetimeMs: 180_000,
@@ -79,6 +84,8 @@ describe("installed HTTP broker", () => {
     try {
       const startedAt = Date.now();
       const input = await b.hostedHttp.resolve(authority, new AbortController().signal);
+      assertEquals(input.session.observeAttachment, observeAttachment);
+      assertEquals(attachments, 0);
       const finishedAt = Date.now();
       const { hardDeadlineAt, prepareDeadlineAt } = input.session.request;
       // Lifetimes start after authorization, which can cross a clock tick.

src/server/isolated-http/installed-http-broker.ts

@@ -11,7 +11,7 @@
   type ExecutorHttpApplicationConfiguration,
   snapshotExecutorHttpApplicationConfiguration,
 } from "./application-configuration.ts";
-import { createHostedHttpBroker } from "./hosted-http-broker.ts";
+import { createHostedHttpBroker, type HostedHttpBrokerOptions } from "./hosted-http-broker.ts";
 import type {
   HostedHttpIngressOptions,
   HostedHttpRequestAuthority,
@@ -26,6 +26,10 @@
   expectedBrokerInstanceId: string;
   allocator: HostedExecutorSessionOptions["allocator"];
   connectTransport: HostedExecutorSessionOptions["connectTransport"];
+  /** Trusted observer for a newly authenticated executor, once per attachment. */
+  observeAttachment?: HostedExecutorSessionOptions["observeAttachment"];
+  /** Trusted observer for each dispatched request, including warm reuse. */
+  observeDispatch?: HostedHttpBrokerOptions["observeDispatch"];
   /** Check current project/environment/release authority against the source API on every request. */
   authorize: {
     (authority: HostedHttpRequestAuthority, signal: AbortSignal): Promise<void>;
@@ -112,8 +116,13 @@
   const authorize = options.authorize.bind(options);
   const allocator = options.allocator;
   const connectTransport = options.connectTransport;
+  const observeAttachment = options.observeAttachment;
   const owner = new AbortController();
-  const broker = createHostedHttpBroker({ maxActive: options.maxActive, signal: owner.signal });
+  const broker = createHostedHttpBroker({
+    maxActive: options.maxActive,
+    signal: owner.signal,
+    ...(options.observeDispatch === undefined ? {} : { observeDispatch: options.observeDispatch }),
+  });
   let pendingAuthorizations = 0;
   const authorizationDrain = Promise.withResolvers<void>();
   const finishAuthorization = () => {
@@ -173,6 +182,7 @@
         expectedImage: selected.image,
         allocator,
         connectTransport,
+        ...(observeAttachment === undefined ? {} : { observeAttachment }),
         ownerSignal: owner.signal,
       },
       installation: {