fix(advisor): support manual consultations during web search

docs-site/src/content/docs/reference/configuration/advisor.md

@@ -142,6 +142,12 @@ After consultation exhaustion the `advisor` tool is removed. A repeated call rec
 paired limit result; if the worker calls it again, typed 502 `advisor_continuation_limit` ends the
 request without another hidden worker call. The bound is shared with empty-completion retries.
 
+Fetch-based web-search turns support the same manual consultation path. The search loop intercepts
+`web_search` and `advisor` calls in worker order, so enabling hosted search does not hide the
+advisor tool. Search and Advisor iterations have separate allowances, and a real client tool call
+still returns control to the client. Run-turn and image/video sidecar turns remain subject to their existing limitations.
+Usage accounting includes each worker turn once, including the leg rejected by the Advisor continuation limit.
+
 Suppression does not read Advisor strings. Automatic dedup is the server-owned ledger. A
 developer message, including one that copies the transport text, does not suppress preflight.
 

scripts/test-layout/layout-responses.json

@@ -1,4 +1,5 @@
 {
+  "advisor-web-search.test.ts": "advisor",
   "passthrough-media-type.test.ts": "responses",
   "passthrough-override.test.ts": "responses",
   "passthrough-headers.test.ts": "responses",

src/advisor/runtime.ts

@@ -139,7 +139,11 @@ export function createAdvisorRuntimePlan(deps: AdvisorRuntimeDeps): AdvisorRunti
     reason: AdvisorTrigger,
     question: string | undefined,
     evidence?: AdvisorRepairEvidence,
+    abortSignal?: AbortSignal,
   ): Promise<AdvisorConsultOutcome> => {
+    const consultationSignal = abortSignal && deps.abortSignal
+      ? AbortSignal.any([deps.abortSignal, abortSignal])
+      : abortSignal ?? deps.abortSignal;
     const current = liveSettings();
     if (!current.enabled || current.model.trim() === "") {
       return {
@@ -187,7 +191,7 @@ export function createAdvisorRuntimePlan(deps: AdvisorRuntimeDeps): AdvisorRunti
         content: formatAdvisorUnavailable(reason === "manual" ? "manual" : "preflight", result.error),
       };
     }
-    if (deps.abortSignal?.aborted) return { ok: false, isError: true, cancelled: true, content: "" };
+    if (consultationSignal?.aborted) return { ok: false, isError: true, cancelled: true, content: "" };
     const budgetKey = identify(parsed).budget;
     const requestCapped = consultations >= MAX_ADVISOR_CONSULTATIONS_PER_REQUEST;
     if (requestCapped
@@ -216,7 +220,7 @@ export function createAdvisorRuntimePlan(deps: AdvisorRuntimeDeps): AdvisorRunti
         deps.config,
         current.effort,
         current.timeoutMs,
-        deps.abortSignal,
+        consultationSignal,
         deps.baseUrlOverride,
       deps.admissionScope,
       );
@@ -334,7 +338,7 @@ export function createAdvisorRuntimePlan(deps: AdvisorRuntimeDeps): AdvisorRunti
   };
 
   const plan: AdvisorPlan = {
-    consult: (parsed, reason, question) => runConsultation(parsed, reason, question),
+    consult: (parsed, reason, question, abortSignal) => runConsultation(parsed, reason, question, undefined, abortSignal),
     // The guard's own failure/limit text goes through the same runtime-owned, marker-neutralized
     // formatter so no guard path can emit text that looks like a genuine advice wrapper.
     formatUnavailable: (kind, error) => formatAdvisorUnavailable(kind, error),

src/server/responses/advisor-plan-slot.ts

@@ -1,6 +1,7 @@
 /** Core-owned, config-scoped registration. An inactive install has no Advisor factory. */
 import type { OcxConfig, OcxParsedRequest, OcxTool } from "../../types";
 import type { AdmissionModelScope } from "../admission-model-scope";
+import type { AdvisorPlan } from "./advisor-slot";
 
 export interface AdvisorPlanInput {
   config: OcxConfig;
@@ -13,6 +14,8 @@ export interface AdvisorPlanInput {
 
 export interface RegisteredAdvisorPlan {
   tool: OcxTool;
+  consult?: AdvisorPlan["consult"];
+  formatUnavailable?: AdvisorPlan["formatUnavailable"];
   preflightInject(parsed: OcxParsedRequest): Promise<boolean>;
   attachGuard(parsed: OcxParsedRequest): void;
 }

src/server/responses/advisor-slot.ts

@@ -69,6 +69,7 @@ export interface AdvisorPlan {
     parsed: OcxParsedRequest,
     reason: "manual",
     question: string | undefined,
+    abortSignal?: AbortSignal,
   ): Promise<AdvisorConsultOutcome>;
   /**
    * Runtime-owned text for a failed or limited consultation. The guard never composes failure
@@ -78,6 +79,68 @@ export interface AdvisorPlan {
   formatUnavailable(kind: "manual" | "limit", error: string): string;
 }
 
+export interface AdvisorSidecarCall {
+  id: string;
+  name: string;
+  question?: string;
+  providerMetadata?: import("../../types").OcxProviderOpaqueToolCallMetadata;
+}
+
+/**
+ * Shares continuation/consultation bounds between the normal guard and fetch sidecar loops.
+ * Each leg is reserved before any expert await, while repeated calls receive one bounded limit
+ * result instead of starting another consultation. The tool result preserves the worker call id.
+ */
+export function createAdvisorSidecarStep(plan: AdvisorPlan) {
+  let consultations = 0;
+  let continuations = 0;
+  let finalContinuationUsed = false;
+  return {
+    beginLeg(): Extract<AdapterEvent, { type: "error" }> | undefined {
+      const final = consultations >= MAX_ADVISOR_CONSULTATIONS_PER_REQUEST;
+      if (continuations >= MAX_ADVISOR_CONTINUATIONS_PER_REQUEST || finalContinuationUsed) {
+        return {
+          type: "error",
+          status: 502,
+          errorType: "advisor_continuation_limit",
+          message: "Advisor worker continuation limit reached; no further worker calls were sent.",
+        };
+      }
+      continuations += 1;
+      finalContinuationUsed ||= final;
+      return undefined;
+    },
+    async consultCall(parsed: OcxParsedRequest, call: AdvisorSidecarCall, abortSignal?: AbortSignal) {
+      let outcome: AdvisorConsultOutcome;
+      if (consultations >= MAX_ADVISOR_CONSULTATIONS_PER_REQUEST) {
+        outcome = {
+          ok: false,
+          isError: true,
+          content: plan.formatUnavailable("limit", "consultation limit reached for this request"),
+        };
+      } else {
+        consultations += 1;
+        try {
+          outcome = await plan.consult(parsed, "manual", call.question, abortSignal);
+        } catch (error) {
+          if (isTranslatorBudgetExceededError(error)) throw error;
+          outcome = {
+            ok: false,
+            isError: true,
+            content: plan.formatUnavailable("manual", error instanceof Error ? error.message : String(error)),
+          };
+        }
+      }
+      return { outcome, message: advisorToolResult(call, outcome, Date.now()) };
+    },
+    exhausted(): boolean {
+      return consultations >= MAX_ADVISOR_CONSULTATIONS_PER_REQUEST;
+    },
+  };
+}
+
+export type AdvisorSidecarStep = ReturnType<typeof createAdvisorSidecarStep>;
+
 interface HeldAdvisorCall {
   id: string;
   name: string;
@@ -90,10 +153,10 @@ interface HeldAdvisorCall {
 
 function parseAdvisorArgs(call: HeldAdvisorCall, retention: AdvisorRetention): { question?: string } {
   // Reserve both the joined input and parsed JSON before either is materialized.
-  return retention.copy(call.argumentBytes * 2 + 4, () => parseArgs(call.args.join("")));
+  return retention.copy(call.argumentBytes * 2 + 4, () => parseAdvisorQuestionArgs(call.args.join("")));
 }
 
-function parseArgs(argsBuf: string): { question?: string } {
+export function parseAdvisorQuestionArgs(argsBuf: string): { question?: string } {
   if (argsBuf.trim() === "") return {};
   try {
     const parsed: unknown = JSON.parse(argsBuf);
@@ -129,7 +192,7 @@ function assistantMessageFromLeg(
 }
 
 function advisorToolResult(
-  call: HeldAdvisorCall,
+  call: AdvisorSidecarCall,
   outcome: AdvisorConsultOutcome,
   timestamp: number,
 ): OcxToolResultMessage {
@@ -163,11 +226,8 @@ export function createAdvisorStreamGuard(options: AdvisorGuardOptions): AsyncGen
 }
 export function createAdvisorGuard(plan: AdvisorPlan): NonNullable<OcxParsedRequest["_advisorGuard"]> {
   // Shared across invocations of this request's guard, including an empty-completion retry.
-  let consultations = 0;
-  let continuations = 0;
-  let finalContinuationUsed = false;
+  const sidecar = createAdvisorSidecarStep(plan);
   return async function* guardAdvisorStream(options: Omit<AdvisorGuardOptions, "plan">): AsyncGenerator<AdapterEvent> {
-    const maxConsultations = MAX_ADVISOR_CONSULTATIONS_PER_REQUEST;
     let parsed = options.parsed;
     let accumulatedUsage: OcxUsage | undefined;
     let source: AsyncIterable<AdapterEvent> = options.firstEvents;
@@ -275,61 +335,28 @@ export function createAdvisorGuard(plan: AdvisorPlan): NonNullable<OcxParsedRequ
 
       // Consume this leg's done — the continuation replaces it.
       accumulatedUsage = retention.usage(accumulatedUsage, terminalEvent?.usage);
-      if (continuations >= MAX_ADVISOR_CONTINUATIONS_PER_REQUEST || finalContinuationUsed) {
-        yield {
-          type: "error",
-          status: 502,
-          errorType: "advisor_continuation_limit",
-          message: "Advisor worker continuation limit reached; no further worker calls were sent.",
-          ...(accumulatedUsage ? { usage: accumulatedUsage } : {}),
-        };
+      const continuationError = sidecar.beginLeg();
+      if (continuationError) {
+        yield accumulatedUsage ? { ...continuationError, usage: accumulatedUsage } : continuationError;
         return;
       }
-      // A repeated call after tool removal gets one last paired limit result, never a loop.
-      const isFinalContinuation = consultations >= maxConsultations;
-      // Reserve before any consultation await, so simultaneous guard entries share the bound.
-      continuations += 1;
-      finalContinuationUsed ||= isFinalContinuation;
       const timestamp = Date.now();
       const assistant = retention.history(retention.replayReservation(), () => assistantMessageFromLeg(legEvents, advisorCalls, timestamp, retention));
       const messages: OcxMessage[] = [...parsed.context.messages];
       if (assistant) messages.push(assistant);
 
       for (const call of advisorCalls) {
-        if (consultations >= maxConsultations) {
-          // Runtime-owned limit text: not advice, and never a genuine advice wrapper.
-          const outcome = {
-            ok: false,
-            isError: true,
-            content: plan.formatUnavailable("limit", "consultation limit reached for this request"),
-          };
-          messages.push(retention.history(jsonUtf8Bytes(outcome) + jsonUtf8Bytes(call.id) + 256, () => advisorToolResult(call, outcome, timestamp))!);
-          continue;
-        }
-        consultations += 1;
         // Later calls in the same leg must see what came before them: the rebuilt worker turn
         // and every earlier advisor result in this leg's accumulated messages.
         const consultParsed: OcxParsedRequest = {
           ...parsed,
           context: { ...parsed.context, messages },
         };
-        let outcome: AdvisorConsultOutcome;
-        try {
-          outcome = await plan.consult(consultParsed, "manual", call.question);
-        } catch (error) {
-          if (isTranslatorBudgetExceededError(error)) throw error;
-          // Runtime-owned failure text only: the implementation neutralizes untrusted exception
-          // text so it can never forge a genuine advice wrapper.
-          outcome = {
-            ok: false,
-            isError: true,
-            content: plan.formatUnavailable("manual", error instanceof Error ? error.message : String(error)),
-          };
-        }
-        messages.push(retention.history(jsonUtf8Bytes(outcome) + jsonUtf8Bytes(call.id) + 256, () => advisorToolResult(call, outcome, timestamp))!);
+        const consulted = await sidecar.consultCall(consultParsed, call);
+        messages.push(retention.history(jsonUtf8Bytes(consulted.outcome) + jsonUtf8Bytes(call.id) + 256, () => consulted.message)!);
       }
 
-      const exhausted = consultations >= maxConsultations;
+      const exhausted = sidecar.exhausted();
       const tools = exhausted
         ? parsed.context.tools?.filter(tool => !tool.advisor && tool.name !== ADVISOR_TOOL_NAME)
         : parsed.context.tools;

src/server/responses/sidecar-execution.ts

@@ -9,7 +9,7 @@ import type { ResponsesSendBudget } from "./request-send-budget";
 import { formatErrorResponse } from "../../bridge";
 import { planWebSearch, buildWebSearchTool, runWithWebSearch } from "../../web-search";
 import { createRegisteredAdvisorPlan } from "./advisor-plan-slot";
-import { ADVISOR_TOOL_NAME } from "./advisor-slot";
+import { ADVISOR_TOOL_NAME, createAdvisorSidecarStep } from "./advisor-slot";
 import { buildToolBridgeMaps } from "./collaboration";
 import {
   planImageBridge,
@@ -655,7 +655,21 @@ export async function executeResponsesSidecars(
   // Placed BEFORE the runTurn early-return for non-runTurn adapters so dual-tool turns dispatch
   // through web-search instead of being swallowed. runTurn adapters never enter this branch.
   if (canRunWebSearch && wsPlan) {
-    parsed.context.tools = [...(parsed.context.tools ?? []), buildWebSearchTool()];
+    const webSearchAdvisor = advisorPlan && !routedCompaction
+      && typeof advisorPlan.consult === "function" && typeof advisorPlan.formatUnavailable === "function"
+      ? advisorPlan
+      : undefined;
+    parsed.context.tools = [
+      ...(parsed.context.tools ?? []).filter(tool => !webSearchAdvisor || (!tool.advisor && tool.name !== ADVISOR_TOOL_NAME)),
+      buildWebSearchTool(),
+      ...(webSearchAdvisor ? [webSearchAdvisor.tool] : []),
+    ];
+    if (webSearchAdvisor) {
+      requestState.toolBridgeMaps = buildToolBridgeMaps(parsed);
+      translatorBudget.chargeRetained(new TextEncoder().encode(ADVISOR_TOOL_NAME).byteLength * 2, {
+        kind: "retained_collectors",
+      });
+    }
     // Resolve the mutable route at send time: a 429 rotation replaces route.provider, so retaining
     // one pre-rotation providerFetch would keep the old credential and transport pin.
     const routedProviderFetch = ((input: Parameters<typeof globalThis.fetch>[0], init?: RequestInit) =>
@@ -666,6 +680,7 @@ export async function executeResponsesSidecars(
       })(input, init)) as typeof globalThis.fetch;
     const loopResponse = await runWithWebSearch({
       parsed, adapter: transportState.adapter,
+      ...(webSearchAdvisor ? { advisor: createAdvisorSidecarStep(webSearchAdvisor as Parameters<typeof createAdvisorSidecarStep>[0]) } : {}),
       chatRawReasoningVisible: options.chatRawReasoningVisible,
       ...(transportState.adapter.name === "anthropic" ? { thinkingRecovery } : {}),
       clientToolCatalogRestriction: route.provider.authMode === "forward" ? undefined : requestState.clientToolCatalogRestriction,
@@ -773,9 +788,8 @@ export async function executeResponsesSidecars(
 
   // Advisor synthetic tool + stream guard: attach ONLY when this function is about to hand the
   // request back to the normal translated exchange (no sidecar loop claimed the turn, not a
-  // run-turn adapter, not compaction). The web-search/image loops own their own event handling;
-  // run-turn adapters execute their own loop; both would leak the synthetic tool call they
-  // cannot intercept. Preflight (above) still applies to those paths — only the tool does not.
+  // run-turn adapter, not compaction). Web-search handles advisor in its own loop above; the
+  // image/video and run-turn loops cannot intercept it. Preflight still applies to those paths.
   if (advisorPlan && !routedCompaction && !transportState.adapter.runTurn && !wsPlan && !imgPlan && !vidPlan) {
     // A client-declared tool named "advisor" would collide with the synthetic injection (one
     // wire name, two schemas); the synthetic runtime owns the name for this turn.