From 12da1223af0ce622b87f7ccb980817183d656582 Mon Sep 17 00:00:00 2001 From: HeavenllyDemon <160774215+HeavenllyDemon@users.noreply.github.com> Date: Tue, 18 Aug 2026 14:11:33 +0100 Subject: [PATCH] [05/06] feat(core): integrate deferred functions exec --- packages/core/src/agent/engine.ts | 235 +++++++++++++++++- packages/core/src/agent/gate.ts | 2 +- .../core/src/agent/tools/functions-exec.ts | 40 +++ packages/core/src/daemon.ts | 2 + packages/core/src/functions-exec/runtime.ts | 8 +- packages/core/src/providers/codex-oauth.ts | 5 +- .../core/src/providers/openai-compatible.ts | 42 +++- packages/core/src/providers/responses-sse.ts | 4 +- packages/core/src/providers/types.ts | 5 +- .../test/agent/engine-functions-exec.test.ts | 195 +++++++++++++++ packages/core/test/agent/engine-steer.test.ts | 5 +- .../test/agent/mode-toolset-census.test.ts | 4 +- .../test/agent/tools/functions-exec.test.ts | 36 +++ .../core/test/providers/codex-oauth.test.ts | 28 +++ .../test/providers/openai-compatible.test.ts | 32 ++- .../core/test/providers/responses-sse.test.ts | 8 + 16 files changed, 633 insertions(+), 18 deletions(-) create mode 100644 packages/core/src/agent/tools/functions-exec.ts create mode 100644 packages/core/test/agent/engine-functions-exec.test.ts create mode 100644 packages/core/test/agent/tools/functions-exec.test.ts diff --git a/packages/core/src/agent/engine.ts b/packages/core/src/agent/engine.ts index e8383984..48feb42d 100644 --- a/packages/core/src/agent/engine.ts +++ b/packages/core/src/agent/engine.ts @@ -41,6 +41,20 @@ import { guardAgentName, type AgentStatus, type BackgroundAgentRegistry, type Re import type { HookResult } from "../plugins/hook-runner"; import type { ComputerUseService } from "./computer-use"; import { SubagentTranscripts } from "./subagent-transcript"; +import { FunctionsExecCells } from "../functions-exec/cells"; +import { + FunctionsExecRuntime, + type FunctionsExecNestedCall, + type FunctionsExecRuntimeBridge, + type FunctionsExecRuntimeDeps, +} from "../functions-exec/runtime"; +import { type CellFrame, type JsonValue } from "../functions-exec/protocol"; +import { + FUNCTIONS_EXEC_TOOL, + FUNCTIONS_WAIT_TOOL, + functionsExecArgs, + functionsWaitArgs, +} from "./tools/functions-exec"; import { resolveModelAlias } from "./model-aliases"; import type { LspManager } from "./lsp/manager"; import { autoDiagnosticsSuffix, AUTO_DIAG_TOOL_NAMES } from "./lsp/auto-diagnostics"; @@ -63,6 +77,21 @@ export interface BgTaskLister { list(sessionId: string): Array<{ status: string }>; } +interface FunctionsExecNestedDispatchContext { + sessionId: string; + threadId: string; + outerCallId: string; + cwd: string; + signal: AbortSignal; + loaded: Set; + pins: Set; + rootsOverride?: string[]; + visionCapable?: boolean; + excludeTools?: Set; + allowTools?: Set; + input: TurnInputItem[]; +} + const MAIN_THREAD = "main"; /** The default tool-iteration bound for a CHILD (subagent) thread — NOT a global ceiling. The MAIN * thread has NO iteration bound at all (see `effectiveMaxIterations` in runThread): a fixed turn @@ -828,6 +857,9 @@ export interface EngineConfig { // resolves from when unset, and as what `contextWindow`'s default-Infinity ModelInfo lookup // falls back to matching when `live` is absent. provider: { provider: Provider; model: string; live?: () => { model: string; reasoningEffort?: string } }; + // The production factory constructs the Seatbelt runtime. Keeping the engine-facing contract + // narrow also permits a remote executor to preserve the same parent-owned tool bridge. + functionsExecRuntimeFactory?: (deps: Pick) => FunctionsExecRuntimeBridge; assembler: ContextAssembler; compactor: Compactor; mcp?: McpManager; @@ -1166,6 +1198,10 @@ export class AgentEngine { // concurrent session's dangerous-domain block, which is the identical cross-session bleed that // map already had to defend against, on a far worse payload. private readonly browserApprovedCallIds = new Set(); + private readonly functionsExecCells = new FunctionsExecCells(); + private readonly functionsExecNested = new Map Promise>(); + private readonly functionsExecRuntime: FunctionsExecRuntimeBridge; + private readonly pendingFunctionsExecMedia = new Map>>(); // CC-parity subagent transcript writer (subagent-transcript.ts) — constructed in the constructor // BODY (not a field initializer) so it can close over `this.cfg`, which parameter-property // assignment guarantees is already set by the time the body runs. @@ -1175,6 +1211,15 @@ export class AgentEngine { // bridge (below) and task-stop.ts's plain tool can share it; nothing engine-local is needed here. constructor(private readonly cfg: EngineConfig) { this.subagentTranscripts = new SubagentTranscripts((sessionId) => this.cfg.tmpDirOf?.(sessionId)); + const functionsExecRuntimeDeps: Pick = { + callTool: async (call) => { + const dispatch = this.functionsExecNested.get(call.cellId); + if (!dispatch) throw new Error("functions.exec cell is unavailable"); + return dispatch(call); + }, + onFrame: (sessionId, cellId, frame) => this.functionsExecCells.recordFrame(sessionId, cellId, frame), + }; + this.functionsExecRuntime = this.cfg.functionsExecRuntimeFactory?.(functionsExecRuntimeDeps) ?? new FunctionsExecRuntime(functionsExecRuntimeDeps); } /** session-activity-hygiene T5: a top-level turn for this session has SETTLED — every terminal @@ -1372,6 +1417,8 @@ export class AgentEngine { /** Abort the in-flight turn for a session, if any. Idempotent — safe to call when idle. */ interrupt(sessionId: string): { wasRunning: boolean } { const ac = this.aborters.get(sessionId); + this.functionsExecCells.cancelSession(sessionId); + this.clearFunctionsExecMedia(sessionId); if (!ac) return { wasRunning: false }; ac.abort(); return { wasRunning: true }; @@ -1587,6 +1634,7 @@ export class AgentEngine { // this session — running OR terminal (unlike task_stop's running-only pin above), since a // FINISHED agent must stay collectable via agent_output without a ToolSearch load. if (this.cfg.bgAgents?.list(sessionId).length) { pins.add("agent_list"); pins.add("agent_output"); } + if (this.functionsExecCells.hasWaitable(sessionId)) pins.add(FUNCTIONS_WAIT_TOOL); // Task B4: /ultracode pins the deferred Workflow tool for the turn — mirrors exit_plan_mode's // own pin just above (a state-required deferred built-in forced visible without touching the // sticky loadedTools set). `ultracodeActive` is computed ONCE per turn by turn() itself (its own @@ -2849,8 +2897,9 @@ export class AgentEngine { // requestApproval invocation triggered by THIS round's calls, further down. `opts // .ultracodeActive` (Task B4) is NOT recomputed per round like the rest of this Set — it's // the one fixed-for-the-whole-turn value turn() already decided, just applied at this seam too. - const pins = tsEnabled ? this.pinnedTools(sessionId, meta, cwd, opts.ultracodeActive) : new Set(); - const effectiveLoaded = pins.size ? new Set([...loaded, ...pins]) : loaded; + const pins = tsEnabled ? this.pinnedTools(sessionId, meta, cwd, opts.ultracodeActive) : new Set(); + const effectiveLoaded = pins.size ? new Set([...loaded, ...pins]) : loaded; + this.drainFunctionsExecMedia(sessionId, threadId, input); for await (const ev of this.cfg.provider.provider.streamTurn({ model: opts.model, @@ -3944,8 +3993,14 @@ export class AgentEngine { } for (const call of calls) { - this.emit(sessionId, { type: "tool_call", sessionId, threadId, callId: call.callId, name: call.name, argsJson: call.argsJson }); - input.push({ type: "function_call", callId: call.callId, name: call.name, argsJson: call.argsJson }); + const persistedArgsJson = call.name === FUNCTIONS_EXEC_TOOL + ? JSON.stringify({ source: "[functions.exec source omitted]" }) + : call.argsJson; + this.emit(sessionId, { type: "tool_call", sessionId, threadId, callId: call.callId, name: call.name, argsJson: persistedArgsJson }); + // Never replay executable source into the model context. Besides keeping later history + // stable, this prevents a source literal from smuggling a media data URL around the + // transient media channel. The worker still receives `call.argsJson` below. + input.push({ type: "function_call", callId: call.callId, name: call.name, argsJson: persistedArgsJson }); // diff-tabs Task 5: `ToolOutcome` (registry.ts) is the single source of truth for the // {output, isError, fileDiff?} shape — reused here rather than hand-copied so a future @@ -4065,6 +4120,21 @@ export class AgentEngine { input.push({ type: "tool_result", callId: call.callId, output: outcome.output, isError: outcome.isError }); continue; } + if (call.name === FUNCTIONS_EXEC_TOOL) { + outcome = await this.runFunctionsExec(call, { + sessionId, threadId, outerCallId: call.callId, cwd, signal, loaded, pins, rootsOverride, visionCapable, + excludeTools, allowTools, input, + }); + this.emit(sessionId, { type: "tool_result", sessionId, threadId, callId: call.callId, output: outcome.output, isError: outcome.isError }); + input.push({ type: "tool_result", callId: call.callId, output: outcome.output, isError: outcome.isError }); + continue; + } + if (call.name === FUNCTIONS_WAIT_TOOL) { + outcome = await this.runFunctionsWait(call, sessionId, signal); + this.emit(sessionId, { type: "tool_result", sessionId, threadId, callId: call.callId, output: outcome.output, isError: outcome.isError }); + input.push({ type: "tool_result", callId: call.callId, output: outcome.output, isError: outcome.isError }); + continue; + } let decision = this.cfg.gate.evaluate(call.name, meta.approvalPolicy); // Worktree tools are MUTATING (gate.ts), so under `ask` policy (the DEFAULT) `decision` is // "ask", not "allow" — checked here, BEFORE the generic `decision === "ask"` branch below, @@ -4821,6 +4891,7 @@ export class AgentEngine { // — the model sees them on the NEXT round's provider call. Placed here (round end, not per // call) so an image never splits a function_call/tool_result pair. No-op when nothing staged. this.drainRoundImages(sessionId, threadId, calls, input); + this.drainFunctionsExecMedia(sessionId, threadId, input); } // CHILD-ONLY terminal. The main thread's `effectiveMaxIterations` is Infinity, so its loop above @@ -6104,6 +6175,162 @@ export class AgentEngine { } } + private functionsExecMediaKey(sessionId: string, threadId: string): string { + return `${sessionId}\u0000${threadId}`; + } + + private stageFunctionsExecFrame( + sessionId: string, + threadId: string, + cellId: string, + frame: CellFrame, + ): void { + if (frame.type === "notification") { + const content = `\n${cellId}\nprogress\n${this.sanitizeForReminder(frame.text)}\n`; + this.emit(sessionId, { type: "notification_requested", sessionId, threadId, title: "Norma", message: frame.text }); + this.cfg.hub.append(sessionId, { type: "task_notification", sessionId, threadId, content }); + if (this.isRunning(sessionId)) this.retriggerPending.add(sessionId); + else void this.runTurn(sessionId).catch((error) => console.error("functions.exec notification turn failed:", error)); + return; + } + if (frame.type !== "image" && frame.type !== "audio") return; + if (frame.type === "audio" && !/^data:audio\/(?:mpeg|wav);base64,/iu.test(frame.dataUrl)) return; + const key = this.functionsExecMediaKey(sessionId, threadId); + const pending = this.pendingFunctionsExecMedia.get(key) ?? []; + pending.push(frame.type === "image" + ? { type: "image", imageUrl: frame.dataUrl, detail: frame.detail } + : { type: "audio", dataUrl: frame.dataUrl }); + this.pendingFunctionsExecMedia.set(key, pending); + } + + private drainFunctionsExecMedia(sessionId: string, threadId: string, input: TurnInputItem[]): void { + const key = this.functionsExecMediaKey(sessionId, threadId); + const pending = this.pendingFunctionsExecMedia.get(key); + if (!pending) return; + this.pendingFunctionsExecMedia.delete(key); + input.push(...pending); + } + + private clearFunctionsExecMedia(sessionId: string): void { + for (const key of this.pendingFunctionsExecMedia.keys()) { + if (key.startsWith(`${sessionId}\u0000`)) this.pendingFunctionsExecMedia.delete(key); + } + } + + private async runFunctionsExec( + call: { callId: string; argsJson: string }, + context: FunctionsExecNestedDispatchContext, + ): Promise { + let raw: unknown; + try { raw = JSON.parse(call.argsJson || "{}"); } + catch { return { output: "tool arguments were not valid JSON", isError: true }; } + const parsed = functionsExecArgs.safeParse(raw); + if (!parsed.success) return { output: "invalid arguments for functions.exec", isError: true }; + const roots = this.writableRoots(context.sessionId, context.cwd ? repoRootFor(context.cwd) : null, context.rootsOverride); + let cellId = ""; + try { + cellId = this.functionsExecCells.start({ + sessionId: context.sessionId, + run: (id) => this.functionsExecRuntime.execute({ + sessionId: context.sessionId, + cellId: id, + source: parsed.data.source, + protectedRoots: roots, + ...(parsed.data.timeoutMs === undefined ? {} : { timeoutMs: parsed.data.timeoutMs }), + }), + cancel: (id) => { this.functionsExecRuntime.cancel(context.sessionId, id); }, + onFrame: (sessionId, id, frame) => this.stageFunctionsExecFrame(sessionId, context.threadId, id, frame), + onRemoved: (_sessionId, id) => { this.functionsExecNested.delete(id); }, + }); + this.functionsExecNested.set(cellId, (nested) => this.dispatchFunctionsExecNested(nested, cellId, context)); + const checkpoint = await this.functionsExecCells.next(context.sessionId, cellId, context.signal); + return { output: JSON.stringify({ cellId, ...checkpoint }), isError: checkpoint.status === "failed" }; + } catch (error) { + if (cellId) this.functionsExecCells.cancel(context.sessionId, cellId); + return { output: error instanceof Error ? error.message : String(error), isError: true }; + } + } + + private async runFunctionsWait( + call: { argsJson: string }, + sessionId: string, + signal: AbortSignal, + ): Promise { + let raw: unknown; + try { raw = JSON.parse(call.argsJson || "{}"); } + catch { return { output: "tool arguments were not valid JSON", isError: true }; } + const parsed = functionsWaitArgs.safeParse(raw); + if (!parsed.success) return { output: "invalid arguments for functions.wait", isError: true }; + if (!this.functionsExecCells.canWait(sessionId, parsed.data.cellId)) { + return { output: `functions.exec cell ${parsed.data.cellId} has not yielded`, isError: true }; + } + try { + const checkpoint = await this.functionsExecCells.next(sessionId, parsed.data.cellId, signal); + return { output: JSON.stringify({ cellId: parsed.data.cellId, ...checkpoint }), isError: checkpoint.status === "failed" }; + } catch (error) { + return { output: error instanceof Error ? error.message : String(error), isError: true }; + } + } + + private async dispatchFunctionsExecNested( + nested: FunctionsExecNestedCall, + cellId: string, + context: FunctionsExecNestedDispatchContext, + ): Promise { + const call = { + callId: `${context.outerCallId}:fx:${cellId}:${nested.callId}`, + name: nested.name, + argsJson: JSON.stringify(nested.args), + }; + this.emit(context.sessionId, { type: "tool_call", sessionId: context.sessionId, threadId: context.threadId, ...call }); + context.input.push({ type: "function_call", ...call }); + let outcome: ToolOutcome & { deniedByHuman?: boolean }; + if (context.excludeTools?.has(call.name) || (context.allowTools && !context.allowTools.has(call.name))) { + outcome = { output: `tool ${call.name} is not available in this session`, isError: true }; + } else { + const meta = this.cfg.store.meta(context.sessionId); + const decision = this.cfg.gate.evaluate(call.name, meta.approvalPolicy); + const webFetchCard = call.name === "web_fetch" ? this.webFetchGate(call, context.cwd) : null; + if (decision === "deny") { + outcome = { output: `Denied automatically — ${call.name} is not permitted in the current approval mode`, isError: true }; + } else if (webFetchCard && meta.approvalPolicy !== "bypass") { + outcome = meta.approvalPolicy === "dont-ask" + ? { output: "web_fetch denied — dont-ask mode declines a fetch to a dangerous domain with no standing WebFetch rule.", isError: true } + : await this.requestApproval(call, context.cwd, context.sessionId, context.threadId, context.signal, { + timeoutMs: this.approvalTimeoutFor(meta), + summary: webFetchCard.summary, + options: webFetchCard.options, + }, context.loaded, undefined, context.pins, context.rootsOverride, context.visionCapable, context.excludeTools, context.allowTools); + } else if (decision === "ask" || (call.name === "bash" && (nested.args as { dangerouslyDisableSandbox?: unknown }).dangerouslyDisableSandbox === true && this.cfg.store.meta(context.sessionId).approvalPolicy !== "bypass")) { + outcome = await this.requestApproval(call, context.cwd, context.sessionId, context.threadId, context.signal, { + timeoutMs: this.approvalTimeoutFor(meta), + summary: approvalCardSummary(call), + }, context.loaded, undefined, context.pins, context.rootsOverride, context.visionCapable, context.excludeTools, context.allowTools); + } else { + outcome = await this.executeCall( + call, + context.cwd, + context.sessionId, + context.threadId, + context.signal, + new Set([...context.loaded, call.name]), + context.pins, + context.rootsOverride, + context.visionCapable, + context.excludeTools, + context.allowTools, + ); + } + } + this.emit(context.sessionId, { type: "tool_result", sessionId: context.sessionId, threadId: context.threadId, callId: call.callId, output: outcome.output, isError: outcome.isError }); + context.input.push({ type: "tool_result", callId: call.callId, output: outcome.output, isError: outcome.isError }); + if (outcome.deniedByHuman) { + this.functionsExecCells.cancel(context.sessionId, cellId); + throw new Error(outcome.output); + } + return outcome.output; + } + private async executeCall( call: { callId: string; name: string; argsJson: string }, cwd: string, diff --git a/packages/core/src/agent/gate.ts b/packages/core/src/agent/gate.ts index 1990137e..d327d634 100644 --- a/packages/core/src/agent/gate.ts +++ b/packages/core/src/agent/gate.ts @@ -95,7 +95,7 @@ export type SessionApprovalPolicy = "plan" | "dont-ask" | "ask" | "accept-edits" // (dispatch's own default — server.ts's session.dispatch) — a card a headless coordinator can never // answer, so in practice a silent hang/timeout-deny on every real call. Same fix shape as // task_stop's own entry above. -const READ_ONLY = new Set(["read", "glob", "grep", "ls", "bash_output", "Skill", "ToolSearch", "ask_user", "AskQuestion", "task_create", "task_update", "task_list", "task_get", "exit_plan_mode", "enter_plan_mode", "spawn_agent", "send_message", "task_stop", "agent_list", "agent_output", "lsp", "push_notification", "list_sessions", "manage_session"]); +const READ_ONLY = new Set(["read", "glob", "grep", "ls", "bash_output", "Skill", "ToolSearch", "functions.exec", "functions.wait", "ask_user", "AskQuestion", "task_create", "task_update", "task_list", "task_get", "exit_plan_mode", "enter_plan_mode", "spawn_agent", "send_message", "task_stop", "agent_list", "agent_output", "lsp", "push_notification", "list_sessions", "manage_session"]); // `computer` (Phase 5 CU) is MUTATING: a computer-use action drives real mouse/keyboard/screen, so // it must pass the gate on EVERY call (spec §4.6: "every CU action passes the permission gate") — // ask → per-action approval card, auto → allow, plan → deny (CU makes changes). Note this is the diff --git a/packages/core/src/agent/tools/functions-exec.ts b/packages/core/src/agent/tools/functions-exec.ts new file mode 100644 index 00000000..cf87197b --- /dev/null +++ b/packages/core/src/agent/tools/functions-exec.ts @@ -0,0 +1,40 @@ +import { z } from "zod"; +import { MAX_CELL_ID_CHARS, MAX_FUNCTIONS_EXEC_SOURCE_CHARS } from "../../functions-exec/protocol"; +import { functionsExecSandboxAvailable } from "../../functions-exec/sandbox"; +import type { ToolRegistry } from "./registry"; + +export const FUNCTIONS_EXEC_TOOL = "functions.exec"; +export const FUNCTIONS_WAIT_TOOL = "functions.wait"; + +export const functionsExecArgs = z.object({ + source: z.string().min(1).max(MAX_FUNCTIONS_EXEC_SOURCE_CHARS), + timeoutMs: z.number().int().min(1).max(60_000).optional(), +}).strict(); + +export const functionsWaitArgs = z.object({ + cellId: z.string().min(1).max(MAX_CELL_ID_CHARS), +}).strict(); + +/** + * Registers only the model-facing handles. AgentEngine owns cell state and runs both through its + * normal dispatch path; the worker itself never gains a direct filesystem or network capability. + */ +export function registerFunctionsExecTools(registry: ToolRegistry, supported = functionsExecSandboxAvailable()): void { + if (!supported) return; + registry.register({ + name: FUNCTIONS_EXEC_TOOL, + description: "Run bounded JavaScript in an isolated worker. Load this deferred tool with ToolSearch first. JavaScript has no direct filesystem, process, or network access; use tools.bash, tools.read, tools.web_fetch, or tools.web_search, which each use Norma's normal permission path. Use tools.text(), tools.image(), tools.audio(), tools.notify(), and await tools.yield().", + args: functionsExecArgs, + modes: ["code"], + deferred: true, + run() { throw new Error("functions.exec requires the AgentEngine runtime bridge"); }, + }); + registry.register({ + name: FUNCTIONS_WAIT_TOOL, + description: "Wait for the next checkpoint from a yielded functions.exec cell. This deferred tool is available only while a cell has yielded or has a pending terminal result.", + args: functionsWaitArgs, + modes: ["code"], + deferred: true, + run() { throw new Error("functions.wait requires the AgentEngine runtime bridge"); }, + }); +} diff --git a/packages/core/src/daemon.ts b/packages/core/src/daemon.ts index 1c1c50c9..f70b4963 100644 --- a/packages/core/src/daemon.ts +++ b/packages/core/src/daemon.ts @@ -26,6 +26,7 @@ import { registerBashTool } from "./agent/tools/bash"; import { registerBackgroundTools } from "./agent/tools/background"; import { registerSkillTools } from "./agent/tools/skill"; import { registerToolSearchTool } from "./agent/tools/toolsearch"; +import { registerFunctionsExecTools } from "./agent/tools/functions-exec"; import { registerAskUserTool } from "./agent/tools/ask-user"; import { registerAskQuestionTool } from "./agent/tools/ask-question"; import { registerTaskTools } from "./agent/tools/tasks"; @@ -670,6 +671,7 @@ export async function startDaemon(opts: { registerBackgroundTools(registry, { bgRegistry }, { deferred: true }); registerSkillTools(registry, { skills: skillStore }); registerToolSearchTool(registry); + registerFunctionsExecTools(registry); questions = new QuestionBroker(); taskStore = new TaskStore(); registerAskUserTool(registry); diff --git a/packages/core/src/functions-exec/runtime.ts b/packages/core/src/functions-exec/runtime.ts index eb44103d..67186b44 100644 --- a/packages/core/src/functions-exec/runtime.ts +++ b/packages/core/src/functions-exec/runtime.ts @@ -41,6 +41,12 @@ export interface FunctionsExecRuntimeDeps { hasSeatbelt?: boolean; } +/** The engine-facing worker contract, kept narrow so alternate process hosts can preserve the same bridge. */ +export interface FunctionsExecRuntimeBridge { + cancel(sessionId: string, cellId: string): boolean; + execute(input: FunctionsExecRuntimeInput): Promise; +} + const DEFAULT_TIMEOUT_MS = 10_000; const MAX_TIMEOUT_MS = 60_000; @@ -61,7 +67,7 @@ function activeKey(sessionId: string, cellId: string): string { return sessionId + "\u0000" + cellId; } -export class FunctionsExecRuntime { +export class FunctionsExecRuntime implements FunctionsExecRuntimeBridge { private readonly active = new Map void>(); constructor(private readonly deps: FunctionsExecRuntimeDeps = {}) {} diff --git a/packages/core/src/providers/codex-oauth.ts b/packages/core/src/providers/codex-oauth.ts index 0666fd7e..e582992c 100644 --- a/packages/core/src/providers/codex-oauth.ts +++ b/packages/core/src/providers/codex-oauth.ts @@ -1,7 +1,7 @@ import type { SecretStore } from "../auth/secret-store"; import type { ModelInfo, Provider, ProviderEvent, TurnRequest } from "./types"; import { ResponsesSseParser } from "./responses-sse"; -import { buildRequestBody, mapHttpError } from "./openai-compatible"; +import { buildRequestBody, mapHttpError, wireToolNames } from "./openai-compatible"; import { refreshTokens, type OAuthTokens } from "./pkce"; import { CODEX, CODEX_MODELS } from "./codex-config"; @@ -86,7 +86,8 @@ export class CodexOAuthProvider implements Provider { } if (!res.ok) { yield await mapHttpError(res.status, res.headers.get("retry-after"), res.text()); return; } - const parser = new ResponsesSseParser(); + const names = wireToolNames(req.tools); + const parser = new ResponsesSseParser((name) => names.get(name) ?? name); const reader = res.body!.getReader(); try { while (true) { diff --git a/packages/core/src/providers/openai-compatible.ts b/packages/core/src/providers/openai-compatible.ts index a717b838..f4b25ef2 100644 --- a/packages/core/src/providers/openai-compatible.ts +++ b/packages/core/src/providers/openai-compatible.ts @@ -1,3 +1,4 @@ +import { createHash } from "node:crypto"; import type { ModelInfo, Provider, ProviderEvent, TurnInputItem, TurnRequest, ToolSpec } from "./types"; import { ResponsesSseParser } from "./responses-sse"; import { parseProviderErrorCode } from "./errors"; @@ -25,7 +26,28 @@ export interface OpenAICompatibleConfig { * Responses API ResponseItem shape (not yet live-verified against the codex backend; * the parity doc only covers message/function_call_output — Task 12's live gate confirms). */ -export function mapInput(items: TurnInputItem[]): unknown[] { +const RESPONSES_FUNCTION_NAME = /^[A-Za-z0-9_-]{1,64}$/; + +/** Encode namespaced internal tools into the restricted Responses API function-name grammar. */ +export function wireToolName(name: string): string { + if (RESPONSES_FUNCTION_NAME.test(name)) return name; + const encoded = `norma_${Buffer.from(name, "utf8").toString("base64url")}`; + if (RESPONSES_FUNCTION_NAME.test(encoded)) return encoded; + return `norma_${createHash("sha256").update(name).digest("hex").slice(0, 58)}`; +} + +/** Maps the constrained Responses wire names back to their internal tool names for one request. */ +export function wireToolNames(tools: ToolSpec[] | undefined): Map { + const names = new Map(); + for (const tool of tools ?? []) { + const wire = wireToolName(tool.name); + if (names.has(wire)) throw new Error(`tool names collide after Responses transport encoding: ${tool.name}`); + names.set(wire, tool.name); + } + return names; +} + +export function mapInput(items: TurnInputItem[], toWireName: (name: string) => string = wireToolName): unknown[] { return items.map((i) => { if (i.type === "message") { // Map role to the appropriate content item type per the Responses API schema. @@ -34,7 +56,7 @@ export function mapInput(items: TurnInputItem[]): unknown[] { return { type: "message", role: i.role, content: [{ type: contentType, text: i.content }] }; } if (i.type === "function_call") { - return { type: "function_call", call_id: i.callId, name: i.name, arguments: i.argsJson }; + return { type: "function_call", call_id: i.callId, name: toWireName(i.name), arguments: i.argsJson }; } if (i.type === "reasoning") return JSON.parse(i.itemJson); // opaque passthrough — never inspected // Computer-use image (Phase 5 CU): a user message carrying an input_image. This is the ONLY @@ -44,15 +66,24 @@ export function mapInput(items: TurnInputItem[]): unknown[] { if (i.type === "image") { const content: unknown[] = []; if (i.alt) content.push({ type: "input_text", text: i.alt }); - content.push({ type: "input_image", image_url: i.imageUrl }); + content.push({ type: "input_image", image_url: i.imageUrl, ...(i.detail === undefined ? {} : { detail: i.detail }) }); return { type: "message", role: "user", content }; } + if (i.type === "audio") { + const match = /^data:audio\/(wav|mpeg);base64,([a-z0-9+/=]+)$/iu.exec(i.dataUrl); + if (!match) throw new Error("audio input requires a base64 audio/mpeg or audio/wav data URL"); + return { + type: "message", + role: "user", + content: [{ type: "input_audio", input_audio: { data: match[2], format: match[1]!.toLowerCase() === "mpeg" ? "mp3" : "wav" } }], + }; + } return { type: "function_call_output", call_id: i.callId, output: i.output }; }); } export function mapTools(tools: ToolSpec[] | undefined): unknown[] { - return (tools ?? []).map((t) => ({ type: "function", name: t.name, description: t.description, parameters: t.parameters, strict: false })); + return (tools ?? []).map((t) => ({ type: "function", name: wireToolName(t.name), description: t.description, parameters: t.parameters, strict: false })); } /** @@ -135,7 +166,8 @@ export class OpenAICompatibleProvider implements Provider { } if (!res.ok) { yield await mapHttpError(res.status, res.headers.get("retry-after"), res.text()); return; } - const parser = new ResponsesSseParser(); + const names = wireToolNames(req.tools); + const parser = new ResponsesSseParser((name) => names.get(name) ?? name); const reader = res.body!.getReader(); try { while (true) { diff --git a/packages/core/src/providers/responses-sse.ts b/packages/core/src/providers/responses-sse.ts index d0463fb0..663eadf5 100644 --- a/packages/core/src/providers/responses-sse.ts +++ b/packages/core/src/providers/responses-sse.ts @@ -9,6 +9,8 @@ export class ResponsesSseParser { private decoder = new TextDecoder(); private sawToolCall = false; + constructor(private readonly fromWireToolName: (name: string) => string = (name) => name) {} + push(chunk: Uint8Array): ProviderEvent[] { // Normalize after appending so \r\n split across two push() calls is handled correctly. this.buf = (this.buf + this.decoder.decode(chunk, { stream: true })).replace(/\r\n/g, "\n"); @@ -61,7 +63,7 @@ export class ResponsesSseParser { return [{ type: "tool_call", callId: String(data.item.call_id), - name: String(data.item.name), + name: this.fromWireToolName(String(data.item.name)), argsJson: String(data.item.arguments ?? ""), }]; } diff --git a/packages/core/src/providers/types.ts b/packages/core/src/providers/types.ts index ff45fd0c..71e14a16 100644 --- a/packages/core/src/providers/types.ts +++ b/packages/core/src/providers/types.ts @@ -48,7 +48,10 @@ export type TurnInputItem = // in-turn by the engine's image drain (never a persisted session event — see engine.ts's // pendingImages), so `eventToInput` has no case for it and cross-turn history never reconstructs // a past image. - | { type: "image"; imageUrl: string; alt?: string }; + | { type: "image"; imageUrl: string; alt?: string; detail?: "auto" | "low" | "high" | "original" } + // Transient functions.exec audio. Like images, it exists only in the in-memory continuation for + // the next provider request and is never reconstructed from session history. + | { type: "audio"; dataUrl: string }; export interface ToolSpec { name: string; diff --git a/packages/core/test/agent/engine-functions-exec.test.ts b/packages/core/test/agent/engine-functions-exec.test.ts new file mode 100644 index 00000000..2ba6cf0a --- /dev/null +++ b/packages/core/test/agent/engine-functions-exec.test.ts @@ -0,0 +1,195 @@ +import { describe, expect, test } from "bun:test"; +import { z } from "zod"; +import { ToolRegistry } from "../../src/agent/tools/registry"; +import { registerToolSearchTool } from "../../src/agent/tools/toolsearch"; +import { registerFunctionsExecTools } from "../../src/agent/tools/functions-exec"; +import { FakeProvider } from "../../src/agent/fake-provider"; +import { setupEngine } from "./engine-steer.test"; +import type { FunctionsExecNestedCall, FunctionsExecRuntimeDeps } from "../../src/functions-exec/runtime"; +import { ImageDetail } from "../../src/functions-exec/protocol"; + +const done = (reason: "end_turn" | "tool_calls") => ({ type: "done" as const, stopReason: reason }); + +function runtimeForNestedCall(nested: Pick) { + return (deps: Pick) => ({ + cancel: () => true, + async execute(input: { sessionId: string; cellId: string }) { + if (!deps.callTool || !deps.onFrame) throw new Error("expected engine bridge callbacks"); + const result = await deps.callTool({ ...input, callId: "nested-1", ...nested }); + deps.onFrame(input.sessionId, input.cellId, { type: "text", text: typeof result === "string" ? result : JSON.stringify(result) }); + return { status: "completed" as const, cellId: input.cellId, frames: [] }; + }, + }); +} + +describe("engine functions.exec integration", () => { + test("requires ToolSearch and keeps executable source opaque in events and provider input", async () => { + const registry = new ToolRegistry(); + registerToolSearchTool(registry); + registerFunctionsExecTools(registry, true); + const source = "tools.text('private source')"; + const provider = new FakeProvider([ + [{ type: "tool_call", callId: "load", name: "ToolSearch", argsJson: JSON.stringify({ query: "select:functions.exec" }) }, done("tool_calls")], + [{ type: "tool_call", callId: "run", name: "functions.exec", argsJson: JSON.stringify({ source }) }, done("tool_calls")], + [{ type: "text_delta", delta: "done" }, done("end_turn")], + ]); + const { engine, events, sessionId } = setupEngine(provider, { registry, toolSearch: {} }); + + await engine.runTurn(sessionId); + + expect(provider.requests[0]!.tools?.map((tool) => tool.name)).toContain("ToolSearch"); + expect(provider.requests[0]!.tools?.map((tool) => tool.name)).not.toContain("functions.exec"); + expect(provider.requests[1]!.tools?.map((tool) => tool.name)).toContain("functions.exec"); + const storedCall = events.find((event) => event.type === "tool_call" && event.callId === "run"); + if (!storedCall || storedCall.type !== "tool_call") throw new Error("expected functions.exec tool call"); + expect(storedCall.argsJson).toBe(JSON.stringify({ source: "[functions.exec source omitted]" })); + expect(storedCall.argsJson).not.toContain("private source"); + expect(provider.requests.flatMap((request) => request.input).some((item) => item.type === "function_call" && item.argsJson.includes("private source"))).toBe(false); + }); + + test("routes a nested bash call through the ordinary auto-policy tool execution path", async () => { + const registry = new ToolRegistry(); + registerToolSearchTool(registry); + registerFunctionsExecTools(registry, true); + const calls: string[] = []; + registry.register({ + name: "bash", + description: "test bash", + args: z.object({ command: z.string() }), + async run({ command }) { calls.push(command); return "bash ran"; }, + }); + const provider = new FakeProvider([ + [{ type: "tool_call", callId: "load", name: "ToolSearch", argsJson: JSON.stringify({ query: "select:functions.exec" }) }, done("tool_calls")], + [{ type: "tool_call", callId: "run", name: "functions.exec", argsJson: JSON.stringify({ source: "ignored" }) }, done("tool_calls")], + [{ type: "text_delta", delta: "done" }, done("end_turn")], + ]); + const { engine, events, sessionId } = setupEngine(provider, { + registry, + toolSearch: {}, + functionsExecRuntimeFactory: runtimeForNestedCall({ name: "bash", args: { command: "pwd" } }), + }); + + await engine.runTurn(sessionId); + + expect(calls).toEqual(["pwd"]); + expect(events.some((event) => event.type === "tool_call" && event.name === "bash")).toBe(true); + expect(events.some((event) => event.type === "tool_result" && event.output === "bash ran")).toBe(true); + }); + + test("applies the dangerous-domain floor to nested web_fetch calls", async () => { + const registry = new ToolRegistry(); + registerToolSearchTool(registry); + registerFunctionsExecTools(registry, true); + let fetches = 0; + registry.register({ + name: "web_fetch", + description: "test fetch", + args: z.object({ url: z.string() }), + async run() { fetches += 1; return "should not run"; }, + }); + const provider = new FakeProvider([ + [{ type: "tool_call", callId: "load", name: "ToolSearch", argsJson: JSON.stringify({ query: "select:functions.exec" }) }, done("tool_calls")], + [{ type: "tool_call", callId: "run", name: "functions.exec", argsJson: JSON.stringify({ source: "ignored" }) }, done("tool_calls")], + [{ type: "text_delta", delta: "done" }, done("end_turn")], + ]); + const { engine, events, sessionId } = setupEngine(provider, { + registry, + policy: "dont-ask", + toolSearch: {}, + functionsExecRuntimeFactory: runtimeForNestedCall({ name: "web_fetch", args: { url: "https://ngrok.io/collect" } }), + }); + + await engine.runTurn(sessionId); + + expect(fetches).toBe(0); + expect(events.some((event) => event.type === "tool_result" && event.output.includes("web_fetch denied"))).toBe(true); + expect(events.some((event) => event.type === "approval_requested")).toBe(false); + }); + + test("interrupt cancels a yielded cell after its outer turn has already settled", async () => { + const registry = new ToolRegistry(); + registerToolSearchTool(registry); + registerFunctionsExecTools(registry, true); + let cancelled = false; + const provider = new FakeProvider([ + [{ type: "tool_call", callId: "load", name: "ToolSearch", argsJson: JSON.stringify({ query: "select:functions.exec" }) }, done("tool_calls")], + [{ type: "tool_call", callId: "run", name: "functions.exec", argsJson: JSON.stringify({ source: "ignored" }) }, done("tool_calls")], + [{ type: "text_delta", delta: "done" }, done("end_turn")], + ]); + const { engine, sessionId } = setupEngine(provider, { + registry, + toolSearch: {}, + functionsExecRuntimeFactory: (deps) => ({ + cancel: () => { cancelled = true; return true; }, + async execute(input) { + deps.onFrame?.(input.sessionId, input.cellId, { type: "yield" }); + return await new Promise(() => {}); + }, + }), + }); + + await engine.runTurn(sessionId); + + expect(engine.interrupt(sessionId)).toEqual({ wasRunning: false }); + expect(cancelled).toBe(true); + }); + + test("forwards image and audio only into the next provider request without persisting their data URLs", async () => { + const registry = new ToolRegistry(); + registerToolSearchTool(registry); + registerFunctionsExecTools(registry, true); + const imageUrl = "data:image/png;base64,aGVsbG8="; + const audioUrl = "data:audio/wav;base64,aGVsbG8="; + const provider = new FakeProvider([ + [{ type: "tool_call", callId: "load", name: "ToolSearch", argsJson: JSON.stringify({ query: "select:functions.exec" }) }, done("tool_calls")], + [{ type: "tool_call", callId: "run", name: "functions.exec", argsJson: JSON.stringify({ source: "ignored" }) }, done("tool_calls")], + [{ type: "text_delta", delta: "done" }, done("end_turn")], + ]); + const { engine, events, sessionId } = setupEngine(provider, { + registry, + toolSearch: {}, + functionsExecRuntimeFactory: (deps) => ({ + cancel: () => true, + async execute(input) { + deps.onFrame?.(input.sessionId, input.cellId, { type: "image", dataUrl: imageUrl, detail: ImageDetail.High }); + deps.onFrame?.(input.sessionId, input.cellId, { type: "audio", dataUrl: audioUrl }); + return { status: "completed", cellId: input.cellId, frames: [] }; + }, + }), + }); + + await engine.runTurn(sessionId); + + expect(provider.requests[2]!.input).toContainEqual({ type: "image", imageUrl, detail: "high" }); + expect(provider.requests[2]!.input).toContainEqual({ type: "audio", dataUrl: audioUrl }); + expect(JSON.stringify(events)).not.toContain(imageUrl); + expect(JSON.stringify(events)).not.toContain(audioUrl); + }); + + test("delivers notify frames as an immediate task notification", async () => { + const registry = new ToolRegistry(); + registerToolSearchTool(registry); + registerFunctionsExecTools(registry, true); + const provider = new FakeProvider([ + [{ type: "tool_call", callId: "load", name: "ToolSearch", argsJson: JSON.stringify({ query: "select:functions.exec" }) }, done("tool_calls")], + [{ type: "tool_call", callId: "run", name: "functions.exec", argsJson: JSON.stringify({ source: "ignored" }) }, done("tool_calls")], + [{ type: "text_delta", delta: "done" }, done("end_turn")], + ]); + const { engine, events, sessionId } = setupEngine(provider, { + registry, + toolSearch: {}, + functionsExecRuntimeFactory: (deps) => ({ + cancel: () => true, + async execute(input) { + deps.onFrame?.(input.sessionId, input.cellId, { type: "notification", text: "cell progress" }); + return { status: "completed", cellId: input.cellId, frames: [] }; + }, + }), + }); + + await engine.runTurn(sessionId); + + expect(events.some((event) => event.type === "notification_requested" && event.message === "cell progress")).toBe(true); + expect(events.some((event) => event.type === "task_notification" && event.content.includes("cell progress"))).toBe(true); + }); +}); diff --git a/packages/core/test/agent/engine-steer.test.ts b/packages/core/test/agent/engine-steer.test.ts index 686c4537..411a63c8 100644 --- a/packages/core/test/agent/engine-steer.test.ts +++ b/packages/core/test/agent/engine-steer.test.ts @@ -23,6 +23,7 @@ import { Compactor } from "../../src/agent/compactor"; import type { McpManager } from "../../src/agent/mcp/manager"; import type { BashReviewer } from "../../src/agent/reviewer"; import type { PermissionRules } from "../../src/agent/permission-rules"; +import type { FunctionsExecRuntimeBridge, FunctionsExecRuntimeDeps } from "../../src/functions-exec/runtime"; import { writeDiff, type DiffHeader } from "../../src/diffs/store"; // Mirrors packages/core/test/agent/engine.test.ts's setup(). Exported so other engine test @@ -40,7 +41,7 @@ export function setupEngine(provider: Provider, opts?: { // fileDiff tests register an arbitrary fake tool name — gate.ts's evaluate() fails an // unclassified name closed to "ask" under every OTHER policy, and "bypass" is the one verdict // that resolves "allow" with no card, so those tests can reach registry.execute() directly). - reviewer?: BashReviewer; reviewerEnabled?: boolean | (() => boolean | undefined); reviewerAllow?: string[]; policy?: "ask" | "auto" | "plan" | "dont-ask" | "bypass"; + reviewer?: BashReviewer; reviewerEnabled?: boolean | (() => boolean | undefined); reviewerAllow?: string[]; policy?: "ask" | "auto" | "plan" | "dont-ask" | "accept-edits" | "bypass"; // phase 5e T3: per-class review on/off — undefined (every pre-5e-T3 test) leaves every class // enabled, unchanged. See EngineConfig.reviewerClasses's own doc comment. Also accepts a getter // directly (hot-settings T2's engine-hot-config.test.ts passes `() => live.reviewer?.classes` @@ -85,6 +86,7 @@ export function setupEngine(provider: Provider, opts?: { // write/edit/notebook_edit resolve through this harness exactly as before this opt existed — // only a test that explicitly opts in (tools-file-diff.test.ts's engine-level test) sees it. diffHome?: string; + functionsExecRuntimeFactory?: (deps: Pick) => FunctionsExecRuntimeBridge; }) { const home = mkdtempSync(join(tmpdir(), "norma-engine-steer-")); const cwd = opts?.cwd ?? realpathSync(mkdtempSync(join(tmpdir(), "norma-engine-steer-cwd-"))); @@ -131,6 +133,7 @@ export function setupEngine(provider: Provider, opts?: { store, hub, registry, broker, gate: new PermissionGate(), provider: { provider, model: "gated-1", live: opts?.live }, + functionsExecRuntimeFactory: opts?.functionsExecRuntimeFactory, dirs, approvalTimeoutMs: 500, assembler, diff --git a/packages/core/test/agent/mode-toolset-census.test.ts b/packages/core/test/agent/mode-toolset-census.test.ts index 7df6a5cd..52c0dff4 100644 --- a/packages/core/test/agent/mode-toolset-census.test.ts +++ b/packages/core/test/agent/mode-toolset-census.test.ts @@ -7,6 +7,7 @@ import { startDaemon, type RunningDaemon } from "../../src/daemon"; import { FileSecretStore } from "../../src/auth/secret-store"; import { FakeProvider } from "../../src/agent/fake-provider"; import { PermissionGate, isGateClassified } from "../../src/agent/gate"; +import { functionsExecSandboxAvailable } from "../../src/functions-exec/sandbox"; /** * R-T3 whole-branch review, Important 1 (FIX 1). mode-toolset-equivalence.test.ts pins the EXACT @@ -94,7 +95,7 @@ describe("daemon tool census (R-T3 whole-branch review FIX 1): real registration // chat's schema carrying only the read verbs — is invisible to `namesForMode` (which reports // ELIGIBILITY, exactly as the dispatch block below records for deferral) and is pinned separately // in test/agent/tools/browser.test.ts. - test("code mode is offered EXACTLY the full daemon tool surface (36 tools)", async () => { + test("code mode is offered EXACTLY the full daemon tool surface for this host's sandbox capability", async () => { const d = await boot(); expect(d.registry).not.toBeNull(); const offered = [...d.registry!.namesForMode("code", { builtinDeferral: true })]; @@ -122,6 +123,7 @@ describe("daemon tool census (R-T3 whole-branch review FIX 1): real registration "list_mcp_resources", "read_mcp_resource", "Workflow", "browser", + ...(functionsExecSandboxAvailable() ? ["functions.exec", "functions.wait"] : []), ].sort(), ); }); diff --git a/packages/core/test/agent/tools/functions-exec.test.ts b/packages/core/test/agent/tools/functions-exec.test.ts new file mode 100644 index 00000000..65770dde --- /dev/null +++ b/packages/core/test/agent/tools/functions-exec.test.ts @@ -0,0 +1,36 @@ +import { describe, expect, test } from "bun:test"; +import { ToolRegistry } from "../../../src/agent/tools/registry"; +import { + FUNCTIONS_EXEC_TOOL, + FUNCTIONS_WAIT_TOOL, + functionsExecArgs, + registerFunctionsExecTools, +} from "../../../src/agent/tools/functions-exec"; +import { registerToolSearchTool } from "../../../src/agent/tools/toolsearch"; + +describe("functions.exec tool registration", () => { + test("is a deferred code-only catalog entry when Seatbelt support is present", () => { + const registry = new ToolRegistry(); + registerToolSearchTool(registry); + registerFunctionsExecTools(registry, true); + + expect(registry.specs(undefined, { builtinDeferral: true, mode: "code" }).map((spec) => spec.name)).toEqual(["ToolSearch"]); + expect(registry.specFor(FUNCTIONS_EXEC_TOOL, undefined, "code")?.name).toBe(FUNCTIONS_EXEC_TOOL); + expect(registry.specFor(FUNCTIONS_WAIT_TOOL, undefined, "code")?.name).toBe(FUNCTIONS_WAIT_TOOL); + expect(registry.namesForMode("dispatch", { builtinDeferral: true })).toEqual(new Set()); + expect(registry.namesForMode("chat", { builtinDeferral: true })).toEqual(new Set()); + }); + + test("does not advertise unsupported worker execution", () => { + const registry = new ToolRegistry(); + registerToolSearchTool(registry); + registerFunctionsExecTools(registry, false); + expect(registry.specFor(FUNCTIONS_EXEC_TOOL, undefined, "code")).toBeUndefined(); + expect(registry.specFor(FUNCTIONS_WAIT_TOOL, undefined, "code")).toBeUndefined(); + }); + + test("accepts a meaningful bounded source budget for raw patch calls", () => { + expect(functionsExecArgs.safeParse({ source: "x".repeat(2_048) }).success).toBe(true); + expect(functionsExecArgs.safeParse({ source: "x".repeat(8_193) }).success).toBe(false); + }); +}); diff --git a/packages/core/test/providers/codex-oauth.test.ts b/packages/core/test/providers/codex-oauth.test.ts index 031345bd..28a92c71 100644 --- a/packages/core/test/providers/codex-oauth.test.ts +++ b/packages/core/test/providers/codex-oauth.test.ts @@ -5,6 +5,7 @@ import { join } from "node:path"; import { FileSecretStore } from "../../src/auth/secret-store"; import { CodexAuthStore, CodexOAuthProvider } from "../../src/providers/codex-oauth"; import { CODEX_MODELS, DEFAULT_CODEX_MODEL } from "../../src/providers/codex-config"; +import { wireToolName } from "../../src/providers/openai-compatible"; let server: ReturnType | null = null; afterEach(() => { server?.stop(true); server = null; }); @@ -98,6 +99,33 @@ describe("CodexOAuthProvider", () => { for await (const e of p.streamTurn({ model: "m", input: [] })) events.push(e); expect(events).toEqual([{ type: "error", code: "auth", message: expect.stringContaining("norma login") }]); }); + + test("uses valid Responses names on the wire and restores dotted internal tool names", async () => { + const internalName = "functions.exec"; + const wireName = wireToolName(internalName); + let body: any; + server = Bun.serve({ + port: 0, + async fetch(req) { + body = await req.json(); + return new Response( + `data: ${JSON.stringify({ type: "response.output_item.done", item: { type: "function_call", call_id: "call-1", name: wireName, arguments: "{}" } })}\n\n` + + `data: ${JSON.stringify({ type: "response.completed", response: {} })}\n\n`, + { headers: { "content-type": "text/event-stream" } }, + ); + }, + }); + const p = new CodexOAuthProvider({ authStore: await seeded(), backendUrl: `http://localhost:${server.port}` }); + const events = []; + for await (const event of p.streamTurn({ + model: "gpt-5.6-sol", + input: [{ type: "message", role: "user", content: "hi" }], + tools: [{ name: internalName, description: "run a cell", parameters: { type: "object" } }], + })) events.push(event); + + expect(body.tools[0].name).toBe(wireName); + expect(events).toContainEqual({ type: "tool_call", callId: "call-1", name: internalName, argsJson: "{}" }); + }); }); describe("use_responses_lite — never-send pin", () => { diff --git a/packages/core/test/providers/openai-compatible.test.ts b/packages/core/test/providers/openai-compatible.test.ts index b6c1d322..5119443e 100644 --- a/packages/core/test/providers/openai-compatible.test.ts +++ b/packages/core/test/providers/openai-compatible.test.ts @@ -1,7 +1,7 @@ import { afterEach, describe, expect, test } from "bun:test"; import { readFileSync } from "node:fs"; import { join } from "node:path"; -import { OpenAICompatibleProvider, buildRequestBody, mapInput } from "../../src/providers/openai-compatible"; +import { OpenAICompatibleProvider, buildRequestBody, mapInput, wireToolName } from "../../src/providers/openai-compatible"; let server: ReturnType | null = null; afterEach(() => { server?.stop(true); server = null; }); @@ -91,6 +91,28 @@ describe("OpenAICompatibleProvider", () => { expect(body.input[2]).toEqual({ type: "function_call_output", call_id: "call_1", output: "a.txt" }); }); + test("encodes dotted tools on the Responses wire and restores them on the inbound stream", async () => { + let body: any; + server = Bun.serve({ + port: 0, + async fetch(req) { + body = await req.json(); + const wire = wireToolName("functions.exec"); + const sse = `data: {"type":"response.output_item.done","item":{"type":"function_call","call_id":"call_fx","name":"${wire}","arguments":"{\\"source\\":\\"tools.text('x')\\"}"}}\n\ndata: {"type":"response.completed","response":{"usage":{"input_tokens":1,"output_tokens":1}}}\n\n`; + return new Response(sse, { headers: { "content-type": "text/event-stream" } }); + }, + }); + const p = new OpenAICompatibleProvider({ baseUrl: `http://localhost:${server.port}`, apiKey: "sk" }); + const events = await collect(p.streamTurn({ + model: "m", + input: [{ type: "function_call", callId: "old", name: "functions.exec", argsJson: "{}" }], + tools: [{ name: "functions.exec", description: "run", parameters: { type: "object" } }], + })); + expect(body.tools[0].name).toBe(wireToolName("functions.exec")); + expect(body.input[0].name).toBe(wireToolName("functions.exec")); + expect(events[0]).toEqual({ type: "tool_call", callId: "call_fx", name: "functions.exec", argsJson: '{"source":"tools.text(\'x\')"}' }); + }); + test("consumer break mid-stream cancels the reader (no connection leak)", async () => { // Bun's HTTP server does not propagate client-side reader.cancel() to the server-side // ReadableStream.cancel() callback within a short window, so we spy on the client reader @@ -260,3 +282,11 @@ describe("mapInput image (Phase 5 CU)", () => { expect(typeof out[0].content[0].image_url).toBe("string"); }); }); + +describe("mapInput audio", () => { + test("maps transient WAV audio into the Responses input_audio part", () => { + expect(mapInput([{ type: "audio", dataUrl: "data:audio/wav;base64,AAAA" }])).toEqual([ + { type: "message", role: "user", content: [{ type: "input_audio", input_audio: { data: "AAAA", format: "wav" } }] }, + ]); + }); +}); diff --git a/packages/core/test/providers/responses-sse.test.ts b/packages/core/test/providers/responses-sse.test.ts index 467ba854..773aa4d5 100644 --- a/packages/core/test/providers/responses-sse.test.ts +++ b/packages/core/test/providers/responses-sse.test.ts @@ -31,6 +31,14 @@ describe("ResponsesSseParser", () => { ]); }); + test("restores an encoded Responses tool name", () => { + const parser = new ResponsesSseParser((name) => name === "norma_ZnVuY3Rpb25zLmV4ZWM" ? "functions.exec" : name); + const chunk = new TextEncoder().encode( + 'data: {"type":"response.output_item.done","item":{"type":"function_call","call_id":"call_fx","name":"norma_ZnVuY3Rpb25zLmV4ZWM","arguments":"{}"}}\n\n', + ); + expect(parser.push(chunk)).toEqual([{ type: "tool_call", callId: "call_fx", name: "functions.exec", argsJson: "{}" }]); + }); + test("unknown event types are ignored (forward compat)", () => { const p = new ResponsesSseParser(); const chunk = new TextEncoder().encode('event: response.shiny.new\ndata: {"type":"response.shiny.new"}\n\n');