diff --git a/.changeset/native-prepared-request.md b/.changeset/native-prepared-request.md new file mode 100644 index 000000000..619930096 --- /dev/null +++ b/.changeset/native-prepared-request.md @@ -0,0 +1,5 @@ +--- +'@openai/agents-codex': minor +--- + +feat: Add per-operation prepared-request editing for native HTTP root conversations while preserving original tool and history authority. diff --git a/packages/agents-codex/examples/prepared-request.ts b/packages/agents-codex/examples/prepared-request.ts new file mode 100644 index 000000000..a22aac109 --- /dev/null +++ b/packages/agents-codex/examples/prepared-request.ts @@ -0,0 +1,118 @@ +import assert from 'node:assert/strict'; +import { z } from 'zod'; +import { Agent, Runner, ToolOutput, tool } from '@openai/agents-codex'; +import { + text, + toolCall, + withScriptedResponses, +} from '@openai/agents-codex/testing'; + +/** + * Exercise the public request, argument and result controls without live credentials. + * The scripted provider uses explicit HTTP and a standard Responses (non-Lite) model. + * Select eager ordinary tools for each sampling step of a direct root. + * beforeStep may narrow a mixed opening catalog. Deferred sampled tools fail before the editor. + * Environment, code mode, children and automatic compaction are outside this profile. + * Pre-turn or mid-turn automatic compaction fails before any hidden model request. + * Shorten the input or start a new operation without the callback. Earlier tool effects remain. + * Post-turn compaction is refused while preserving the completed answer. + * Supply new input to Agent.run when using prepareRequest. A resumed turn cannot add an editor. + * Sampling edits do not change canonical history, executable schemas or policies. + * For structured output, use native mode or prompted mode with maxRetries: 0. + */ +export async function runPreparedRequestExample(): Promise { + const effects: string[] = []; + const lookup = tool({ + name: 'lookup', + description: 'Look up a synthetic record.', + parameters: z.object({ key: z.enum(['allowed']), label: z.string() }), + before: ({ callId, originalArguments }) => { + assert.equal(callId, 'lookup'); + assert.deepEqual(originalArguments, { + key: 'allowed', + label: 'original', + }); + return { + action: 'replaceArguments', + arguments: { key: 'allowed' as const, label: 'corrected' }, + }; + }, + execute: ({ key, label }) => { + effects.push(label); + return `Raw record ${key}`; + }, + after: (_output, { callId, effectiveArguments }) => { + assert.equal(callId, 'lookup'); + assert.equal(effectiveArguments.label, 'corrected'); + return new ToolOutput([{ type: 'inputText', text: 'SHAPED_RESULT' }]); + }, + }); + return withScriptedResponses( + [ + [ + toolCall( + 'lookup', + { key: 'allowed', label: 'original' }, + { callId: 'lookup' }, + ), + ], + [text('The record is allowed.')], + ], + async (peer) => { + const result = await Runner.run( + new Agent({ + model: 'gpt-5.5', + instructions: 'Call lookup, then explain its result.', + provider: peer.provider, + tools: [lookup], + }), + 'Look up the record.', + { + prepareRequest({ request, signal }) { + signal.throwIfAborted(); + // Recovery can advance the index without a result. Inspect native input. + if ( + !request.input.some( + (item) => + item !== null && + typeof item === 'object' && + !Array.isArray(item) && + item.type === 'function_call_output' && + item.call_id === 'lookup', + ) + ) + return; + request.instructions = 'Explain the lookup result in one sentence.'; + request.input.push({ + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'SAMPLING_ONLY_NOTE' }], + }); + request.tools[0].description = 'Sampled record description.'; + request.tools[0].parameters.properties = { + key: { type: 'string', enum: ['allowed', 'sampled-only'] }, + label: { type: 'string' }, + }; + return request; + }, + }, + ); + assert.deepEqual(effects, ['corrected']); + assert.equal(peer.toolOutput('lookup'), 'SHAPED_RESULT'); + assert.equal(peer.requests.length, 2); + assert.match( + JSON.stringify(peer.requests[1].input), + /SAMPLING_ONLY_NOTE/, + ); + assert.match( + JSON.stringify(peer.requests[1].tools), + /Sampled record description/, + ); + assert.match( + JSON.stringify(peer.requests[1].input), + /Explain the lookup result in one sentence/, + ); + return result.finalOutput; + }, + ); +} diff --git a/packages/agents-codex/native/Cargo.lock b/packages/agents-codex/native/Cargo.lock index 8259d48e1..c976b31fc 100644 --- a/packages/agents-codex/native/Cargo.lock +++ b/packages/agents-codex/native/Cargo.lock @@ -1698,8 +1698,8 @@ dependencies = [ "serde", "serde_ignored", "serde_json", - "serde_yaml", "serde_path_to_error", + "serde_yaml", "sha2 0.10.9", "strum", "strum_macros 0.28.0", @@ -2412,6 +2412,7 @@ dependencies = [ "codex-utils-absolute-path", "codex-utils-path-uri", "futures", + "serde", "serde_json", ] diff --git a/packages/agents-codex/native/index.d.ts b/packages/agents-codex/native/index.d.ts index 358ecb696..267969402 100644 --- a/packages/agents-codex/native/index.d.ts +++ b/packages/agents-codex/native/index.d.ts @@ -165,6 +165,7 @@ export class NativeThread { traceJson?: string | null, tracing_context?: string, ): Promise; + _reply_prepared_request(responseJson: string): void; _reply_step(response_json: string): void; _check_step(response_json: string): boolean; _reply_child_step(response_json: string): boolean; diff --git a/packages/agents-codex/native/patches/0054-native-prepared-request.patch b/packages/agents-codex/native/patches/0054-native-prepared-request.patch new file mode 100644 index 000000000..91700b6b4 --- /dev/null +++ b/packages/agents-codex/native/patches/0054-native-prepared-request.patch @@ -0,0 +1,612 @@ +diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock +index 295d1f826b19efa6e781761ca0c3d8bba6dd0d8a..61d3e12d4c59edaed9778a9bcc2cc4c65bf7e4e2 100644 +--- a/codex-rs/Cargo.lock ++++ b/codex-rs/Cargo.lock +@@ -3360,6 +3360,7 @@ dependencies = [ + "codex-utils-path-uri", + "futures", + "pretty_assertions", ++ "serde", + "serde_json", + "tokio", + ] +diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs +index 3d02d17fa16a24eee88d36407542b3ae97ab4f84..b5b634ea44ac24009e6f29a665bd606fc1bb6bb6 100644 +--- a/codex-rs/core/src/client.rs ++++ b/codex-rs/core/src/client.rs +@@ -1787,6 +1787,9 @@ impl ModelClientSession { + ) + .with_telemetry(Some(request_telemetry), Some(sse_telemetry)); + crate::model_request::wait_if_sampling_stopped(prompt.sampling_stop.as_deref()).await; ++ if let Some(admission) = &prompt.prepared_request_admission { ++ admission.admit_request().map_err(CodexErr::Fatal)?; ++ } + let stream_result = client.stream_request(request, options).await; + + match stream_result { +diff --git a/codex-rs/core/src/client_common.rs b/codex-rs/core/src/client_common.rs +index 6188306e2ae26a96b758f2359697dc4ad4654b36..764b7b82562a831a482ce8c76187d4e1bc46e288 100644 +--- a/codex-rs/core/src/client_common.rs ++++ b/codex-rs/core/src/client_common.rs +@@ -34,6 +34,10 @@ pub struct Prompt { + /// Stops new provider requests for this turn, including prompts without tools. + pub(crate) sampling_stop: Option>, + ++ /// Request-scoped host ownership, checked independently of response policies. ++ pub(crate) prepared_request_admission: ++ Option>, ++ + /// Whether parallel tool calls are permitted for this prompt. + pub(crate) parallel_tool_calls: bool, + +@@ -55,6 +59,7 @@ impl Default for Prompt { + tools: Arc::default(), + tool_admission: None, + sampling_stop: None, ++ prepared_request_admission: None, + parallel_tool_calls: false, + base_instructions: BaseInstructions::default(), + output_schema: None, +diff --git a/codex-rs/core/src/compact_remote_v2_attempt.rs b/codex-rs/core/src/compact_remote_v2_attempt.rs +index c302aac22a6b055739501c55033aec814eb7a248..6843656823fe6a779c6d12ad191728510d1e1e98 100644 +--- a/codex-rs/core/src/compact_remote_v2_attempt.rs ++++ b/codex-rs/core/src/compact_remote_v2_attempt.rs +@@ -98,6 +98,7 @@ pub(super) async fn run_remote_compact_v2_attempt( + Arc::clone(tool_router), + )), + sampling_stop: crate::model_request::turn_sampling_stop(&turn_context.extension_data), ++ prepared_request_admission: None, + parallel_tool_calls: true, + base_instructions, + output_schema: None, +diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs +index 258b670961950f71e3a8dc29edb8a961158bdb6d..2f8c5526d90c21d44f61df78069a532209005ac6 100644 +--- a/codex-rs/core/src/lib.rs ++++ b/codex-rs/core/src/lib.rs +@@ -11,6 +11,7 @@ mod apps; + mod client; + mod client_common; + mod model_request; ++mod prepared_request; + mod realtime_context; + mod realtime_conversation; + mod realtime_history; +diff --git a/codex-rs/core/src/prepared_request.rs b/codex-rs/core/src/prepared_request.rs +new file mode 100644 +index 0000000000000000000000000000000000000000..a70e5faf6d3b77f8fa6d90f9c92276aed1cb8ec0 +--- /dev/null ++++ b/codex-rs/core/src/prepared_request.rs +@@ -0,0 +1,289 @@ ++//! Sampling-copy editing, separate from the selected execution plan and canonical history. ++use crate::client_common::Prompt; ++use codex_extension_api::{PreparedRequestData, PreparedToolPresentation}; ++use codex_protocol::models::ResponseItem; ++use codex_tools::{JsonSchema, ResponsesApiNamespaceTool, ResponsesApiTool, ToolSpec}; ++use serde::Serialize; ++use serde::de::DeserializeOwned; ++use serde_json::Value; ++ ++const PROFILE: &str = "prepare_request requires a standard Responses (non-Lite) model, Responses HTTP (supports_websockets=false), and direct-root eager function/MCP tools"; ++ ++/// Preserve submitted keys, including null values, while allowing native defaults. ++fn preserves(submitted: &Value, native: &Value) -> bool { ++ match (submitted, native) { ++ (Value::Object(a), Value::Object(b)) => a ++ .iter() ++ .all(|(key, value)| b.get(key).is_some_and(|actual| preserves(value, actual))), ++ (Value::Array(a), Value::Array(b)) => { ++ a.len() == b.len() && a.iter().zip(b).all(|(a, b)| preserves(a, b)) ++ } ++ _ => submitted == native, ++ } ++} ++ ++fn parse_preserving(value: Value) -> Result { ++ let parsed: T = serde_json::from_value(value.clone()) ++ .map_err(|_| "invalid native prepared-request value".to_string())?; ++ let canonical = serde_json::to_value(&parsed) ++ .map_err(|_| "cannot serialize prepared-request value".to_string())?; ++ if !preserves(&value, &canonical) { ++ return Err("prepared-request value loses fields in native serialization; omit skipped optional fields instead of explicit null".into()); ++ } ++ Ok(parsed) ++} ++ ++fn presentation( ++ namespace: Option<&str>, ++ tool: &ResponsesApiTool, ++) -> Result { ++ if tool.defer_loading == Some(true) { ++ return Err(PROFILE.into()); ++ } ++ Ok(PreparedToolPresentation { ++ namespace: namespace.map(str::to_owned), ++ name: tool.name.clone(), ++ kind: "function".into(), ++ strict: tool.strict, ++ defer_loading: tool.defer_loading, ++ description: tool.description.clone(), ++ parameters: serde_json::to_value(&tool.parameters) ++ .map_err(|_| "cannot serialize native schema".to_string())?, ++ }) ++} ++ ++pub(crate) fn snapshot(prompt: &Prompt) -> Result { ++ let mut tools = Vec::new(); ++ for spec in prompt.tools.iter() { ++ match spec { ++ ToolSpec::Function(tool) => tools.push(presentation(None, tool)?), ++ ToolSpec::Namespace(namespace) => { ++ for leaf in &namespace.tools { ++ match leaf { ++ ResponsesApiNamespaceTool::Function(tool) => { ++ tools.push(presentation(Some(&namespace.name), tool)?) ++ } ++ _ => return Err(PROFILE.into()), ++ } ++ } ++ } ++ _ => return Err(PROFILE.into()), ++ } ++ } ++ Ok(PreparedRequestData { ++ instructions: prompt.base_instructions.text.clone(), ++ input: prompt ++ .input ++ .iter() ++ .map(serde_json::to_value) ++ .collect::>() ++ .map_err(|_| "cannot serialize native input".to_string())?, ++ tools, ++ }) ++} ++ ++fn edit_tool( ++ namespace: Option<&str>, ++ tool: &mut ResponsesApiTool, ++ edit: &PreparedToolPresentation, ++) -> Result<(), String> { ++ let original = presentation(namespace, tool)?; ++ if edit.namespace != original.namespace ++ || edit.name != original.name ++ || edit.kind != original.kind ++ || edit.strict != original.strict ++ || edit.defer_loading != original.defer_loading ++ { ++ return Err( ++ "prepared-request tool identity, order, kind, strict and defer_loading are immutable" ++ .into(), ++ ); ++ } ++ if !edit.parameters.is_object() { ++ return Err("prepared-request parameters must be a supported schema object".into()); ++ } ++ if edit.parameters != original.parameters { ++ let mut schema: JsonSchema = parse_preserving(edit.parameters.clone())?; ++ // Retain native-only MCP bookkeeping; the executable owner remains untouched. ++ schema.mcp_input_schema_max_bytes = tool.parameters.mcp_input_schema_max_bytes; ++ tool.parameters = schema; ++ } ++ tool.description = edit.description.clone(); ++ Ok(()) ++} ++ ++pub(crate) fn apply(original: &Prompt, data: PreparedRequestData) -> Result { ++ let before = snapshot(original)?; ++ if before.tools.len() != data.tools.len() { ++ return Err("prepared-request tool count is immutable".into()); ++ } ++ let input: Vec = data ++ .input ++ .into_iter() ++ .enumerate() ++ .map(|(index, value)| { ++ parse_preserving(value) ++ .map_err(|category| format!("prepared-request input[{index}]: {category}")) ++ }) ++ .collect::>()?; ++ let mut tools = original.tools.to_vec(); ++ let mut edits = data.tools.iter(); ++ for spec in &mut tools { ++ match spec { ++ ToolSpec::Function(tool) => { ++ edit_tool(None, tool, edits.next().ok_or("missing tool presentation")?)? ++ } ++ ToolSpec::Namespace(namespace) => { ++ for leaf in &mut namespace.tools { ++ match leaf { ++ ResponsesApiNamespaceTool::Function(tool) => edit_tool( ++ Some(&namespace.name), ++ tool, ++ edits.next().ok_or("missing tool presentation")?, ++ )?, ++ _ => return Err(PROFILE.into()), ++ } ++ } ++ } ++ _ => return Err(PROFILE.into()), ++ } ++ } ++ let mut result = original.clone(); ++ result.input = input; ++ result.base_instructions.text = data.instructions; ++ result.tools = tools.into(); ++ Ok(result) ++} ++ ++#[cfg(test)] ++mod tests { ++ use super::*; ++ use serde_json::json; ++ ++ #[test] ++ fn native_text_function_and_mcp_items_preserve_submitted_fields() { ++ for item in [ ++ json!({"type":"message","role":"user","content":[{"type":"input_text","text":"note"}]}), ++ json!({"type":"function_call","name":"reserve","arguments":"{\"seats\":2}","call_id":"call-1"}), ++ json!({"type":"function_call_output","call_id":"call-1","output":"SHAPED_RESULT"}), ++ json!({"type":"function_call","namespace":"mcp__local","name":"reserve","arguments":"{}","call_id":"mcp-1"}), ++ json!({"type":"function_call_output","call_id":"mcp-1","output":"ok"}), ++ ] { ++ assert!(parse_preserving::(item).is_ok()); ++ } ++ } ++ ++ #[test] ++ fn unknown_nested_null_and_skipped_optional_null_are_rejected() { ++ for item in [ ++ json!({"type":"message","role":"user","content":[{"type":"input_text","text":"note","unknown":null}]}), ++ json!({"type":"message","role":"user","id":null,"content":[]}), ++ ] { ++ assert!(parse_preserving::(item).is_err()); ++ } ++ } ++ ++ #[test] ++ fn native_schema_rejects_discarded_keywords_without_sanitizing() { ++ assert!(parse_preserving::(json!({"type":"integer","maximum":99})).is_err()); ++ assert!( ++ parse_preserving::(json!({"type":"string","enum":["one","two"]})).is_ok() ++ ); ++ } ++ ++ fn function(name: &str) -> ResponsesApiTool { ++ ResponsesApiTool { ++ name: name.into(), ++ description: "original".into(), ++ strict: false, ++ defer_loading: None, ++ parameters: serde_json::from_value(json!({"type":"string","enum":["one"]})).unwrap(), ++ output_schema: None, ++ } ++ } ++ ++ #[test] ++ fn edit_changes_only_detached_sampling_presentation_and_input() { ++ let original = Prompt { ++ tools: vec![ ++ ToolSpec::Function(function("local")), ++ ToolSpec::Namespace(codex_tools::ResponsesApiNamespace { ++ name: "mcp__local".into(), ++ description: "container".into(), ++ tools: vec![ResponsesApiNamespaceTool::Function(function("remote"))], ++ }), ++ ] ++ .into(), ++ ..Default::default() ++ }; ++ let original_tools = original.tools.clone(); ++ let mut data = snapshot(&original).unwrap(); ++ data.instructions = "edited".into(); ++ data.input.push(json!({"type":"message","role":"user","content":[{"type":"input_text","text":"sampling only"}]})); ++ for tool in &mut data.tools { ++ tool.description = "edited description".into(); ++ tool.parameters = json!({"type":"string","enum":["one","two"]}); ++ } ++ let edited = apply(&original, data).unwrap(); ++ assert_eq!( ++ snapshot(&edited).unwrap().tools[1].namespace.as_deref(), ++ Some("mcp__local") ++ ); ++ assert_eq!(edited.base_instructions.text, "edited"); ++ assert_eq!(edited.input.len(), 1); ++ assert!(original.input.is_empty()); ++ assert_eq!(original.tools, original_tools); ++ assert_eq!( ++ snapshot(&original).unwrap().tools[0].description, ++ "original" ++ ); ++ match &edited.tools[1] { ++ ToolSpec::Namespace(namespace) => assert_eq!(namespace.description, "container"), ++ _ => panic!("namespace was lost"), ++ } ++ } ++ ++ #[test] ++ fn tool_mutations_and_invalid_schemas_apply_nothing() { ++ let original = Prompt { ++ tools: vec![ToolSpec::Function(function("local"))].into(), ++ ..Default::default() ++ }; ++ let before = snapshot(&original).unwrap(); ++ for mutation in 0..6 { ++ let mut data = before.clone(); ++ match mutation { ++ 0 => data.tools[0].name = "invented".into(), ++ 1 => data.tools[0].strict = true, ++ 2 => data.tools[0].defer_loading = Some(true), ++ 3 => data.tools.clear(), ++ 4 => data.tools[0].parameters = json!({"maximum":99}), ++ _ => data.tools[0].namespace = Some("other".into()), ++ } ++ assert!(apply(&original, data).is_err()); ++ assert_eq!(snapshot(&original).unwrap(), before); ++ } ++ } ++ ++ #[test] ++ fn unsupported_selected_presentations_fail_before_callback_entry() { ++ let mut tool = function("deferred"); ++ tool.defer_loading = Some(true); ++ let original = Prompt { ++ tools: vec![ToolSpec::Function(tool)].into(), ++ ..Default::default() ++ }; ++ assert!(snapshot(&original).is_err()); ++ } ++ ++ #[test] ++ fn defaults_may_be_added_but_array_shape_and_scalar_values_must_stay() { ++ assert!(preserves( ++ &json!({"a":[1]}), ++ &json!({"a":[1],"default":null}) ++ )); ++ assert!(!preserves(&json!({"a":null}), &json!({}))); ++ assert!(!preserves(&json!([1]), &json!([1, 2]))); ++ assert!(!preserves(&json!(1), &json!(2))); ++ } ++} +diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs +index 030a65b21d390af8a7f90bdd674039d09f370316..9c43cb1685b0f4ea8a6a300282a8d54f7a2aafd5 100644 +--- a/codex-rs/core/src/session/turn.rs ++++ b/codex-rs/core/src/session/turn.rs +@@ -1,3 +1,4 @@ ++use std::borrow::Cow; + use std::collections::HashMap; + use std::collections::HashSet; + use std::sync::Arc; +@@ -1465,6 +1466,20 @@ async fn run_auto_compact( + phase: CompactionPhase, + ) -> CodexResult<()> { + let turn_context = &step_context.turn; ++ if sess ++ .services ++ .extensions ++ .model_request_contributors() ++ .iter() ++ .any(|contributor| contributor.prepares_request(&turn_context.extension_data)) ++ { ++ if matches!(phase, CompactionPhase::PostTurn) { ++ return Ok(()); ++ } ++ let err = CodexErr::Fatal("automatic compaction is unsupported for an operation with prepare_request; shorten the input or start a new operation".into()); ++ return Err(err); ++ } ++ + if matches!(phase, CompactionPhase::PreTurn) + && crate::guardian::is_basic_session_source(&turn_context.session_source) + && !crate::guardian::should_compact_guardian_input(sess)? +@@ -1608,6 +1623,7 @@ pub(crate) fn build_prompt( + parallel_tool_calls: true, + tool_admission: None, + sampling_stop: None, ++ prepared_request_admission: None, + base_instructions, + output_schema: turn_context.final_output_json_schema.clone(), + output_schema_strict: !crate::guardian::is_basic_session_source( +@@ -1659,6 +1675,7 @@ async fn run_sampling_request( + ); + let max_retries = turn_context.provider.info().stream_max_retries(); + let mut retry_state = ResponsesStreamRetryState::default(); ++ let mut recovery_attempt = 0; + let mut initial_input = Some(input); + let mut original_input = None; + let mut executed_tool_calls_by_output = HashMap::new(); +@@ -1691,6 +1708,27 @@ async fn run_sampling_request( + let responses_metadata = sess + .responses_metadata(step_context.as_ref(), CodexResponsesRequestKind::Turn) + .await; ++ let editors: Vec<_> = sess ++ .services ++ .extensions ++ .model_request_contributors() ++ .iter() ++ .filter(|contributor| contributor.prepares_request(&turn_context.extension_data)) ++ .collect(); ++ if !editors.is_empty() { ++ if editors.len() != 1 ++ || turn_context.provider.info().supports_websockets ++ || step_context.settings.model_info.use_responses_lite ++ || step_context.realtime.active ++ || turn_context.config.features.enabled(Feature::CodeMode) ++ || matches!( ++ turn_context.session_source, ++ SessionSource::SubAgent(_) | SessionSource::Internal(_) ++ ) ++ { ++ return Err(CodexErr::Fatal("prepare_request requires a standard Responses (non-Lite) model, Responses HTTP (supports_websockets=false), and direct-root eager function/MCP tools".into())); ++ } ++ } + if crate::guardian::is_basic_session_source(&turn_context.session_source) { + crate::guardian::prepare_guardian_prompt( + &sess, +@@ -1700,6 +1738,34 @@ async fn run_sampling_request( + ) + .await?; + } ++ // The original prompt remains authoritative for history and retry bookkeeping. ++ let mut sampling_prompt = Cow::Borrowed(&prompt); ++ if !editors.is_empty() { ++ let data = crate::prepared_request::snapshot(&prompt).map_err(CodexErr::Fatal)?; ++ let admission = prompt.tool_admission.clone().ok_or_else(|| { ++ CodexErr::Fatal("prepared request has no selected tool admission".into()) ++ })?; ++ let output = editors[0] ++ .prepare_request(codex_extension_api::PreparedRequestInput { ++ admission, ++ recovery_attempt, ++ data, ++ }) ++ .or_cancel(&preempt) ++ .or_cancel(&cancellation_token) ++ .await?? ++ .map_err(CodexErr::Fatal)?; ++ if cancellation_token.is_cancelled() || preempt.is_cancelled() { ++ return Err(CodexErr::TurnAborted); ++ } ++ output.admission.admit_request().map_err(CodexErr::Fatal)?; ++ if let Some(data) = output.data { ++ sampling_prompt = Cow::Owned( ++ crate::prepared_request::apply(&prompt, data).map_err(CodexErr::Fatal)?, ++ ); ++ } ++ sampling_prompt.to_mut().prepared_request_admission = Some(output.admission); ++ } + let err = match try_run_sampling_request( + tool_runtime.clone(), + Arc::clone(&sess), +@@ -1708,7 +1774,7 @@ async fn run_sampling_request( + client_session, + &responses_metadata, + Arc::clone(&turn_diff_tracker), +- &prompt, ++ &sampling_prompt, + cancellation_token.child_token(), + ) + .await +@@ -1765,6 +1831,7 @@ async fn run_sampling_request( + )); + } + retry??; ++ recovery_attempt += 1; + turn_context.turn_timing_state.record_sampling_retry(); + } + } +diff --git a/codex-rs/ext/extension-api/Cargo.toml b/codex-rs/ext/extension-api/Cargo.toml +index 4b487226eb7deb0a44ad5d9ae16e53f90fd7ec8e..3e19d0358c23532df40aca4dff37125b201c938c 100644 +--- a/codex-rs/ext/extension-api/Cargo.toml ++++ b/codex-rs/ext/extension-api/Cargo.toml +@@ -27,6 +27,7 @@ codex-protocol = { workspace = true } + codex-tools = { workspace = true } + codex-utils-absolute-path = { workspace = true } + codex-utils-path-uri = { workspace = true } ++serde = { workspace = true } + serde_json = { workspace = true } + + [dev-dependencies] +diff --git a/codex-rs/ext/extension-api/src/lib.rs b/codex-rs/ext/extension-api/src/lib.rs +index f2dbed8d89823999a82e89e43576ecf7efab7a5f..54d70d95f627509f85c177d38173c8568ebd5666 100644 +--- a/codex-rs/ext/extension-api/src/lib.rs ++++ b/codex-rs/ext/extension-api/src/lib.rs +@@ -126,6 +126,12 @@ pub use model_request::ModelToolAdmission; + pub use model_request::ResponseEvent; + pub use model_request::TurnSamplingStop; + ++pub use model_request::PreparedRequestAdmission; ++pub use model_request::PreparedRequestData; ++pub use model_request::PreparedRequestInput; ++pub use model_request::PreparedRequestOutput; ++pub use model_request::PreparedToolPresentation; ++ + /// SDK host context for recording the settings used to prepare a sampling request. + pub struct SdkTurnReceiptSeed { + pub codex_revision: &'static str, +diff --git a/codex-rs/ext/extension-api/src/model_request.rs b/codex-rs/ext/extension-api/src/model_request.rs +index 5263f85cd4517aa8b72186b02ac91d9c08d0b2e2..34d02776f0119a72955a827c3a5cf48c3bfda7e5 100644 +--- a/codex-rs/ext/extension-api/src/model_request.rs ++++ b/codex-rs/ext/extension-api/src/model_request.rs +@@ -3,7 +3,10 @@ use crate::ExtensionData; + pub use codex_api::ApiError as ModelResponseError; + pub use codex_api::ResponseEvent; + use codex_protocol::models::ResponseItem; ++use futures::future::BoxFuture; + use futures::stream::BoxStream; ++use serde::{Deserialize, Serialize}; ++use serde_json::Value; + use std::collections::HashMap; + use std::sync::Arc; + use std::sync::atomic::AtomicBool; +@@ -78,9 +81,25 @@ pub struct ModelRequestInput<'a> { + pub model: &'a str, + } + +-/// Creates request-scoped interceptors without delaying inference. ++/// Contributes request-scoped preparation and response interception. + /// Return None for requests this host does not manage. + pub trait ModelRequestContributor: Send + Sync + std::fmt::Debug { ++ /// Whether this exact operation opted into prepared-request editing. ++ /// Keep this captured marker true through retirement; it is not an owner-liveness check. ++ /// Hosts must reject unsupported operation settings before starting work. ++ fn prepares_request(&self, _turn_store: &ExtensionData) -> bool { ++ false ++ } ++ ++ /// Prepare an owned sampling copy. Dropping the future revokes its pending reply; ++ /// the host still owns cancellation and joining of application callbacks. ++ fn prepare_request( ++ &self, ++ _input: PreparedRequestInput, ++ ) -> BoxFuture<'_, Result> { ++ Box::pin(async { Err("prepared-request contributor is not implemented".into()) }) ++ } ++ + fn request(&self, input: ModelRequestInput<'_>) -> Option>; + } + +@@ -114,6 +133,46 @@ pub trait ModelResponseInterceptor: Send + Sync { + fn intercept(self: Box, stream: ModelResponseStream) -> ModelResponseStream; + } + ++/// Detached native sampling data. Unknown fields are rejected at the host reply boundary. ++#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] ++#[serde(deny_unknown_fields)] ++pub struct PreparedRequestData { ++ pub instructions: String, ++ pub input: Vec, ++ pub tools: Vec, ++} ++ ++/// Ordered function leaf. Only description and parameters may change. ++#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] ++#[serde(deny_unknown_fields)] ++pub struct PreparedToolPresentation { ++ pub namespace: Option, ++ pub name: String, ++ pub kind: String, ++ pub strict: bool, ++ pub defer_loading: Option, ++ pub description: String, ++ pub parameters: Value, ++} ++ ++/// Native ownership evidence; hosts add their generation and one-shot nonce. ++pub struct PreparedRequestInput { ++ pub admission: Arc, ++ /// Outer sampling recovery count; lower-level transport retries reuse this preparation. ++ pub recovery_attempt: usize, ++ pub data: PreparedRequestData, ++} ++ ++/// A settled reply retains an owner check even when it makes no edit. ++pub struct PreparedRequestOutput { ++ pub data: Option, ++ pub admission: Arc, ++} ++ ++pub trait PreparedRequestAdmission: Send + Sync + std::fmt::Debug { ++ fn admit_request(&self) -> Result<(), String>; ++} ++ + #[cfg(test)] + mod sampling_stop_tests { + use super::TurnSamplingStop; diff --git a/packages/agents-codex/native/patches/series.json b/packages/agents-codex/native/patches/series.json index 3f62a4307..a4cb02204 100644 --- a/packages/agents-codex/native/patches/series.json +++ b/packages/agents-codex/native/patches/series.json @@ -263,7 +263,11 @@ { "path": "0059-terminal-authentication-retry.patch", "sha256": "2064b337e86352296df7ac5bde428a11ad145ca568478f977dcfb2ef3a87cff8" + }, + { + "path": "0054-native-prepared-request.patch", + "sha256": "616c979bd3accbfdfa978713e2a82f0049645cd3ae9c72ffacff1abcd2c517c2" } ], - "result_tree": "77c98d47f44a9f13525a2c7622e7682e3a607d64" + "result_tree": "1c65094c37a0a4099f035256a8bd0cebb66d6bab" } diff --git a/packages/agents-codex/native/rust/checkpoint.rs b/packages/agents-codex/native/rust/checkpoint.rs index 92ff538e9..4e24cc834 100644 --- a/packages/agents-codex/native/rust/checkpoint.rs +++ b/packages/agents-codex/native/rust/checkpoint.rs @@ -501,6 +501,9 @@ async fn export_inner( .thread .thread_extension_data() .get::(); + if let Some(tools) = &tools { + tools.check_prepared_checkpoint()?; + } let managed_tool_specs = if let Some(tools) = &tools { if let Some(key) = pending { check_capture(state, key)?; diff --git a/packages/agents-codex/native/rust/extensions.rs b/packages/agents-codex/native/rust/extensions.rs index b291ae7db..0658036dd 100644 --- a/packages/agents-codex/native/rust/extensions.rs +++ b/packages/agents-codex/native/rust/extensions.rs @@ -334,6 +334,9 @@ pub fn build( skills_registry.build(), crate::native_interception::DirectOutput::Json, ); + builder.model_request_contributor(Arc::new( + crate::python_tools::prepared_requests::Contributor, + )); builder.model_request_contributor(recovery); builder.model_request_contributor(Arc::new(crate::response_integrity::Contributor( interception.clone(), diff --git a/packages/agents-codex/native/rust/lib.rs b/packages/agents-codex/native/rust/lib.rs index 645440d8c..39f33a56b 100644 --- a/packages/agents-codex/native/rust/lib.rs +++ b/packages/agents-codex/native/rust/lib.rs @@ -2111,6 +2111,21 @@ impl NativeThread { .map_err(input_error) } + #[napi(js_name = "_reply_prepared_request")] + pub fn _reply_prepared_request(&self, response_json: String) -> napi::Result<()> { + let inner = self.current()?; + let response: python_tools::prepared_requests::Reply = serde_json::from_str(&response_json) + .map_err(|_| input_error("Invalid prepared request reply"))?; + let tools = inner + .thread + .thread_extension_data() + .get::() + .ok_or_else(|| input_error("Prepared request requires managed Session"))?; + tools + .reply_prepared(&inner.id.to_string(), response) + .map_err(input_error) + } + #[napi(js_name = "_reply_step")] pub fn _reply_step(&self, response_json: String) -> napi::Result<()> { let inner = self.current()?; @@ -2911,6 +2926,14 @@ impl NativeThread { .transpose()?; // A restored root can use step preparation to enforce its remaining // SDK budget. The SDK never serializes an application callback. + if selection + .as_ref() + .is_some_and(|selection| selection.prepare_request) + { + return Err(native_error( + "manual continuation recovery cannot restore request preparation", + )); + } let inner = self.current()?; let _operation = inner.engine.operation(false).map_err(native_error)?; inner diff --git a/packages/agents-codex/native/rust/prepared_requests.rs b/packages/agents-codex/native/rust/prepared_requests.rs new file mode 100644 index 000000000..dc418913b --- /dev/null +++ b/packages/agents-codex/native/rust/prepared_requests.rs @@ -0,0 +1,366 @@ +//! One issuing turn owns each prepared sampling reply and dispatch capability. +use super::*; +use codex_extension_api::{ + ModelRequestContributor, ModelRequestInput, ModelResponseInterceptor, ModelToolAdmission, + PreparedRequestAdmission, PreparedRequestData, PreparedRequestInput, PreparedRequestOutput, +}; +use futures_util::future::BoxFuture; +use std::sync::atomic::AtomicBool; + +pub(super) struct Owner { + tools: Weak, + generation: Arc, + turn: String, + thread: String, + store: usize, + live: AtomicBool, + replies: Mutex>>>, +} +impl Owner { + fn check(&self) -> std::result::Result, String> { + let tools = self + .tools + .upgrade() + .ok_or("Prepared request owner closed")?; + if !self.live.load(Ordering::Acquire) + || !Arc::ptr_eq(&tools.snapshot(), &self.generation) + || !tools + .prepared + .lock() + .expect("prepared owner poisoned") + .as_ref() + .is_some_and(|current| std::ptr::eq(current.as_ref(), self)) + { + return Err("Prepared request owner retired".into()); + } + Ok(tools) + } + pub(super) fn queued(&self, id: u64) -> bool { + self.check().is_ok() + && self + .replies + .lock() + .expect("prepared replies poisoned") + .get(&id) + .is_some_and(|reply| !reply.is_closed()) + } + pub(super) fn retire(&self) { + self.live.store(false, Ordering::Release); + self.replies + .lock() + .expect("prepared replies poisoned") + .clear(); + } +} +#[derive(Default, Debug)] +pub(crate) struct Contributor; +impl ModelRequestContributor for Contributor { + fn request(&self, _: ModelRequestInput<'_>) -> Option> { + None + } + fn prepares_request(&self, turn: &ExtensionData) -> bool { + turn.get::().is_some() + } + fn prepare_request( + &self, + input: PreparedRequestInput, + ) -> BoxFuture<'_, std::result::Result> { + Box::pin(async move { + let owner = input + .admission + .turn_store() + .get::() + .ok_or("Missing prepared request owner")?; + let tools = owner.check()?; + if owner.store != input.admission.turn_store() as *const ExtensionData as usize + || owner.turn != input.admission.turn_store().level_id() + || owner.thread != input.admission.thread_id() + || tools.instance.is_some() + { + return Err("Expired prepared request store".into()); + } + let presentation = tools + .schema_receipts + .prepared(input.data) + .map_err(|_| "Invalid prepared schema owner")?; + let id = tools.family.next_step.fetch_add(1, Ordering::Relaxed); + let (send, receive) = oneshot::channel(); + { + let mut pending = owner.replies.lock().expect("prepared replies poisoned"); + owner.check()?; + pending.insert(id, send); + } + let wait = Wait { + owner: owner.clone(), + id, + }; + tools.family.requests.send(FamilyRequest::Prepared(Arc::downgrade(&owner), id, json!({ + "id": owner.turn, "msg": {"type":"_sdk_prepare_request", "turn_id":owner.turn, + "thread":input.admission.thread_id(), "generation":owner.generation.id.to_string(), + "request_id":id.to_string(), "recovery_attempt":input.recovery_attempt,"data":presentation.data} + }))).map_err(|_| "Prepared request reader closed")?; + let mut data = receive + .await + .map_err(|_| "Prepared request callback failed or cancelled")?; + owner.check()?; + let sampling_schemas = data + .as_mut() + .map(|data| { + tools + .schema_receipts + .reconcile_prepared(&presentation, data) + }) + .transpose() + .map_err(|_| "Invalid prepared schema presentation")? + .flatten(); + drop(wait); + Ok(PreparedRequestOutput { + data, + admission: Arc::new(Admission { + owner, + original: input.admission, + _sampling_schemas: sampling_schemas, + }), + }) + }) + } +} +struct Wait { + owner: Arc, + id: u64, +} +impl Drop for Wait { + fn drop(&mut self) { + self.owner + .replies + .lock() + .expect("prepared replies poisoned") + .remove(&self.id); + } +} +struct Admission { + owner: Arc, + original: Arc, + // A sampling copy keeps only its own immutable presentation receipts alive. + _sampling_schemas: Option>, +} +impl std::fmt::Debug for Admission { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("PreparedRequestAdmission") + } +} +impl PreparedRequestAdmission for Admission { + fn admit_request(&self) -> std::result::Result<(), String> { + self.owner.check()?; + if self.owner.store != self.original.turn_store() as *const ExtensionData as usize + || !self + .original + .turn_store() + .get::() + .is_some_and(|current| Arc::ptr_eq(¤t, &self.owner)) + { + return Err("Expired prepared request admission".into()); + } + Ok(()) + } +} +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct Reply { + pub(crate) thread: String, + turn_id: String, + generation: String, + request_id: String, + // A failed callback closes the native wait without treating failure as no edit. + failed: bool, + data: Option, +} +impl PythonTools { + pub(super) fn begin_preparation(self: &Arc, input: &TurnStartInput<'_>) { + let enabled = input + .turn_store + .get::>() + .is_some_and(|operation| { + operation + .as_ref() + .as_ref() + .is_some_and(|operation| operation.prepare_request) + }); + if !enabled { + self.retire_preparation(None); + return; + } + let owner = input.turn_store.get_or_init(|| Owner { + tools: Arc::downgrade(self), + generation: self.snapshot(), + turn: input.turn_id.to_owned(), + thread: input.thread_store.level_id().to_owned(), + store: input.turn_store as *const ExtensionData as usize, + live: AtomicBool::new(true), + replies: Mutex::new(HashMap::new()), + }); + let previous = self + .prepared + .lock() + .expect("prepared owner poisoned") + .replace(owner.clone()); + if let Some(previous) = previous.filter(|previous| !Arc::ptr_eq(previous, &owner)) { + previous.retire(); + } + } + pub(crate) fn reply_prepared(&self, thread: &str, response: Reply) -> Result<()> { + let owner = self + .prepared + .lock() + .expect("prepared owner poisoned") + .clone() + .context("Expired prepared request")?; + owner.check().map_err(anyhow::Error::msg)?; + ensure!( + response.thread == thread + && response.turn_id == owner.turn + && response.generation == owner.generation.id.to_string(), + "Expired prepared request owner" + ); + let sender = owner + .replies + .lock() + .expect("prepared replies poisoned") + .remove(&response.request_id.parse()?) + .context("Expired prepared request reply")?; + if response.failed { + drop(sender); + return Ok(()); + } + sender + .send(response.data) + .map_err(|_| anyhow::anyhow!("Expired prepared request reply")) + } + pub(crate) fn check_prepared_checkpoint(&self) -> Result<()> { + ensure!( + self.prepared + .lock() + .expect("prepared owner poisoned") + .is_none(), + "Cannot capture a prepared-request operation" + ); + Ok(()) + } + pub(super) fn retire_preparation(&self, turn: Option<&str>) { + let retired = { + let mut owner = self.prepared.lock().expect("prepared owner poisoned"); + if owner + .as_ref() + .is_some_and(|owner| turn.is_none_or(|turn| turn == owner.turn)) + { + owner.take() + } else { + None + } + }; + if let Some(owner) = retired { + owner.retire(); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use codex_protocol::models::ResponseItem; + + #[derive(Debug)] + struct Original { + turn: Arc, + } + impl ModelToolAdmission for Original { + fn thread_id(&self) -> &str { + "root" + } + fn turn_store(&self) -> &ExtensionData { + &self.turn + } + fn admits(&self, _: &ResponseItem) -> bool { + true + } + } + fn fixture() -> (Arc, Arc, Arc) { + let tools = super::super::pending_tests::tools(); + let turn = Arc::new(ExtensionData::new("turn")); + let owner = turn.get_or_init(|| Owner { + tools: Arc::downgrade(&tools), + generation: tools.snapshot(), + turn: "turn".into(), + thread: "root".into(), + store: turn.as_ref() as *const ExtensionData as usize, + live: AtomicBool::new(true), + replies: Mutex::new(HashMap::new()), + }); + *tools.prepared.lock().unwrap() = Some(owner.clone()); + (tools, turn, owner) + } + fn input(turn: Arc) -> PreparedRequestInput { + PreparedRequestInput { + admission: Arc::new(Original { turn }), + recovery_attempt: 0, + data: PreparedRequestData { + instructions: "original".into(), + input: Vec::new(), + tools: Vec::new(), + }, + } + } + fn reply(event: &Value) -> Reply { + serde_json::from_value(json!({ "thread":"root", "turn_id":"turn", "generation":"0", "request_id":event["msg"]["request_id"], "failed":false,"data":null })).unwrap() + } + #[tokio::test] + async fn no_edit_retains_dispatch_admission_and_reply_is_one_shot() { + let (tools, turn, owner) = fixture(); + let contributor = Contributor; + let mut prepare = contributor.prepare_request(input(turn)); + assert!(futures_util::poll!(&mut prepare).is_pending()); + let event = tools.next_request().await; + tools.reply_prepared("root", reply(&event)).unwrap(); + assert!(tools.reply_prepared("root", reply(&event)).is_err()); + let output = prepare.await.unwrap(); + assert!(output.data.is_none()); + assert!(output.admission.admit_request().is_ok()); + tools.retire_preparation(Some("turn")); + assert!(!owner.live.load(Ordering::Acquire)); + assert!(output.admission.admit_request().is_err()); + } + #[tokio::test] + async fn dropped_wait_revokes_receipt_and_active_capture_is_refused() { + let (tools, turn, _) = fixture(); + assert!(tools.check_prepared_checkpoint().is_err()); + let contributor = Contributor; + let mut prepare = contributor.prepare_request(input(turn)); + assert!(futures_util::poll!(&mut prepare).is_pending()); + let event = tools.next_request().await; + drop(prepare); + assert!(tools.reply_prepared("root", reply(&event)).is_err()); + tools.retire_preparation(None); + assert!(tools.check_prepared_checkpoint().is_ok()); + } + #[tokio::test] + async fn superseded_owner_and_exact_store_mismatch_cannot_dispatch() { + let (tools, turn, owner) = fixture(); + let contributor = Contributor; + let mut prepare = contributor.prepare_request(input(turn.clone())); + assert!(futures_util::poll!(&mut prepare).is_pending()); + let event = tools.next_request().await; + tools.reply_prepared("root", reply(&event)).unwrap(); + let output = prepare.await.unwrap(); + let impostor = Arc::new(ExtensionData::new("turn")); + // Even the same marker and textual ID cannot change the original store identity. + let wrong = Admission { + owner: owner.clone(), + original: Arc::new(Original { turn: impostor }), + _sampling_schemas: None, + }; + assert!(wrong.admit_request().is_err()); + *tools.prepared.lock().unwrap() = None; + assert!(output.admission.admit_request().is_err()); + assert!(contributor.prepares_request(&turn)); + } +} diff --git a/packages/agents-codex/native/rust/python_tools.rs b/packages/agents-codex/native/rust/python_tools.rs index b403afaeb..2f25e39b6 100644 --- a/packages/agents-codex/native/rust/python_tools.rs +++ b/packages/agents-codex/native/rust/python_tools.rs @@ -38,6 +38,9 @@ pub(crate) use manual::{Continuation, ManualLedger, ManualSlot}; #[path = "schema_fidelity.rs"] mod schema_fidelity; +#[path = "prepared_requests.rs"] +pub(crate) mod prepared_requests; + struct Entry { name: ToolName, spec: ToolSpec, @@ -115,11 +118,13 @@ pub struct PythonTools { manual: Mutex, selection: Mutex>, step_replies: Mutex>, + prepared: Mutex>>, } // The family inbox transports requests; reply authority stays with each issuer. enum FamilyRequest { Step(Weak, u64, Value), + Prepared(Weak, u64, Value), Call(Weak, PendingKey, Value), Retired { thread: String, instance: String }, } @@ -451,6 +456,7 @@ impl PythonTools { manual: Mutex::new(ManualLedger::default()), selection: Mutex::new(None), step_replies: Mutex::new(HashMap::new()), + prepared: Mutex::new(None), }; let seed = Arc::new(tools.prepare(0, specs)?); tools.schema_receipts.register(&seed); @@ -503,6 +509,7 @@ impl PythonTools { manual: Mutex::new(parent.child_manual_ledger()), selection: Mutex::new(None), step_replies: Mutex::new(HashMap::new()), + prepared: Mutex::new(None), }; // Validate the inherited catalog against this child's current ceiling. let parent_generation = parent.snapshot(); @@ -704,6 +711,11 @@ impl PythonTools { let mut receiver = self.family.receiver.lock().await; loop { match receiver.recv().await { + Some(FamilyRequest::Prepared(owner, id, request)) => { + if owner.upgrade().is_some_and(|owner| owner.queued(id)) { + return request; + } + } Some(FamilyRequest::Step(owner, id, request)) => { if owner.upgrade().is_some_and(|tools| { if tools.instance.is_some() { @@ -952,6 +964,7 @@ impl ThreadLifecycleContributor for PythonContributor { progress.0.send_replace(()); } if let Some(tools) = input.thread_store.get::() { + tools.retire_preparation(None); tools.retire_manual_wait(); tools .step_replies @@ -998,6 +1011,8 @@ pub(crate) struct OperationSelection { // Names select across all native families; slots remain the legacy JS ceiling. pub names: Option>, pub before_step: bool, + #[serde(default)] + pub prepare_request: bool, } #[derive(Clone)] struct ActiveSelection { @@ -1006,6 +1021,7 @@ struct ActiveSelection { slots: Option>, names: Option>, before_step: bool, + prepare_request: bool, } struct PendingStep { turn: String, @@ -1105,6 +1121,7 @@ impl Drop for StepWait { impl PythonTools { fn retire_selection(&self, turn: &str) { + self.retire_preparation(Some(turn)); let mut active = self .selection .lock() @@ -1132,6 +1149,7 @@ impl PythonTools { slots: selection.slots.clone(), names: selection.names.clone(), before_step: selection.before_step, + prepare_request: selection.prepare_request, }) } @@ -1226,6 +1244,7 @@ impl PythonTools { slots: None, names: None, before_step: true, + prepare_request: false, }); } operation.map(|selection| ActiveSelection { @@ -1234,6 +1253,7 @@ impl PythonTools { slots: selection.slots.clone(), names: selection.names.clone(), before_step: selection.before_step, + prepare_request: selection.prepare_request, }) } @@ -1662,6 +1682,7 @@ impl TurnLifecycleContributor for PythonContributor { .selection .lock() .expect("JavaScript selection poisoned") = selection; + tools.begin_preparation(&input); tools.begin_native_manual_turn(input); } }) @@ -1681,6 +1702,7 @@ impl TurnLifecycleContributor for PythonContributor { .lock() .expect("JavaScript step replies poisoned") .retain(|_, pending| pending.turn != input.turn_store.level_id()); + tools.retire_preparation(Some(input.turn_store.level_id())); tools.stop_native_manual_turn(input); } }) @@ -1761,7 +1783,7 @@ mod pending_tests { // Exercise the actual registry and its lifetime owners without starting Core, // a runtime, a network service, or a native thread. - fn tools() -> Arc { + pub(super) fn tools() -> Arc { let tools = Arc::new(PythonTools { local_store: None, custom_store: None, @@ -1785,6 +1807,7 @@ mod pending_tests { manual: Mutex::default(), selection: Mutex::new(None), step_replies: Mutex::new(HashMap::new()), + prepared: Mutex::new(None), }); let specs = vec![DynamicToolSpec::Function(DynamicToolFunctionSpec { name: "decision".into(), @@ -2055,6 +2078,7 @@ mod pending_tests { slots: None, names, before_step: true, + prepare_request: false, }); (thread, tools) } @@ -2076,6 +2100,7 @@ mod pending_tests { slots: Some(Vec::new()), names: None, before_step: false, + prepare_request: false, }; *child.selection.lock().unwrap() = child.turn_selection("child-turn", Some(&root_operation)); @@ -2371,6 +2396,7 @@ mod pending_tests { slots: Some(Vec::new()), names: None, before_step: false, + prepare_request: false, }); let successor = ExtensionData::new("successor-step"); assert!( @@ -2499,6 +2525,7 @@ mod pending_tests { slots: operation.slots, names: operation.names, before_step: operation.before_step, + prepare_request: operation.prepare_request, }); let continued = tools.continuation_selection("turn").unwrap(); assert_eq!(continued.slots, expected_slots); diff --git a/packages/agents-codex/native/rust/schema_fidelity.rs b/packages/agents-codex/native/rust/schema_fidelity.rs index 2b5f8a95f..fe013bd49 100644 --- a/packages/agents-codex/native/rust/schema_fidelity.rs +++ b/packages/agents-codex/native/rust/schema_fidelity.rs @@ -7,11 +7,13 @@ use std::collections::HashMap; use std::sync::{Arc, LazyLock, Mutex, Weak}; use anyhow::{Result, ensure}; +use codex_extension_api::{PreparedRequestData, PreparedToolPresentation}; use codex_protocol::ThreadId; use codex_tools::{ResponsesApiNamespaceTool, ToolSpec}; use serde_json::Value; -use super::{Generation, PythonTools, functions}; +use super::{Entry, Generation, PythonTools, functions}; +use codex_protocol::dynamic_tools::{DynamicToolNamespaceTool, DynamicToolSpec}; static MARKER: LazyLock = LazyLock::new(|| format!("\n[agents-schema:{}/", ThreadId::new())); @@ -21,6 +23,19 @@ struct Receipt { historical: Option>, } +struct ManagedSchema { + generation: Arc, + slot: usize, +} + +/// One clean callback view plus its private original presentation owners. +/// This metadata never supplies executors or changes the live tool generation. +pub(super) struct PreparedSchemas { + pub(super) data: PreparedRequestData, + original: Vec, + managed: Vec>, +} + #[derive(Default)] pub(super) struct Receipts(Mutex>); @@ -162,12 +177,128 @@ impl Receipts { Ok(()) } - fn function(&self, tool: &mut Value, namespace: &str) -> Result<()> { + pub(super) fn prepared(&self, mut data: PreparedRequestData) -> Result { + let original = data.tools.clone(); + let mut managed = Vec::with_capacity(data.tools.len()); + for tool in &mut data.tools { + let owner = self.resolve( + &serde_json::json!({ + "name":tool.name, "description":tool.description, "parameters":tool.parameters, + }), + tool.namespace.as_deref().unwrap_or("functions"), + )?; + if let Some(owner) = &owner { + let originals = functions(&owner.generation.specs); + tool.description = originals[owner.slot].1.description.clone(); + // Core's projection omits raw JSON Schema keywords. + tool.parameters = originals[owner.slot].1.input_schema.clone(); + } + managed.push(owner); + } + Ok(PreparedSchemas { + data, + original, + managed, + }) + } + + pub(super) fn reconcile_prepared( + &self, + captured: &PreparedSchemas, + data: &mut PreparedRequestData, + ) -> Result>> { + ensure!( + data.tools.len() == captured.data.tools.len(), + "Prepared tool count changed" + ); + let token = ThreadId::new().to_string(); + let mut entries = Vec::new(); + let mut specs = Vec::new(); + for (index, tool) in data.tools.iter_mut().enumerate() { + // Only this runtime's actual private syntax is reserved. Ordinary + // documentation that mentions agents-schema is still plain text. + ensure!( + !tool.description.contains(MARKER.as_str()), + "Prepared description contains a reserved schema reference" + ); + let Some(owner) = &captured.managed[index] else { + continue; + }; + let public = &captured.data.tools[index]; + let description_changed = tool.description != public.description; + let parameters_changed = tool.parameters != public.parameters; + if !parameters_changed { + // Return the unchanged declaration in Core's original form. + tool.parameters = captured.original[index].parameters.clone(); + } + if !description_changed && !parameters_changed { + tool.description = captured.original[index].description.clone(); + continue; + } + let original_entry = &owner.generation.entries[owner.slot]; + let originals = functions(&owner.generation.specs); + let (namespace, original) = originals[owner.slot]; + let mut raw = original.clone(); + raw.description = tool.description.clone(); + if parameters_changed { + raw.input_schema = tool.parameters.clone(); + } + let mut projected = original_entry.spec.clone(); + let function = match &mut projected { + ToolSpec::Function(function) => function, + ToolSpec::Namespace(namespace) => match namespace.tools.first_mut() { + Some(ResponsesApiNamespaceTool::Function(function)) => function, + _ => anyhow::bail!("Invalid managed prepared namespace"), + }, + _ => anyhow::bail!("Invalid managed prepared function"), + }; + // Core validates the native projection. This captured sampling + // receipt restores the exact raw schema only after tool identity + // and projected declaration checks succeed at serialization. + if parameters_changed { + function.parameters = serde_json::from_value(tool.parameters.clone()) + .map_err(|_| anyhow::anyhow!("Invalid prepared schema projection"))?; + tool.parameters = serde_json::to_value(&function.parameters) + .map_err(|_| anyhow::anyhow!("Invalid prepared schema projection"))?; + } + function.description = Self::description(&tool.description, &token, entries.len()); + tool.description = function.description.clone(); + entries.push(Entry { + name: original_entry.name.clone(), + spec: projected, + exposure: original_entry.exposure, + }); + specs.push(match namespace { + None => DynamicToolSpec::Function(raw), + Some(namespace) => { + let mut namespace = namespace.clone(); + namespace.tools = vec![DynamicToolNamespaceTool::Function(raw)]; + DynamicToolSpec::Namespace(namespace) + } + }); + } + if entries.is_empty() { + return Ok(None); + } + let generation = Arc::new(Generation { + id: 0, + schema_token: Some(token), + entries, + specs, + defaults: Vec::new(), + }); + // Register only presentation metadata. No executable plan, current + // generation, search-history retention, or application policy changes. + self.register(&generation); + Ok(Some(generation)) + } + + fn resolve(&self, tool: &Value, namespace: &str) -> Result> { let Some(description) = tool.get("description").and_then(Value::as_str) else { - return Ok(()); + return Ok(None); }; let Some((_, suffix)) = description.rsplit_once(MARKER.as_str()) else { - return Ok(()); + return Ok(None); }; let (token, slot) = suffix .split_once(']') @@ -213,6 +344,25 @@ impl Receipts { && tool.get("parameters") == Some(&serde_json::to_value(&projected.parameters)?), "Managed tool schema declaration does not match its native owner" ); + Ok(Some(ManagedSchema { generation, slot })) + } + + fn function(&self, tool: &mut Value, namespace: &str) -> Result<()> { + let Some(ManagedSchema { generation, slot }) = self.resolve(tool, namespace)? else { + return Ok(()); + }; + let description = tool + .get("description") + .and_then(Value::as_str) + .ok_or_else(|| anyhow::anyhow!("Invalid managed tool description"))?; + let projected = match &generation.entries[slot].spec { + ToolSpec::Function(function) => function, + ToolSpec::Namespace(namespace) => match namespace.tools.first() { + Some(ResponsesApiNamespaceTool::Function(function)) => function, + _ => unreachable!("resolved namespace function"), + }, + _ => unreachable!("resolved function"), + }; let originals = functions(&generation.specs); let (_, original) = originals .get(slot) @@ -419,3 +569,179 @@ mod tests { } } } + +#[cfg(test)] +mod prepared_tests { + use super::*; + use codex_protocol::dynamic_tools::DynamicToolFunctionSpec; + use codex_tools::{ToolExposure, ToolName, dynamic_tool_to_responses_api_tool}; + use serde_json::json; + + fn fixture(receipts: &Receipts) -> (Arc, PreparedRequestData) { + let raw = DynamicToolFunctionSpec { + name: "lookup".into(), + description: "Original description.".into(), + input_schema: json!({"type":"object","properties":{"key":{"type":"string","minLength":3}},"required":["key"],"additionalProperties":false}), + defer_loading: false, + }; + let token = ThreadId::new().to_string(); + let mut projected = dynamic_tool_to_responses_api_tool(&raw).unwrap(); + projected.description = Receipts::description(&raw.description, &token, 0); + let data = PreparedRequestData { + instructions: "original".into(), + input: Vec::new(), + tools: vec![PreparedToolPresentation { + namespace: None, + name: projected.name.clone(), + kind: "function".into(), + strict: projected.strict, + defer_loading: projected.defer_loading, + description: projected.description.clone(), + parameters: serde_json::to_value(&projected.parameters).unwrap(), + }], + }; + let generation = Arc::new(Generation { + id: 7, + schema_token: Some(token), + entries: vec![Entry { + name: ToolName::plain("lookup"), + spec: ToolSpec::Function(projected), + exposure: ToolExposure::Direct, + }], + specs: vec![DynamicToolSpec::Function(raw)], + defaults: vec![0], + }); + receipts.register(&generation); + (generation, data) + } + fn wire(data: &PreparedRequestData) -> Value { + json!({"tools":data.tools.iter().map(|tool|json!({"type":"function","name":tool.name,"description":tool.description,"parameters":serde_json::to_value(serde_json::from_value::(tool.parameters.clone()).unwrap()).unwrap()})).collect::>()}) + } + #[test] + fn field_intent_retains_raw_schema_until_parameters_actually_change() { + for (description_changed, parameters_changed) in + [(false, false), (true, false), (false, true), (true, true)] + { + let receipts = Receipts::default(); + let (original, data) = fixture(&receipts); + let view = receipts.prepared(data).unwrap(); + assert_eq!(view.data.tools[0].description, "Original description."); + assert!(!view.data.tools[0].description.contains(MARKER.as_str())); + assert_eq!( + view.data.tools[0].parameters, + functions(&original.specs)[0].1.input_schema + ); + let mut edited = view.data.clone(); + if description_changed { + edited.tools[0].description = "Edited description.".into(); + } + if parameters_changed { + edited.tools[0].parameters = json!({"type":"object","properties":{"key":{"type":"string","enum":["allowed","wider"]}},"required":["key"],"additionalProperties":false}); + } + let returned_parameters = edited.tools[0].parameters.clone(); + let metadata = receipts.reconcile_prepared(&view, &mut edited).unwrap(); + if parameters_changed { + let projected_parameters: codex_tools::JsonSchema = + serde_json::from_value(returned_parameters.clone()).unwrap(); + assert_eq!( + edited.tools[0].parameters, + serde_json::to_value(projected_parameters).unwrap() + ); + } else { + assert_eq!(edited.tools[0].parameters, view.original[0].parameters); + } + assert_eq!( + metadata.is_some(), + description_changed || parameters_changed + ); + let mut request = wire(&edited); + receipts.restore(&mut request).unwrap(); + assert_eq!( + request["tools"][0]["description"], + if description_changed { + "Edited description." + } else { + "Original description." + } + ); + if parameters_changed { + assert_eq!(request["tools"][0]["parameters"], returned_parameters); + } else { + assert_eq!( + request["tools"][0]["parameters"], + functions(&original.specs)[0].1.input_schema + ); + } + if let Some(metadata) = metadata { + let mut stale = wire(&edited); + drop(metadata); + assert!(receipts.restore(&mut stale).is_err()); + } + } + } + #[test] + fn partial_parameter_edits_keep_original_raw_keywords() { + let receipts = Receipts::default(); + let (original, data) = fixture(&receipts); + let view = receipts.prepared(data).unwrap(); + assert_eq!( + view.data.tools[0].parameters["properties"]["key"]["minLength"], + json!(3) + ); + let mut edited = view.data.clone(); + edited.tools[0].parameters["properties"]["key"]["maxLength"] = json!(9); + let expected = edited.tools[0].parameters.clone(); + let _metadata = receipts.reconcile_prepared(&view, &mut edited).unwrap(); + let mut request = wire(&edited); + receipts.restore(&mut request).unwrap(); + assert_eq!(request["tools"][0]["parameters"], expected); + assert_eq!( + functions(&original.specs)[0].1.input_schema["properties"]["key"]["maxLength"], + Value::Null + ); + } + #[test] + fn managed_schema_projection_keeps_immutable_identity_for_core() { + let receipts = Receipts::default(); + let (_original, data) = fixture(&receipts); + let view = receipts.prepared(data).unwrap(); + let mut edited = view.data.clone(); + edited.tools[0].parameters["properties"]["key"]["unknown_null_key"] = Value::Null; + edited.tools[0].name = "renamed".into(); + edited.tools[0].namespace = Some("changed_namespace".into()); + edited.tools[0].kind = "changed_kind".into(); + edited.tools[0].strict = !edited.tools[0].strict; + edited.tools[0].defer_loading = Some(true); + let submitted = edited.tools[0].clone(); + let _metadata = receipts.reconcile_prepared(&view, &mut edited).unwrap(); + assert_ne!(edited.tools[0].parameters, submitted.parameters); + assert!( + edited.tools[0].parameters["properties"]["key"] + .get("unknown_null_key") + .is_none() + ); + assert_eq!(edited.tools[0].name, submitted.name); + assert_eq!(edited.tools[0].namespace, submitted.namespace); + assert_eq!(edited.tools[0].kind, submitted.kind); + assert_eq!(edited.tools[0].strict, submitted.strict); + assert_eq!(edited.tools[0].defer_loading, submitted.defer_loading); + } + #[test] + fn prepared_receipts_remain_strict_and_only_actual_private_syntax_is_reserved() { + let receipts = Receipts::default(); + let (_original, data) = fixture(&receipts); + let private = data.tools[0].description.clone(); + let view = receipts.prepared(data).unwrap(); + let mut forged = view.data.clone(); + forged.tools[0].description = private; + assert!(receipts.reconcile_prepared(&view, &mut forged).is_err()); + let mut edited = view.data.clone(); + edited.tools[0].description = "Docs mention [agents-schema:example] syntax.".into(); + let _metadata = receipts.reconcile_prepared(&view, &mut edited).unwrap(); + let mut request = wire(&edited); + request["tools"][0]["parameters"]["properties"]["key"]["type"] = json!("number"); + let before = request.clone(); + assert!(receipts.restore(&mut request).is_err()); + assert_eq!(request, before); + } +} diff --git a/packages/agents-codex/src/index.ts b/packages/agents-codex/src/index.ts index f1c12d483..2376f6cee 100644 --- a/packages/agents-codex/src/index.ts +++ b/packages/agents-codex/src/index.ts @@ -140,7 +140,15 @@ export { OutputValidationError } from './runtime/output'; export type { OutputSpec } from './runtime/output'; export type { ToolRef, ToolSelector } from './runtime/tools'; -export type { OperationOptions, BeforeStepContext } from './runtime/session'; +export type { + PreparedRequestData, + PreparedToolPresentation, +} from './runtime/prepared-request'; +export type { + OperationOptions, + BeforeStepContext, + PrepareRequestContext, +} from './runtime/session'; export * as telemetry from './runtime/telemetry'; export type { diff --git a/packages/agents-codex/src/runtime/prepared-request.ts b/packages/agents-codex/src/runtime/prepared-request.ts new file mode 100644 index 000000000..b922f3ff6 --- /dev/null +++ b/packages/agents-codex/src/runtime/prepared-request.ts @@ -0,0 +1,94 @@ +import { z } from 'zod'; +import { losslessJsonValueSchema, type JsonValue } from './binding'; + +/** Sampling presentation only; executable declarations keep their original authority. */ +export interface PreparedToolPresentation { + readonly namespace: string | null; + readonly name: string; + readonly kind: 'function'; + readonly strict: boolean; + readonly deferLoading: boolean | null; + description: string; + parameters: { [key: string]: JsonValue }; +} +/** Native Responses input items use native omission for optional values. + * Managed function parameter edits retain raw schema fields through their captured + * sampling receipt. Other tools require a lossless native schema round trip. + */ +export interface PreparedRequestData { + instructions: string; + input: JsonValue[]; + tools: PreparedToolPresentation[]; +} +const toolSchema = z + .object({ + namespace: z.string().nullable(), + name: z.string(), + kind: z.literal('function'), + strict: z.boolean(), + deferLoading: z.boolean().nullable(), + description: z.string(), + parameters: z.record(z.string(), z.unknown()), + }) + .strict(); +const requestSchema = z + .object({ + instructions: z.string(), + input: z.array(z.unknown()), + tools: z.array(toolSchema), + }) + .strict(); + +/** Validate own keys before mapping so undefined-valued unknown fields cannot vanish. */ +function ownKeys(value: unknown, keys: readonly string[]): void { + if (value === null || typeof value !== 'object' || Array.isArray(value)) + throw new TypeError('Expected a prepared request object'); + if ( + Reflect.ownKeys(value).some( + (key) => typeof key !== 'string' || !keys.includes(key), + ) + ) + throw new TypeError('Unknown prepared request field'); +} +export function capturePreparedRequest(value: unknown): string { + ownKeys(value, ['instructions', 'input', 'tools']); + const tools = (value as PreparedRequestData).tools; + if (!Array.isArray(tools)) throw new TypeError('Expected prepared tools'); + for (const tool of tools) + ownKeys(tool, [ + 'namespace', + 'name', + 'kind', + 'strict', + 'deferLoading', + 'description', + 'parameters', + ]); + // Copy first: schema parsing alone can drop undefined/non-JSON nested values. + const captured = requestSchema.parse( + losslessJsonValueSchema.parse(value), + ) as PreparedRequestData; + return JSON.stringify({ + ...captured, + tools: captured.tools.map(({ deferLoading, ...tool }) => ({ + ...tool, + defer_loading: deferLoading, + })), + }); +} +export function preparedRequestFromNative(value: unknown): PreparedRequestData { + const captured = losslessJsonValueSchema.parse(value) as unknown as { + instructions: string; + input: JsonValue[]; + tools: (Omit & { + defer_loading: boolean | null; + })[]; + }; + return { + ...captured, + tools: captured.tools.map(({ defer_loading, ...tool }) => ({ + ...tool, + deferLoading: defer_loading, + })), + }; +} diff --git a/packages/agents-codex/src/runtime/public-agent.ts b/packages/agents-codex/src/runtime/public-agent.ts index bceffaa99..26a9cb6a0 100644 --- a/packages/agents-codex/src/runtime/public-agent.ts +++ b/packages/agents-codex/src/runtime/public-agent.ts @@ -109,6 +109,11 @@ export interface RunOptions extends Omit< > { /** Root model preparation limit for a new run. Resumed runs inherit their saved budget. */ maxModelSteps?: OperationOptions['maxModelSteps']; + /** + * Edit requests for a new run with input. A resumed turn cannot add an editor. + * Manual tools require onInteraction because an edited run cannot save an active checkpoint. + */ + prepareRequest?: import('./session').OperationOptions['prepareRequest']; returnState?: boolean; onNativeQuestion?: ( request: NativeQuestion, @@ -378,6 +383,8 @@ export class Agent { throw new TypeError( 'Cannot replace the model step budget when resuming a run', ); + if (input === null && options.prepareRequest !== undefined) + throw new TypeError('Cannot add prepareRequest when resuming a run'); const preparedInput = input === null ? null : nativeInput(input); const factory = this.session; const sessionOptions: OwnedSessionOptions = { @@ -397,6 +404,7 @@ export class Agent { ], [ownedRun]: { input: preparedInput, + prepareRequest: options.prepareRequest, onNativeQuestion: options.onNativeQuestion, onRequiredAction: options.onInteraction ? (action, signal) => @@ -559,6 +567,42 @@ export class Agent { ? Object.freeze({ ...declaredOutput }) : declaredOutput; const output = opening ? opening.output : prepareOutput(outputType); + if ( + (options as OwnedSessionOptions)[ownedRun]?.prepareRequest !== + undefined + ) { + const prepareRequest = (options as OwnedSessionOptions)[ + ownedRun + ]!.prepareRequest; + if (typeof prepareRequest !== 'function') + throw new TypeError('prepareRequest must be callable'); + if (output.repair) + throw new TypeError( + 'prepareRequest does not support output correction attempts; use native output or prompted maxRetries: 0', + ); + if ( + (provider.supportsWebsockets ?? provider.baseUrl === undefined) !== + false + ) + throw new TypeError( + 'prepareRequest requires NativeProvider({ supportsWebsockets: false })', + ); + if ( + programmaticToolCalling || + environmentInput || + declarations.nativeAgents || + declarations.nativeGoals || + declarations.checklist || + declarations.webSearch || + declarations.tools.some((tool) => tool.deferLoading) || + declarations.mcp.some( + (tool) => tool.deferLoading || isHostedRegistration(tool), + ) + ) + throw new TypeError( + 'prepareRequest supports only direct root eager function and ordinary MCP tools', + ); + } const { instructions, files } = !opening && options.instructionFiles !== undefined ? await loadInstructionFiles( diff --git a/packages/agents-codex/src/runtime/session.ts b/packages/agents-codex/src/runtime/session.ts index 5e2dc91ba..980f1b97c 100644 --- a/packages/agents-codex/src/runtime/session.ts +++ b/packages/agents-codex/src/runtime/session.ts @@ -1,3 +1,8 @@ +import { + capturePreparedRequest, + preparedRequestFromNative, + type PreparedRequestData, +} from './prepared-request'; import { captureTracing, getCurrentSpan, @@ -274,6 +279,7 @@ export interface ManualCheckpointEntry { /** Options used by an owned Agent.run Session to drive manual replies. */ export type OneShotOptions = { input: string | JsonValue[] | null; + prepareRequest?: OperationOptions['prepareRequest']; onNativeQuestion?: ( request: NativeQuestion, signal: AbortSignal, @@ -306,6 +312,20 @@ export interface OperationOptions { writableRoots?: readonly string[]; /** Omission uses Session defaults; an empty list selects no tools. */ tools?: readonly ToolSelector[]; + /** + * Edit detached native sampling data; original tools and history retain authority. + * Requires a standard Responses (non-Lite) model, explicit HTTP, and eager root function/MCP tools. + * Unselected deferred tools may remain in the opening catalog. beforeStep may narrow a mixed ceiling. + * Undefined keeps the original. + * Recovery may prepare again; avoid business effects in this callback. Active checkpoints refuse it. + * Pre-turn or mid-turn automatic compaction fails the operation before any hidden request. + * Shorten the input or start a new operation without this callback. Earlier tool effects remain. + * Post-turn compaction is also refused, but the operation keeps its completed answer. + */ + prepareRequest?: ( + context: PrepareRequestContext, + ) => + PreparedRequestData | undefined | Promise; /** * Select a subset of the operation ceiling before each model preparation. * Returning undefined uses that ceiling. Indices start at zero. @@ -317,6 +337,14 @@ export interface OperationOptions { | undefined | Promise[] | undefined>; } +export interface PrepareRequestContext extends Omit< + BeforeStepContext, + 'stepIndex' +> { + readonly request: PreparedRequestData; + readonly preparationIndex: number; + readonly recoveryAttempt: number; +} export interface BeforeStepContext { readonly context: Context; readonly submission: Turn; @@ -342,6 +370,8 @@ interface OperationSelection { readonly names: readonly ReturnType[] | undefined; readonly maxModelSteps?: number; readonly beforeStep?: OperationOptions['beforeStep']; + readonly prepareRequest?: OperationOptions['prepareRequest']; + preparationIndex: number; generation?: bigint; stepIndex: number; } @@ -2486,7 +2516,10 @@ export class Session { false, undefined, false, - this.captureSelection({ maxModelSteps }), + this.captureSelection({ + maxModelSteps, + prepareRequest: this.oneShot?.prepareRequest, + }), callerSignal, ); @@ -2504,7 +2537,7 @@ export class Session { throw new TypeError('submitWithOutput requires an output contract'); const parts = nativeInput(input); const output = prepareOutput(outputType); - const selection = this.captureSelection(options); + const selection = this.captureSelection(options, output); const parentTrace = captureTraceContext(options?.parentTrace) ?? this.options.parentTrace; const run = this.startTurn( @@ -2559,7 +2592,7 @@ export class Session { const parts = nativeInput(input); const output = outputType === undefined ? this.defaultOutput : prepareOutput(outputType); - const selection = this.captureSelection(options); + const selection = this.captureSelection(options, output); const parentTrace = captureTraceContext(options?.parentTrace) ?? this.options.parentTrace; const writableRoots = @@ -2646,6 +2679,7 @@ export class Session { options?.tools !== undefined || options?.beforeStep !== undefined || options?.maxModelSteps !== undefined || + options?.prepareRequest !== undefined || options?.tracing !== undefined ) throw new TypeError('Operation selection applies only to new roots'); @@ -2729,20 +2763,48 @@ export class Session { private captureSelection( options?: OperationOptions, + output = this.defaultOutput, ): OperationSelection | undefined { if ( options?.tools === undefined && options?.beforeStep === undefined && - options?.maxModelSteps === undefined + options?.maxModelSteps === undefined && + options?.prepareRequest === undefined ) return; + if (options.prepareRequest !== undefined) { + if (typeof options.prepareRequest !== 'function') + throw new TypeError('prepareRequest must be callable'); + if (output.repair) + throw new TypeError( + 'prepareRequest does not support output correction attempts; use native output or prompted maxRetries: 0', + ); + if (this.options.websocketsAllowed !== false) + throw new TypeError( + 'prepareRequest requires NativeProvider({ supportsWebsockets: false })', + ); + if ( + this.options.nativeAgents || + this.codeModeEnabled || + this.environment || + this.options.webSearch || + this.options.nativeGoals || + this.options.checklist || + this.openingMcpTools.some(isHostedRegistration) + ) + throw new TypeError( + 'prepareRequest supports only direct root eager function and ordinary MCP tools', + ); + } return { + prepareRequest: options.prepareRequest, names: options.tools === undefined ? undefined : this.selectionNames([...options.tools]), maxModelSteps: validateMaxModelSteps(options.maxModelSteps), beforeStep: options.beforeStep, + preparationIndex: 0, stepIndex: 0, }; } @@ -3239,7 +3301,7 @@ export class Session { !interactive && this.defaultTools.some((tool) => tool.kind === 'manual') && !this.oneShot?.onRequiredAction && - !this.canCheckpointManual(output) + !this.canCheckpointManual(output, selection) ) throw new InteractionRequired(); const roots = @@ -3276,6 +3338,10 @@ export class Session { private dispatchStart(run: NativeTurn, parts: JsonValue[]): void { const { output } = submissions.get(run)!; const selection = run.selection; + if (selection?.prepareRequest && output.repair) + throw new TypeError( + 'prepareRequest does not support output correction attempts; use native output or prompted maxRetries: 0', + ); if (selection) selection.generation = this.catalog.generation; const wire = encode({ ...(selection @@ -3286,6 +3352,7 @@ export class Session { before_step: selection.beforeStep !== undefined || selection.maxModelSteps !== undefined, + prepare_request: selection.prepareRequest !== undefined, }, } : {}), @@ -4492,6 +4559,11 @@ export class Session { void this.trackNativeQuestion(run, event).catch(() => {}); return; } + if (kind === '_sdk_prepare_request') { + if (msg.turn_id !== run.id) return; + this.track(run, this.prepareRequest(run, event)); + return; + } if (kind === '_sdk_select_step') { if (msg.turn_id !== run.id) return; this.track(run, this.selectStep(run, event)); @@ -4725,6 +4797,81 @@ export class Session { ); } } + private prepareRequest(run: NativeTurn, event: NativeEvent): Promise { + const msg = event.msg; + const selection = run.selection; + const preparation = new AbortController(); + const signal = AbortSignal.any([run.controller.signal, preparation.signal]); + run.preparation = preparation; + return this.invokeCallback( + run, + async () => { + let replied = false; + const reply = (data: string | undefined, failed: boolean) => { + if (replied) return; + replied = true; + // The replacement was captured before this boundary; never stringify an application object here. + const receipt = `{"thread":${JSON.stringify(this.id)},"turn_id":${JSON.stringify(msg.turn_id)},"generation":${JSON.stringify(msg.generation)},"request_id":${JSON.stringify(msg.request_id)},"failed":${failed},"data":${data ?? 'null'}}`; + if (failed) { + try { + this.#native._reply_prepared_request(receipt); + } catch { + /* Already revoked. */ + } + } else this.#native._reply_prepared_request(receipt); + }; + const abort = () => reply(undefined, true); + signal.addEventListener('abort', abort, { once: true }); + try { + if ( + !selection?.prepareRequest || + !this.current(run) || + msg.turn_id !== run.id || + msg.generation !== this.catalog.generation.toString() || + selection.generation !== this.catalog.generation + ) + throw new NativeError('Expired prepared request owner'); + signal.throwIfAborted(); + if (this.graphStops.size) + throw new TurnAborted('Request preparation stopped'); + const returned = selection.prepareRequest({ + context: this.options.context as Context, + submission: this.#turnFor(this.rootRun(run)), + preparationIndex: selection.preparationIndex++, + recoveryAttempt: msg.recovery_attempt as number, + signal, + request: preparedRequestFromNative(msg.data), + }); + const replacement = + returned && + typeof (returned as Promise).then === + 'function' + ? await returned + : returned; + // Capture immediately after settlement, before cancellation/owner checks or another await. + const captured = + replacement === undefined + ? undefined + : capturePreparedRequest(replacement); + signal.throwIfAborted(); + if ( + !this.current(run) || + selection.generation !== this.catalog.generation + ) + throw new NativeError('Prepared request is no longer active'); + reply(captured, false); + } catch (error) { + reply(undefined, true); + throw error; + } finally { + signal.removeEventListener('abort', abort); + if (run.preparation === preparation) run.preparation = undefined; + } + }, + signal, + ); + } + private selectStep(run: NativeTurn, event: NativeEvent): Promise { const msg = event.msg; const selection = run.selection; @@ -4821,6 +4968,7 @@ export class Session { selection = { names: undefined, generation: this.catalog.generation, + preparationIndex: 0, stepIndex: 0, }; this.childSelections.set(owner, selection); @@ -5486,8 +5634,12 @@ export class Session { .catch((error) => this.readerFailed(error)); } - private canCheckpointManual(output = this.defaultOutput): boolean { + private canCheckpointManual( + output = this.defaultOutput, + selection?: OperationSelection, + ): boolean { return ( + selection?.prepareRequest === undefined && !output.prompt && this.nativeManualCheckpoint && supportsManualCheckpoint(this.threadOptions, !!this.environment) @@ -5499,7 +5651,10 @@ export class Session { beforeSeal?: (entries: readonly ManualCheckpointEntry[]) => void, signal?: AbortSignal, ): Promise { - if (run.selection?.beforeStep !== undefined) + if ( + run.selection?.beforeStep !== undefined || + run.selection?.prepareRequest !== undefined + ) return Promise.reject( new NativeError('Cannot capture a callback-selected operation'), ); @@ -5826,6 +5981,7 @@ export class Session { ), maxModelSteps: remaining, generation: this.catalog.generation, + preparationIndex: 0, stepIndex: 0, }; this.#turnFor(run); @@ -7222,7 +7378,10 @@ export class Session { if (this.lifecycle.kind !== 'turn') return checkpointStatus('notReady', 'No settled pending manual boundary'); const run = this.lifecycle.run; - if (run.selection?.beforeStep !== undefined) + if ( + run.selection?.beforeStep !== undefined || + run.selection?.prepareRequest !== undefined + ) return checkpointStatus( 'notReady', 'Cannot capture a callback-selected operation', diff --git a/packages/agents-codex/test/native/fixtures/model-server.ts b/packages/agents-codex/test/native/fixtures/model-server.ts index ac965fe11..e6ad2884c 100644 --- a/packages/agents-codex/test/native/fixtures/model-server.ts +++ b/packages/agents-codex/test/native/fixtures/model-server.ts @@ -4,7 +4,9 @@ import { createServer, type IncomingHttpHeaders } from 'node:http'; import type { JsonObject, JsonValue } from '../../../src/runtime/binding'; import type { NativeConfig } from '../../../src/runtime/config'; -type ResponseValue = JsonObject[] | { status: number; body: string }; +type ResponseValue = + | JsonObject[] + | { status: number; body: string; headers?: Record }; type Response = | ResponseValue | ((request: JsonObject) => ResponseValue | Promise); @@ -65,7 +67,7 @@ export class ModelServer { const output = typeof selected === 'function' ? await selected(requestBody) : selected; if (!Array.isArray(output)) { - response.writeHead(output.status).end(output.body); + response.writeHead(output.status, output.headers).end(output.body); return; } const id = `response-${requestIndex + 1}`; diff --git a/packages/agents-codex/test/native/fixtures/tracing-parity-child.ts b/packages/agents-codex/test/native/fixtures/tracing-parity-child.ts index 44e1ebcbc..e2e243a0e 100644 --- a/packages/agents-codex/test/native/fixtures/tracing-parity-child.ts +++ b/packages/agents-codex/test/native/fixtures/tracing-parity-child.ts @@ -1159,6 +1159,41 @@ async function credentialsCase() { await telemetry.shutdown(); console.log(JSON.stringify({ traceIds })); } +async function preparedRequestCase() { + const server = new ModelServer([[message('traced answer')]]); + const config = await server.config(); + await using provider = new NativeProvider({ + baseUrl: (config['model_providers.fixture'] as { base_url: string }) + .base_url, + apiKey: 'synthetic-editor-trace', + supportsWebsockets: false, + }); + const agent = new Agent({ + instructions: 'Answer.', + model: 'gpt-5.5', + provider, + }); + let callbacks = 0; + await telemetry.withTrace( + { name: 'prepared-request-workflow' }, + async (workflow) => { + const result = await agent.run('answer', { + prepareRequest() { + callbacks++; + assert.equal(telemetry.getCurrentTrace(), workflow); + const callback = telemetry.getCurrentSpan(); + assert.equal(callback?.context.traceId, workflow.context.traceId); + assert.notEqual(callback?.context.spanId, workflow.context.spanId); + return undefined; + }, + }); + assert.equal(result.finalOutput, 'traced answer'); + }, + ); + assert.equal(callbacks, 1); + assert.equal(server.requests.length, 1); +} + async function main() { const mode = process.argv[2]; if (mode === 'custom-only') { @@ -1247,6 +1282,7 @@ async function main() { else if (mode === 'control') await controlCase(collector.batches); else if (mode === 'processors') await processorCase(collector.batches); else if (mode === 'restore') await restoreCase(collector.batches); + else if (mode === 'model-preparation') await preparedRequestCase(); else if (mode === 'parent-policy') await parentPolicyCase(collector); else if (mode === 'retained') await retainedCase(collector.batches, collector.authorization); diff --git a/packages/agents-codex/test/native/prepared-request-native.test.ts b/packages/agents-codex/test/native/prepared-request-native.test.ts new file mode 100644 index 000000000..fd3fb0fc6 --- /dev/null +++ b/packages/agents-codex/test/native/prepared-request-native.test.ts @@ -0,0 +1,1438 @@ +import { afterEach, expect, it, vi } from 'vitest'; +import { z } from 'zod'; +import { createServer } from 'node:http'; +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { + jsonObject, + type JsonObject, + type NativeThread, +} from '../../src/runtime/binding'; +import { + Agent, + NativeProvider, + Runner, + ToolOutput, + tool, + manualTool, + toolRef, + HostedVerificationMCP, + InteractionRequired, + HttpMCP, + localStorage, +} from '@openai/agents-codex'; +import type { PreparedRequestData } from '@openai/agents-codex'; +import { ModelServer, call, message } from './fixtures/model-server'; + +// This fixture observes actual HTTP bytes and native tool execution, not a mock model loop. +const closePeers: (() => Promise)[] = []; +afterEach(async () => { + await Promise.all(closePeers.splice(0).map((close) => close())); + await ModelServer.closeAll(); +}); +async function peer() { + const calls: JsonObject[] = []; + const effects: unknown[] = []; + const server = createServer(async (request, response) => { + if (request.method !== 'POST') { + response.writeHead(405).end(); + return; + } + const chunks: Buffer[] = []; + for await (const chunk of request) chunks.push(Buffer.from(chunk)); + const rpc = jsonObject.parse(JSON.parse(Buffer.concat(chunks).toString())); + if (rpc.id === undefined) { + response.writeHead(202).end(); + return; + } + let result: JsonObject = {}; + if (rpc.method === 'initialize') + result = { + protocolVersion: jsonObject.parse(rpc.params).protocolVersion, + capabilities: { tools: {} }, + serverInfo: { name: 'editor-peer', version: '1' }, + }; + else if (rpc.method === 'tools/list') + result = { + tools: [ + { + name: 'lookup', + description: 'Original remote description.', + inputSchema: { + type: 'object', + properties: { value: { type: 'string', enum: ['allowed'] } }, + required: ['value'], + additionalProperties: false, + }, + }, + ], + }; + else if (rpc.method === 'tools/call') { + calls.push(jsonObject.parse(rpc.params)); + effects.push(calls.at(-1)!.arguments); + result = { content: [{ type: 'text', text: 'MCP_RESULT' }] }; + } + response + .writeHead(200, { 'content-type': 'application/json' }) + .end(JSON.stringify({ jsonrpc: '2.0', id: rpc.id, result })); + }); + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + closePeers.push(async () => { + server.closeAllConnections(); + await new Promise((resolve, reject) => + server.close((error) => (error ? reject(error) : resolve())), + ); + }); + const address = server.address(); + if (!address || typeof address === 'string') + throw new Error('Missing peer address'); + return { url: `http://127.0.0.1:${address.port}/mcp`, calls, effects }; +} +async function providerFor(server: ModelServer) { + const config = await server.config(); + return new NativeProvider({ + baseUrl: (config['model_providers.fixture'] as { base_url: string }) + .base_url, + apiKey: 'synthetic-editor', + supportsWebsockets: false, + }); +} + +it('combines prepared edits, argument correction and result shaping through public Runner.run', async () => { + const remote = await peer(); + const effects: string[] = []; + const stages: string[] = []; + const reserve = tool({ + name: 'reserve', + description: 'Original declaration.', + parameters: z.object({ value: z.enum(['allowed', 'corrected']) }), + before: ({ callId, originalArguments }) => { + expect(callId).toBe('first'); + expect(originalArguments).toEqual({ value: 'allowed' }); + stages.push('before'); + return { + action: 'replaceArguments', + arguments: { value: 'corrected' as const }, + }; + }, + execute: ({ value }) => { + stages.push('execute'); + effects.push(value); + return 'RAW_FUNCTION_RESULT'; + }, + after: (output, { callId, effectiveArguments }) => { + expect(callId).toBe('first'); + expect(effectiveArguments).toEqual({ value: 'corrected' }); + expect(output.contentItems).toEqual([ + { type: 'inputText', text: 'RAW_FUNCTION_RESULT' }, + ]); + stages.push('after'); + return new ToolOutput([ + { type: 'inputText', text: 'SHAPED_FUNCTION_RESULT' }, + ]); + }, + }); + const mcp = new HttpMCP({ + name: 'records', + url: remote.url, + tools: ['lookup'], + approvalMode: 'approve', + approval: ({ call, effectiveArguments }) => { + expect(call.callId).toBe('remote'); + expect(effectiveArguments).toEqual({ value: 'allowed' }); + stages.push('mcp-approval'); + return effectiveArguments.value === 'allowed'; + }, + }); + const server = new ModelServer([ + [call('reserve', { value: 'allowed' }, { callId: 'first' })], + [ + call( + 'lookup', + { value: 'allowed' }, + { callId: 'remote', namespace: 'mcp__records' }, + ), + ], + [message('done')], + ]); + await using provider = await providerFor(server); + const snapshots: PreparedRequestData[] = []; + const context = { callback: () => 'resource' }; + const result = await Runner.run( + new Agent({ + model: 'gpt-5.5', + instructions: 'Original instructions.', + provider, + tools: [reserve, mcp], + }), + 'go', + { + context, + returnState: true, + prepareRequest({ + request, + preparationIndex, + recoveryAttempt, + context: supplied, + }) { + expect(supplied).toBe(context); + expect(recoveryAttempt).toBe(0); + expect(preparationIndex).toBe(snapshots.length); + snapshots.push(structuredClone(request)); + const input = z.array(jsonObject).parse(request.input); + if ( + !input.some( + (item) => + item.type === 'function_call_output' && item.call_id === 'first', + ) + ) + return; + request.instructions = 'Edited after the tool.'; + request.input.push({ + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'sampling-only-note' }], + }); + for (const presentation of request.tools) { + presentation.description = 'Sampled widened presentation.'; + presentation.parameters = { + type: 'object', + properties: { + value: { + type: 'string', + enum: ['allowed', 'corrected', 'forbidden'], + }, + }, + required: ['value'], + additionalProperties: false, + }; + } + return request; + }, + }, + ); + expect(result.finalOutput).toBe('done'); + expect(effects).toEqual(['corrected']); + expect(stages).toEqual(['before', 'execute', 'after', 'mcp-approval']); + expect(remote.calls).toHaveLength(1); + expect(remote.calls[0].name).toBe('lookup'); + expect(remote.calls[0].arguments).toEqual({ value: 'allowed' }); + expect(remote.effects).toEqual([{ value: 'allowed' }]); + expect(snapshots).toHaveLength(3); + const functionOutput = z + .array(jsonObject) + .parse(server.requests[1].input) + .filter( + (item) => + item.type === 'function_call_output' && item.call_id === 'first', + ); + expect(functionOutput).toHaveLength(1); + expect(JSON.stringify(functionOutput[0].output)).toContain( + 'SHAPED_FUNCTION_RESULT', + ); + expect(JSON.stringify(functionOutput[0].output)).not.toContain( + 'RAW_FUNCTION_RESULT', + ); + const mcpOutput = z + .array(jsonObject) + .parse(server.requests[2].input) + .filter( + (item) => + item.type === 'function_call_output' && item.call_id === 'remote', + ); + expect(mcpOutput).toHaveLength(1); + expect(JSON.stringify(mcpOutput[0].output)).toContain('MCP_RESULT'); + for (const request of server.requests.slice(1)) { + expect(JSON.stringify(request)).toContain('sampling-only-note'); + expect(JSON.stringify(request.input)).toContain('Edited after the tool.'); + const selected = JSON.stringify(request.tools); + expect(selected).toContain('Sampled widened presentation.'); + expect(selected).toContain('reserve'); + expect(selected).toContain('lookup'); + expect(selected).toContain('forbidden'); + expect(request).not.toHaveProperty('previous_response_id'); + } + expect(JSON.stringify(result.state)).not.toContain('sampling-only-note'); + expect(reserve.description).toBe('Original declaration.'); +}); + +it.each(['none', 'description', 'partial'] as const)( + 'preserves raw managed schemas through %s preparation', + async (edit) => { + const execute = vi.fn(() => 'unused'); + const lookup = tool({ + name: 'lookup', + description: 'Original lookup.', + parameters: z.object({ value: z.string().min(3) }), + execute, + }); + const server = new ModelServer([[message('done')]]); + await using provider = await providerFor(server); + const agent = new Agent({ + model: 'fixture-standard', + instructions: 'Answer.', + provider, + tools: [lookup], + }); + const result = await agent.run('go', { + prepareRequest({ request }) { + expect(request.tools[0].parameters).toMatchObject({ + properties: { value: { minLength: 3 } }, + }); + if (edit === 'description') + request.tools[0].description = 'Edited lookup.'; + if (edit === 'partial') { + const properties = jsonObject.parse( + request.tools[0].parameters.properties, + ); + request.tools[0].parameters = { + ...request.tools[0].parameters, + properties: { + ...properties, + value: { ...jsonObject.parse(properties.value), maxLength: 9 }, + }, + }; + } + return request; + }, + }); + expect(result.finalOutput).toBe('done'); + expect(server.requests).toHaveLength(1); + const declarations = JSON.stringify(server.requests[0].tools); + expect(declarations).toContain('"minLength":3'); + if (edit === 'partial') expect(declarations).toContain('"maxLength":9'); + if (edit === 'description') + expect(declarations).toContain('Edited lookup.'); + expect(execute).not.toHaveBeenCalled(); + }, +); + +it('keeps the original function validator when a sampled enum is widened', async () => { + const execute = vi.fn(() => 'effect'); + const reserve = tool({ + name: 'reserve', + parameters: z.object({ value: z.enum(['allowed']) }), + execute, + }); + const server = new ModelServer([ + [call('reserve', { value: 'forbidden' }, { callId: 'widened' })], + ]); + await using provider = await providerFor(server); + const agent = new Agent({ + model: 'gpt-5.5', + instructions: 'Use reserve.', + provider, + tools: [reserve], + }); + await expect( + agent.run('go', { + prepareRequest({ request }) { + request.tools[0].parameters = { + type: 'object', + properties: { + value: { type: 'string', enum: ['allowed', 'forbidden'] }, + }, + required: ['value'], + additionalProperties: false, + }; + return request; + }, + }), + ).rejects.toThrow(); + expect(execute).not.toHaveBeenCalled(); + expect(server.requests).toHaveLength(1); + expect(JSON.stringify(server.requests[0])).toContain('forbidden'); +}, 10_000); + +it('fails callback errors before dispatch, then permits a fresh operation', async () => { + const server = new ModelServer([[message('fresh')]]); + await using provider = await providerFor(server); + await using session = await new Agent({ + model: 'gpt-5.5', + instructions: 'Answer.', + provider, + }).session(); + const failed = await session.submit('first', { + prepareRequest() { + throw new Error('synthetic callback failure'); + }, + }); + await expect(failed.wait()).rejects.toThrow('synthetic callback failure'); + expect(server.requests).toHaveLength(0); + const fresh = await session.submit('next', { + prepareRequest: () => undefined, + }); + expect((await fresh.wait()).finalOutput).toBe('fresh'); +}); + +it('joins a cancelled uncooperative callback before admitting the next root', async () => { + const server = new ModelServer([[message('fresh')]]); + await using provider = await providerFor(server); + await using session = await new Agent({ + model: 'gpt-5.5', + instructions: 'Answer.', + provider, + }).session(); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + let signal!: AbortSignal; + const first = await session.submit('old', { + async prepareRequest(context) { + signal = context.signal; + entered.resolve(); + await release.promise; + return context.request; + }, + }); + let cancellation: Promise | undefined; + try { + await entered.promise; + let joined = false; + cancellation = first.cancel().then(() => { + joined = true; + }); + await vi.waitFor(() => expect(signal.aborted).toBe(true)); + expect(joined).toBe(false); + expect(server.requests).toHaveLength(0); + } finally { + release.resolve(); + await cancellation; + await first.cancel(); + } + await expect(first.wait()).rejects.toThrow(); + expect(server.requests).toHaveLength(0); + const next = await session.submit('fresh', { + prepareRequest: () => undefined, + }); + expect((await next.wait()).finalOutput).toBe('fresh'); +}); + +it.each(['empty', 'eager-reference'] as const)( + 'prepares the selected %s roster when the opening catalog has a deferred tool', + async (selection) => { + const eager = tool({ + name: 'eager', + parameters: z.object({}), + execute: () => 'unused', + }); + const deferred = tool({ + name: 'deferred', + parameters: z.object({}), + deferLoading: true, + execute: () => 'unused', + }); + const server = new ModelServer([[message('done')]]); + await using provider = await providerFor(server); + await using session = await new Agent({ + model: 'gpt-5.5', + instructions: 'Answer.', + provider, + tools: [eager, deferred], + }).session(); + const editor = vi.fn(({ request }: { request: PreparedRequestData }) => { + expect(request.tools.map((tool) => tool.name)).toEqual( + selection === 'empty' ? [] : ['eager'], + ); + return request; + }); + const turn = await session.submit('go', { + tools: selection === 'empty' ? [] : [toolRef('eager')], + prepareRequest: editor, + }); + expect((await turn.wait()).finalOutput).toBe('done'); + expect(editor).toHaveBeenCalledTimes(1); + expect(server.requests[0].tools).toEqual( + selection === 'empty' + ? [] + : [expect.objectContaining({ type: 'function', name: 'eager' })], + ); + }, +); + +it('refuses a selected deferred tool before the editor, HTTP and tool effects', async () => { + const execute = vi.fn(() => 'unused'); + const deferred = tool({ + name: 'deferred', + parameters: z.object({}), + deferLoading: true, + execute, + }); + const server = new ModelServer([]); + await using provider = await providerFor(server); + await using session = await new Agent({ + model: 'gpt-5.5', + instructions: 'Answer.', + provider, + tools: [deferred], + }).session(); + const editor = vi.fn(() => undefined); + const turn = await session.submit('go', { + tools: [toolRef('deferred')], + prepareRequest: editor, + }); + await expect(turn.wait()).rejects.toThrow(/prepare_request.*eager/i); + expect(editor).not.toHaveBeenCalled(); + expect(execute).not.toHaveBeenCalled(); + expect(server.requests).toHaveLength(0); +}); + +it('checks the roster after beforeStep and preserves effects before a later deferred selection', async () => { + const execute = vi.fn(() => 'original effect'); + const eager = tool({ name: 'eager', parameters: z.object({}), execute }); + const deferredExecute = vi.fn(() => 'unused'); + const deferred = tool({ + name: 'deferred', + parameters: z.object({}), + deferLoading: true, + execute: deferredExecute, + }); + const server = new ModelServer([[call('eager', {}, { callId: 'effect' })]]); + await using provider = await providerFor(server); + await using session = await new Agent({ + model: 'gpt-5.5', + instructions: 'Call eager.', + provider, + tools: [eager, deferred], + }).session(); + const editor = vi.fn(({ request }: { request: PreparedRequestData }) => { + expect(request.tools.map((tool) => tool.name)).toEqual(['eager']); + return request; + }); + const beforeStep = vi.fn(({ stepIndex }: { stepIndex: number }) => [ + toolRef(stepIndex === 0 ? 'eager' : 'deferred'), + ]); + const turn = await session.submit('go', { + beforeStep, + prepareRequest: editor, + }); + await expect(turn.wait()).rejects.toThrow(/prepare_request.*eager/i); + expect(beforeStep).toHaveBeenCalledTimes(2); + expect(editor).toHaveBeenCalledTimes(1); + expect(execute).toHaveBeenCalledTimes(1); + expect(deferredExecute).not.toHaveBeenCalled(); + expect(server.requests).toHaveLength(1); + expect(server.requests[0].tools).toEqual([ + expect.objectContaining({ type: 'function', name: 'eager' }), + ]); +}); + +it('keeps the operation ceiling when beforeStep tries to select an excluded deferred tool', async () => { + const eager = tool({ + name: 'eager', + parameters: z.object({}), + execute: () => 'unused', + }); + const deferred = tool({ + name: 'deferred', + parameters: z.object({}), + deferLoading: true, + execute: () => 'unused', + }); + const server = new ModelServer([]); + await using provider = await providerFor(server); + await using session = await new Agent({ + model: 'gpt-5.5', + instructions: 'Answer.', + provider, + tools: [eager, deferred], + }).session(); + const editor = vi.fn(() => undefined); + const turn = await session.submit('go', { + tools: [toolRef('eager')], + beforeStep: () => [toolRef('deferred')], + prepareRequest: editor, + }); + await expect(turn.wait()).rejects.toThrow(); + expect(editor).not.toHaveBeenCalled(); + expect(server.requests).toHaveLength(0); +}); + +it('keeps fixed hosted MCP registration outside the editor profile even with no selected tools', async () => { + const remote = await peer(); + const hosted = new HostedVerificationMCP({ + url: remote.url, + tools: ['lookup'], + deferLoading: false, + }); + const server = new ModelServer([]); + await using provider = await providerFor(server); + await using session = await new Agent({ + model: 'gpt-5.5', + instructions: 'Answer.', + provider, + tools: [hosted], + }).session(); + const editor = vi.fn(() => undefined); + await expect( + session.submit('go', { tools: [], prepareRequest: editor }), + ).rejects.toThrow(/eager function/i); + expect(editor).not.toHaveBeenCalled(); + expect(remote.effects).toHaveLength(0); + expect(server.requests).toHaveLength(0); +}); + +it.each([false, true])( + 'requires a one-shot interaction handler with an editor and manual tools (handler=%s)', + async (handled) => { + const approval = manualTool({ + name: 'approval', + description: 'Wait.', + parameters: z.object({}), + result: z.string(), + }); + const server = new ModelServer([ + [call('approval', {}, { callId: 'manual' })], + [message('continued')], + ]); + await using provider = await providerFor(server); + const editor = vi.fn(() => undefined); + const result = new Agent({ + model: 'gpt-5.5', + instructions: 'Ask.', + provider, + tools: [approval], + }).run('go', { + prepareRequest: editor, + ...(handled + ? { + onInteraction: (action: Parameters[0]) => + approval.match(action)!.reply('approved'), + } + : {}), + }); + if (handled) { + await expect(result).resolves.toMatchObject({ finalOutput: 'continued' }); + expect(editor).toHaveBeenCalledTimes(2); + expect(server.requests).toHaveLength(2); + } else { + await expect(result).rejects.toBeInstanceOf(InteractionRequired); + expect(editor).not.toHaveBeenCalled(); + expect(server.requests).toHaveLength(0); + } + }, +); + +it('rejects a new request editor before restoring an admitted manual turn', async () => { + const approval = manualTool({ + name: 'approval', + description: 'Wait.', + parameters: z.object({}), + result: z.string(), + }); + const server = new ModelServer([ + [call('approval', {}, { callId: 'manual' })], + [message('continued')], + ]); + await using provider = await providerFor(server); + const agent = new Agent({ + model: 'gpt-5.5', + instructions: 'Ask.', + provider, + tools: [approval], + }); + await using session = await agent.session(); + const turn = await session.submit('go'); + const pending = await turn.wait({ until: 'request' }); + expect(pending).toHaveProperty('requests'); + const state = await session.suspend(); + await session.close(); + const editor = vi.fn(() => undefined); + const onInteraction = vi.fn((action: Parameters[0]) => + approval.match(action)!.reply('approved'), + ); + await expect( + agent.run(null, { state, prepareRequest: editor, onInteraction }), + ).rejects.toThrow('Cannot add prepareRequest when resuming a run'); + expect(editor).not.toHaveBeenCalled(); + expect(onInteraction).not.toHaveBeenCalled(); + expect(server.requests).toHaveLength(1); + await expect( + agent.run(null, { state, onInteraction }), + ).resolves.toMatchObject({ + finalOutput: 'continued', + }); + expect(server.requests).toHaveLength(2); + expect(JSON.stringify(server.requests[1]!.input)).toContain('approved'); +}); + +it('refuses active manual checkpoint readiness and export after preparation settles', async () => { + const approval = manualTool({ + name: 'approval', + description: 'Wait.', + parameters: z.object({}), + result: z.string(), + }); + const server = new ModelServer([ + [call('approval', {}, { callId: 'manual' })], + [message('continued')], + ]); + await using provider = await providerFor(server); + await using session = await new Agent({ + model: 'gpt-5.5', + instructions: 'Ask.', + provider, + tools: [approval], + }).session(); + const indices: number[] = []; + const turn = await session.submit('go', { + prepareRequest: ({ preparationIndex }) => { + indices.push(preparationIndex); + return undefined; + }, + }); + const pending = await turn.wait({ until: 'request' }); + if (!('requests' in pending)) throw new Error('Expected manual request'); + expect((await session.checkpointStatus({ kind: 'manualWait' })).status).toBe( + 'notReady', + ); + await expect(session.suspend()).rejects.toThrow( + 'callback-selected operation', + ); + await session.respond(approval.match(pending.requests[0])!.reply('approved')); + expect((await turn.wait()).finalOutput).toBe('continued'); + expect(indices).toEqual([0, 1]); +}); + +it('edits the combined function/MCP conversation after both results while original MCP policy blocks widened arguments before RPC', async () => { + const remote = await peer(); + const approvals: unknown[] = []; + const mcp = new HttpMCP({ + name: 'records', + url: remote.url, + tools: ['lookup'], + approvalMode: 'approve', + approval: ({ effectiveArguments }) => { + approvals.push(effectiveArguments.value); + return effectiveArguments.value === 'allowed' + ? true + : { action: 'stop' as const }; + }, + }); + let effects = 0; + const local = tool({ + name: 'reserve', + parameters: z.object({ value: z.enum(['allowed']) }), + execute: () => { + effects++; + return 'FUNCTION_RESULT'; + }, + }); + const server = new ModelServer([ + [call('reserve', { value: 'allowed' }, { callId: 'local' })], + [ + call( + 'lookup', + { value: 'allowed' }, + { namespace: 'mcp__records', callId: 'remote' }, + ), + ], + [ + call( + 'lookup', + { value: 'forbidden' }, + { namespace: 'mcp__records', callId: 'denied' }, + ), + ], + [message('done')], + ]); + await using provider = await providerFor(server); + await using session = await new Agent({ + model: 'gpt-5.5', + instructions: 'Use the selected tools.', + provider, + tools: [local, mcp], + }).session(); + const turn = await session.submit('go', { + prepareRequest({ request, preparationIndex }) { + if (preparationIndex < 2) return; + const history = JSON.stringify(request.input); + expect(history).toContain('FUNCTION_RESULT'); + expect(history).toContain('MCP_RESULT'); + request.instructions = 'Edited combined request.'; + request.input.push({ + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'sampling-only-combined-note' }], + }); + for (const presentation of request.tools.filter((tool) => + ['reserve', 'lookup'].includes(tool.name), + )) { + presentation.description = 'Edited selected description.'; + presentation.parameters = { + type: 'object', + properties: { + value: { type: 'string', enum: ['allowed', 'forbidden'] }, + }, + required: ['value'], + additionalProperties: false, + }; + } + return request; + }, + }); + await expect(turn.wait()).rejects.toThrow(); + expect(effects).toBe(1); + expect(approvals).toEqual(['allowed', 'forbidden']); + expect(remote.calls).toHaveLength(1); + expect(server.requests).toHaveLength(3); + const edited = JSON.stringify(server.requests[2]); + expect(edited).toContain('sampling-only-combined-note'); + expect(edited).toContain('Edited selected description.'); + expect(edited).toContain('forbidden'); + expect(server.requests[2]).not.toHaveProperty('previous_response_id'); +}); + +it('uses one captured edit for an HTTP retry and detaches retained callback objects', async () => { + const server = new ModelServer([ + { status: 500, body: 'synthetic retry' }, + [message('done')], + ]); + await using provider = await providerFor(server); + const lookup = tool({ + name: 'lookup', + description: 'Original lookup.', + parameters: z.object({ value: z.string().min(3) }), + execute: () => 'unused', + }); + let preparations = 0; + let retained!: PreparedRequestData; + const result = await new Agent({ + model: 'gpt-5.5', + instructions: 'Answer.', + provider, + tools: [lookup], + }).run('go', { + prepareRequest({ request }) { + preparations++; + request.instructions = 'Captured edit.'; + request.tools[0].description = 'Captured description.'; + retained = request; + setTimeout(() => { + retained.instructions = 'Late retained mutation.'; + retained.tools[0].description = 'Late description mutation.'; + }, 0); + return request; + }, + }); + expect(result.finalOutput).toBe('done'); + expect(preparations).toBe(1); + expect(server.requests).toHaveLength(2); + for (const request of server.requests) { + expect(JSON.stringify(request)).toContain('Captured edit.'); + expect(JSON.stringify(request)).not.toContain('Late retained mutation.'); + expect(JSON.stringify(request.tools)).toContain('Captured description.'); + expect(JSON.stringify(request.tools)).toContain('minLength'); + expect(JSON.stringify(request.tools)).not.toContain( + 'Late description mutation.', + ); + } +}); + +it('exports idle canonical history and requires a fresh editor after restore', async () => { + const server = new ModelServer([[message('first')], [message('second')]]); + await using provider = await providerFor(server); + const agent = new Agent({ + model: 'gpt-5.5', + instructions: 'Answer.', + provider, + }); + const first = await agent.run('go', { + returnState: true, + prepareRequest({ request }) { + request.input.push({ + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'ephemeral-note' }], + }); + return request; + }, + }); + expect(JSON.stringify(first.state)).not.toContain('ephemeral-note'); + const editor = vi.fn(() => undefined); + const second = await agent.run('next', { + state: first.state, + prepareRequest: editor, + }); + expect(second.finalOutput).toBe('second'); + expect(editor).toHaveBeenCalledTimes(1); + expect(JSON.stringify(server.requests[1])).not.toContain('ephemeral-note'); +}); + +it.each([true, false])( + 'refuses hidden model-change compaction only with editor=%s', + async (edited) => { + const server = new ModelServer([ + [message('original')], + [message('summary')], + [message('changed')], + ]); + await using provider = await providerFor(server); + await using session = await new Agent({ + instructions: 'Answer.', + provider, + }).session(); + await (await session.submit('establish history')).wait(); + await session.update({ model: 'gpt-5.5' }); + const editor = vi.fn(() => undefined); + const turn = await session.submit( + 'change model', + edited ? { prepareRequest: editor } : undefined, + ); + if (edited) { + await expect(turn.wait()).rejects.toThrow(/compaction/i); + const terminals = []; + for await (const event of turn.events()) + if (event.type === 'turn_finished') terminals.push(event); + expect(terminals).toHaveLength(1); + expect(terminals[0].status).toBe('failed'); + expect(server.requests).toHaveLength(1); + expect(editor).not.toHaveBeenCalled(); + } else { + expect((await turn.wait()).finalOutput).toBe('changed'); + expect(server.requests).toHaveLength(3); + } + }, +); + +it.each([true, false])( + 'refuses post-tool rollover before hidden dispatch only with editor=%s', + async (edited) => { + // The native factory is the existing low-level capacity seam. The public operation and all HTTP/tool work remain real. + const { binding } = await import('../../src/runtime/binding'); + const native = await binding(); + const nativeErrors: unknown[] = []; + const nextEvent = native.NativeThread.prototype._next_session_event; + const observer = vi + .spyOn(native.NativeThread.prototype, '_next_session_event') + .mockImplementation(async function (this: NativeThread) { + const raw = await nextEvent.call(this); + const event = JSON.parse(raw); + if ( + event.msg.type === 'error' && + String(event.msg.message).includes('automatic compaction') + ) + nativeErrors.push(event.msg); + return raw; + }); + const create = native.create_engine; + const spy = vi + .spyOn(native, 'create_engine') + .mockImplementation((config, ...args) => { + const parsed = JSON.parse(config); + return create( + JSON.stringify({ + ...parsed, + model_context_window: 200000, + model_auto_compact_token_limit: 10000, + }), + ...args, + ); + }); + let effects = 0; + const reserve = tool({ + name: 'reserve', + parameters: z.object({}), + execute: () => { + effects++; + return 'completed tool effect'; + }, + }); + const server = new ModelServer( + [ + [call('reserve', {}, { callId: 'effect' })], + [message('summary')], + [message('done')], + ], + ['{"input_tokens":12000,"output_tokens":1,"total_tokens":12001}'], + ); + try { + await using provider = await providerFor(server); + await using session = await new Agent({ + model: 'gpt-5.5', + instructions: 'Use reserve once.', + provider, + tools: [reserve], + }).session(); + const editor = vi.fn(() => undefined); + const turn = await session.submit( + 'go', + edited ? { prepareRequest: editor } : undefined, + ); + if (edited) { + await expect(turn.wait()).rejects.toThrow(/compaction/i); + const terminals = []; + for await (const event of turn.events()) + if (event.type === 'turn_finished') terminals.push(event); + expect(terminals).toHaveLength(1); + expect(terminals[0].status).toBe('failed'); + expect(nativeErrors).toHaveLength(1); + expect(server.requests).toHaveLength(1); + expect(editor).toHaveBeenCalledTimes(1); + } else { + expect((await turn.wait()).finalOutput).toBe('done'); + expect(nativeErrors).toHaveLength(0); + expect(server.requests).toHaveLength(3); + } + expect(effects).toBe(1); + } finally { + observer.mockRestore(); + spy.mockRestore(); + } + }, +); + +it.each([true, false])( + 'preserves the completed answer at the post-turn threshold with editor=%s', + async (edited) => { + const { binding } = await import('../../src/runtime/binding'); + const native = await binding(); + const create = native.create_engine; + const spy = vi + .spyOn(native, 'create_engine') + .mockImplementation((config, ...args) => + create( + JSON.stringify({ + ...JSON.parse(config), + model_context_window: 200000, + model_auto_compact_token_limit: 10000, + model_post_turn_compact_threshold_percent: 1, + }), + ...args, + ), + ); + const server = new ModelServer( + [[message('completed answer')], [message('summary')]], + ['{"input_tokens":12000,"output_tokens":1,"total_tokens":12001}'], + ); + try { + await using provider = await providerFor(server); + const editor = vi.fn(() => undefined); + const result = await new Agent({ + model: 'fixture-standard', + instructions: 'Answer.', + provider, + }).run('go', edited ? { prepareRequest: editor } : undefined); + expect(result.finalOutput).toBe('completed answer'); + expect(server.requests).toHaveLength(edited ? 1 : 2); + expect(editor).toHaveBeenCalledTimes(edited ? 1 : 0); + } finally { + spy.mockRestore(); + } + }, +); + +it('rejects a retired native reply capability while a later root awaits its own callback', async () => { + const { binding } = await import('../../src/runtime/binding'); + const native = await binding(); + const original = native.NativeThread.prototype._reply_prepared_request; + let retained: + | { owner: typeof native.NativeThread.prototype; receipt: string } + | undefined; + const spy = vi + .spyOn(native.NativeThread.prototype, '_reply_prepared_request') + .mockImplementation(function ( + this: typeof native.NativeThread.prototype, + receipt, + ) { + retained ??= { owner: this, receipt }; + return original.call(this, receipt); + }); + const release = Promise.withResolvers(); + try { + const server = new ModelServer([[message('old')], [message('new')]]); + await using provider = await providerFor(server); + await using session = await new Agent({ + model: 'gpt-5.5', + instructions: 'Answer.', + provider, + }).session(); + await ( + await session.submit('old', { prepareRequest: () => undefined }) + ).wait(); + const entered = Promise.withResolvers(); + const next = await session.submit('new', { + async prepareRequest() { + entered.resolve(); + await release.promise; + return undefined; + }, + }); + try { + await entered.promise; + expect(retained).toBeDefined(); + expect(() => original.call(retained!.owner, retained!.receipt)).toThrow( + /Expired prepared request/, + ); + expect(server.requests).toHaveLength(1); + } finally { + release.resolve(); + } + expect((await next.wait()).finalOutput).toBe('new'); + expect(server.requests).toHaveLength(2); + } finally { + release.resolve(); + spy.mockRestore(); + } +}); + +it('reopens completed idle LocalStorage history with only a freshly supplied editor', async () => { + const home = await mkdtemp(join(tmpdir(), 'native-editor-storage-')); + try { + const storage = localStorage(join(home, 'profile')); + const server = new ModelServer([ + [message('persisted answer')], + [message('fresh answer')], + ]); + await using provider = await providerFor(server); + const agent = new Agent({ + model: 'gpt-5.5', + instructions: 'Canonical instructions.', + provider, + }); + const oldEditor = vi.fn(({ request }: { request: PreparedRequestData }) => { + request.instructions = 'Old sampling instructions.'; + request.input.push({ + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'sampling-only-local-note' }], + }); + return request; + }); + let id!: string; + let canonical!: unknown[]; + { + await using first = await agent.session({ storage }); + id = first.id; + expect( + ( + await ( + await first.submit('original user input', { + prepareRequest: oldEditor, + }) + ).wait() + ).finalOutput, + ).toBe('persisted answer'); + canonical = []; + for await (const item of first.items()) canonical.push(item); + expect(JSON.stringify(canonical)).not.toContain( + 'sampling-only-local-note', + ); + expect(JSON.stringify(canonical)).not.toContain( + 'Old sampling instructions.', + ); + } + await using reopened = await agent.session({ storage, resume: id }); + expect(reopened.id).toBe(id); + const restored = []; + for await (const item of reopened.items()) restored.push(item); + expect(restored).toEqual(canonical); + const freshEditor = vi.fn( + ({ + request, + preparationIndex, + recoveryAttempt, + }: { + request: PreparedRequestData; + preparationIndex: number; + recoveryAttempt: number; + }) => { + expect(preparationIndex).toBe(0); + expect(recoveryAttempt).toBe(0); + expect(JSON.stringify(request)).not.toContain( + 'sampling-only-local-note', + ); + expect(request.instructions).not.toContain( + 'Old sampling instructions.', + ); + request.instructions = 'Fresh sampling instructions.'; + return request; + }, + ); + const next = await reopened.submit('next user input', { + prepareRequest: freshEditor, + }); + expect((await next.wait()).finalOutput).toBe('fresh answer'); + expect(oldEditor).toHaveBeenCalledTimes(1); + expect(freshEditor).toHaveBeenCalledTimes(1); + expect(JSON.stringify(server.requests[1])).toContain('original user input'); + expect(JSON.stringify(server.requests[1])).toContain('persisted answer'); + expect(JSON.stringify(server.requests[1])).toContain( + 'Fresh sampling instructions.', + ); + expect(JSON.stringify(server.requests[1])).not.toContain( + 'sampling-only-local-note', + ); + server.assertExhausted(); + } finally { + await rm(home, { recursive: true, force: true }); + } +}); + +it('prepares again when outer stream recovery rebuilds the prompt', async () => { + // A failed event on a successful HTTP response exercises outer stream recovery. + const failure = JSON.stringify({ + type: 'response.failed', + response: { + error: { + type: 'server_error', + code: 'server_error', + message: 'retry synthetic stream', + }, + }, + }); + const server = new ModelServer([ + { + status: 200, + headers: { 'content-type': 'text/event-stream' }, + body: `data: ${failure}\n\n`, + }, + [message('recovered')], + ]); + await using provider = await providerFor(server); + const indices: [number, number][] = []; + const result = await new Agent({ + model: 'gpt-5.5', + instructions: 'Answer.', + provider, + }).run('Try', { + prepareRequest({ request, preparationIndex, recoveryAttempt }) { + indices.push([preparationIndex, recoveryAttempt]); + return request; + }, + }); + expect(result.finalOutput).toBe('recovered'); + expect(indices).toEqual([ + [0, 0], + [1, 1], + ]); + expect(server.requests).toHaveLength(2); + expect( + server.requests.every((request) => !('previous_response_id' in request)), + ).toBe(true); + server.assertExhausted(); +}, 30_000); + +it('rejects correction-enabled Session defaults and per-operation overrides before native submission', async () => { + const server = new ModelServer([]); + await using provider = await providerFor(server); + const outputType = { + mode: 'prompted' as const, + schema: z.object({ answer: z.string() }), + maxRetries: 1, + }; + const editor = vi.fn(() => undefined); + await using repairing = await new Agent({ + model: 'gpt-5.5', + instructions: 'Answer.', + provider, + outputType, + }).session(); + await expect( + repairing.submit('default', { prepareRequest: editor }), + ).rejects.toThrow('output correction attempts'); + await using plain = await new Agent({ + model: 'gpt-5.5', + instructions: 'Answer.', + provider, + }).session(); + await expect( + plain.submitWithOutput('override', outputType, { prepareRequest: editor }), + ).rejects.toThrow('output correction attempts'); + await expect( + plain.enqueue('queued override', outputType, { prepareRequest: editor }), + ).rejects.toThrow('output correction attempts'); + expect(server.requests).toHaveLength(0); + expect(editor).not.toHaveBeenCalled(); +}); + +it.each(['native', 'prompted-zero'] as const)( + 'supports edited %s output without correction requests', + async (profile) => { + const server = new ModelServer([[message('{"answer":"valid"}')]]); + await using provider = await providerFor(server); + const schema = z.object({ answer: z.string() }); + const outputType = + profile === 'native' + ? { mode: 'native' as const, schema } + : { mode: 'prompted' as const, schema, maxRetries: 0 }; + const editor = vi.fn(() => undefined); + const result = await new Agent({ + model: 'gpt-5.5', + instructions: 'Answer.', + provider, + outputType, + }).run('go', { prepareRequest: editor }); + expect(result.finalOutput).toEqual({ answer: 'valid' }); + expect(editor).toHaveBeenCalledTimes(1); + expect(server.requests).toHaveLength(1); + }, +); + +it('preserves ordinary output correction when prepareRequest is omitted', async () => { + const server = new ModelServer([ + [message('invalid output')], + [message('{"answer":"corrected"}')], + ]); + await using provider = await providerFor(server); + const outputType = { + mode: 'prompted' as const, + schema: z.object({ answer: z.string() }), + maxRetries: 1, + }; + const result = await new Agent({ + model: 'gpt-5.5', + instructions: 'Answer.', + provider, + outputType, + }).run('go'); + expect(result.finalOutput).toEqual({ answer: 'corrected' }); + expect(server.requests).toHaveLength(2); +}); + +it.each(['no-edit', 'unchanged', 'description', 'parameters', 'both'] as const)( + 'keeps clean presentation and field-specific schema fidelity for %s', + async (edit) => { + const selected = tool({ + name: 'lookup', + description: 'Original raw-schema description.', + parameters: z.object({ value: z.string().min(3) }), + execute: () => 'unused', + }); + const server = new ModelServer([[message('done')]]); + await using provider = await providerFor(server); + const replacement = { + type: 'object', + properties: { value: { type: 'string', enum: ['allowed', 'wider'] } }, + required: ['value'], + additionalProperties: false, + }; + const result = await new Agent({ + model: 'gpt-5.5', + instructions: 'Answer.', + provider, + tools: [selected], + }).run('go', { + prepareRequest({ request }) { + expect(request.tools[0].description).toBe( + 'Original raw-schema description.', + ); + expect(request.tools[0].description).not.toContain('[agents-schema:'); + if (edit === 'no-edit') return; + if (edit === 'description' || edit === 'both') + request.tools[0].description = 'Edited clean description.'; + if (edit === 'parameters' || edit === 'both') + request.tools[0].parameters = replacement; + return request; + }, + }); + expect(result.finalOutput).toBe('done'); + const declaration = z + .array(jsonObject) + .parse(server.requests[0].tools) + .find((tool) => tool.name === 'lookup'); + expect(declaration).toBeDefined(); + expect(declaration!.description).toBe( + edit === 'description' || edit === 'both' + ? 'Edited clean description.' + : 'Original raw-schema description.', + ); + expect(JSON.stringify(declaration)).not.toContain('[agents-schema:'); + expect(declaration!.parameters).toEqual( + edit === 'parameters' || edit === 'both' + ? replacement + : selected.parameters, + ); + expect(server.requests).toHaveLength(1); + }, +); + +it('preserves raw edited bounds and null schema keys while original execution validation remains authoritative', async () => { + const effects: number[] = []; + const selected = tool({ + name: 'lookup', + parameters: z.object({ value: z.number().max(10) }), + execute: ({ value }) => { + effects.push(value); + return 'effect'; + }, + }); + const replacement = { + type: 'object', + properties: { + value: { type: 'number', maximum: 20, unknown_null_key: null }, + }, + required: ['value'], + additionalProperties: false, + }; + const server = new ModelServer([ + [call('lookup', { value: 7 }, { callId: 'allowed' })], + [call('lookup', { value: 15 }, { callId: 'above-original-maximum' })], + ]); + await using provider = await providerFor(server); + const result = new Agent({ + model: 'gpt-5.5', + instructions: 'Call lookup then answer.', + provider, + tools: [selected], + }).run('go', { + prepareRequest({ request }) { + request.tools[0].parameters = structuredClone(replacement); + return request; + }, + }); + await expect(result).rejects.toMatchObject({ + issues: [ + expect.objectContaining({ + code: 'too_big', + maximum: 10, + path: ['value'], + }), + ], + }); + expect(effects).toEqual([7]); + expect(server.requests).toHaveLength(2); + for (const request of server.requests) { + expect(request.tools).toEqual([ + expect.objectContaining({ + name: 'lookup', + type: 'function', + parameters: replacement, + }), + ]); + expect(JSON.stringify(request.tools)).not.toContain('[agents-schema:'); + } +}); + +it('rejects a lossy ordinary MCP schema edit without a managed raw receipt before HTTP or effects', async () => { + const remote = await peer(); + const selected = new HttpMCP({ + name: 'records', + url: remote.url, + tools: ['lookup'], + deferLoading: false, + }); + const server = new ModelServer([]); + await using provider = await providerFor(server); + await expect( + new Agent({ + model: 'gpt-5.5', + instructions: 'Answer.', + provider, + tools: [selected], + }).run('go', { + prepareRequest({ request }) { + request.tools[0].parameters = { + type: 'object', + properties: { + value: { type: 'number', maximum: 20, unknown_null_key: null }, + }, + required: ['value'], + }; + return request; + }, + }), + ).rejects.toThrow(/loses fields|prepared.request/i); + expect(remote.effects).toHaveLength(0); + expect(server.requests).toHaveLength(0); +}); diff --git a/packages/agents-codex/test/native/tracing-parity.test.ts b/packages/agents-codex/test/native/tracing-parity.test.ts index ae9ad87d2..598c6876e 100644 --- a/packages/agents-codex/test/native/tracing-parity.test.ts +++ b/packages/agents-codex/test/native/tracing-parity.test.ts @@ -17,6 +17,7 @@ for (const mode of [ 'processors', 'restore', 'retained', + 'model-preparation', 'parent-policy', 'no-export', 'custom-only', diff --git a/packages/agents-codex/test/prepared-request.test.ts b/packages/agents-codex/test/prepared-request.test.ts new file mode 100644 index 000000000..f42e93698 --- /dev/null +++ b/packages/agents-codex/test/prepared-request.test.ts @@ -0,0 +1,206 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; +import * as nativeBinding from '../src/runtime/binding'; +import * as turnReceipt from '../src/runtime/turn-receipt'; +import { Agent, NativeProvider, tool, HttpMCP, checklist } from '../src'; +import { z } from 'zod'; +import { + capturePreparedRequest, + preparedRequestFromNative, + type PreparedRequestData, +} from '../src/runtime/prepared-request'; + +afterEach(() => vi.restoreAllMocks()); + +const native = () => ({ + instructions: 'original', + input: [ + { + type: 'message', + role: 'user', + content: [{ type: 'input_text', text: 'hello' }], + }, + ], + tools: [ + { + namespace: null, + name: 'reserve', + kind: 'function', + strict: true, + defer_loading: null, + description: 'reserve allowed', + parameters: { + type: 'object', + properties: { value: { type: 'string', enum: ['allowed'] } }, + }, + }, + ], +}); + +describe('prepared request capture boundary', () => { + it('detaches native input and preserves public camelCase evidence through strict capture', () => { + const source = native(); + const request = preparedRequestFromNative(source); + request.instructions = 'sampling only'; + (request.input[0] as { role: string }).role = 'system'; + request.tools[0].description = 'changed'; + expect(source).toEqual(native()); + const captured = capturePreparedRequest(request); + request.instructions = 'retained mutation'; + expect(JSON.parse(captured)).toMatchObject({ + instructions: 'sampling only', + tools: [{ defer_loading: null, description: 'changed' }], + }); + expect(captured).not.toContain('deferLoading'); + }); + it.each(['extra', 'defer_loading'])( + 'rejects unknown public fields even when undefined: %s', + (key) => { + const request = preparedRequestFromNative(native()); + Object.assign(request, { [key]: undefined }); + expect(() => capturePreparedRequest(request)).toThrow( + 'Unknown prepared request field', + ); + const tool = preparedRequestFromNative(native()); + Object.assign(tool.tools[0], { [key]: undefined }); + expect(() => capturePreparedRequest(tool)).toThrow( + 'Unknown prepared request field', + ); + }, + ); + it.each([undefined, NaN, Infinity, new Date(), { toJSON: () => ({}) }])( + 'rejects lossy nested JSON: %s', + (invalid) => { + const request = preparedRequestFromNative(native()); + request.input.push({ unknown: invalid } as never); + expect(() => capturePreparedRequest(request)).toThrow(); + }, + ); + it('rejects omitted fields, cycles, sparse arrays, and non-object schemas', () => { + const request = preparedRequestFromNative(native()); + delete (request as Partial).instructions; + expect(() => capturePreparedRequest(request)).toThrow(); + const cyclic = preparedRequestFromNative(native()); + cyclic.input.push(cyclic as never); + expect(() => capturePreparedRequest(cyclic)).toThrow(); + const sparse = preparedRequestFromNative(native()); + sparse.input = new Array(2); + expect(() => capturePreparedRequest(sparse)).toThrow(); + const schema = preparedRequestFromNative(native()); + schema.tools[0].parameters = [] as never; + expect(() => capturePreparedRequest(schema)).toThrow(); + }); +}); + +it.each([ + 'websocket', + 'deferred-function', + 'deferred-mcp', + 'checklist', +] as const)( + 'rejects unsupported %s before provider acquisition or editor invocation', + async (profile) => { + const provider = new NativeProvider( + profile === 'websocket' ? {} : { supportsWebsockets: false }, + ); + const acquire = vi.spyOn(provider, '_acquire'); + const editor = vi.fn(() => undefined); + const tools = + profile === 'deferred-function' + ? [ + tool({ + name: 'lookup', + deferLoading: true, + parameters: z.object({}), + execute: () => 'unreachable', + }), + ] + : profile === 'deferred-mcp' + ? [ + new HttpMCP({ + name: 'remote', + tools: ['lookup'], + deferLoading: true, + url: 'http://127.0.0.1:1/mcp', + }), + ] + : profile === 'checklist' + ? [checklist()] + : []; + await expect( + new Agent({ instructions: 'Answer.', provider, tools }).run('go', { + prepareRequest: editor, + }), + ).rejects.toThrow( + profile === 'websocket' ? 'supportsWebsockets: false' : 'eager function', + ); + expect(acquire).not.toHaveBeenCalled(); + expect(editor).not.toHaveBeenCalled(); + await provider.close(); + }, +); + +it.each(['default', 'override'] as const)( + 'rejects resolved %s output correction before provider acquisition', + async (source) => { + const provider = new NativeProvider({ supportsWebsockets: false }); + const acquire = vi.spyOn(provider, '_acquire'); + const loadFiles = vi + .spyOn(turnReceipt, 'loadInstructionFiles') + .mockRejectedValue( + new Error('instruction loading must follow editor preflight'), + ); + const prepareRequest = vi.fn(() => undefined); + const outputType = { + mode: 'prompted' as const, + schema: z.object({ answer: z.string() }), + maxRetries: 1, + }; + const agent = new Agent({ + instructions: 'Answer.', + provider, + ...(source === 'default' ? { outputType } : {}), + }); + const view = + source === 'default' ? agent : agent.withOutputType(outputType); + await expect( + view.run('go', { + prepareRequest, + instructionFiles: ['unreachable-instructions.md'], + }), + ).rejects.toThrow('output correction attempts'); + expect(loadFiles).not.toHaveBeenCalled(); + expect(acquire).not.toHaveBeenCalled(); + expect(prepareRequest).not.toHaveBeenCalled(); + await provider.close(); + }, +); + +it.each(['native', 'prompted-zero', 'omitted-editor'] as const)( + 'retains public admission for %s output', + async (profile) => { + const provider = new NativeProvider({ supportsWebsockets: false }); + const boundary = new Error('native loader boundary'); + const loader = vi + .spyOn(nativeBinding, 'binding') + .mockRejectedValueOnce(boundary); + const acquire = vi.spyOn(provider, '_acquire'); + const outputType = + profile === 'native' + ? { mode: 'native' as const, schema: z.object({ answer: z.string() }) } + : { + mode: 'prompted' as const, + schema: z.object({ answer: z.string() }), + maxRetries: profile === 'prompted-zero' ? 0 : 1, + }; + const agent = new Agent({ instructions: 'Answer.', provider, outputType }); + await expect( + agent.run( + 'go', + profile === 'omitted-editor' ? {} : { prepareRequest: () => undefined }, + ), + ).rejects.toBe(boundary); + expect(loader).toHaveBeenCalledTimes(1); + expect(acquire).not.toHaveBeenCalled(); + await provider.close(); + }, +);