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();
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};
use tokio::task::JoinHandle;
struct Owned {
child: Contained,
pipe: Pipe,
exited: Option<JoinHandle<anyhow::Result<i32>>>,
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 = spawn(seat)?;
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;
while let Some(frame) = owned.pipe.input.recv().await {
match frame {
Err(error) => cleanup.push(Err(error)),
Ok(None) => break,
Ok(Some(_)) => {}
}
}
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,
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,
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,
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 result = drive(owned, &mut active, term, interrupt, input).await;
close_active(owned, active, result).await
}
async fn drive(
owned: &mut Owned,
active: &mut Active,
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 door = owned
.link
.as_mut()
.ok_or_else(|| anyhow::anyhow!("door_link_absent"))?;
loop {
let deadline = active.pump.opening_deadline();
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,
()=opening_deadline(deadline), if deadline.is_some()=>anyhow::bail!("acp_session_open_timeout"),
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 close_active(
owned: &mut Owned,
mut active: Active,
result: anyhow::Result<i32>,
) -> anyhow::Result<i32> {
active.pump.closing = true;
let mut cleanup = owned.child.shutdown(&mut owned.exited).await;
if !active.stdout_closed {
while let Some(frame) = owned.pipe.input.recv().await {
match frame {
Ok(Some(super::wire::Frame::Notification { method, params }))
if method == "session/update" =>
{
cleanup.push(active.pump.closing_update(params).await);
}
Ok(None) => break,
Err(error) => cleanup.push(Err(error)),
Ok(Some(_)) => {}
}
}
}
let status = result.as_ref().copied().unwrap_or(1);
let status = if status == 0 && cleanup.iter().any(Result::is_err) {
1
} else {
status
};
cleanup.push(active.pump.end(status).await);
drop(active.pump);
if !active.publisher_seen {
cleanup.push(
active
.publisher
.await
.map_err(|_| anyhow::anyhow!("publisher_panicked"))
.and_then(std::convert::identity),
);
}
if !owned.door_writer_seen {
let door = owned
.link
.as_ref()
.ok_or_else(|| anyhow::anyhow!("door_link_absent"))?;
cleanup.push(door.output.flush().await);
}
Ok(report(result, cleanup))
}
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 executable = super::catalogue::executable(&seat.harness.program)?;
let mut command = Command::new(executable);
super::environment::apply(&mut command, &seat.harness.environment)?;
command
.args(&seat.harness.args)
.current_dir(seat.cwd()?)
.process_group(0)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::null())
.kill_on_drop(true);
Contained::new(
command
.spawn()
.map_err(|error| anyhow::anyhow!("harness_spawn_failed: {}", error.kind()))?,
)
}
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,
))
}