fix(agents): one root call per sub-agent broadcast, none without recipients (#2252)

.changeset/facet-broadcast-root-calls.md

@@ -0,0 +1,5 @@
+---
+"agents": patch
+---
+
+Sub-agent broadcasts cost at most one root call, and none when nobody would receive them. Outside the client frame that started the work, every facet broadcast used to resolve the root through `getAgentByName()` (an `__unsafe_ensureInitialized` round trip) before calling `_cf_broadcastToSubAgent`, so a streamed answer made two billed root requests per chunk. The root endpoints facets call now start the root's lifecycle themselves, so facets use a plain stub, and a root that was evicted still runs `onStart()` before serving the call. A facet also skips the root entirely when its hydrated connection mirror shows no connection outside `without`; until the mirror is hydrated, broadcasts route to the root as before.

packages/agents/src/dynamic-agents/dynamic-agents.ts

@@ -19,9 +19,7 @@ import {
   parseSubAgentPath,
   type AgentPathStep
 } from "../sub-routing";
-import { getAgentByName } from "../agent-routing";
 import { camelCaseToKebabCase, isInternalJsStubProp } from "../utils";
-import type { Agent } from "../index";
 import { agentPathKey, isValidParentPath } from "./identity";
 import { DynamicAgentRegistry } from "./registry";
 import {
@@ -104,6 +102,14 @@ export class DynamicAgentsInternal extends LifecycleCapability {
    */
   #virtualConnections = new Map<string, StoredDynamicAgentConnection>();
 
+  /**
+   * Whether `#virtualConnections` has been seeded from the root's sockets
+   * since this instance started. Until then the mirror can miss hibernated
+   * root sockets, so it cannot prove a broadcast has no recipients.
+   */
+  #virtualConnectionsHydrated = false;
+  #virtualConnectionsHydration?: Promise<void>;
+
   /** Per-connection operation queues (send/setState/close ordering). */
   #connectionOperationTails = new Map<string, Promise<void>>();
 
@@ -189,7 +195,7 @@ export class DynamicAgentsInternal extends LifecycleCapability {
       if (this.#host._isFacet) {
         await (await this.rootAlarmOwner())._cf_cleanupFacetPrefix(stalePath);
       } else {
-        await this.#host._cf_cleanupFacetPrefix(stalePath);
+        await this.cleanupPrefix(stalePath);
       }
       return false;
     }
@@ -234,10 +240,12 @@ export class DynamicAgentsInternal extends LifecycleCapability {
       );
     }
 
-    return (await getAgentByName<Cloudflare.Env, Agent>(
-      binding as unknown as DurableObjectNamespace<Agent>,
-      root.name
-    )) as unknown as RootFacetRpcSurface;
+    // Every root endpoint on this surface starts the root's lifecycle itself,
+    // so a plain stub is enough. `getAgentByName()` would spend an extra
+    // `__unsafe_ensureInitialized` round trip on every broadcast and send.
+    return binding.get(
+      binding.idFromName(root.name)
+    ) as unknown as RootFacetRpcSurface;
   }
 
   rootResolvesToSelf(): boolean {
@@ -521,7 +529,7 @@ export class DynamicAgentsInternal extends LifecycleCapability {
     // upfront so we don't have to make an extra round trip back from
     // each intermediate hop.
     if (this.#host._parentPath.length === 0) {
-      await this.#host._cf_cleanupFacetPrefix(targetPath);
+      await this.cleanupPrefix(targetPath);
     }
 
     if (selfPath.length === targetPath.length - 1) {
@@ -725,7 +733,7 @@ export class DynamicAgentsInternal extends LifecycleCapability {
       const root = await this.rootAlarmOwner();
       await root._cf_cleanupFacetPrefix(childPath);
     } else {
-      await this.#host._cf_cleanupFacetPrefix(childPath);
+      await this.cleanupPrefix(childPath);
     }
 
     // Idempotent: make `ctx.facets.delete` tolerant of missing keys.
@@ -746,6 +754,7 @@ export class DynamicAgentsInternal extends LifecycleCapability {
   /** Drop all facet-side virtual connections (test/rehydration hook). */
   clearVirtualConnections(): void {
     this.#virtualConnections.clear();
+    this.#virtualConnectionsHydrated = false;
   }
 
   /** Facet-side lookup of a virtual connection by id. */
@@ -889,9 +898,41 @@ export class DynamicAgentsInternal extends LifecycleCapability {
     message: string | ArrayBuffer | ArrayBufferView,
     without?: string[]
   ): Promise<void> {
+    if (this.#broadcastHasNoRecipients(without)) return;
     await this.routeBroadcast(this.#host.selfPath, message, without);
   }
 
+  /**
+   * Whether a hydrated mirror proves no root socket targets this facet
+   * outside `without`. Stale mirror entries only cost a root call; a socket
+   * missing from a hydrated mirror has not had its connect forwarded here
+   * yet, so it has not been through `onConnect`. Synchronous so a skipped
+   * broadcast never reorders later operations.
+   */
+  #broadcastHasNoRecipients(without: string[] | undefined): boolean {
+    if (!this.#virtualConnectionsHydrated) {
+      this.#hydrateVirtualConnectionsInBackground();
+      return false;
+    }
+    for (const id of this.#virtualConnections.keys()) {
+      if (!without?.includes(id)) return false;
+    }
+    return true;
+  }
+
+  #hydrateVirtualConnectionsInBackground(): void {
+    if (this.#virtualConnectionsHydration) return;
+    const hydration = this.hydrateConnectionsFromRoot()
+      .catch(() => {
+        // Best-effort: the next broadcast retries.
+      })
+      .finally(() => {
+        this.#virtualConnectionsHydration = undefined;
+      });
+    this.#virtualConnectionsHydration = hydration;
+    this.#host.ctx.waitUntil(hydration);
+  }
+
   async broadcastToPath(
     ownerPath: ReadonlyArray<AgentPathStep>,
     message: string | ArrayBuffer | ArrayBufferView,
@@ -1446,8 +1487,12 @@ export class DynamicAgentsInternal extends LifecycleCapability {
     const root = await this.rootAlarmOwner();
     const metas = await root._cf_subAgentConnectionMetas(this.#host.selfPath);
     for (const meta of metas) {
-      this.#virtualConnections.set(meta.id, { meta });
+      // A connection forwarded while the root read was in flight is fresher.
+      if (!this.#virtualConnections.has(meta.id)) {
+        this.#virtualConnections.set(meta.id, { meta });
+      }
     }
+    this.#virtualConnectionsHydrated = true;
   }
 
   getRawConnectionState(connection: Connection): unknown {

packages/agents/src/dynamic-agents/host.ts

@@ -43,9 +43,6 @@ export interface DynamicAgentHostPort {
   ): boolean;
   hasSubAgent(className: string, name: string): boolean;
   _cf_resolveSubAgent(className: string, name: string): Promise<unknown>;
-  _cf_cleanupFacetPrefix(
-    ownerPath: ReadonlyArray<AgentPathStep>
-  ): Promise<void>;
   _cf_routeLifecycle(
     target: LifecycleRouteAddress | undefined,
     envelope: LifecycleRouteEnvelope

packages/agents/src/index.ts

@@ -3169,10 +3169,11 @@ export class Agent<
   }
 
   /** Single native-RPC aperture for routed Lifecycle capabilities. */
-  _cf_routeLifecycle(
+  async _cf_routeLifecycle(
     target: LifecycleRouteAddress | undefined,
     envelope: LifecycleRouteEnvelope
   ): Promise<unknown> {
+    await this.__unsafe_ensureInitialized();
     return this._dynamicAgents.routeLifecycle(target, envelope);
   }
 
@@ -3195,6 +3196,7 @@ export class Agent<
   async _cf_cleanupFacetPrefix(
     ownerPath: ReadonlyArray<AgentPathStep>
   ): Promise<void> {
+    await this.__unsafe_ensureInitialized();
     await this._dynamicAgents.cleanupPrefix(ownerPath);
   }
 
@@ -3204,9 +3206,10 @@ export class Agent<
    * physical alarm, so this lets facet work use the root alarm heartbeat.
    * @internal
    */
-  _cf_acquireFacetKeepAlive(
+  async _cf_acquireFacetKeepAlive(
     ownerPath: ReadonlyArray<AgentPathStep>
   ): Promise<string> {
+    await this.__unsafe_ensureInitialized();
     return this._dynamicAgents.acquireKeepAlive(ownerPath);
   }
 
@@ -3215,8 +3218,9 @@ export class Agent<
    * Idempotent so disposer calls can safely race or run twice.
    * @internal
    */
-  _cf_releaseFacetKeepAlive(token: string): Promise<void> {
-    return this._dynamicAgents.releaseKeepAlive(token);
+  async _cf_releaseFacetKeepAlive(token: string): Promise<void> {
+    await this.__unsafe_ensureInitialized();
+    await this._dynamicAgents.releaseKeepAlive(token);
   }
 
   /**
@@ -3225,22 +3229,24 @@ export class Agent<
    * The facet remains authoritative for snapshots and recovery hooks.
    * @internal
    */
-  _cf_registerFacetRun(
+  async _cf_registerFacetRun(
     ownerPath: ReadonlyArray<AgentPathStep>,
     runId: string
   ): Promise<void> {
-    return this._dynamicAgents.registerRun(ownerPath, runId);
+    await this.__unsafe_ensureInitialized();
+    await this._dynamicAgents.registerRun(ownerPath, runId);
   }
 
   /**
    * Remove a completed facet fiber from the root-side index.
    * @internal
    */
-  _cf_unregisterFacetRun(
+  async _cf_unregisterFacetRun(
     ownerPath: ReadonlyArray<AgentPathStep>,
     runId: string
   ): Promise<void> {
-    return this._dynamicAgents.unregisterRun(ownerPath, runId);
+    await this.__unsafe_ensureInitialized();
+    await this._dynamicAgents.unregisterRun(ownerPath, runId);
   }
 
   /**
@@ -4720,10 +4726,11 @@ export class Agent<
    * cleanup as `parent.deleteSubAgent(Cls, name)` from the parent.
    * @internal
    */
-  _cf_destroyDescendantFacet(
+  async _cf_destroyDescendantFacet(
     targetPath: ReadonlyArray<AgentPathStep>
   ): Promise<void> {
-    return this._dynamicAgents.destroyDescendant(targetPath);
+    await this.__unsafe_ensureInitialized();
+    await this._dynamicAgents.destroyDescendant(targetPath);
   }
 
   /**
@@ -5072,34 +5079,39 @@ export class Agent<
     message: string | ArrayBuffer | ArrayBufferView,
     without?: string[]
   ): Promise<void> {
+    await this.__unsafe_ensureInitialized();
     await this._dynamicAgents.broadcastToPath(ownerPath, message, without);
   }
 
-  _cf_subAgentConnectionMetas(
+  async _cf_subAgentConnectionMetas(
     ownerPath: ReadonlyArray<AgentPathStep>
   ): Promise<SubAgentConnectionMeta[]> {
+    await this.__unsafe_ensureInitialized();
     return this._dynamicAgents.connectionMetas(ownerPath);
   }
 
-  _cf_sendToSubAgentConnection(
+  async _cf_sendToSubAgentConnection(
     connectionId: string,
     message: string | ArrayBuffer | ArrayBufferView
   ): Promise<void> {
-    return this._dynamicAgents.sendToConnection(connectionId, message);
+    await this.__unsafe_ensureInitialized();
+    await this._dynamicAgents.sendToConnection(connectionId, message);
   }
 
-  _cf_closeSubAgentConnection(
+  async _cf_closeSubAgentConnection(
     connectionId: string,
     code?: number,
     reason?: string
   ): Promise<void> {
-    return this._dynamicAgents.closeConnection(connectionId, code, reason);
+    await this.__unsafe_ensureInitialized();
+    await this._dynamicAgents.closeConnection(connectionId, code, reason);
   }
 
-  _cf_setSubAgentConnectionState(
+  async _cf_setSubAgentConnectionState(
     connectionId: string,
     state: unknown
   ): Promise<unknown> {
+    await this.__unsafe_ensureInitialized();
     return this._dynamicAgents.setConnectionState(connectionId, state);
   }
 

packages/agents/src/tests/agents/sub-agent.ts

@@ -1230,11 +1230,40 @@ export class TestSubAgentParent extends Agent {
     this._subAgentBroadcastFailuresRemaining += 1;
   }
 
+  private _subAgentBroadcastCalls = 0;
+  private _startsInThisInstance = 0;
+
+  onStart(): void {
+    this._startsInThisInstance += 1;
+  }
+
+  /** Read without an RPC so the probe itself cannot start the agent. */
+  get startsInThisInstance(): number {
+    return this._startsInThisInstance;
+  }
+
+  subAgentBroadcastCallCount(): number {
+    return this._subAgentBroadcastCalls;
+  }
+
+  async broadcastFromSubAgentDetached(
+    childName: string,
+    messages: string[]
+  ): Promise<void> {
+    const child = await this.subAgent(SlowReplySubAgent, childName);
+    await child.broadcastDetached(messages);
+  }
+
+  abortSlowReplySubAgent(childName: string): void {
+    this.abortSubAgent(SlowReplySubAgent, childName);
+  }
+
   override async _cf_broadcastToSubAgent(
     ownerPath: ReadonlyArray<{ className: string; name: string }>,
     message: string | ArrayBuffer | ArrayBufferView,
     without?: string[]
   ): Promise<void> {
+    this._subAgentBroadcastCalls += 1;
     if (this._subAgentBroadcastFailuresRemaining > 0) {
       this._subAgentBroadcastFailuresRemaining -= 1;
       throw new Error("TestSubAgentParent broadcast forwarding failed");
@@ -2621,6 +2650,39 @@ export class SlowReplySubAgent extends Agent {
     return "broadcast";
   }
 
+  /** Broadcasts and then sends directly, both during the current frame. */
+  @callable()
+  broadcastThenSendNow(broadcast: string, direct: string): string {
+    const { connection } = getCurrentAgent();
+    if (!connection) {
+      throw new Error(
+        "SlowReplySubAgent.broadcastThenSendNow requires an active connection"
+      );
+    }
+
+    this.broadcast(broadcast);
+    connection.send(direct);
+    return "sent";
+  }
+
+  /** Broadcasts after the current frame completes, optionally skipping the caller. */
+  @callable()
+  broadcastMessagesAfterDelay(messages: string[], withoutSelf = false): string {
+    const { connection } = getCurrentAgent();
+    const without = withoutSelf && connection ? [connection.id] : undefined;
+    this.ctx.waitUntil(
+      new Promise((resolve) => setTimeout(resolve, 50)).then(() => {
+        for (const message of messages) this.broadcast(message, without);
+      })
+    );
+    return "scheduled";
+  }
+
+  /** Broadcasts from a parent RPC, outside any client frame. */
+  broadcastDetached(messages: string[]): void {
+    for (const message of messages) this.broadcast(message);
+  }
+
   /** Schedules consecutive messages after the current frame completes. */
   @callable()
   sendConnectionMessagesAfterDelay(messages: string[]): string {