kcode_k1_codex_websearch/
lib.rs1#![forbid(unsafe_code)]
2#![doc = include_str!("../Documentation.md")]
3#[cfg(not(unix))]
4compile_error!("kcode-k1-codex-websearch requires Unix process groups");
5
6use nix::{
7 sys::signal::{Signal, killpg},
8 unistd::Pid,
9};
10use serde_json::Value;
11use std::{ffi::OsString, io, os::unix::process::ExitStatusExt, path::PathBuf, time::Instant};
12use tokio::{
13 io::{AsyncRead, AsyncReadExt, AsyncWriteExt},
14 process::Command,
15 sync::oneshot,
16};
17
18const ARG_ERROR: &str =
19 "WebSearch failed: model or reasoning_effort is not representable as a process argument";
20const PREFLIGHT: &str =
21 "WebSearch failed: Codex compatibility check failed in kcode-k1-codex-websearch preflight";
22const EXEC_IO_ERROR: &str = "WebSearch failed: kcode-k1-codex-websearch could not start or communicate with the Codex execution subprocess";
23const EXEC_CANCELLED: &str =
24 "WebSearch failed: kcode-k1-codex-websearch Codex execution subprocess was cancelled";
25const EXEC_ERROR: &str = "WebSearch failed: Codex execution failed";
26const BAD_JSON: &str = "WebSearch failed: invalid Codex JSONL output";
27const INCOMPLETE_ERROR: &str = "WebSearch failed: Codex response was incomplete";
28const MESSAGE_ERROR: &str = "WebSearch failed: Codex returned no agent message";
29
30const STDERR_RULES: &[(&[&str], &str)] = &[
31 (
32 &[
33 "unexpected argument",
34 "unrecognized argument",
35 "unknown argument",
36 "unexpected option",
37 "unrecognized option",
38 "unknown option",
39 "configuration",
40 "config key",
41 "config value",
42 "strict config",
43 ],
44 "CLI argument/configuration",
45 ),
46 (
47 &[
48 "authentication",
49 "not logged in",
50 "login required",
51 "credential",
52 ],
53 "authentication",
54 ),
55 (&["unsupported model", "unknown model"], "unsupported model"),
56 (&["rate limit", "quota"], "quota/rate limit"),
57 (
58 &["network", "connection", "dns", "transport"],
59 "network/transport",
60 ),
61 (
62 &["upstream", "provider", "service unavailable"],
63 "provider/upstream",
64 ),
65 (&["codex failed", "codex error"], "other Codex failure"),
66];
67
68const CONFIG: &str = r#"web_search="live"
69tools.web_search=true
70tools.view_image=false
71apps._default.enabled=false
72agents.enabled=false
73features.apps=false
74features.code_mode.enabled=false
75features.goals=false
76features.hooks=false
77features.memories=false
78features.multi_agent=false
79features.remote_plugin=false
80features.shell_snapshot=false
81features.shell_tool=false
82features.skill_mcp_dependency_install=false
83features.unified_exec=false
84memories.generate_memories=false
85memories.use_memories=false
86history.persistence="none"
87check_for_update_on_startup=false
88feedback.enabled=false
89analytics.enabled=false
90allow_login_shell=false
91skills.config=[]
92mcp_servers={}
93plugins={}
94marketplaces={}
95hooks={}"#;
96
97const EXEC_PREFIX: &str = r#"exec
98--search
99--ephemeral
100--ignore-user-config
101--ignore-rules
102--json
103--sandbox
104read-only
105--ask-for-approval
106never
107--skip-git-repo-check
108--no-daemon
109--strict-config
110--model"#;
111
112#[derive(Clone)]
113pub struct Runner {
114 executable: PathBuf,
115}
116
117pub struct Request {
118 pub query: String,
119 pub model: String,
120 pub reasoning_effort: String,
121 pub deadline: Instant,
122}
123
124impl Runner {
125 pub fn new(executable: PathBuf) -> Self {
126 Self { executable }
127 }
128
129 pub async fn run(&self, request: Request) -> Result<String, String> {
130 if request.model.contains('\0') || request.reasoning_effort.contains('\0') {
131 return Err(ARG_ERROR.to_owned());
132 }
133 let deadline = request.deadline;
134 if Instant::now() >= deadline {
135 return Err(deadline_error("before compatibility preflight"));
136 }
137 let mut help = self
138 .check_help(&["--help"], "codex --help", deadline)
139 .await?;
140 help.extend(
141 self.check_help(&["exec", "--help"], "codex exec --help", deadline)
142 .await?,
143 );
144 let help = String::from_utf8_lossy(&help);
145 let missing = required_flags()
146 .into_iter()
147 .filter(|flag| !help.contains(flag.as_str()))
148 .collect::<Vec<_>>();
149 if !missing.is_empty() {
150 return Err(format!(
151 "{PREFLIGHT}: combined required-flag validation omitted {}",
152 missing.join(", ")
153 ));
154 }
155 if Instant::now() >= deadline {
156 return Err(deadline_error("after compatibility preflight"));
157 }
158 let output = run_process(
159 self.executable.clone(),
160 exec_args(&request.model, &request.reasoning_effort),
161 request.query.into_bytes(),
162 deadline,
163 )
164 .await
165 .map_err(|error| match error {
166 ProcError::Timeout => deadline_error("during Codex execution"),
167 ProcError::Io => EXEC_IO_ERROR.to_owned(),
168 ProcError::Cancelled => EXEC_CANCELLED.to_owned(),
169 })?;
170 if !output.status.success() {
171 return Err(execution_exit_error(output.status, &output.stderr));
172 }
173 parse_jsonl(&output.stdout)
174 }
175
176 async fn check_help(
177 &self,
178 args: &[&str],
179 phase: &str,
180 deadline: Instant,
181 ) -> Result<Vec<u8>, String> {
182 let output = run_process(
183 self.executable.clone(),
184 args.iter().map(OsString::from).collect(),
185 Vec::new(),
186 deadline,
187 )
188 .await
189 .map_err(|error| match error {
190 ProcError::Timeout => {
191 deadline_error(&format!("during compatibility preflight ({phase})"))
192 }
193 ProcError::Io => preflight_error(phase, "subprocess setup, spawn, or I/O failed"),
194 ProcError::Cancelled => preflight_error(phase, "subprocess was cancelled"),
195 })?;
196 if !output.status.success() {
197 return Err(preflight_error(phase, "subprocess exited unsuccessfully"));
198 }
199 let mut text = output.stdout;
200 text.extend(output.stderr);
201 Ok(text)
202 }
203}
204
205fn deadline_error(phase: &str) -> String {
206 format!("WebSearch failed: absolute operation deadline exceeded {phase}")
207}
208
209fn preflight_error(phase: &str, cause: &str) -> String {
210 format!("{PREFLIGHT}: {phase} {cause}")
211}
212
213fn execution_exit_error(status: std::process::ExitStatus, stderr: &[u8]) -> String {
214 let termination = status.code().map_or_else(
215 || {
216 format!(
217 "terminated by signal {}",
218 status.signal().unwrap_or_default()
219 )
220 },
221 |code| format!("exited with numeric exit code {code}"),
222 );
223 format!(
224 "WebSearch failed: kcode-k1-codex-websearch Codex execution phase {termination}; stderr classification: {}",
225 classify_stderr(stderr)
226 )
227}
228
229fn classify_stderr(bytes: &[u8]) -> &'static str {
230 if bytes.is_empty() {
231 return "empty";
232 }
233 let Ok(text) = std::str::from_utf8(bytes) else {
234 return "non-UTF8";
235 };
236 let text = text.to_ascii_lowercase();
237 STDERR_RULES
238 .iter()
239 .find_map(|(needles, category)| {
240 needles
241 .iter()
242 .any(|needle| text.contains(needle))
243 .then_some(*category)
244 })
245 .unwrap_or("unclassified")
246}
247
248fn exec_args(model: &str, effort: &str) -> Vec<OsString> {
249 let mut args = EXEC_PREFIX.lines().map(OsString::from).collect::<Vec<_>>();
250 args.push(model.into());
251 args.push("-c".into());
252 args.push(format!("model_reasoning_effort={}", toml_quote(effort)).into());
253 for config in CONFIG.lines() {
254 args.push("-c".into());
255 args.push(config.into());
256 }
257 args.push("-".into());
258 args
259}
260
261fn required_flags() -> Vec<String> {
262 let mut flags = exec_args("", "")
263 .into_iter()
264 .filter_map(|arg| arg.into_string().ok())
265 .filter(|arg| arg.starts_with("--"))
266 .collect::<Vec<_>>();
267 flags.push("--config".into());
268 flags
269}
270
271fn toml_quote(value: &str) -> String {
272 let mut output = String::from("\"");
273 for character in value.chars() {
274 match character {
275 '"' => output.push_str("\\\""),
276 '\\' => output.push_str("\\\\"),
277 '\u{8}' => output.push_str("\\b"),
278 '\t' => output.push_str("\\t"),
279 '\n' => output.push_str("\\n"),
280 '\u{c}' => output.push_str("\\f"),
281 '\r' => output.push_str("\\r"),
282 value if value.is_control() => output.push_str(&format!("\\u{:04X}", value as u32)),
283 value => output.push(value),
284 }
285 }
286 output.push('"');
287 output
288}
289
290fn parse_jsonl(bytes: &[u8]) -> Result<String, String> {
291 let mut turn_completed = false;
292 let mut turn_failed = false;
293 let mut message = None;
294 for raw in bytes.split(|byte| *byte == b'\n') {
295 let line = raw.trim_ascii();
296 if line.is_empty() {
297 continue;
298 }
299 let value: Value = serde_json::from_slice(line).map_err(|_| BAD_JSON.to_owned())?;
300 match value.get("type").and_then(Value::as_str) {
301 Some("turn.completed") => turn_completed = true,
302 Some("turn.failed") => turn_failed = true,
303 Some("item.completed")
304 if value.pointer("/item/type").and_then(Value::as_str) == Some("agent_message") =>
305 {
306 let text = value
307 .pointer("/item/text")
308 .and_then(Value::as_str)
309 .ok_or_else(|| BAD_JSON.to_owned())?;
310 message = Some(text.to_owned());
311 }
312 _ => {}
313 }
314 }
315 if turn_failed {
316 return Err(EXEC_ERROR.to_owned());
317 }
318 if !turn_completed {
319 return Err(INCOMPLETE_ERROR.to_owned());
320 }
321 message.ok_or_else(|| MESSAGE_ERROR.to_owned())
322}
323
324enum ProcError {
325 Io,
326 Timeout,
327 Cancelled,
328}
329
330async fn drain<R: AsyncRead + Unpin>(mut reader: R) -> io::Result<Vec<u8>> {
331 let mut bytes = Vec::new();
332 reader.read_to_end(&mut bytes).await?;
333 Ok(bytes)
334}
335
336async fn run_process(
337 executable: PathBuf,
338 args: Vec<OsString>,
339 input: Vec<u8>,
340 until: Instant,
341) -> Result<std::process::Output, ProcError> {
342 if Instant::now() >= until {
343 return Err(ProcError::Timeout);
344 }
345 let directory = tempfile::tempdir().map_err(|_| ProcError::Io)?;
346 let mut command = Command::new(executable);
347 command
348 .args(args)
349 .current_dir(directory.path())
350 .stdin(std::process::Stdio::piped())
351 .stdout(std::process::Stdio::piped())
352 .stderr(std::process::Stdio::piped())
353 .kill_on_drop(true)
354 .process_group(0);
355 let mut child = command.spawn().map_err(|_| ProcError::Io)?;
356 let pid = Pid::from_raw(child.id().ok_or(ProcError::Io)? as i32);
357 let mut stdin = child.stdin.take().ok_or(ProcError::Io)?;
358 let stdout = child.stdout.take().ok_or(ProcError::Io)?;
359 let stderr = child.stderr.take().ok_or(ProcError::Io)?;
360 let (cancel_tx, mut cancel_rx) = oneshot::channel::<()>();
361 let worker = tokio::spawn(async move {
362 let _directory = directory;
363 let complete = async move {
364 let write = async move {
365 if input.is_empty() {
366 return Ok(());
367 }
368 stdin.write_all(&input).await?;
369 stdin.shutdown().await
370 };
371 let (status, written, stdout, stderr) =
372 tokio::join!(child.wait(), write, drain(stdout), drain(stderr));
373 written.map_err(|_| ProcError::Io)?;
374 Ok::<_, ProcError>(std::process::Output {
375 status: status.map_err(|_| ProcError::Io)?,
376 stdout: stdout.map_err(|_| ProcError::Io)?,
377 stderr: stderr.map_err(|_| ProcError::Io)?,
378 })
379 };
380 tokio::pin!(complete);
381 let timer = tokio::time::sleep_until(tokio::time::Instant::from_std(until));
382 tokio::pin!(timer);
383 let error = tokio::select! {
384 result = &mut complete => return result,
385 _ = &mut timer => ProcError::Timeout,
386 _ = &mut cancel_rx => ProcError::Cancelled,
387 };
388 let _ = killpg(pid, Signal::SIGKILL);
389 let _ = (&mut complete).await;
390 Err(error)
391 });
392 let result = worker.await.map_err(|_| ProcError::Io)?;
393 drop(cancel_tx);
394 result
395}