use super::{config::Seat, containment::Contained, wire::Pipe};
use std::process::Stdio;
use tokio::process::Command;
use tokio::signal::unix::{SignalKind, signal};
pub(super) async fn serve(seat: &Seat) -> anyhow::Result<i32> {
let mut term = signal(SignalKind::terminate())?;
let mut interrupt = signal(SignalKind::interrupt())?;
let input = super::input::eof(seat.harness.acp_frame_bytes);
tokio::pin!(input);
let config = seat.clone();
let mut startup = tokio::task::spawn_blocking(move || {
let mut identity = super::tools::Identity::launch(&config)?;
let owed = super::exit::Owed::begin(&config, &identity)?;
let acquired = super::tools::acquire(
&config.issuer,
&identity.agent,
&identity.launch,
config.harness.issuer_response_bytes,
)
.map(|pass| identity.pass = pass);
Ok::<_, anyhow::Error>((identity, owed, acquired))
});
let (early, ready) = tokio::select! {
_=term.recv()=>(Some(Ok(143)),None),
_=interrupt.recv()=>(Some(Ok(130)),None),
result=&mut input=>(Some(result.map(|()|0)),None),
ready=&mut startup=>(None,Some(ready)),
};
let (identity, owed, acquired) = match ready {
Some(ready) => ready,
None => startup.await,
}
.map_err(|_| anyhow::anyhow!("assistant_startup_worker_failed"))??;
let result = match (early, acquired) {
(Some(status), _) => status,
(None, Err(error)) => Err(error),
(None, Ok(())) => run(seat, &identity, &mut term, &mut interrupt, input.as_mut()).await,
};
let revoked = owed.close().await;
match (result, revoked) {
(result, Ok(())) => result,
(Ok(status), Err(error)) if status != 0 => {
eprintln!("haematite assistant: {error}; revoke remains owed");
Ok(status)
}
(Ok(_), Err(error)) => Err(error),
(Err(error), Err(revoke)) => Err(anyhow::anyhow!("{error}; {revoke}; revoke remains owed")),
}
}
use super::{containment::observe, door_link::Link, pump::Pump, termination::End};
use tokio::task::JoinHandle;
struct Owned {
child: Contained,
pipe: Pipe,
exited: Option<JoinHandle<anyhow::Result<End>>>,
link: Option<Link>,
pipe_writer_seen: bool,
pipe_reader_seen: bool,
door_writer_seen: bool,
}
struct Active {
pump: Pump,
publisher: JoinHandle<anyhow::Result<()>>,
publisher_seen: bool,
stdout_closed: bool,
}
async fn run(
seat: &Seat,
identity: &super::tools::Identity,
term: &mut tokio::signal::unix::Signal,
interrupt: &mut tokio::signal::unix::Signal,
input: std::pin::Pin<&mut impl std::future::Future<Output = anyhow::Result<()>>>,
) -> anyhow::Result<i32> {
let mut child = match spawn(seat) {
Ok(child) => child,
Err(error) => {
let recorded = startup::record(seat, identity, &error).await;
return match recorded {
Ok(()) => Err(error),
Err(record) => Err(anyhow::anyhow!(
"{error}; startup_ended_not_recorded: {record}"
)),
};
}
};
let pipe = match pipes(&mut child, seat) {
Ok(pipe) => pipe,
Err(error) => {
let mut exited = Some(child.observe_exit());
let mut cleanup = child.shutdown(&mut exited).await;
cleanup.push(child.reap().await);
return Ok(report(Err(error), cleanup));
}
};
let exited = Some(child.observe_exit());
let mut owned = Owned {
child,
pipe,
exited,
link: None,
pipe_writer_seen: false,
pipe_reader_seen: false,
door_writer_seen: false,
};
let result = converse(&mut owned, seat, identity, term, interrupt, input).await;
let mut cleanup = owned.child.shutdown(&mut owned.exited).await;
cleanup.push(close::discard(&mut owned).await);
if !owned.pipe_reader_seen {
cleanup.push(
super::closing::stop(&mut owned.pipe.reader, "acp_reader")
.await
.map(|_| ()),
);
}
if !owned.pipe_writer_seen {
cleanup.push(
super::closing::stop(&mut owned.pipe.writer, "acp_writer")
.await
.and_then(|result| result.unwrap_or(Ok(()))),
);
}
if let Some(door) = owned.link.as_mut() {
cleanup.push(
super::closing::stop(&mut door.reader, "door_reader")
.await
.map(|_| ()),
);
while let Ok(frame) = door.input.try_recv() {
if let Err(error) = frame {
cleanup.push(Err(error));
}
}
if !owned.door_writer_seen {
cleanup.push(
super::closing::stop(&mut door.writer, "door_writer")
.await
.and_then(|result| result.unwrap_or(Ok(()))),
);
}
}
cleanup.push(owned.child.reap().await);
Ok(report(result, cleanup))
}
async fn converse(
owned: &mut Owned,
seat: &Seat,
identity: &super::tools::Identity,
term: &mut tokio::signal::unix::Signal,
interrupt: &mut tokio::signal::unix::Signal,
mut input: std::pin::Pin<&mut impl std::future::Future<Output = anyhow::Result<()>>>,
) -> anyhow::Result<i32> {
let session = tokio::select! {
biased;
_=term.recv()=>return Ok(143),
_=interrupt.recv()=>return Ok(130),
result=&mut input=>{result?;return Ok(0);},
result=observe(&mut owned.exited)=>return result.map(|end|end.status),
result=super::handshake::open(&mut owned.pipe,seat)=>result?,
};
eprintln!(
"ACP negotiated protocol {}; resume capability {}; resumable {}",
session.version.number(),
session.capability,
session.resumable
);
owned.link = Some(tokio::select! {
biased;
_=term.recv()=>return Ok(143),
_=interrupt.recv()=>return Ok(130),
result=&mut input=>{result?;return Ok(0);},
result=observe(&mut owned.exited)=>return result.map(|end|end.status),
result=Link::connect(seat,identity,&session)=>result?,
});
let door = owned
.link
.as_mut()
.ok_or_else(|| anyhow::anyhow!("door_link_absent"))?;
tokio::select! {
biased;
_=term.recv()=>return Ok(143),
_=interrupt.recv()=>return Ok(130),
result=&mut input=>{result?;return Ok(0);},
result=observe(&mut owned.exited)=>return result.map(|end|end.status),
result=door.welcome(seat,identity)=>result?,
}
let (publications, queued) = tokio::sync::mpsc::channel(seat.harness.stream_queue);
let recorder = super::record::Record::new(
seat.clone(),
identity,
door.person.clone(),
serde_json::json!({"id":seat.harness.id,"protocol":session.version.number(),"resume_capability":session.capability,"info":session.info}),
door.output.clone(),
)?;
let pump = Pump::new(
session,
seat.clone(),
identity.agent.clone(),
door.person.clone(),
owned.pipe.output.clone(),
publications,
super::permission::Issuer::new(seat, identity)?,
)
.await?;
let mut active = Active {
pump,
publisher: tokio::task::spawn_blocking(move || recorder.run(queued)),
publisher_seen: false,
stdout_closed: false,
};
let mut exited = owned.exited.take();
let result = tokio::select! {
biased;
_=term.recv()=>Ok(End::signalled(rustix::process::Signal::TERM.as_raw())),
_=interrupt.recv()=>Ok(End::signalled(rustix::process::Signal::INT.as_raw())),
result=&mut input=>result.map(|()|End::exited(0)),
result=observe(&mut exited)=>result,
result=drive(owned,&mut active)=>result.map(End::exited),
};
owned.exited = exited;
close::active(owned, active, result).await
}
async fn drive(owned: &mut Owned, active: &mut Active) -> anyhow::Result<i32> {
let door = owned
.link
.as_mut()
.ok_or_else(|| anyhow::anyhow!("door_link_absent"))?;
loop {
let deadline = active.pump.opening_deadline();
tokio::select! {
biased;
()=opening_deadline(deadline), if deadline.is_some()=>active.pump.fail_session("acp_session_open_timeout").await?,
result=&mut active.publisher=>{active.publisher_seen=true;result.map_err(|_|anyhow::anyhow!("publisher_panicked"))??;anyhow::bail!("publisher_closed");},
result=&mut door.writer=>{owned.door_writer_seen=true;result.map_err(|_|anyhow::anyhow!("door_writer_panicked"))??;anyhow::bail!("door_writer_closed");},
result=&mut owned.pipe.reader, if !owned.pipe_reader_seen=>{owned.pipe_reader_seen=true;result.map_err(|_|anyhow::anyhow!("acp_reader_panicked"))?;},
result=&mut owned.pipe.writer=>{owned.pipe_writer_seen=true;result.map_err(|_|anyhow::anyhow!("acp_writer_panicked"))??;anyhow::bail!("acp_writer_closed");},
value=door.input.recv()=>{
let value=value.transpose()?.flatten().ok_or_else(||anyhow::anyhow!("assistant_door_closed"))?;
active.pump.operation(value).await?;
}
receipt=owned.pipe.written.recv()=>active.pump.written(receipt.ok_or_else(||anyhow::anyhow!("acp_written_receipts_closed"))?).await?,
answer=active.pump.approvals.events.recv()=>active.pump.approved(&answer.ok_or_else(||anyhow::anyhow!("permission_watch_closed"))??)?,
Some(held)=active.pump.drafts.join_next(), if !active.pump.drafts.is_empty()=>{
let (key,result)=held.map_err(|_|anyhow::anyhow!("permission_hold_worker_failed"))?;
active.pump.drafted(&key,result).await?;
}
frame=owned.pipe.input.recv(), if !active.stdout_closed=>{
if let Some(frame)=frame.transpose()?.flatten(){active.pump.frame(frame).await?;}else{active.stdout_closed=true;}
}
}
}
}
async fn opening_deadline(deadline: Option<tokio::time::Instant>) {
match deadline {
Some(deadline) => tokio::time::sleep_until(deadline).await,
None => std::future::pending().await,
}
}
fn report(result: anyhow::Result<i32>, cleanup: Vec<anyhow::Result<()>>) -> i32 {
let outcome = super::closing::gather(result, cleanup);
for error in outcome.errors {
eprintln!("haematite assistant: {error}");
}
outcome.status
}
fn spawn(seat: &Seat) -> anyhow::Result<Contained> {
let mut command = command(seat)?;
command.stderr(Stdio::null());
Contained::new(
crate::process_spawn::locked(|| command.spawn())
.map_err(|error| anyhow::anyhow!("harness_spawn_failed: {error}"))?,
)
}
pub(super) fn command(seat: &Seat) -> anyhow::Result<Command> {
let executable = super::catalogue::executable(&seat.harness.program)?;
let mut command = Command::new(executable);
super::environment::apply(&mut command, &seat.harness.environment)?;
super::environment::harness_home(&mut command, &seat.harness)?;
command
.args(&seat.harness.args)
.current_dir(seat.cwd()?)
.process_group(0)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.kill_on_drop(true);
if seat.harness.id == "codex" {
match std::env::var("OPENAI_BASE_URL") {
Ok(base) => {
let encoded = serde_json::to_string(&base)
.map_err(|_| {
anyhow::anyhow!("harness_proxy_invalid: model proxy cannot be encoded")
})?
.replace('\u{7f}', "\\u007f");
command.args(["-c", &format!("model_providers.lys={{name=\"OpenAI through Lys\",base_url={encoded},wire_api=\"responses\",requires_openai_auth=true,supports_websockets=false}}"), "-c", "model_provider=\"lys\""]);
}
Err(std::env::VarError::NotPresent) => {}
Err(std::env::VarError::NotUnicode(_)) => {
anyhow::bail!("harness_proxy_invalid: OPENAI_BASE_URL must be text")
}
}
}
Ok(command)
}
fn pipes(child: &mut Contained, seat: &Seat) -> anyhow::Result<Pipe> {
let stdout = child
.child()
.stdout
.take()
.ok_or_else(|| anyhow::anyhow!("ACP stdout absent"))?;
let stdin = child
.child()
.stdin
.take()
.ok_or_else(|| anyhow::anyhow!("ACP stdin absent"))?;
Ok(Pipe::start(
stdout,
stdin,
seat.harness.acp_frame_bytes,
seat.harness.write_queue,
seat.harness.read_queue,
))
}
#[path = "host_close.rs"]
mod close;
#[path = "host_startup.rs"]
mod startup;