diff --git a/src/config.ts b/src/config.ts index 106dadd2b6a..88bcf1fafd2 100644 --- a/src/config.ts +++ b/src/config.ts @@ -1262,6 +1262,8 @@ const configSchema = z.object({ providerContextCapValues: z.record(z.string(), z.number().int().positive()).optional(), contextCapValue: z.number().int().positive().optional(), multiAgentGuidanceEnabled: z.boolean().optional(), + // Invalid hand edits disable only this experimental opt-in. + v2RoutedDelegationBridge: z.boolean().optional().catch(undefined), // Invalid optional recovery config must not discard unrelated provider/account state. agentTaskRecovery: agentTaskRecoverySchema.optional().catch(undefined), // Same rationale: a bad notify section must not cost the operator their providers. diff --git a/src/responses/state.ts b/src/responses/state.ts index f9195196a25..dcbdceca55f 100644 --- a/src/responses/state.ts +++ b/src/responses/state.ts @@ -2314,6 +2314,12 @@ export function markBodyNonPersistable(body: unknown): void { if (body && typeof body === "object") nonPersistableBodies.add(body as object); } +export function copyBodyNonPersistableMarker(source: unknown, target: unknown): void { + if (!source || typeof source !== "object" || Array.isArray(source)) return; + if (!target || typeof target !== "object" || Array.isArray(target)) return; + if (nonPersistableBodies.has(source as object)) nonPersistableBodies.add(target as object); +} + export function rememberResponseState( requestBody: unknown, response: { id?: unknown; output?: unknown; status?: unknown; incomplete_details?: unknown }, diff --git a/src/server/management/agent-settings-routes.ts b/src/server/management/agent-settings-routes.ts index d9757d5b112..6a92a8d4db5 100644 --- a/src/server/management/agent-settings-routes.ts +++ b/src/server/management/agent-settings-routes.ts @@ -253,6 +253,7 @@ export async function handleAgentSettingsRoutes(ctx: ManagementContext): Promise // max_depth is V1-only upstream; this is the global-flag statement, derived // server-side so no client can present it as an effective V2 limit. agentsMaxDepthAppliesWhenV2Disabled: !enabled, + v2RoutedDelegationBridge: config.v2RoutedDelegationBridge === true, }); } if (url.pathname === "/api/v2" && req.method === "PUT") { @@ -265,6 +266,7 @@ export async function handleAgentSettingsRoutes(ctx: ManagementContext): Promise agentsMaxDepth?: unknown; subagentDeveloperInstructions?: unknown; multiAgentModeHintText?: unknown; + v2RoutedDelegationBridge?: unknown; }; try { body = await readManagementJsonBody(req); } catch (error) { rethrowManagementBodyTooLarge(error); return jsonResponse({ error: "invalid JSON body" }, 400); } const wantsFlag = body.enabled !== undefined; @@ -275,8 +277,9 @@ export async function handleAgentSettingsRoutes(ctx: ManagementContext): Promise const wantsMaxDepth = body.agentsMaxDepth !== undefined; const wantsSubagentInstructions = body.subagentDeveloperInstructions !== undefined; const wantsModeHintText = body.multiAgentModeHintText !== undefined; - if (!wantsFlag && !wantsThreads && !wantsMode && !wantsKeepNative && !wantsAgentsEnabled && !wantsMaxDepth && !wantsSubagentInstructions && !wantsModeHintText) { - return jsonResponse({ error: "body must set enabled, multiAgentMode, keepNativeChatGptOnV1, maxConcurrentThreadsPerSession, agentsEnabled, agentsMaxDepth, subagentDeveloperInstructions, and/or multiAgentModeHintText" }, 400); + const wantsV2RoutedDelegationBridge = body.v2RoutedDelegationBridge !== undefined; + if (!wantsFlag && !wantsThreads && !wantsMode && !wantsKeepNative && !wantsAgentsEnabled && !wantsMaxDepth && !wantsSubagentInstructions && !wantsModeHintText && !wantsV2RoutedDelegationBridge) { + return jsonResponse({ error: "body must set enabled, multiAgentMode, keepNativeChatGptOnV1, maxConcurrentThreadsPerSession, agentsEnabled, agentsMaxDepth, subagentDeveloperInstructions, multiAgentModeHintText, and/or v2RoutedDelegationBridge" }, 400); } if (wantsFlag && typeof body.enabled !== "boolean") return jsonResponse({ error: "body.enabled must be a boolean" }, 400); if (wantsMode && body.multiAgentMode !== "v1" && body.multiAgentMode !== "default" && body.multiAgentMode !== "v2") { @@ -285,6 +288,9 @@ export async function handleAgentSettingsRoutes(ctx: ManagementContext): Promise if (wantsKeepNative && typeof body.keepNativeChatGptOnV1 !== "boolean") { return jsonResponse({ error: "body.keepNativeChatGptOnV1 must be a boolean" }, 400); } + if (wantsV2RoutedDelegationBridge && typeof body.v2RoutedDelegationBridge !== "boolean") { + return jsonResponse({ error: "body.v2RoutedDelegationBridge must be a boolean" }, 400); + } if (wantsThreads && (typeof body.maxConcurrentThreadsPerSession !== "number" || !Number.isInteger(body.maxConcurrentThreadsPerSession) || body.maxConcurrentThreadsPerSession < 1)) { return jsonResponse({ error: "body.maxConcurrentThreadsPerSession must be an integer >= 1" }, 400); } @@ -386,6 +392,19 @@ export async function handleAgentSettingsRoutes(ctx: ManagementContext): Promise // asserts this route file contains no direct write primitive, and matches on the // symbol name even inside a comment. const scalarWrites: Array<{ field: string; run: () => { ok: true; changed: boolean } | { ok: false; error: string } }> = []; + if (wantsV2RoutedDelegationBridge) scalarWrites.push({ + field: "v2RoutedDelegationBridge", + run: () => { + try { + if (body.v2RoutedDelegationBridge === false) deleteConfigTopLevelKey(config, "v2RoutedDelegationBridge"); + else config.v2RoutedDelegationBridge = true; + saveConfigPreservingClaudeCode(config); + return { ok: true, changed: true }; + } catch (error) { + return { ok: false, error: error instanceof Error ? error.message : String(error) }; + } + }, + }); if (wantsAgentsEnabled) scalarWrites.push({ field: "agentsEnabled", run: () => setAgentsEnabled(body.agentsEnabled as boolean | null) }); if (wantsMaxDepth) scalarWrites.push({ field: "agentsMaxDepth", run: () => setAgentsMaxDepth(body.agentsMaxDepth as number | null) }); if (wantsSubagentInstructions) scalarWrites.push({ field: "subagentDeveloperInstructions", run: () => setSubagentDeveloperInstructions(body.subagentDeveloperInstructions as string | null) }); @@ -425,6 +444,7 @@ export async function handleAgentSettingsRoutes(ctx: ManagementContext): Promise multiAgentModeHintText: getMultiAgentModeHintText(), multiAgentModeHintRecommendation: MULTI_AGENT_MODE_HINT_RECOMMENDATION, agentsMaxDepthAppliesWhenV2Disabled: !enabled, + v2RoutedDelegationBridge: config.v2RoutedDelegationBridge === true, warnings, catalogRefresh, }); diff --git a/src/server/responses/core.ts b/src/server/responses/core.ts index cccd942026e..35311316d60 100644 --- a/src/server/responses/core.ts +++ b/src/server/responses/core.ts @@ -41,11 +41,13 @@ import { FORWARD_HEADERS, sanitizeReasoningInputContent } from "../../adapters/o import { XaiToolSchemaCompatibilityError } from "../../adapters/xai-tool-schema"; import { copyPreviousResponseReplayProvenance, + copyBodyNonPersistableMarker, expandPreviousResponseInput, markBodyNonPersistable, previousResponseProviderState, previousResponseReplayFailure, previousResponseScopeMismatch, + previousResponseReplayPrefixLength, rememberResponseState, } from "../../responses/state"; import { @@ -384,6 +386,14 @@ import { import type { EffectiveSubagentRoster, SpawnAgentSurface } from "../../codex/catalog"; import { buildToolBridgeMaps, collabSurface, injectDeveloperMessage, multiAgentGuidanceText } from "./collaboration"; +import { isMultiAgentV2Enabled } from "../../codex/features"; +import { + createV2RoutedDelegationSseRewrite, + injectV2RoutedDelegationBridge, + rewriteV2RoutedDelegationCallsInJson, + type V2RoutedDelegationBridgeContext, +} from "./v2-routed-delegation-bridge"; +import { decideV2RoutedDelegationBridge } from "./v2-routed-delegation-policy"; import { mapCodexAuthContextErrorToResponse, nativeMainRefreshFailureResponse } from "./codex-auth-error"; import { hasUnreadableEncryptedAgentTask, looksLikeBackendCiphertext, sanitizeEncryptedContentInPlace } from "./encrypted-payload"; import { fetchWithHeaderTimeout, providerFetch, safeHostLabel, safeOriginLabel, storedPoolReplayDispatchNotifier, type ProviderFetchOptions } from "./fetch-helpers"; @@ -509,6 +519,18 @@ function readProviderContinuationOwner( return { kind: "valid", owner: { ...owner } }; } +function requestHasSubagentMarker(headers: Headers): boolean { + if (headers.has("x-openai-subagent")) return true; + const metadata = headers.get("x-codex-turn-metadata"); + if (!metadata) return false; + try { + const parsedMetadata = JSON.parse(metadata) as { subagent_kind?: unknown }; + return typeof parsedMetadata.subagent_kind === "string" && parsedMetadata.subagent_kind.length > 0; + } catch { + return false; + } +} + function providerContinuationPayload( state: OcxProviderContinuationState | undefined, ): OcxProviderContinuationState | undefined { @@ -3502,6 +3524,7 @@ async function handleResponsesInner( return formatErrorResponse(400, "invalid_request_error", err instanceof Error ? err.message : String(err)); } options.onRequestBodyRead?.(); + let v2RoutedDelegationBridge: V2RoutedDelegationBridgeContext | undefined; const responseStateOptions = (force = false): { force?: boolean; clientThreadId?: string } => ({ ...(force ? { force: true } : {}), ...(parsed._clientThreadId ? { clientThreadId: parsed._clientThreadId } : {}), @@ -3546,6 +3569,7 @@ async function handleResponsesInner( let route: RouteResult; let credentialDomainWasRewritten = false; + let shadowIntercepted = false; try { // A `compaction_trigger` turn may name a bare native model the operator has // no canonical OpenAI route for (#2901). Only the initial compaction route @@ -3567,6 +3591,7 @@ async function handleResponsesInner( } catch { /* Native Codex helper calls remain OpenAI-owned without an enabled OpenAI route. */ } const targetRoute = resolveRoute(_sci.model); if (shouldInterceptShadowCall(parsed.modelId, _sci.sourceModels, sourceIdentity, targetRoute)) { + shadowIntercepted = true; credentialDomainWasRewritten = true; const _sciOriginal = parsed.modelId; parsed.modelId = _sci.model; @@ -3899,6 +3924,44 @@ async function handleResponsesInner( ); } + const bridgeEnabled = config.v2RoutedDelegationBridge === true; + const bridgeDecision = bridgeEnabled ? decideV2RoutedDelegationBridge({ + enabled: true, + inboundWire, + multiAgentMode: config.multiAgentMode, + upstreamV2Enabled: isMultiAgentV2Enabled(), + canonicalNativeRoute: isCanonicalOpenAiForwardProvider(route.provider), + hasSubagentMarker: requestHasSubagentMarker(req.headers), + threadSpawn, + comboAttempt: options.comboAttempt === true, + compaction: parsed._compactionRequest === true, + shadowRoute: shadowIntercepted, + collaborationSurface: collabSurface(parsed), + body: parsed._rawBody, + replayPrefixLength: previousResponseReplayPrefixLength(parsed._rawBody), + }) : undefined; + if (bridgeDecision) { + Object.assign(logCtx, { + v2BridgeDecision: bridgeDecision.decision, + ...(bridgeDecision.active ? { v2BridgeScope: bridgeDecision.scope } : {}), + }); + } + if (bridgeDecision?.active) { + try { + v2RoutedDelegationBridge = injectV2RoutedDelegationBridge(parsed); + if (v2RoutedDelegationBridge) { + copyPreviousResponseReplayProvenance( + parsed._rawBody, + v2RoutedDelegationBridge.requestStateBody, + ); + copyBodyNonPersistableMarker(parsed._rawBody, v2RoutedDelegationBridge.requestStateBody); + toolBridgeMaps = buildToolBridgeMaps(parsed, translatorBudget); + } + } catch (error) { + return formatErrorResponse(400, "invalid_request_error", error instanceof Error ? error.message : String(error)); + } + } + // Captured before normalization: whether the CLIENT asked for SSE. The // transport-neutral upstream-streaming policy below may force a bounded JSON // upstream for reliability (#875); the answer must then be reframed to SSE @@ -4700,7 +4763,7 @@ async function handleResponsesInner( && (!parsed.previousResponseId || parsed._previousResponseInputExpanded === true); const rememberPassthroughResponse = passthroughRecordEligible ? (response: { id?: unknown; output?: unknown; status?: unknown }) => - rememberResponseState(parsed._rawBody, response, undefined, responseStateOptions(true)) + rememberResponseState(v2RoutedDelegationBridge?.requestStateBody ?? parsed._rawBody, response, undefined, responseStateOptions(true)) : undefined; if (parsed.previousResponseId && !parsed._previousResponseInputExpanded) { console.warn( @@ -4968,8 +5031,15 @@ async function handleResponsesInner( response: { id?: unknown; output?: unknown; status?: unknown; model?: unknown }, ) => { if (inspectionSawUndeclaredTool) return; + const namespaceRestored = restoreRoutedNamespaceCalls(response, routedNamespaceToolAliases).value; + const bridgeRestored = v2RoutedDelegationBridge + ? JSON.parse(rewriteV2RoutedDelegationCallsInJson( + JSON.stringify(namespaceRestored), + v2RoutedDelegationBridge, + )) as typeof response + : namespaceRestored; const restored = restoreRoutedCustomCalls( - restoreAuthorizedBareNamespaceToolCalls(restoreRoutedNamespaceCalls(response, routedNamespaceToolAliases).value), + restoreAuthorizedBareNamespaceToolCalls(bridgeRestored), routedCustomToolNames, routedCustomToolRepairNames, declaredWireToolNames, @@ -5984,7 +6054,9 @@ async function handleResponsesInner( // injection at the block level, after payload rewrites. Defaults come // from the finalized OUTBOUND body — the normalized internal tool shapes // are not the Responses wire shapes the snapshot must mirror. + const bridgeSseRewrite = createV2RoutedDelegationSseRewrite(v2RoutedDelegationBridge); const blockRewrites = [ + bridgeSseRewrite ? payloadRewriteAsBlockRewrite(bridgeSseRewrite) : undefined, payloadRewrites.length > 0 ? payloadRewriteAsBlockRewrite(composeSsePayloadRewrites(...payloadRewrites)) : undefined, @@ -6235,8 +6307,12 @@ async function handleResponsesInner( restoredNamespace, authorizedBareNamespaceToolAliases, ); - const restored = restoreRoutedCustomCallsInJson( + const bridgeNormalized = rewriteV2RoutedDelegationCallsInJson( restoredAuthorizedBareNamespace, + v2RoutedDelegationBridge, + ); + const restored = restoreRoutedCustomCallsInJson( + bridgeNormalized, routedCustomToolNames, routedCustomToolRepairNames, declaredWireToolNames, diff --git a/src/server/responses/v2-routed-delegation-bridge.ts b/src/server/responses/v2-routed-delegation-bridge.ts new file mode 100644 index 00000000000..3c89a6bccb1 --- /dev/null +++ b/src/server/responses/v2-routed-delegation-bridge.ts @@ -0,0 +1,265 @@ +import type { OcxParsedRequest, OcxTool } from "../../types"; +import type { SsePayloadRewrite } from "../sse-payload-rewrite"; +import { MIRRORABLE_COLLABORATION_OPERATIONS, isRecord } from "./v2-routed-delegation-shared"; + +const MIRROR_NAMESPACE = "ocx_agents"; +const NATIVE_NAMESPACE = "collaboration"; +const GUIDANCE = "Use this routed-child mirror for collaboration operations."; +const MAX_SSE_BINDINGS = 128; +const injectedGroups = new WeakSet(); + +type RecordValue = Record; + +export interface V2RoutedDelegationBridgeContext { + readonly names: ReadonlySet; + /** Request snapshot taken before mirror injection, for continuation-cache persistence. */ + readonly requestStateBody: unknown; +} + +function rawToolLists(body: unknown, replayPrefixLength: number): unknown[][] { + if (!isRecord(body)) return []; + const lists: unknown[][] = []; + if (Array.isArray(body.tools)) lists.push(body.tools); + if (Array.isArray(body.input)) { + for (const item of body.input.slice(Math.max(0, Math.min(replayPrefixLength, body.input.length)))) { + if (isRecord(item) && item.type === "additional_tools" && Array.isArray(item.tools)) lists.push(item.tools); + } + } + return lists; +} + +function requestStateBody(body: unknown): unknown { + if (!isRecord(body)) return body; + const cloneTools = (tools: unknown[]) => tools.map(tool => ( + isRecord(tool) && tool.type === "namespace" && Array.isArray(tool.tools) + ? { ...tool, tools: [...tool.tools] } + : tool + )); + if (!Array.isArray(body.tools) && !Array.isArray(body.input)) return body; + // Only tool catalogs are mutated below. Clone that narrow path so the continuation + // cache retains the caller's catalog without copying unrelated context. + return { + ...body, + ...(Array.isArray(body.tools) ? { tools: cloneTools(body.tools) } : {}), + ...(Array.isArray(body.input) ? { input: body.input.map(item => ( + isRecord(item) && item.type === "additional_tools" && Array.isArray(item.tools) + ? { ...item, tools: cloneTools(item.tools) } + : item + )) } : {}), + }; +} + +function mirrorTool(tool: RecordValue): RecordValue { + const parameters = isRecord(tool.parameters) ? tool.parameters : undefined; + const properties = isRecord(parameters?.properties) ? parameters.properties : undefined; + const message = isRecord(properties?.message) ? properties.message : undefined; + const { encrypted: _, ...plaintextMessage } = message ?? {}; + return { + ...tool, + description: `${GUIDANCE} ${tool.name}.`, + ...(message && Object.hasOwn(message, "encrypted") ? { + parameters: { ...parameters, properties: { ...properties, message: plaintextMessage } }, + } : {}), + }; +} + +function mirrorChildren(group: RecordValue): RecordValue[] { + if (!Array.isArray(group.tools)) return []; + return group.tools.filter((tool): tool is RecordValue => ( + isRecord(tool) && tool.type === "function" && typeof tool.name === "string" && MIRRORABLE_COLLABORATION_OPERATIONS.has(tool.name) + )).map(mirrorTool); +} + +function mirrorGroup(group: RecordValue): RecordValue { + return { type: "namespace", name: MIRROR_NAMESPACE, description: GUIDANCE, tools: mirrorChildren(group) }; +} + +/** + * Add request-local plaintext collaboration mirrors to raw and parsed Responses catalogs. + * The caller has already proved this request is eligible; this helper owns no routing policy. + */ +export function injectV2RoutedDelegationBridge( + parsed: OcxParsedRequest, +): V2RoutedDelegationBridgeContext | undefined { + const stateBody = requestStateBody(parsed._rawBody); + const lists = rawToolLists(parsed._rawBody, parsed._replayPrefixLen ?? 0); + const nativeGroups: Array<{ list: unknown[]; index: number; group: RecordValue }> = []; + const existing: Array<{ list: unknown[]; index: number; group: RecordValue }> = []; + for (const list of lists) { + list.forEach((tool, index) => { + if (!isRecord(tool) || tool.type !== "namespace") return; + if (tool.name === NATIVE_NAMESPACE) nativeGroups.push({ list, index, group: tool }); + if (tool.name === MIRROR_NAMESPACE) existing.push({ list, index, group: tool }); + }); + } + if (nativeGroups.length === 0) return undefined; + + const names = new Set(); + for (const { group } of nativeGroups) { + for (const child of mirrorChildren(group)) names.add(child.name as string); + } + for (const { group } of existing) { + for (const child of mirrorChildren(group)) names.add(child.name as string); + } + if (names.size === 0) return undefined; + + const expected = nativeGroups.map(({ list, index, group }) => ({ list, index: index + 1, group: mirrorGroup(group) })); + const idempotent = existing.length === nativeGroups.length && existing.every(({ list, index, group }) => ( + injectedGroups.has(group) + && nativeGroups.some(native => native.list === list && native.index + 1 === index) + )); + if (existing.length > 0 && !idempotent) { + throw new Error("v2 routed delegation bridge namespace collision"); + } + if (!idempotent) { + for (const { list, index, group } of [...expected].reverse()) { + injectedGroups.add(group); + list.splice(index, 0, group); + } + } + + const mirrorTools: OcxTool[] = []; + for (const name of names) { + const source = parsed.context.tools?.find(tool => tool.namespace === NATIVE_NAMESPACE && tool.name === name); + if (source) { + mirrorTools.push({ ...mirrorTool(source as unknown as RecordValue), namespace: MIRROR_NAMESPACE } as OcxTool); + continue; + } + const raw = nativeGroups.flatMap(entry => mirrorChildren(entry.group)).find(tool => tool.name === name); + mirrorTools.push({ + name, + namespace: MIRROR_NAMESPACE, + description: `${GUIDANCE} ${name}.`, + parameters: isRecord(raw?.parameters) ? raw.parameters : {}, + }); + } + for (const { group } of nativeGroups) { + if (Array.isArray(group.tools)) group.tools = group.tools.filter(tool => ( + !isRecord(tool) + || tool.type !== "function" + || typeof tool.name !== "string" + || !MIRRORABLE_COLLABORATION_OPERATIONS.has(tool.name) + )); + } + if (mirrorTools.length > 0) { + parsed.context.tools = (parsed.context.tools ?? []).filter(tool => ( + tool.namespace !== NATIVE_NAMESPACE + || tool.freeform === true + || !MIRRORABLE_COLLABORATION_OPERATIONS.has(tool.name) + )); + const present = new Set((parsed.context.tools ?? []) + .filter(tool => tool.namespace === MIRROR_NAMESPACE) + .map(tool => tool.name)); + if (mirrorTools.some(tool => !present.has(tool.name))) { + parsed.context.tools = [...(parsed.context.tools ?? []), ...mirrorTools.filter(tool => !present.has(tool.name))]; + } + } + return names.size > 0 ? { names, requestStateBody: stateBody } : undefined; +} + +function rewriteValue( + value: unknown, + active: V2RoutedDelegationBridgeContext, + authorizedIds?: ReadonlySet, +): { value: unknown; changed: boolean } { + if (Array.isArray(value)) { + let changed = false; + const next = value.map(entry => { + const rewritten = rewriteValue(entry, active, authorizedIds); + changed ||= rewritten.changed; + return rewritten.value; + }); + return changed ? { value: next, changed: true } : { value, changed: false }; + } + if (!isRecord(value)) return { value, changed: false }; + let changed = false; + const entries = Object.entries(value).map(([key, entry]) => { + const rewritten = rewriteValue(entry, active, authorizedIds); + changed ||= rewritten.changed; + return [key, rewritten.value]; + }); + const next: RecordValue = Object.fromEntries(entries); + const armed = + value.type === "function_call" + && value.namespace === MIRROR_NAMESPACE + && typeof value.name === "string" + && active.names.has(value.name); + if (armed && authorizedIds !== undefined) { + if (typeof value.id !== "string" || !authorizedIds.has(value.id)) { + const capped = authorizedIds.size >= MAX_SSE_BINDINGS; + throw Object.assign(new Error(capped + ? `v2 routed delegation bridge exceeded ${MAX_SSE_BINDINGS} SSE call bindings` + : "v2 routed delegation bridge received an unbound SSE call"), { + ...(capped ? { code: "translation_buffer_limit" } : {}), + }); + } + } + if (armed) { + next.namespace = NATIVE_NAMESPACE; + next.encrypted_function_args = []; + changed = true; + } + return changed ? { value: next, changed: true } : { value, changed: false }; +} + +/** Normalize only mirror calls armed by this request in a complete JSON response. */ +export function rewriteV2RoutedDelegationCallsInJson( + json: string, + active: V2RoutedDelegationBridgeContext | undefined, +): string { + if (!active || active.names.size === 0) return json; + let parsed: unknown; + try { parsed = JSON.parse(json); } catch { return json; } + const rewritten = rewriteValue(parsed, active); + return rewritten.changed ? JSON.stringify(rewritten.value) : json; +} + +/** Stateful payload rewrite for SSE; item ids bind later argument events to their mirror call. */ +export function createV2RoutedDelegationSseRewrite( + active: V2RoutedDelegationBridgeContext | undefined, +): SsePayloadRewrite | undefined { + if (!active || active.names.size === 0) return undefined; + const admittedIds = new Set(); + const openArgumentIds = new Set(); + const bind = (itemId: unknown): void => { + if (typeof itemId !== "string" || itemId.trim().length === 0) return; + if (admittedIds.has(itemId)) return; + if (admittedIds.size >= MAX_SSE_BINDINGS) { + throw Object.assign( + new Error(`v2 routed delegation bridge exceeded ${MAX_SSE_BINDINGS} SSE call bindings`), + { code: "translation_buffer_limit" }, + ); + } + admittedIds.add(itemId); + openArgumentIds.add(itemId); + }; + return payload => { + let event: unknown; + try { event = JSON.parse(payload); } catch { return payload; } + if (!isRecord(event)) return payload; + const type = event.type; + const item = isRecord(event.item) ? event.item : undefined; + const armed = !!item && item.type === "function_call" && item.namespace === MIRROR_NAMESPACE + && typeof item.name === "string" && active.names.has(item.name); + const added = type === "response.output_item.added"; + const itemDone = type === "response.output_item.done"; + if (added && armed) bind(item?.id); + const admittedSnapshot = itemDone && armed && typeof item?.id === "string" && admittedIds.has(item.id); + const argumentEvent = type === "response.function_call_arguments.delta" || type === "response.function_call_arguments.done"; + const matchedArgument = argumentEvent && typeof event.item_id === "string" && openArgumentIds.has(event.item_id); + const failedTerminal = type === "response.failed" || type === "response.incomplete"; + const rewritten = rewriteValue(event, active, admittedIds); + if (type === "response.function_call_arguments.done" && matchedArgument) openArgumentIds.delete(event.item_id as string); + if (itemDone && admittedSnapshot) openArgumentIds.delete(item!.id as string); + if (type === "response.completed" || failedTerminal) { + admittedIds.clear(); + openArgumentIds.clear(); + } + if (matchedArgument) { + const next = rewritten.changed && isRecord(rewritten.value) ? rewritten.value : { ...event }; + next.encrypted_function_args = []; + return JSON.stringify(next); + } + return rewritten.changed ? JSON.stringify(rewritten.value) : payload; + }; +} diff --git a/src/server/responses/v2-routed-delegation-policy.ts b/src/server/responses/v2-routed-delegation-policy.ts new file mode 100644 index 00000000000..3a0f735922b --- /dev/null +++ b/src/server/responses/v2-routed-delegation-policy.ts @@ -0,0 +1,82 @@ +import { MIRRORABLE_COLLABORATION_OPERATIONS, catalogLists, isRecord } from "./v2-routed-delegation-shared"; + +export type V2RoutedDelegationBridgeScope = "root" | "child"; + +export type V2RoutedDelegationBridgeInactiveReason = + | "disabled" + | "not_v2" + | "non_native_route" + | "maintenance_turn" + | "no_collaboration_catalog" + | "combo" + | "compaction" + | "shadow_route"; + +export type V2RoutedDelegationBridgeDecision = + | { active: true; decision: "active"; scope: V2RoutedDelegationBridgeScope } + | { active: false; decision: V2RoutedDelegationBridgeInactiveReason }; + +export interface V2RoutedDelegationBridgePolicyInput { + enabled: boolean; + inboundWire: string; + multiAgentMode: string | undefined; + upstreamV2Enabled: boolean; + canonicalNativeRoute: boolean; + hasSubagentMarker: boolean; + threadSpawn: boolean; + comboAttempt: boolean; + compaction: boolean; + shadowRoute: boolean; + collaborationSurface: "v1" | "v2" | null; + body: unknown; + replayPrefixLength?: number; +} + +/** The caller-supplied current-turn catalog is the delegation authority. */ +export function hasMirrorableV2CollaborationCatalog( + body: unknown, + replayPrefixLength = 0, +): boolean { + return catalogLists(body, replayPrefixLength).some(list => list.some(group => ( + isRecord(group) + && group.type === "namespace" + && group.name === "collaboration" + && Array.isArray(group.tools) + && group.tools.some(tool => ( + isRecord(tool) + && tool.type === "function" + && typeof tool.name === "string" + && MIRRORABLE_COLLABORATION_OPERATIONS.has(tool.name) + )) + ))); +} + +/** Decide eligibility after fallback/recovery has settled the physical route. */ +export function decideV2RoutedDelegationBridge( + input: V2RoutedDelegationBridgePolicyInput, +): V2RoutedDelegationBridgeDecision { + if (!input.enabled) return { active: false, decision: "disabled" }; + if ( + input.inboundWire !== "responses" + || input.multiAgentMode !== "v2" + || !input.upstreamV2Enabled + ) return { active: false, decision: "not_v2" }; + if (!input.canonicalNativeRoute) return { active: false, decision: "non_native_route" }; + if (input.comboAttempt) return { active: false, decision: "combo" }; + if (input.compaction) return { active: false, decision: "compaction" }; + if (input.shadowRoute) return { active: false, decision: "shadow_route" }; + if (input.hasSubagentMarker && !input.threadSpawn) { + return { active: false, decision: "maintenance_turn" }; + } + if ( + input.collaborationSurface !== "v2" + || !hasMirrorableV2CollaborationCatalog(input.body, input.replayPrefixLength) + ) { + return { active: false, decision: "no_collaboration_catalog" }; + } + return { + active: true, + decision: "active", + scope: input.threadSpawn ? "child" : "root", + }; +} diff --git a/src/server/responses/v2-routed-delegation-shared.ts b/src/server/responses/v2-routed-delegation-shared.ts new file mode 100644 index 00000000000..0291b239b38 --- /dev/null +++ b/src/server/responses/v2-routed-delegation-shared.ts @@ -0,0 +1,21 @@ +export const MIRRORABLE_COLLABORATION_OPERATIONS = new Set([ + "spawn_agent", + "send_message", + "followup_task", +]); + +export function isRecord(value: unknown): value is Record { + return !!value && typeof value === "object" && !Array.isArray(value); +} + +export function catalogLists(body: unknown, replayPrefixLength: number): unknown[][] { + if (!isRecord(body)) return []; + const lists: unknown[][] = []; + if (Array.isArray(body.tools)) lists.push(body.tools); + if (!Array.isArray(body.input)) return lists; + const start = Math.max(0, Math.min(replayPrefixLength, body.input.length)); + for (const item of body.input.slice(start)) { + if (isRecord(item) && item.type === "additional_tools" && Array.isArray(item.tools)) lists.push(item.tools); + } + return lists; +} diff --git a/src/types/config.ts b/src/types/config.ts index 499ca59eb15..c8685960221 100644 --- a/src/types/config.ts +++ b/src/types/config.ts @@ -634,6 +634,8 @@ export interface OcxConfig { * Routed parents get v2 tools; Sol/Terra can still spawn Grok/Claude (issue #92). */ keepNativeChatGptOnV1?: boolean; + /** Experimental plaintext delegation bridge for eligible native V2 root and thread-spawn child turns. */ + v2RoutedDelegationBridge?: boolean; /** Experimental, default-off ChatGPT recovery for encrypted V2 routed tasks. */ agentTaskRecovery?: { enabled?: boolean; diff --git a/tests/config/config-load-degrade.test.ts b/tests/config/config-load-degrade.test.ts index 4d8cca68f76..6233926c5ee 100644 --- a/tests/config/config-load-degrade.test.ts +++ b/tests/config/config-load-degrade.test.ts @@ -139,3 +139,13 @@ test("Fast rows default on for fresh and omitted config; explicit false and malf expect(loaded.providers.xai.note).toBe("keep me"); } }); + +test("V2 routed delegation bridge degrades independently when hand edited", () => { + for (const [value, expected] of [[true, true], [false, false], ["invalid", undefined]] as const) { + const config = { ...candidate({}), v2RoutedDelegationBridge: value }; + writeFileSync(getConfigPath(), JSON.stringify(config), "utf8"); + const loaded = loadConfig(); + expect(loaded.v2RoutedDelegationBridge).toBe(expected); + expect(loaded.providers.xai.note).toBe("keep me"); + } +}); diff --git a/tests/responses/responses-v2-routed-delegation-bridge.test.ts b/tests/responses/responses-v2-routed-delegation-bridge.test.ts new file mode 100644 index 00000000000..42d6e9903e1 --- /dev/null +++ b/tests/responses/responses-v2-routed-delegation-bridge.test.ts @@ -0,0 +1,561 @@ +import { afterAll, beforeAll, describe, expect, test } from "bun:test"; +import { mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { expandPreviousResponseInput } from "../../src/responses/state"; +import { handleResponses, handleResponsesCompact } from "../../src/server/responses"; +import type { RequestLogContext } from "../../src/server/request-log"; +import type { OcxConfig } from "../../src/types"; +import { fakeChatGptJwt } from "../helpers/agent-task-recovery"; + +const savedCodexHome = process.env.CODEX_HOME; +const codexHome = mkdtempSync(join(tmpdir(), "ocx-v2-routed-delegation-bridge-")); + +beforeAll(() => { + writeFileSync(join(codexHome, "config.toml"), "[features.multi_agent_v2]\nenabled = true\n"); + process.env.CODEX_HOME = codexHome; +}); + +afterAll(() => { + if (savedCodexHome === undefined) delete process.env.CODEX_HOME; + else process.env.CODEX_HOME = savedCodexHome; + rmSync(codexHome, { recursive: true, force: true }); +}); + +function config(): OcxConfig { + return { + defaultProvider: "openai", + multiAgentMode: "v2", + v2RoutedDelegationBridge: true, + providers: { + openai: { + adapter: "openai-responses", + baseUrl: "https://chatgpt.com/backend-api/codex", + authMode: "forward", + codexAccountMode: "direct", + }, + }, + } as OcxConfig; +} + +function request(body: Record, headers: Record = {}): Request { + return new Request("http://localhost/v1/responses", { + method: "POST", + headers: { + "content-type": "application/json", + authorization: `Bearer ${fakeChatGptJwt("acct-bridge")}`, + "chatgpt-account-id": "acct-bridge", + originator: "codex_cli_rs", + ...headers, + }, + body: JSON.stringify(body), + }); +} + +function rootBody(extra: Record = {}): Record { + return { + model: "gpt-5.5", + input: [{ type: "message", role: "user", content: [{ type: "input_text", text: "delegate" }] }], + tools: [{ + type: "namespace", + name: "collaboration", + tools: [ + { type: "function", name: "spawn_agent", parameters: { type: "object" } }, + { type: "function", name: "send_message", parameters: { type: "object" } }, + ], + }], + stream: false, + ...extra, + }; +} + +describe("Responses V2 routed delegation bridge runtime", () => { + test("injects a canonical V2 root and normalizes its JSON mirror call before caching", async () => { + const requests: Array> = []; + const originalFetch = globalThis.fetch; + globalThis.fetch = (async (_url: unknown, init?: RequestInit) => { + requests.push(JSON.parse(String(init?.body ?? "{}")) as Record); + return new Response(JSON.stringify({ + id: "resp_bridge", + status: "completed", + output: [{ + type: "function_call", + id: "fc_bridge", + call_id: "call_bridge", + namespace: "ocx_agents", + name: "spawn_agent", + arguments: "{\"task\":\"sentinel\"}", + }], + }), { headers: { "content-type": "application/json" } }); + }) as typeof fetch; + try { + const logCtx: RequestLogContext = { model: "", provider: "" }; + const response = await handleResponses(request(rootBody()), config(), logCtx); + const output = await response.json() as { output: Array> }; + + expect(requests).toHaveLength(1); + expect(requests[0]?.stream).toBe(false); + expect((requests[0]?.tools as Array>)[0]?.name).toBe("collaboration"); + expect((requests[0]?.tools as Array>)[1]).toMatchObject({ + type: "namespace", name: "ocx_agents", tools: [{ name: "spawn_agent" }, { name: "send_message" }], + }); + expect(output.output[0]).toMatchObject({ + namespace: "collaboration", name: "spawn_agent", encrypted_function_args: [], + }); + expect(logCtx.v2BridgeDecision).toBe("active"); + expect(logCtx.v2BridgeScope).toBe("root"); + expect(logCtx.v2BridgeStateDurability).toBeUndefined(); + } finally { + globalThis.fetch = originalFetch; + } + }); + + test("replays JSON continuation state with the restored collaboration namespace", async () => { + const requests: Array> = []; + const originalFetch = globalThis.fetch; + globalThis.fetch = (async (_url: unknown, init?: RequestInit) => { + const body = JSON.parse(String(init?.body ?? "{}")) as Record; + requests.push(body); + return Response.json({ + id: requests.length === 1 ? "resp_bridge_json_replay" : "resp_bridge_json_followup", + status: "completed", + output: requests.length === 1 ? [{ + type: "function_call", id: "fc_bridge_json_replay", call_id: "call_bridge_json_replay", + namespace: "ocx_agents", name: "spawn_agent", arguments: "{\"task\":\"continue\"}", + }] : [], + }); + }) as typeof fetch; + try { + await handleResponses(request(rootBody()), config(), { model: "", provider: "" }); + const followup = await handleResponses(request(rootBody({ + previous_response_id: "resp_bridge_json_replay", + input: [{ type: "message", role: "user", content: [{ type: "input_text", text: "follow up" }] }], + })), config(), { model: "", provider: "" }); + + expect(followup.status).toBe(200); + expect(JSON.stringify(requests[1]?.input)).toContain('"namespace":"collaboration"'); + expect(JSON.stringify(requests[1]?.input)).not.toContain('"namespace":"ocx_agents"'); + } finally { + globalThis.fetch = originalFetch; + } + }); + + test("uses the same JSON normalization for the canonical websocket path", async () => { + const originalFetch = globalThis.fetch; + globalThis.fetch = (async () => Response.json({ id: "resp_ws", status: "completed", output: [{ + type: "function_call", id: "fc_ws", call_id: "call_ws", namespace: "ocx_agents", name: "spawn_agent", arguments: "{}", + }] })) as typeof fetch; + try { + const response = await handleResponses(request(rootBody()), config(), { model: "", provider: "" }, { inboundTransport: "websocket" }); + expect((await response.json() as { output: Array> }).output[0]).toMatchObject({ + namespace: "collaboration", encrypted_function_args: [], + }); + } finally { + globalThis.fetch = originalFetch; + } + }); + + test("mirrors additional_tools and rebuilds the parsed authorization catalog", async () => { + const requests: Array> = []; + const originalFetch = globalThis.fetch; + globalThis.fetch = (async (_url: unknown, init?: RequestInit) => { + requests.push(JSON.parse(String(init?.body ?? "{}")) as Record); + return Response.json({ id: "resp_additional", status: "completed", output: [{ + type: "function_call", id: "fc_additional", call_id: "call_additional", namespace: "ocx_agents", name: "spawn_agent", arguments: "{}", + }] }); + }) as typeof fetch; + try { + const native = { type: "namespace", name: "collaboration", tools: [{ type: "function", name: "spawn_agent", parameters: { type: "object" } }, { type: "function", name: "send_message", parameters: { type: "object" } }] }; + const response = await handleResponses(request(rootBody({ tools: [], input: [{ type: "additional_tools", tools: [native] }] })), config(), { model: "", provider: "" }); + const outbound = requests[0]!; + expect((outbound.input as Array>)[0]?.tools).toEqual([{ ...native, tools: [] }, { + type: "namespace", name: "ocx_agents", description: "Use this routed-child mirror for collaboration operations.", tools: [{ type: "function", name: "spawn_agent", description: "Use this routed-child mirror for collaboration operations. spawn_agent.", parameters: { type: "object" } }, { type: "function", name: "send_message", description: "Use this routed-child mirror for collaboration operations. send_message.", parameters: { type: "object" } }], + }]); + expect((await response.json() as { output: Array> }).output[0]).toMatchObject({ namespace: "collaboration", encrypted_function_args: [] }); + } finally { globalThis.fetch = originalFetch; } + }); + + test("keeps replayed additional_tools out of a fresh current-turn mirror catalog", async () => { + const requests: Array> = []; + const native = { type: "namespace", name: "collaboration", tools: [ + { type: "function", name: "spawn_agent", parameters: { type: "object" } }, + { type: "function", name: "send_message", parameters: { type: "object" } }, + ] }; + const originalNative = structuredClone(native); + const originalFetch = globalThis.fetch; + globalThis.fetch = (async (_url: unknown, init?: RequestInit) => { + requests.push(JSON.parse(String(init?.body ?? "{}")) as Record); + return Response.json({ id: requests.length === 1 ? "resp_additional_replay" : "resp_additional_after", status: "completed", output: [] }); + }) as typeof fetch; + try { + await handleResponses(request(rootBody({ tools: [], input: [{ type: "additional_tools", tools: [native] }] })), config(), { model: "", provider: "" }); + const second = await handleResponses(request(rootBody({ + tools: [], + previous_response_id: "resp_additional_replay", + input: [{ type: "additional_tools", tools: [native] }], + })), config(), { model: "", provider: "" }); + + expect(second.status).toBe(200); + const catalogs = (requests[1]?.input as Array>) + .filter(item => item.type === "additional_tools") + .map(item => item.tools); + expect(catalogs).toEqual([[originalNative], [{ ...native, tools: [] }, expect.objectContaining({ name: "ocx_agents" })]]); + } finally { globalThis.fetch = originalFetch; } + }); + + test("replayed additional_tools alone cannot arm a later bridge turn", async () => { + const requests: Array> = []; + const native = { type: "namespace", name: "collaboration", tools: [ + { type: "function", name: "spawn_agent", parameters: { type: "object" } }, + { type: "function", name: "send_message", parameters: { type: "object" } }, + ] }; + const originalFetch = globalThis.fetch; + globalThis.fetch = (async (_url: unknown, init?: RequestInit) => { + requests.push(JSON.parse(String(init?.body ?? "{}")) as Record); + return Response.json({ id: requests.length === 1 ? "resp_stale_catalog" : "resp_stale_after", status: "completed", output: [] }); + }) as typeof fetch; + try { + await handleResponses(request(rootBody({ tools: [], input: [{ type: "additional_tools", tools: [native] }] })), config(), { model: "", provider: "" }); + const second = await handleResponses(request(rootBody({ + tools: [], + previous_response_id: "resp_stale_catalog", + input: [{ type: "message", role: "user", content: [{ type: "input_text", text: "continue" }] }], + })), config(), { model: "", provider: "" }); + + expect(second.status).toBe(200); + expect(JSON.stringify(requests[0]?.input)).toContain('"ocx_agents"'); + expect(JSON.stringify(requests[1]?.input)).not.toContain('"ocx_agents"'); + } finally { globalThis.fetch = originalFetch; } + }); + + test("does not modify a genuine native collaboration response", async () => { + const originalFetch = globalThis.fetch; + globalThis.fetch = (async () => Response.json({ id: "resp_native", status: "completed", output: [{ + type: "function_call", id: "fc_native", call_id: "call_native", namespace: "collaboration", name: "spawn_agent", arguments: "{}", + }] })) as typeof fetch; + try { + const call = (await (await handleResponses(request(rootBody()), config(), { model: "", provider: "" })).json() as { output: Array> }).output[0]!; + expect(call.encrypted_function_args).toBeUndefined(); + expect(call.namespace).toBe("collaboration"); + } finally { globalThis.fetch = originalFetch; } + }); + + test("does not inject on excluded request shapes", async () => { + const requests: Array> = []; + const originalFetch = globalThis.fetch; + globalThis.fetch = (async (_url: unknown, init?: RequestInit) => { + requests.push(JSON.parse(String(init?.body ?? "{}")) as Record); + return Response.json({ id: `resp_${requests.length}`, status: "completed", output: [] }); + }) as typeof fetch; + try { + const routed = config(); + routed.providers.gw = { adapter: "openai-responses", baseUrl: "https://gateway.example/v1", authMode: "key", apiKey: "test" } as never; + const cases: Array<[OcxConfig, Record, Record, Parameters[3]]> = [ + [{ ...config(), multiAgentMode: "v1" }, rootBody(), {}, {}], + [{ ...config(), multiAgentMode: "default" }, rootBody(), {}, {}], + [config(), rootBody({ tools: [] }), {}, {}], + [config(), rootBody(), { "x-openai-subagent": "review" }, {}], + [config(), rootBody(), { "x-codex-turn-metadata": JSON.stringify({ subagent_kind: "review" }) }, {}], + [config(), rootBody({ tools: [{ type: "function", name: "spawn_agent" }, ...(rootBody().tools as unknown[]) ] }), {}, {}], + [config(), rootBody(), {}, { comboAttempt: true }], + ]; + for (const [cfg, body, headers, options] of cases) await handleResponses(request(body, headers), cfg, { model: "", provider: "" }, options); + expect(requests).toHaveLength(cases.length); + for (const outbound of requests) expect(JSON.stringify(outbound.tools ?? outbound.input)).not.toContain('"ocx_agents"'); + expect(routed.providers.gw).toBeDefined(); + } finally { globalThis.fetch = originalFetch; } + }); + + test("bridges a canonical native child so its routed grandchild task stays plaintext", async () => { + const requests: Array> = []; + const originalFetch = globalThis.fetch; + globalThis.fetch = (async (_url: unknown, init?: RequestInit) => { + requests.push(JSON.parse(String(init?.body ?? "{}")) as Record); + return Response.json({ id: "resp_nested_child", status: "completed", output: [{ + type: "function_call", + id: "fc_nested_child", + call_id: "call_nested_child", + namespace: "ocx_agents", + name: "spawn_agent", + arguments: "{\"message\":\"plaintext grandchild assignment\"}", + }] }); + }) as typeof fetch; + try { + const response = await handleResponses( + request(rootBody(), { + "x-openai-subagent": "collab_spawn", + "x-codex-turn-metadata": JSON.stringify({ subagent_kind: "thread_spawn" }), + }), + config(), + { model: "", provider: "" }, + ); + const output = await response.json() as { output: Array> }; + + expect(response.status).toBe(200); + expect(JSON.stringify(requests[0]?.tools)).toContain('"name":"ocx_agents"'); + expect(output.output[0]).toMatchObject({ + namespace: "collaboration", + name: "spawn_agent", + arguments: "{\"message\":\"plaintext grandchild assignment\"}", + encrypted_function_args: [], + }); + } finally { globalThis.fetch = originalFetch; } + }); + + test("runs the root, routed child, and parent continuation lifecycle without leaking mirrors", async () => { + const outbound: Array<{ url: string; body: Record }> = []; + let turn = 0; + const originalFetch = globalThis.fetch; + globalThis.fetch = (async (url: unknown, init?: RequestInit) => { + const body = JSON.parse(String(init?.body ?? "{}")) as Record; + outbound.push({ url: String(url), body }); + turn += 1; + if (turn === 1) return Response.json({ id: "resp_root", status: "completed", output: [{ + type: "function_call", id: "fc_root", call_id: "call_root", namespace: "ocx_agents", name: "spawn_agent", arguments: "{}", + }] }); + if (turn === 2) return Response.json({ choices: [{ message: { role: "assistant", content: "safe-fixed-result" }, finish_reason: "stop" }] }); + return Response.json({ id: "resp_parent", status: "completed", output: [ + { type: "function_call", id: "fc_follow", call_id: "call_follow", namespace: "ocx_agents", name: "followup_task", arguments: "{}" }, + { type: "function_call", id: "fc_wait", call_id: "call_wait", namespace: "collaboration", name: "wait_agent", arguments: "{}" }, + { type: "function_call", id: "fc_list", call_id: "call_list", namespace: "collaboration", name: "list_agents", arguments: "{}" }, + ] }); + }) as typeof fetch; + try { + const root = await handleResponses(request(rootBody()), config(), { model: "", provider: "" }); + expect((await root.json() as { output: Array> }).output[0]).toMatchObject({ namespace: "collaboration", encrypted_function_args: [] }); + + const childConfig = config(); + childConfig.providers.gw = { adapter: "openai-chat", baseUrl: "https://gateway.example/v1", authMode: "key", apiKey: "test" } as never; + const child = await handleResponses(request({ + model: "gw/routed", stream: false, + input: [{ type: "agent_message", author: "/root", recipient: "/root/child", content: [ + { type: "encrypted_content", encrypted_content: "Implement the readable child task." }, + ] }, { type: "function_call_output", call_id: "safe-fixed-call", output: "safe-fixed-result" }], + tools: rootBody().tools, + }, { "x-openai-subagent": "collab_spawn" }), childConfig, { model: "", provider: "" }); + expect(child.status).toBe(200); + + const parent = await handleResponses(request(rootBody({ tools: [{ type: "namespace", name: "collaboration", tools: [ + { type: "function", name: "spawn_agent", parameters: { type: "object" } }, + { type: "function", name: "followup_task", parameters: { type: "object" } }, + { type: "function", name: "wait_agent", parameters: { type: "object" } }, + { type: "function", name: "list_agents", parameters: { type: "object" } }, + ] }] })), config(), { model: "", provider: "" }); + const calls = (await parent.json() as { output: Array> }).output; + + expect(JSON.stringify(outbound[0]?.body.tools)).toContain('"ocx_agents"'); + expect(outbound[1]?.url).toContain("gateway.example"); + expect(outbound[1]?.url).not.toContain("chatgpt.com"); + expect(JSON.stringify(outbound[1]?.body)).toContain("Implement the readable child task."); + expect(JSON.stringify(outbound[1]?.body)).toContain("safe-fixed-result"); + expect(JSON.stringify(outbound[1]?.body)).not.toContain('"ocx_agents"'); + expect(calls[0]).toMatchObject({ namespace: "collaboration", name: "followup_task", encrypted_function_args: [] }); + expect(calls[1]).toMatchObject({ namespace: "collaboration", name: "wait_agent" }); + expect(calls[1]?.encrypted_function_args).toBeUndefined(); + expect(calls[2]).toMatchObject({ namespace: "collaboration", name: "list_agents" }); + expect(calls[2]?.encrypted_function_args).toBeUndefined(); + } finally { globalThis.fetch = originalFetch; } + }); + + test("keeps routed parent to native child to routed grandchild delegation plaintext on V2", async () => { + const outbound: Array<{ url: string; body: Record }> = []; + let turn = 0; + const originalFetch = globalThis.fetch; + globalThis.fetch = (async (url: unknown, init?: RequestInit) => { + const body = JSON.parse(String(init?.body ?? "{}")) as Record; + outbound.push({ url: String(url), body }); + turn += 1; + if (turn === 1) { + return Response.json({ choices: [{ message: { role: "assistant", content: "dispatch native child" }, finish_reason: "stop" }] }); + } + if (turn === 2) { + return Response.json({ id: "resp_native_child", status: "completed", output: [{ + type: "function_call", + id: "fc_native_child", + call_id: "call_native_child", + namespace: "ocx_agents", + name: "spawn_agent", + arguments: "{\"message\":\"plaintext routed grandchild assignment\"}", + }] }); + } + return Response.json({ choices: [{ message: { role: "assistant", content: "grandchild complete" }, finish_reason: "stop" }] }); + }) as typeof fetch; + try { + const mixed = config(); + mixed.providers.gw = { + adapter: "openai-chat", + baseUrl: "https://gateway.example/v1", + authMode: "key", + apiKey: "test", + } as never; + + const parent = await handleResponses( + request(rootBody({ model: "gw/routed" })), + mixed, + { model: "", provider: "" }, + ); + expect(parent.status).toBe(200); + + const child = await handleResponses( + request(rootBody({ + model: "gpt-5.5", + input: [{ + type: "agent_message", + author: "/root", + recipient: "/root/native-child", + content: [{ type: "input_text", text: "Inspect the routed parent result." }], + }], + }), { + "x-openai-subagent": "collab_spawn", + "x-codex-turn-metadata": JSON.stringify({ subagent_kind: "thread_spawn" }), + }), + mixed, + { model: "", provider: "" }, + ); + const childOutput = await child.json() as { output: Array> }; + + const grandchild = await handleResponses( + request({ + model: "gw/routed", + stream: false, + input: [{ + type: "agent_message", + author: "/root/native-child", + recipient: "/root/native-child/routed-grandchild", + content: [{ type: "input_text", text: "plaintext routed grandchild assignment" }], + }], + tools: [], + }, { "x-openai-subagent": "collab_spawn" }), + mixed, + { model: "", provider: "" }, + ); + + expect(child.status).toBe(200); + expect(grandchild.status).toBe(200); + expect(outbound).toHaveLength(3); + expect(outbound[0]?.url).toContain("gateway.example"); + expect(outbound[1]?.url).toContain("chatgpt.com"); + expect(JSON.stringify(outbound[1]?.body.tools)).toContain('"name":"ocx_agents"'); + expect(childOutput.output[0]).toMatchObject({ + namespace: "collaboration", + name: "spawn_agent", + arguments: "{\"message\":\"plaintext routed grandchild assignment\"}", + encrypted_function_args: [], + }); + expect(outbound[2]?.url).toContain("gateway.example"); + expect(JSON.stringify(outbound[2]?.body)).toContain("plaintext routed grandchild assignment"); + expect(JSON.stringify(outbound[2]?.body)).not.toContain("encrypted_content"); + expect(JSON.stringify(outbound[2]?.body)).not.toContain('"ocx_agents"'); + } finally { globalThis.fetch = originalFetch; } + }); + + test("fails a mirror namespace collision before upstream I/O", async () => { + let fetches = 0; + const originalFetch = globalThis.fetch; + globalThis.fetch = (async () => { + fetches += 1; + throw new Error("collision must not fetch"); + }) as typeof fetch; + try { + const response = await handleResponses(request(rootBody({ tools: [ + ...(rootBody().tools as unknown[]), + { type: "namespace", name: "ocx_agents", tools: [] }, + ] })), config(), { model: "", provider: "" }); + + expect(response.status).toBe(400); + expect(fetches).toBe(0); + } finally { + globalThis.fetch = originalFetch; + } + }); + + test("rewrites split SSE calls for eligible roots and both spawned-child marker forms", async () => { + const originalFetch = globalThis.fetch; + globalThis.fetch = (async () => new Response([ + "data: ", JSON.stringify({ type: "response.output_item.added", item: { + type: "function_call", id: "fc_bridge", call_id: "call_bridge", namespace: "ocx_agents", name: "spawn_agent", arguments: "", + } }), "\n\n", + "data: ", JSON.stringify({ type: "response.function_call_arguments.done", item_id: "fc_bridge", arguments: "{}" }), "\n\n", + "data: ", JSON.stringify({ type: "response.completed", response: { id: "resp_sse", status: "completed", output: [{ + type: "function_call", id: "fc_bridge", call_id: "call_bridge", namespace: "ocx_agents", name: "spawn_agent", arguments: "{}", + }] } }), "\n\n", + ].join(""), { headers: { "content-type": "text/event-stream" } })) as typeof fetch; + try { + const markerForms = [ + {}, + { "x-openai-subagent": "collab_spawn" }, + { "x-codex-turn-metadata": JSON.stringify({ subagent_kind: "thread_spawn" }) }, + ]; + for (const headers of markerForms) { + const response = await handleResponses( + request(rootBody({ stream: true }), headers), + config(), + { model: "", provider: "" }, + ); + const text = await response.text(); + expect(text).toContain('"namespace":"collaboration"'); + expect(text).toContain('"encrypted_function_args":[]'); + expect(text).not.toContain('"namespace":"ocx_agents"'); + } + } finally { + globalThis.fetch = originalFetch; + } + }); + + test("replays SSE continuation state with the restored collaboration namespace", async () => { + const requests: Array> = []; + const originalFetch = globalThis.fetch; + globalThis.fetch = (async (_url: unknown, init?: RequestInit) => { + const body = JSON.parse(String(init?.body ?? "{}")) as Record; + requests.push(body); + if (requests.length === 1) { + return new Response([ + `data: ${JSON.stringify({ type: "response.output_item.added", item: { + type: "function_call", id: "fc_bridge_sse_replay", call_id: "call_bridge_sse_replay", + namespace: "ocx_agents", name: "spawn_agent", arguments: "{}", + } })}\n\n`, + `data: ${JSON.stringify({ type: "response.completed", response: { + id: "resp_bridge_sse_replay", status: "completed", output: [{ + type: "function_call", id: "fc_bridge_sse_replay", call_id: "call_bridge_sse_replay", + namespace: "ocx_agents", name: "spawn_agent", arguments: "{}", + }], + } })}\n\n`, + ].join(""), { headers: { "content-type": "text/event-stream" } }); + } + return Response.json({ id: "resp_bridge_sse_followup", status: "completed", output: [] }); + }) as typeof fetch; + try { + const streamed = await handleResponses(request(rootBody({ stream: true })), config(), { model: "", provider: "" }); + expect(streamed.status).toBe(200); + await streamed.text(); + const followup = await handleResponses(request(rootBody({ + previous_response_id: "resp_bridge_sse_replay", + input: [{ type: "message", role: "user", content: [{ type: "input_text", text: "follow up" }] }], + })), config(), { model: "", provider: "" }); + + expect(followup.status).toBe(200); + expect(JSON.stringify(requests[1]?.input)).toContain('"namespace":"collaboration"'); + expect(JSON.stringify(requests[1]?.input)).not.toContain('"namespace":"ocx_agents"'); + } finally { + globalThis.fetch = originalFetch; + } + }); + + test("leaves the canonical catalog unchanged when disabled", async () => { + const requests: Array> = []; + const originalFetch = globalThis.fetch; + globalThis.fetch = (async (_url: unknown, init?: RequestInit) => { + requests.push(JSON.parse(String(init?.body ?? "{}")) as Record); + return Response.json({ id: "resp_disabled", status: "completed", output: [] }); + }) as typeof fetch; + try { + const disabled = config(); + disabled.v2RoutedDelegationBridge = false; + await handleResponses(request(rootBody()), disabled, { model: "", provider: "" }); + + expect(requests[0]?.tools).toHaveLength(1); + } finally { + globalThis.fetch = originalFetch; + } + }); +}); diff --git a/tests/server/v2-routed-delegation-bridge.test.ts b/tests/server/v2-routed-delegation-bridge.test.ts new file mode 100644 index 00000000000..329633cc6e1 --- /dev/null +++ b/tests/server/v2-routed-delegation-bridge.test.ts @@ -0,0 +1,254 @@ +import { describe, expect, test } from "bun:test"; +import { + createV2RoutedDelegationSseRewrite, + injectV2RoutedDelegationBridge, + rewriteV2RoutedDelegationCallsInJson, +} from "../../src/server/responses/v2-routed-delegation-bridge"; +import type { OcxParsedRequest } from "../../src/types"; + +const GUIDANCE = "Use this routed-child mirror for collaboration operations."; + +function parsed(body: Record): OcxParsedRequest { + return { + modelId: "gpt-5.6-terra", + stream: false, + context: { + messages: [], + tools: [{ namespace: "collaboration", name: "spawn_agent", description: "native spawn", parameters: { type: "object", properties: { task: { type: "string" } } } }], + }, + options: {}, + _rawBody: body, + }; +} + +const spawn = { type: "function", name: "spawn_agent", description: "native spawn", parameters: { type: "object", properties: { task: { type: "string" } } }, strict: true }; +const send = { type: "function", name: "send_message", description: "native send", parameters: { type: "object" } }; +const followup = { type: "function", name: "followup_task", description: "native follow", parameters: { type: "object" } }; + +describe("V2 routed delegation bridge", () => { + test("moves message operations to plaintext mirrors while preserving native control operations", () => { + const encrypted = (tool: Record) => ({ + ...tool, + parameters: { type: "object", properties: { + target: { type: "string" }, + message: { type: "string", encrypted: true, description: `${tool.name} message` }, + } }, + }); + const messages = [encrypted(spawn), encrypted(send), encrypted(followup)]; + const controls = ["wait_agent", "interrupt_agent", "list_agents"].map(name => ( + { type: "function", name, parameters: { type: "object", properties: { id: { type: "string" } } } } + )); + const body = { tools: [{ type: "namespace", name: "collaboration", tools: [...messages, ...controls] }] }; + const request = parsed(body); + request.context.tools = [...messages, ...controls].map(tool => ({ + namespace: "collaboration", name: tool.name, description: tool.description, parameters: tool.parameters, + })); + const original = structuredClone(body); + + const active = injectV2RoutedDelegationBridge(request); + + expect(active?.names).toEqual(new Set(["spawn_agent", "send_message", "followup_task"])); + expect(active?.requestStateBody).toEqual(original); + expect(body.tools[0]).toEqual({ type: "namespace", name: "collaboration", tools: controls }); + expect(body.tools).toHaveLength(2); + const mirrored = (body.tools[1] as { tools: Array> }).tools; + expect(mirrored.map(tool => tool.name)).toEqual(["spawn_agent", "send_message", "followup_task"]); + for (const tool of mirrored) { + expect(tool.parameters.properties).toEqual({ + target: { type: "string" }, + message: { type: "string", description: `${tool.name} message` }, + }); + } + expect(request.context.tools?.filter(tool => tool.namespace === "collaboration")).toEqual( + controls.map(tool => ({ namespace: "collaboration", name: tool.name, parameters: tool.parameters })), + ); + expect(request.context.tools?.filter(tool => tool.namespace === "ocx_agents").map(tool => tool.name)) + .toEqual(["spawn_agent", "send_message", "followup_task"]); + }); + + test("mirrors every catalog shape, exactly the available three functions, and is idempotent", () => { + const body = { + tools: [{ type: "namespace", name: "collaboration", tools: [spawn] }], + input: [{ type: "additional_tools", tools: [{ type: "namespace", name: "collaboration", tools: [send, followup] }] }], + }; + const request = parsed(body); + const first = injectV2RoutedDelegationBridge(request); + const raw = structuredClone(body); + const second = injectV2RoutedDelegationBridge(request); + + expect(first?.names).toEqual(new Set(["spawn_agent", "send_message", "followup_task"])); + expect(second?.names).toEqual(first?.names); + expect(body).toEqual(raw); + expect((body.tools[1] as { tools: unknown[] }).tools).toHaveLength(1); + const inputMirror = (body.input[0] as { tools: Array> }).tools[1]!; + expect(inputMirror).toMatchObject({ + type: "namespace", + name: "ocx_agents", + description: GUIDANCE, + }); + expect(inputMirror.tools).toHaveLength(2); + }); + + test("fails closed for a conflicting ocx_agents namespace", () => { + const request = parsed({ tools: [ + { type: "namespace", name: "collaboration", tools: [spawn] }, + { type: "namespace", name: "ocx_agents", tools: [] }, + ] }); + expect(() => injectV2RoutedDelegationBridge(request)).toThrow("v2 routed delegation bridge namespace collision"); + }); + + test("rejects an adjacent mirror whose schema differs from native collaboration", () => { + const request = parsed({ tools: [ + { type: "namespace", name: "collaboration", tools: [spawn] }, + { type: "namespace", name: "ocx_agents", description: GUIDANCE, tools: [{ ...spawn, description: `${GUIDANCE} spawn_agent.`, parameters: { type: "object" } }] }, + ] }); + expect(() => injectV2RoutedDelegationBridge(request)).toThrow("v2 routed delegation bridge namespace collision"); + }); + + test("rejects a caller-supplied canonical mirror beside native control tools", () => { + const request = parsed({ tools: [ + { type: "namespace", name: "collaboration", tools: [{ type: "function", name: "wait_agent", parameters: {} }] }, + { type: "namespace", name: "ocx_agents", description: GUIDANCE, tools: [{ ...spawn, description: `${GUIDANCE} spawn_agent.` }] }, + ] }); + + expect(() => injectV2RoutedDelegationBridge(request)).toThrow("v2 routed delegation bridge namespace collision"); + }); + + test("accepts a key-reordered injected mirror as idempotent", () => { + const body = { tools: [{ type: "namespace", name: "collaboration", tools: [spawn] }] }; + const request = parsed(body); + injectV2RoutedDelegationBridge(request); + const mirror = body.tools[1] as { tools: Array> }; + mirror.tools[0] = Object.fromEntries(Object.entries(mirror.tools[0]!).reverse()); + + expect(() => injectV2RoutedDelegationBridge(request)).not.toThrow(); + }); + + test("leaves a collaboration group with no mirrorable function inactive", () => { + const body = { tools: [{ type: "namespace", name: "collaboration", tools: [{ type: "function", name: "wait_agent", parameters: {} }] }] }; + const request = parsed(body); + const before = structuredClone(body); + const tools = structuredClone(request.context.tools); + + expect(injectV2RoutedDelegationBridge(request)).toBeUndefined(); + expect(body).toEqual(before); + expect(request.context.tools).toEqual(tools); + }); + + test("normalizes only armed mirror calls in JSON", () => { + const active = { names: new Set(["spawn_agent"]) }; + const input = JSON.stringify({ output: [ + { type: "function_call", namespace: "ocx_agents", name: "spawn_agent", id: "fc_1", call_id: "call_1", arguments: "{}", status: "completed" }, + { type: "function_call", namespace: "collaboration", name: "spawn_agent", id: "fc_2", call_id: "call_2", arguments: "{}" }, + { type: "function_call", namespace: "ocx_agents", name: "list_agents", id: "fc_3", call_id: "call_3", arguments: "{}" }, + { type: "function_call", namespace: 4, name: "spawn_agent", id: "fc_4", call_id: "call_4", arguments: "{}" }, + ] }); + + expect(JSON.parse(rewriteV2RoutedDelegationCallsInJson(input, active))).toEqual({ output: [ + { type: "function_call", namespace: "collaboration", name: "spawn_agent", id: "fc_1", call_id: "call_1", arguments: "{}", status: "completed", encrypted_function_args: [] }, + { type: "function_call", namespace: "collaboration", name: "spawn_agent", id: "fc_2", call_id: "call_2", arguments: "{}" }, + { type: "function_call", namespace: "ocx_agents", name: "list_agents", id: "fc_3", call_id: "call_3", arguments: "{}" }, + { type: "function_call", namespace: 4, name: "spawn_agent", id: "fc_4", call_id: "call_4", arguments: "{}" }, + ] }); + }); + + test("preserves owned __proto__ data while normalizing untrusted JSON", () => { + const input = '{"__proto__":{"polluted":true},"type":"function_call","namespace":"ocx_agents","name":"spawn_agent"}'; + const output = JSON.parse(rewriteV2RoutedDelegationCallsInJson(input, { names: new Set(["spawn_agent"]) })); + + expect(Object.hasOwn(output, "__proto__")).toBe(true); + expect(output.__proto__).toEqual({ polluted: true }); + expect(Object.getPrototypeOf(output)).toBe(Object.prototype); + }); + + test("normalizes SSE item snapshots and only matching interleaved argument events", () => { + const rewrite = createV2RoutedDelegationSseRewrite({ names: new Set(["spawn_agent"]) })!; + const added = JSON.stringify({ type: "response.output_item.added", output_index: 3, item: { type: "function_call", namespace: "ocx_agents", name: "spawn_agent", id: "fc_1", call_id: "call_1", arguments: "", status: "in_progress" } }); + const unrelated = JSON.stringify({ type: "response.output_item.added", output_index: 4, item: { type: "function_call", namespace: "ocx_agents", name: "list_agents", id: "fc_2", call_id: "call_2", arguments: "" } }); + + expect(JSON.parse(rewrite(added))).toMatchObject({ item: { namespace: "collaboration", name: "spawn_agent", encrypted_function_args: [] } }); + expect(JSON.parse(rewrite(JSON.stringify({ type: "response.function_call_arguments.delta", item_id: "native_fc", output_index: 3, delta: "{" })))).not.toHaveProperty("encrypted_function_args"); + expect(JSON.parse(rewrite(JSON.stringify({ type: "response.function_call_arguments.delta", item_id: "fc_2", output_index: 4, delta: "{" })))).not.toHaveProperty("encrypted_function_args"); + expect(JSON.parse(rewrite(JSON.stringify({ type: "response.function_call_arguments.done", item_id: "fc_1", output_index: 3, arguments: "{}" })))).toMatchObject({ encrypted_function_args: [] }); + expect(JSON.parse(rewrite(JSON.stringify({ type: "response.function_call_arguments.delta", item_id: "fc_1", output_index: 3, delta: "late" })))).not.toHaveProperty("encrypted_function_args"); + expect(JSON.parse(rewrite(unrelated))).toMatchObject({ item: { namespace: "ocx_agents", name: "list_agents" } }); + }); + + test("fails ID-less mirror snapshots closed and leaves index-only arguments untouched", () => { + const rewrite = createV2RoutedDelegationSseRewrite({ names: new Set(["spawn_agent"]) })!; + const snapshot = JSON.stringify({ type: "response.output_item.added", output_index: 3, item: { type: "function_call", namespace: "ocx_agents", name: "spawn_agent" } }); + + expect(() => rewrite(snapshot)).toThrow("v2 routed delegation bridge received an unbound SSE call"); + const argument = JSON.stringify({ type: "response.function_call_arguments.delta", output_index: 3, delta: "{}" }); + expect(rewrite(argument)).toBe(argument); + }); + + test("fails closed before a capped SSE mirror can escape", () => { + const rewrite = createV2RoutedDelegationSseRewrite({ names: new Set(["spawn_agent"]) })!; + for (let index = 0; index < 128; index++) { + rewrite(JSON.stringify({ type: "response.output_item.added", output_index: index, item: { type: "function_call", namespace: "ocx_agents", name: "spawn_agent", id: `fc_${index}` } })); + } + + expect(() => rewrite(JSON.stringify({ type: "response.output_item.added", output_index: 128, item: { type: "function_call", namespace: "ocx_agents", name: "spawn_agent", id: "fc_128" } }))) + .toThrow("v2 routed delegation bridge exceeded 128 SSE call bindings"); + }); + + test("keeps only authorized nonblank ids through completed aggregates", () => { + const rewrite = createV2RoutedDelegationSseRewrite({ names: new Set(["spawn_agent"]) })!; + const snapshot = (id: string, output_index: number) => JSON.stringify({ type: "response.output_item.added", output_index, item: { type: "function_call", namespace: "ocx_agents", name: "spawn_agent", id } }); + rewrite(snapshot("fc_0", 0)); + expect(() => rewrite(snapshot(" ", 1))).toThrow("v2 routed delegation bridge received an unbound SSE call"); + const completed = JSON.parse(rewrite(JSON.stringify({ type: "response.completed", response: { output: [ + { type: "function_call", namespace: "ocx_agents", name: "spawn_agent", id: "fc_0" }, + ] } }))); + + expect(completed.response.output.map((item: { namespace: string }) => item.namespace)).toEqual(["collaboration"]); + expect(rewrite(JSON.stringify({ type: "response.function_call_arguments.delta", item_id: "fc_0", delta: "late" }))).toContain('"item_id":"fc_0"'); + }); + + test("normalizes admitted calls in failed and incomplete terminal snapshots", () => { + for (const type of ["response.failed", "response.incomplete"]) { + const rewrite = createV2RoutedDelegationSseRewrite({ names: new Set(["spawn_agent"]) })!; + rewrite(JSON.stringify({ type: "response.output_item.added", item: { type: "function_call", namespace: "ocx_agents", name: "spawn_agent", id: "fc_terminal" } })); + + const terminal = JSON.parse(rewrite(JSON.stringify({ type, response: { status: type.slice(9), output: [ + { type: "function_call", namespace: "ocx_agents", name: "spawn_agent", id: "fc_terminal" }, + ] } }))); + + expect(terminal.response.output[0]).toMatchObject({ namespace: "collaboration", encrypted_function_args: [] }); + } + }); + + test("an unknown item-id terminal does not affect an authorized call", () => { + const rewrite = createV2RoutedDelegationSseRewrite({ names: new Set(["spawn_agent"]) })!; + rewrite(JSON.stringify({ type: "response.output_item.added", item: { type: "function_call", namespace: "ocx_agents", name: "spawn_agent", id: "fc_known" } })); + const unknown = JSON.stringify({ type: "response.function_call_arguments.done", item_id: "fc_unknown", arguments: "{}" }); + + expect(rewrite(unknown)).toBe(unknown); + expect(JSON.parse(rewrite(JSON.stringify({ type: "response.function_call_arguments.delta", item_id: "fc_known", delta: "{}" }))).encrypted_function_args).toEqual([]); + }); + + test("does not reauthorize arguments after a call closes", () => { + const rewrite = createV2RoutedDelegationSseRewrite({ names: new Set(["spawn_agent"]) })!; + const added = JSON.stringify({ type: "response.output_item.added", item: { type: "function_call", namespace: "ocx_agents", name: "spawn_agent", id: "fc_closed" } }); + rewrite(added); + rewrite(JSON.stringify({ type: "response.function_call_arguments.done", item_id: "fc_closed", arguments: "{}" })); + rewrite(JSON.stringify({ type: "response.output_item.done", item: { type: "function_call", namespace: "ocx_agents", name: "spawn_agent", id: "fc_closed" } })); + + const late = JSON.stringify({ type: "response.function_call_arguments.delta", item_id: "fc_closed", delta: "late" }); + expect(rewrite(late)).toBe(late); + }); + + test("does not reopen a closed id when its added snapshot repeats", () => { + const rewrite = createV2RoutedDelegationSseRewrite({ names: new Set(["spawn_agent"]) })!; + const added = JSON.stringify({ type: "response.output_item.added", item: { type: "function_call", namespace: "ocx_agents", name: "spawn_agent", id: "fc_duplicate" } }); + rewrite(added); + rewrite(JSON.stringify({ type: "response.function_call_arguments.done", item_id: "fc_duplicate", arguments: "{}" })); + rewrite(JSON.stringify({ type: "response.output_item.done", item: { type: "function_call", namespace: "ocx_agents", name: "spawn_agent", id: "fc_duplicate" } })); + + expect(JSON.parse(rewrite(added))).toMatchObject({ item: { namespace: "collaboration", encrypted_function_args: [] } }); + const late = JSON.stringify({ type: "response.function_call_arguments.delta", item_id: "fc_duplicate", delta: "late" }); + expect(rewrite(late)).toBe(late); + }); + +}); diff --git a/tests/server/v2-routed-delegation-policy.test.ts b/tests/server/v2-routed-delegation-policy.test.ts new file mode 100644 index 00000000000..137388833b3 --- /dev/null +++ b/tests/server/v2-routed-delegation-policy.test.ts @@ -0,0 +1,94 @@ +import { describe, expect, test } from "bun:test"; +import { + decideV2RoutedDelegationBridge, + hasMirrorableV2CollaborationCatalog, + type V2RoutedDelegationBridgePolicyInput, +} from "../../src/server/responses/v2-routed-delegation-policy"; + +const collaboration = { + type: "namespace", + name: "collaboration", + tools: [ + { type: "function", name: "spawn_agent", parameters: { type: "object" } }, + { type: "function", name: "list_agents", parameters: { type: "object" } }, + ], +}; + +function eligible( + patch: Partial = {}, +): V2RoutedDelegationBridgePolicyInput { + return { + enabled: true, + inboundWire: "responses", + multiAgentMode: "v2", + upstreamV2Enabled: true, + canonicalNativeRoute: true, + hasSubagentMarker: false, + threadSpawn: false, + comboAttempt: false, + compaction: false, + shadowRoute: false, + collaborationSurface: "v2", + body: { tools: [collaboration] }, + ...patch, + }; +} + +describe("Routed V2 delegation bridge policy", () => { + test("admits roots and positively classified thread-spawn children", () => { + expect(decideV2RoutedDelegationBridge(eligible())).toEqual({ + active: true, + decision: "active", + scope: "root", + }); + expect(decideV2RoutedDelegationBridge(eligible({ + hasSubagentMarker: true, + threadSpawn: true, + }))).toEqual({ active: true, decision: "active", scope: "child" }); + }); + + test("rejects maintenance markers unless thread_spawn is positive", () => { + expect(decideV2RoutedDelegationBridge(eligible({ hasSubagentMarker: true }))).toEqual({ + active: false, + decision: "maintenance_turn", + }); + expect(decideV2RoutedDelegationBridge(eligible({ + hasSubagentMarker: true, + threadSpawn: true, + })).active).toBe(true); + }); + + test("returns one bounded reason for every safety exclusion", () => { + const cases: Array<[Partial, string]> = [ + [{ enabled: false }, "disabled"], + [{ inboundWire: "chat" }, "not_v2"], + [{ multiAgentMode: "v1" }, "not_v2"], + [{ upstreamV2Enabled: false }, "not_v2"], + [{ canonicalNativeRoute: false }, "non_native_route"], + [{ comboAttempt: true }, "combo"], + [{ compaction: true }, "compaction"], + [{ shadowRoute: true }, "shadow_route"], + [{ body: { tools: [] } }, "no_collaboration_catalog"], + ]; + for (const [patch, decision] of cases) { + expect(decideV2RoutedDelegationBridge(eligible(patch))).toEqual({ active: false, decision }); + } + }); + + test("uses only the current-turn catalog and requires a mirrorable operation", () => { + const replayed = { type: "additional_tools", tools: [collaboration] }; + const leaf = { + type: "namespace", + name: "collaboration", + tools: [{ type: "function", name: "wait_agent", parameters: {} }], + }; + expect(hasMirrorableV2CollaborationCatalog({ + tools: [], + input: [replayed, { type: "additional_tools", tools: [leaf] }], + }, 1)).toBe(false); + expect(hasMirrorableV2CollaborationCatalog({ + tools: [], + input: [replayed, { type: "additional_tools", tools: [collaboration] }], + }, 1)).toBe(true); + }); +});