kcode_codex_runtime_v2/
runtime.rs

1use std::{collections::HashSet, path::PathBuf, process::Stdio, sync::Arc, time::Duration};
2
3use serde_json::{Value, json};
4use tokio::{
5    io::{AsyncBufReadExt, AsyncWrite, AsyncWriteExt, BufReader},
6    process::{Child, Command},
7    sync::{mpsc, watch},
8};
9
10use crate::{
11    AgentEvent, AgentRequest, CompletedTurn, DynamicToolCall, Error, ErrorKind, Result, TokenUsage,
12    ToolResult,
13};
14
15/// Default sandboxed Codex launcher.
16pub const DEFAULT_CODEX_EXECUTABLE: &str = "codex-safe";
17
18const TOOL_OUTPUT_TOKEN_LIMIT: usize = 180_000;
19
20const MAX_INPUT_CHARACTERS: usize = 1_048_576;
21const STARTUP_TIMEOUT: Duration = Duration::from_secs(30);
22
23/// Runtime configuration applied to every Codex app-server process.
24#[derive(Clone, Debug, Eq, PartialEq)]
25pub struct CodexConfig {
26    /// Codex or sandbox-launcher executable.
27    pub executable: String,
28    /// Directory supplied to Codex as its working directory.
29    pub working_directory: PathBuf,
30    /// Fixed base instruction for every fresh or resumed thread.
31    pub base_instruction: String,
32    /// Optional sanitized Codex model-catalog JSON path.
33    pub model_catalog: Option<PathBuf>,
34}
35
36impl Default for CodexConfig {
37    fn default() -> Self {
38        Self {
39            executable: DEFAULT_CODEX_EXECUTABLE.into(),
40            working_directory: std::env::temp_dir(),
41            base_instruction: String::new(),
42            model_catalog: None,
43        }
44    }
45}
46
47/// Cloneable Codex app-server runtime.
48#[derive(Clone, Debug)]
49pub struct Codex {
50    config: Arc<CodexConfig>,
51}
52
53impl Codex {
54    /// Validates configuration and requires a ChatGPT-authenticated Codex CLI.
55    pub async fn open(config: CodexConfig) -> Result<Self> {
56        validate_config(&config)?;
57        validate_chatgpt_login(&config.executable).await?;
58        Ok(Self {
59            config: Arc::new(config),
60        })
61    }
62
63    /// Starts one standard Codex turn and returns its interactive event stream.
64    pub async fn start_turn(&self, request: AgentRequest) -> Result<AgentTurn> {
65        validate_request(&request)?;
66        let child = spawn_app_server(&self.config)?;
67        let (event_sender, events) = mpsc::channel(32);
68        let (tool_results, result_receiver) = mpsc::channel(8);
69        let (cancel, cancelled) = watch::channel(false);
70        let config = self.config.clone();
71        tokio::spawn(async move {
72            let result = tokio::select! {
73                _ = cancellation(cancelled) => Err(Error::new(ErrorKind::Cancelled, "Codex turn was cancelled")),
74                result = tokio::time::timeout(request.timeout, run_protocol(child, &config, request, &event_sender, result_receiver)) => {
75                    match result {
76                        Ok(result) => result,
77                        Err(_) => Err(Error::new(ErrorKind::Timeout, "Codex turn timed out")),
78                    }
79                }
80            };
81            if let Err(error) = result {
82                let _ = event_sender.send(Err(error)).await;
83            }
84        });
85        Ok(AgentTurn {
86            events,
87            tool_results,
88            cancel,
89        })
90    }
91}
92
93/// Interactive handle for a running Codex turn.
94pub struct AgentTurn {
95    events: mpsc::Receiver<Result<AgentEvent>>,
96    tool_results: mpsc::Sender<(String, ToolResult)>,
97    cancel: watch::Sender<bool>,
98}
99
100impl AgentTurn {
101    /// Waits for the next provider-input, tool-call, or completion event.
102    pub async fn next_event(&mut self) -> Option<Result<AgentEvent>> {
103        self.events.recv().await
104    }
105
106    /// Returns the result for the currently pending dynamic tool call.
107    pub async fn respond(&self, call_id: impl Into<String>, result: ToolResult) -> Result<()> {
108        self.tool_results
109            .send((call_id.into(), result))
110            .await
111            .map_err(|_| Error::new(ErrorKind::Cancelled, "Codex turn is no longer running"))
112    }
113
114    /// Stops this turn. Dropping the handle has the same effect.
115    pub fn cancel(&self) {
116        let _ = self.cancel.send(true);
117    }
118}
119
120impl Drop for AgentTurn {
121    fn drop(&mut self) {
122        let _ = self.cancel.send(true);
123    }
124}
125
126async fn cancellation(mut cancelled: watch::Receiver<bool>) {
127    if *cancelled.borrow() {
128        return;
129    }
130    while cancelled.changed().await.is_ok() {
131        if *cancelled.borrow() {
132            return;
133        }
134    }
135}
136
137fn validate_config(config: &CodexConfig) -> Result<()> {
138    if config.executable.trim().is_empty() {
139        return Err(Error::new(
140            ErrorKind::InvalidInput,
141            "Codex executable must not be empty",
142        ));
143    }
144    if config.base_instruction.chars().count() > MAX_INPUT_CHARACTERS {
145        return Err(Error::new(
146            ErrorKind::InvalidInput,
147            "Codex base instruction is too large",
148        ));
149    }
150    Ok(())
151}
152
153fn validate_request(request: &AgentRequest) -> Result<()> {
154    if request.input.trim().is_empty() {
155        return Err(Error::new(
156            ErrorKind::InvalidInput,
157            "Codex turn input must not be empty",
158        ));
159    }
160    if request.input.chars().count() > MAX_INPUT_CHARACTERS {
161        return Err(Error::new(
162            ErrorKind::InvalidInput,
163            format!("Codex turn input exceeds {MAX_INPUT_CHARACTERS} characters"),
164        ));
165    }
166    if request.model.trim().is_empty() {
167        return Err(Error::new(
168            ErrorKind::InvalidInput,
169            "Codex model must not be empty",
170        ));
171    }
172    if request.timeout.is_zero() {
173        return Err(Error::new(
174            ErrorKind::InvalidInput,
175            "Codex timeout must be greater than zero",
176        ));
177    }
178    let mut names = HashSet::new();
179    for tool in &request.tools {
180        if tool.name.trim().is_empty() || tool.description.trim().is_empty() {
181            return Err(Error::new(
182                ErrorKind::InvalidInput,
183                "dynamic tool names and descriptions must not be empty",
184            ));
185        }
186        if !tool.input_schema.is_object() {
187            return Err(Error::new(
188                ErrorKind::InvalidInput,
189                format!(
190                    "dynamic tool '{}' must use an object JSON Schema",
191                    tool.name
192                ),
193            ));
194        }
195        if !names.insert(tool.name.as_str()) {
196            return Err(Error::new(
197                ErrorKind::InvalidInput,
198                format!("dynamic tool '{}' is duplicated", tool.name),
199            ));
200        }
201    }
202    Ok(())
203}
204
205async fn validate_chatgpt_login(executable: &str) -> Result<()> {
206    let output = tokio::time::timeout(
207        STARTUP_TIMEOUT,
208        Command::new(executable)
209            .args(["login", "status"])
210            .env_remove("OPENAI_API_KEY")
211            .env_remove("CODEX_API_KEY")
212            .output(),
213    )
214    .await
215    .map_err(|_| Error::new(ErrorKind::Unavailable, "Codex login check timed out"))?
216    .map_err(|_| {
217        Error::new(
218            ErrorKind::Unavailable,
219            format!("Codex sandbox launcher '{executable}' could not be started"),
220        )
221    })?;
222    let status = format!(
223        "{}\n{}",
224        String::from_utf8_lossy(&output.stdout),
225        String::from_utf8_lossy(&output.stderr)
226    );
227    if !output.status.success() || !status.to_ascii_lowercase().contains("chatgpt") {
228        return Err(Error::new(
229            ErrorKind::Authentication,
230            format!("'{executable}' must be logged in with ChatGPT"),
231        ));
232    }
233    Ok(())
234}
235
236fn spawn_app_server(config: &CodexConfig) -> Result<Child> {
237    app_server_command(config).spawn().map_err(|_| {
238        Error::new(
239            ErrorKind::Unavailable,
240            format!(
241                "Codex sandbox launcher '{}' could not start app-server",
242                config.executable
243            ),
244        )
245    })
246}
247
248fn app_server_command(config: &CodexConfig) -> Command {
249    let mut command = Command::new(&config.executable);
250    command
251        .arg("-c")
252        .arg("web_search=\"disabled\"")
253        .arg("-c")
254        .arg("mcp_servers={}")
255        .arg("-c")
256        .arg("features.shell_tool=false")
257        .arg("-c")
258        .arg("features.apps=false")
259        .arg("-c")
260        .arg("features.browser_use=false")
261        .arg("-c")
262        .arg("features.computer_use=false")
263        .arg("-c")
264        .arg("features.goals=false")
265        .arg("-c")
266        .arg("features.hooks=false")
267        .arg("-c")
268        .arg("features.image_generation=false")
269        .arg("-c")
270        .arg("features.multi_agent=false")
271        .arg("-c")
272        .arg("features.plugins=false")
273        .arg("-c")
274        .arg("features.tool_suggest=false")
275        .arg("-c")
276        .arg("features.remote_plugin=false")
277        .arg("-c")
278        .arg("model_auto_compact_token_limit=9223372036854775807")
279        .arg("-c")
280        .arg(format!("tool_output_token_limit={TOOL_OUTPUT_TOKEN_LIMIT}"));
281    if let Some(path) = &config.model_catalog {
282        command.arg("-c").arg(format!(
283            "model_catalog_json={}",
284            serde_json::to_string(path.to_string_lossy().as_ref())
285                .expect("serializing a path string cannot fail")
286        ));
287    }
288    command
289        .arg("app-server")
290        .arg("--stdio")
291        .current_dir(&config.working_directory)
292        .stdin(Stdio::piped())
293        .stdout(Stdio::piped())
294        .stderr(Stdio::null())
295        .env_remove("OPENAI_API_KEY")
296        .env_remove("CODEX_API_KEY")
297        .kill_on_drop(true);
298    command
299}
300
301async fn run_protocol(
302    mut child: Child,
303    config: &CodexConfig,
304    request: AgentRequest,
305    events: &mpsc::Sender<Result<AgentEvent>>,
306    mut tool_results: mpsc::Receiver<(String, ToolResult)>,
307) -> Result<()> {
308    let stdin = child.stdin.take().ok_or_else(|| {
309        Error::new(
310            ErrorKind::Unavailable,
311            "Codex app-server standard input is unavailable",
312        )
313    })?;
314    let stdout = child.stdout.take().ok_or_else(|| {
315        Error::new(
316            ErrorKind::Unavailable,
317            "Codex app-server standard output is unavailable",
318        )
319    })?;
320    let mut writer = stdin;
321    let mut lines = BufReader::new(stdout).lines();
322
323    write_record(
324        &mut writer,
325        events,
326        json!({
327            "id":1,
328            "method":"initialize",
329            "params":{
330                "clientInfo":{"name":"kcode-codex-runtime-v2","version":env!("CARGO_PKG_VERSION")},
331                "capabilities":{"experimentalApi":true}
332            }
333        }),
334    )
335    .await?;
336    let initialized = read_response(&mut lines, 1).await?;
337    require_result(&initialized, "initialize")?;
338    write_record(
339        &mut writer,
340        events,
341        json!({"method":"initialized","params":{}}),
342    )
343    .await?;
344
345    let thread_method;
346    let thread_params;
347    if let Some(thread_id) = request.previous_thread_id.as_deref() {
348        thread_method = "thread/resume";
349        thread_params = json!({
350            "threadId":thread_id,
351            "model":request.model,
352            "cwd":config.working_directory,
353            "approvalPolicy":"never",
354            "sandbox":"read-only",
355            "baseInstructions":config.base_instruction,
356        });
357    } else {
358        thread_method = "thread/start";
359        let tools = request
360            .tools
361            .iter()
362            .map(|tool| {
363                json!({
364                    "type":"function",
365                    "name":tool.name,
366                    "description":tool.description,
367                    "inputSchema":tool.input_schema,
368                })
369            })
370            .collect::<Vec<_>>();
371        thread_params = json!({
372            "model":request.model,
373            "cwd":config.working_directory,
374            "approvalPolicy":"never",
375            "sandbox":"read-only",
376            "baseInstructions":config.base_instruction,
377            "developerInstructions":"",
378            "dynamicTools":tools,
379            "ephemeral":request.ephemeral,
380            "environments":[],
381        });
382    }
383    write_record(
384        &mut writer,
385        events,
386        json!({"id":2,"method":thread_method,"params":thread_params}),
387    )
388    .await?;
389    let thread_response = read_response(&mut lines, 2).await?;
390    let thread_result = require_result(&thread_response, thread_method)?;
391    let thread_id = thread_result
392        .pointer("/thread/id")
393        .and_then(Value::as_str)
394        .ok_or_else(|| Error::new(ErrorKind::Protocol, "Codex omitted its thread ID"))?
395        .to_owned();
396
397    write_record(
398        &mut writer,
399        events,
400        json!({
401            "id":3,
402            "method":"turn/start",
403            "params":{
404                "threadId":thread_id,
405                "input":[{"type":"text","text":request.input}],
406                "effort":request.reasoning_effort.as_str(),
407                "approvalPolicy":"never",
408            }
409        }),
410    )
411    .await?;
412    let turn_response = read_response(&mut lines, 3).await?;
413    let turn_result = require_result(&turn_response, "turn/start")?;
414    let turn_id = turn_result
415        .pointer("/turn/id")
416        .and_then(Value::as_str)
417        .ok_or_else(|| Error::new(ErrorKind::Protocol, "Codex omitted its turn ID"))?
418        .to_owned();
419
420    let mut answer = String::new();
421    let mut usage = None;
422    while let Some(line) = lines.next_line().await.map_err(|_| {
423        Error::new(
424            ErrorKind::Unavailable,
425            "Codex app-server output could not be read",
426        )
427    })? {
428        let message: Value = serde_json::from_str(&line)
429            .map_err(|_| Error::new(ErrorKind::Protocol, "Codex emitted invalid JSONL"))?;
430        match message.get("method").and_then(Value::as_str) {
431            Some("item/tool/call") => {
432                let id = message.get("id").cloned().ok_or_else(|| {
433                    Error::new(
434                        ErrorKind::Protocol,
435                        "Codex tool request omitted its JSON-RPC ID",
436                    )
437                })?;
438                let params = message.get("params").ok_or_else(|| {
439                    Error::new(ErrorKind::Protocol, "Codex tool request omitted params")
440                })?;
441                let call_id = required_string(params, "callId", "Codex tool request")?;
442                let call = DynamicToolCall {
443                    call_id: call_id.clone(),
444                    tool: required_string(params, "tool", "Codex tool request")?,
445                    arguments: params.get("arguments").cloned().unwrap_or(Value::Null),
446                };
447                events
448                    .send(Ok(AgentEvent::ToolCall(call)))
449                    .await
450                    .map_err(|_| Error::new(ErrorKind::Cancelled, "turn event receiver closed"))?;
451                let (response_call_id, result) = tool_results.recv().await.ok_or_else(|| {
452                    Error::new(ErrorKind::Cancelled, "tool result channel closed")
453                })?;
454                if response_call_id != call_id {
455                    return Err(Error::new(
456                        ErrorKind::InvalidInput,
457                        format!(
458                            "tool result call ID '{response_call_id}' does not match pending call '{call_id}'"
459                        ),
460                    ));
461                }
462                write_record(
463                    &mut writer,
464                    events,
465                    json!({
466                        "id":id,
467                        "result":{
468                            "success":result.success,
469                            "contentItems":[{"type":"inputText","text":result.text}],
470                        }
471                    }),
472                )
473                .await?;
474            }
475            Some("item/completed") => {
476                if let Some(item) = message.pointer("/params/item")
477                    && item.get("type").and_then(Value::as_str) == Some("agentMessage")
478                    && item.get("phase").and_then(Value::as_str) != Some("commentary")
479                    && let Some(text) = item.get("text").and_then(Value::as_str)
480                {
481                    answer = text.to_owned();
482                }
483            }
484            Some("thread/tokenUsage/updated") => {
485                usage = parse_usage(message.pointer("/params/tokenUsage"));
486            }
487            Some("turn/completed") => {
488                if message.pointer("/params/turn/id").and_then(Value::as_str)
489                    != Some(turn_id.as_str())
490                {
491                    continue;
492                }
493                let status = message
494                    .pointer("/params/turn/status")
495                    .and_then(Value::as_str)
496                    .unwrap_or("failed");
497                if status != "completed" {
498                    let detail = message
499                        .pointer("/params/turn/error/message")
500                        .and_then(Value::as_str)
501                        .unwrap_or("Codex turn did not complete");
502                    return Err(Error::new(ErrorKind::Protocol, detail));
503                }
504                events
505                    .send(Ok(AgentEvent::Completed(CompletedTurn {
506                        thread_id,
507                        turn_id,
508                        answer,
509                        usage,
510                    })))
511                    .await
512                    .map_err(|_| Error::new(ErrorKind::Cancelled, "turn event receiver closed"))?;
513                let _ = child.kill().await;
514                return Ok(());
515            }
516            _ => {
517                if message.get("id").is_some() && message.get("method").is_some() {
518                    let id = message.get("id").cloned().unwrap_or(Value::Null);
519                    write_record(
520                        &mut writer,
521                        events,
522                        json!({"id":id,"error":{"code":-32601,"message":"Method is not supported by this client"}}),
523                    )
524                    .await?;
525                }
526            }
527        }
528    }
529    Err(Error::new(
530        ErrorKind::Unavailable,
531        "Codex app-server closed before completing the turn",
532    ))
533}
534
535async fn write_record<W: AsyncWrite + Unpin>(
536    writer: &mut W,
537    events: &mpsc::Sender<Result<AgentEvent>>,
538    value: Value,
539) -> Result<()> {
540    let mut exact = serde_json::to_string(&value).map_err(|_| {
541        Error::new(
542            ErrorKind::InvalidInput,
543            "Codex request could not be encoded",
544        )
545    })?;
546    exact.push('\n');
547    writer.write_all(exact.as_bytes()).await.map_err(|_| {
548        Error::new(
549            ErrorKind::Unavailable,
550            "Codex app-server closed its standard input",
551        )
552    })?;
553    writer.flush().await.map_err(|_| {
554        Error::new(
555            ErrorKind::Unavailable,
556            "Codex app-server input could not be flushed",
557        )
558    })?;
559    events
560        .send(Ok(AgentEvent::ProviderInput(exact)))
561        .await
562        .map_err(|_| Error::new(ErrorKind::Cancelled, "turn event receiver closed"))
563}
564
565async fn read_response<R: tokio::io::AsyncBufRead + Unpin>(
566    lines: &mut tokio::io::Lines<R>,
567    expected_id: u64,
568) -> Result<Value> {
569    while let Some(line) = lines.next_line().await.map_err(|_| {
570        Error::new(
571            ErrorKind::Unavailable,
572            "Codex app-server output could not be read",
573        )
574    })? {
575        let message: Value = serde_json::from_str(&line)
576            .map_err(|_| Error::new(ErrorKind::Protocol, "Codex emitted invalid JSONL"))?;
577        if message.get("id").and_then(Value::as_u64) == Some(expected_id) {
578            return Ok(message);
579        }
580    }
581    Err(Error::new(
582        ErrorKind::Unavailable,
583        "Codex app-server closed during startup",
584    ))
585}
586
587fn require_result<'a>(message: &'a Value, method: &str) -> Result<&'a Value> {
588    if let Some(error) = message.get("error") {
589        let detail = error
590            .get("message")
591            .and_then(Value::as_str)
592            .unwrap_or("unknown protocol error");
593        return Err(Error::new(
594            ErrorKind::Protocol,
595            format!("Codex {method} failed: {detail}"),
596        ));
597    }
598    message.get("result").ok_or_else(|| {
599        Error::new(
600            ErrorKind::Protocol,
601            format!("Codex {method} response omitted result"),
602        )
603    })
604}
605
606fn required_string(value: &Value, key: &str, label: &str) -> Result<String> {
607    value
608        .get(key)
609        .and_then(Value::as_str)
610        .map(str::to_owned)
611        .ok_or_else(|| Error::new(ErrorKind::Protocol, format!("{label} omitted {key}")))
612}
613
614fn parse_usage(value: Option<&Value>) -> Option<TokenUsage> {
615    let value = value?;
616    let total = value.get("total")?;
617    let last = value.get("last");
618    Some(TokenUsage {
619        input_tokens: nonnegative(total.get("inputTokens")),
620        output_tokens: nonnegative(total.get("outputTokens")),
621        cached_input_tokens: nonnegative(total.get("cachedInputTokens")),
622        reasoning_output_tokens: nonnegative(total.get("reasoningOutputTokens")),
623        last_input_tokens: last.map(|value| nonnegative(value.get("inputTokens"))),
624        last_output_tokens: last.map(|value| nonnegative(value.get("outputTokens"))),
625    })
626}
627
628fn nonnegative(value: Option<&Value>) -> u64 {
629    value.and_then(Value::as_i64).unwrap_or_default().max(0) as u64
630}
631
632#[cfg(test)]
633mod tests {
634    use super::*;
635    use crate::DynamicTool;
636
637    #[test]
638    fn validates_tool_names_and_schema() {
639        let mut request = AgentRequest::new("hello", "model");
640        request.tools = vec![
641            DynamicTool::new("call", "Call it", json!({"type":"object"})),
642            DynamicTool::new("call", "Call it again", json!({"type":"object"})),
643        ];
644        assert_eq!(
645            validate_request(&request).unwrap_err().kind(),
646            ErrorKind::InvalidInput
647        );
648    }
649
650    #[test]
651    fn parses_cumulative_and_last_usage() {
652        let usage = parse_usage(Some(&json!({
653            "total":{"inputTokens":20,"outputTokens":7,"cachedInputTokens":4,"reasoningOutputTokens":2},
654            "last":{"inputTokens":8,"outputTokens":3,"cachedInputTokens":1,"reasoningOutputTokens":1}
655        })))
656        .unwrap();
657        assert_eq!(usage.input_tokens, 20);
658        assert_eq!(usage.last_output_tokens, Some(3));
659    }
660
661    #[test]
662    fn app_server_overrides_tool_output_token_limit() {
663        let command = app_server_command(&CodexConfig::default());
664        let arguments = command
665            .as_std()
666            .get_args()
667            .map(|argument| argument.to_string_lossy().into_owned())
668            .collect::<Vec<_>>();
669
670        assert!(arguments.windows(2).any(|arguments| {
671            arguments[0] == "-c"
672                && arguments[1] == format!("tool_output_token_limit={TOOL_OUTPUT_TOKEN_LIMIT}")
673        }));
674    }
675
676    #[cfg(unix)]
677    #[tokio::test]
678    async fn completes_a_native_dynamic_tool_turn() {
679        use std::os::unix::fs::PermissionsExt;
680
681        let path = std::env::temp_dir().join(format!(
682            "kcode-codex-runtime-v2-test-{}",
683            std::process::id()
684        ));
685        std::fs::write(
686            &path,
687            r##"#!/bin/sh
688if [ "$1" = "login" ]; then
689  echo "Logged in using ChatGPT"
690  exit 0
691fi
692read initialize
693echo '{"id":1,"result":{"userAgent":"test","platformFamily":"unix","platformOs":"linux","codexHome":"/tmp"}}'
694read initialized
695read thread_start
696echo '{"id":2,"result":{"thread":{"id":"thread-1"}}}'
697read turn_start
698echo '{"id":3,"result":{"turn":{"id":"turn-1"}}}'
699echo '{"id":77,"method":"item/tool/call","params":{"threadId":"thread-1","turnId":"turn-1","callId":"call-1","tool":"call_ktool","arguments":{"name":"LoadNode","arguments":{"identifier":3}}}}'
700read tool_result
701echo '{"method":"item/completed","params":{"threadId":"thread-1","turnId":"turn-1","completedAtMs":1,"item":{"id":"message-1","type":"agentMessage","phase":"final_answer","text":"Finished."}}}'
702echo '{"method":"thread/tokenUsage/updated","params":{"threadId":"thread-1","turnId":"turn-1","tokenUsage":{"total":{"inputTokens":12,"outputTokens":4,"cachedInputTokens":2,"reasoningOutputTokens":1,"totalTokens":16},"last":{"inputTokens":12,"outputTokens":4,"cachedInputTokens":2,"reasoningOutputTokens":1,"totalTokens":16}}}}'
703echo '{"method":"turn/completed","params":{"threadId":"thread-1","turn":{"id":"turn-1","items":[],"status":"completed"}}}'
704"##,
705        )
706        .unwrap();
707        let mut permissions = std::fs::metadata(&path).unwrap().permissions();
708        permissions.set_mode(0o700);
709        std::fs::set_permissions(&path, permissions).unwrap();
710
711        let config = CodexConfig {
712            executable: path.to_string_lossy().into_owned(),
713            ..CodexConfig::default()
714        };
715        let codex = Codex::open(config).await.unwrap();
716        let mut request = AgentRequest::new("Exact input", "test-model");
717        request.tools.push(DynamicTool::new(
718            "call_ktool",
719            "Call one tool",
720            json!({"type":"object"}),
721        ));
722        let mut turn = codex.start_turn(request).await.unwrap();
723        let mut provider_inputs = String::new();
724        let completed = loop {
725            match turn.next_event().await.unwrap().unwrap() {
726                AgentEvent::ProviderInput(exact) => provider_inputs.push_str(&exact),
727                AgentEvent::ToolCall(call) => {
728                    assert_eq!(call.arguments["name"], "LoadNode");
729                    turn.respond(call.call_id, ToolResult::success("loaded"))
730                        .await
731                        .unwrap();
732                }
733                AgentEvent::Completed(completed) => break completed,
734            }
735        };
736        assert_eq!(completed.answer, "Finished.");
737        assert_eq!(completed.usage.unwrap().input_tokens, 12);
738        assert!(provider_inputs.contains("\"method\":\"turn/start\""));
739        assert!(provider_inputs.contains("\"success\":true"));
740        std::fs::remove_file(path).unwrap();
741    }
742}
kcode_codex_runtime_v2/runtime.rs

kcode_codex_runtime_v2/
runtime.rs
