use super::{
CliError, MAX_INPUT_BYTES, NativeLaunchRequest, configure_harness_runtime, read_nofollow,
};
use serde_json::{Value, json};
use std::{
io::{BufRead, BufReader, Read, Write},
path::Path,
process::{Command, Stdio},
sync::mpsc,
time::{Duration, Instant},
};
pub(super) fn task_text(
request: &NativeLaunchRequest,
project_root: &Path,
) -> Result<String, CliError> {
let task = Path::new(&request.prepare.task_file);
if task.is_absolute()
|| task
.components()
.any(|c| !matches!(c, std::path::Component::Normal(_)))
{
return Err(CliError::message(
"Codex task file must be project-relative",
));
}
let root = project_root.canonicalize().map_err(failure)?;
let path = root.join(task).canonicalize().map_err(failure)?;
if !path.starts_with(root) {
return Err(CliError::message("Codex task file escaped project"));
}
String::from_utf8(read_nofollow(&path, MAX_INPUT_BYTES)?).map_err(failure)
}
fn failure(error: impl std::fmt::Display) -> CliError {
CliError::message(format!("Codex thread bootstrap failed: {error}"))
}
fn thread_parameters(configuration: &[u8], project_root: &Path) -> Result<Value, CliError> {
let role: toml::Value =
toml::from_str(std::str::from_utf8(configuration).map_err(failure)?).map_err(failure)?;
let instructions = role
.get("developer_instructions")
.and_then(toml::Value::as_str)
.filter(|value| !value.is_empty())
.ok_or_else(|| failure("role has no developer instructions"))?;
let sandbox = role
.get("sandbox_mode")
.and_then(toml::Value::as_str)
.filter(|value| matches!(*value, "read-only" | "workspace-write"))
.ok_or_else(|| failure("role sandbox is not bounded"))?;
let mut params = json!({"cwd": project_root, "developerInstructions": instructions,
"sandbox": sandbox, "approvalPolicy": "on-request", "approvalsReviewer": "auto_review", "ephemeral": false});
if let Some(model) = role.get("model").and_then(toml::Value::as_str) {
params["model"] = json!(model);
}
if let Some(effort) = role
.get("model_reasoning_effort")
.and_then(toml::Value::as_str)
{
params["config"] = json!({"model_reasoning_effort": effort});
}
Ok(params)
}
struct Server {
child: std::process::Child,
process_tree: super::process_tree::ProcessTree,
terminated: bool,
receiver: mpsc::Receiver<std::io::Result<String>>,
deadline: Instant,
pending: std::collections::VecDeque<Value>,
}
impl Drop for Server {
fn drop(&mut self) {
let _ = self.terminate();
}
}
impl Server {
fn terminate(&mut self) -> Result<(), CliError> {
if !self.terminated {
self.process_tree.terminate().map_err(failure)?;
self.child.wait().map_err(failure)?;
self.terminated = true;
}
Ok(())
}
fn send(&mut self, value: Value) -> Result<(), CliError> {
writeln!(
self.child
.stdin
.as_mut()
.ok_or_else(|| failure("missing app-server stdin"))?,
"{value}"
)
.map_err(failure)
}
fn next(&mut self) -> Result<Value, CliError> {
if let Some(event) = self.pending.pop_front() {
return Ok(event);
}
self.receive()
}
fn receive(&self) -> Result<Value, CliError> {
let remaining = self
.deadline
.checked_duration_since(Instant::now())
.ok_or_else(|| failure("provider lease expired"))?;
let line = self
.receiver
.recv_timeout(remaining)
.map_err(failure)?
.map_err(failure)?;
serde_json::from_str(&line).map_err(failure)
}
fn response(&mut self, id: u64) -> Result<Value, CliError> {
loop {
let value = self.receive()?;
if value.get("id").and_then(Value::as_u64) != Some(id) {
if self.pending.len() >= 32 {
return Err(failure("too many notifications before RPC response"));
}
self.pending.push_back(value);
continue;
}
if let Some(error) = value.get("error") {
let message = error
.get("message")
.and_then(Value::as_str)
.unwrap_or("unknown protocol error");
return Err(failure(format!(
"app-server rejected RPC {id} (code={}): {}",
error
.get("code")
.and_then(Value::as_i64)
.map_or_else(|| "unknown".into(), |code| code.to_string()),
message.chars().take(512).collect::<String>()
)));
}
return value
.get("result")
.cloned()
.ok_or_else(|| failure("app-server response has no result"));
}
}
}
pub(super) fn run(
request: &mut NativeLaunchRequest,
role: shepherd::dispatch::Role,
context: &crate::ExecutionContext,
home: &Path,
endpoint: &Path,
configuration: &[u8],
parent: &mut crate::BrokerClient,
) -> Result<(), CliError> {
let project_root = &context.workspace_root;
let task = task_text(request, project_root)?;
let params = thread_parameters(configuration, project_root)?;
let launch_id = crate::BrokerLaunchId::fresh().map_err(failure)?;
let mut command = Command::new(&request.executable);
command
.arg("app-server")
.current_dir(project_root)
.env_clear();
for entry in &request.env {
command.env(&entry.name, &entry.value);
}
command
.env("HOME", home)
.env("XDG_CONFIG_HOME", home.join(".config"))
.env("XDG_STATE_HOME", home.join(".state"))
.env("XDG_CACHE_HOME", home.join(".cache"))
.env("SHEPHERD_NATIVE_BROKER_ENDPOINT", endpoint)
.env(
"SHEPHERD_NATIVE_BROKER_LAUNCH_ID",
launch_id.opaque_string(),
)
.env(
"SHEPHERD_NATIVE_BROKER_AGENT_TYPE",
super::native_agent_type(request.harness, role),
)
.env("SHEPHERD_NATIVE_BROKER_CODEX_THREAD", "1");
let run = request
.prepare
.run
.as_ref()
.ok_or_else(|| failure("Codex requires an explicit run"))?;
command.env("SHEPHERD_NATIVE_BROKER_RUN", run);
configure_harness_runtime(&mut command, request.harness, home, "")?;
command
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::inherit());
super::process_tree::configure(&mut command);
let mut child = command.spawn().map_err(failure)?;
let process_tree = match super::process_tree::ProcessTree::attach(&child) {
Ok(tree) => tree,
Err(error) => {
let _ = child.kill();
let _ = child.wait();
return Err(failure(error));
}
};
let stdout = child
.stdout
.take()
.ok_or_else(|| failure("missing app-server stdout"))?;
let (sender, receiver) = mpsc::sync_channel(32);
std::thread::spawn(move || {
let mut reader = BufReader::new(stdout);
loop {
let mut line = String::new();
let read = reader
.by_ref()
.take(MAX_INPUT_BYTES as u64 + 1)
.read_line(&mut line);
match read {
Ok(0) => break,
Ok(n) if n <= MAX_INPUT_BYTES => {
if sender.send(Ok(line)).is_err() {
break;
}
}
Ok(_) => {
let _ = sender.send(Err(std::io::Error::other(
"provider response exceeds bound",
)));
break;
}
Err(error) => {
let _ = sender.send(Err(error));
break;
}
}
}
});
let launch_deadline = Instant::now() + Duration::from_millis(request.timeout_ms);
let mut server = Server {
child,
process_tree,
terminated: false,
receiver,
deadline: Instant::now() + Duration::from_millis(request.timeout_ms.min(30_000)),
pending: std::collections::VecDeque::new(),
};
server.send(json!({"id":1,"method":"initialize","params":{"clientInfo":{"name":"shepherd","version":env!("CARGO_PKG_VERSION")}}}))?;
server.response(1)?;
server.send(json!({"method":"initialized","params":{}}))?;
server.send(json!({"id":2,"method":"thread/start","params":params}))?;
let result = server.response(2)?;
let id = result
.pointer("/thread/id")
.and_then(Value::as_str)
.ok_or_else(|| failure("host returned no thread ID"))?;
shepherd::dispatch::SessionId::new(id).map_err(failure)?;
request.prepare.child_session_id = id.to_owned();
request.prepare.expected_attachment.agent_id = id.to_owned();
let launch = parent
.prepare_reserved(request.prepare.clone(), Some(launch_id))
.map_err(failure)?;
let result = (|| {
parent
.register_child(&launch, server.child.id())
.map_err(failure)?;
server.deadline = launch_deadline;
server.send(json!({"id":3,"method":"turn/start","params":{"threadId":id,"input":[{"type":"text","text":task,"text_elements":[]}]}}))?;
let result = server.response(3)?;
let turn = result
.pointer("/turn/id")
.and_then(Value::as_str)
.ok_or_else(|| failure("host returned no turn ID"))?;
loop {
let event = server.next()?;
if event.get("id").is_some() && event.get("method").is_some() {
return Err(failure(format!(
"provider requested interactive decision: {}",
event
.get("method")
.and_then(Value::as_str)
.unwrap_or("unknown")
.chars()
.take(128)
.collect::<String>()
)));
}
if event.get("method").and_then(Value::as_str) == Some("item/agentMessage/delta")
&& let Some(delta) = event.pointer("/params/delta").and_then(Value::as_str)
{
eprint!("{delta}");
}
if event.get("method").and_then(Value::as_str) != Some("turn/completed") {
continue;
}
if event.pointer("/params/threadId").and_then(Value::as_str) != Some(id)
|| event.pointer("/params/turn/id").and_then(Value::as_str) != Some(turn)
{
return Err(failure("host completed a different thread or turn"));
}
if event.pointer("/params/turn/status").and_then(Value::as_str) != Some("completed") {
return Err(failure("host turn did not complete successfully"));
}
return Ok(());
}
})();
let result = result.and_then(|()| {
server.terminate()?;
super::verify_provider_completion(request, &context.runs_root)?;
parent
.finish_provider(&launch)
.map_err(|error| failure(error.to_string()))?;
Ok(())
});
match result {
Ok(()) => {
server.terminate()?;
Ok(())
}
Err(error) => {
let error = error.message_text().unwrap_or("provider execution failed");
let identity = format!(
"run={} agent={} session={}",
request.prepare.run.as_deref().unwrap_or("<missing>"),
request.prepare.expected_attachment.agent_id,
request.prepare.child_session_id
);
if let Err(cleanup) = server.terminate() {
let cleanup = cleanup.message_text().unwrap_or("process cleanup failed");
return Err(failure(format!(
"{identity}: {error}; process-tree cleanup failed, Native custody retained: {cleanup}"
)));
}
match parent.fail_provider(&launch) {
Ok(Some(record)) => Err(failure(format!(
"{identity}: {error}; Native state={}",
record.state
))),
Ok(None) => Err(failure(format!(
"{identity}: {error}; Native launch state=launch-failed"
))),
Err(cleanup) => Err(failure(format!(
"{identity}: {error}; Native cleanup unresolved, custody retained: {cleanup}"
))),
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn completion_event_before_rpc_reply_is_retained() {
let (sender, receiver) = mpsc::sync_channel(4);
let completion = json!({"method":"turn/completed","params":{"threadId":"real-host-id"}});
sender.send(Ok(completion.to_string())).unwrap();
sender
.send(Ok(
json!({"id":3,"result":{"turn":{"id":"actual-turn"}}}).to_string()
))
.unwrap();
let mut command = Command::new(std::env::current_exe().unwrap());
command
.arg("--help")
.stdout(Stdio::null())
.stderr(Stdio::null());
super::super::process_tree::configure(&mut command);
let child = command.spawn().unwrap();
let process_tree = super::super::process_tree::ProcessTree::attach(&child).unwrap();
let mut server = Server {
child,
process_tree,
terminated: false,
receiver,
deadline: Instant::now() + Duration::from_secs(5),
pending: Default::default(),
};
assert_eq!(server.response(3).unwrap()["turn"]["id"], "actual-turn");
assert_eq!(server.next().unwrap(), completion);
}
#[test]
fn compiler_role_controls_prompt_model_effort_and_permissions() {
let parameters = thread_parameters(
br#"developer_instructions = "invoke skill"
model = "gpt-6"
model_reasoning_effort = "high"
sandbox_mode = "read-only"
"#,
Path::new("/project"),
)
.unwrap();
assert_eq!(parameters["developerInstructions"], "invoke skill");
assert_eq!(parameters["model"], "gpt-6");
assert_eq!(parameters["config"]["model_reasoning_effort"], "high");
assert_eq!(parameters["sandbox"], "read-only");
assert_eq!(parameters["approvalPolicy"], "on-request");
assert_eq!(parameters["ephemeral"], false);
}
#[test]
fn unbounded_or_missing_role_permissions_are_rejected() {
for sandbox in ["danger-full-access", ""] {
let role =
format!("developer_instructions = 'instructions'\nsandbox_mode = '{sandbox}'");
assert!(thread_parameters(role.as_bytes(), Path::new("/project")).is_err());
}
}
}