mod approvals;
mod sink;
use super::startup::RuntimeContext;
use crate::sessions::Session;
use crossbeam_channel::Receiver;
use serde::Deserialize;
use serde_json::{Value, json};
use std::{
sync::{
Arc, Mutex,
atomic::{AtomicBool, Ordering},
},
thread::JoinHandle,
};
#[derive(Debug)]
pub(super) struct RunFailure(pub(super) &'static str);
impl std::fmt::Display for RunFailure {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(self.0)
}
}
impl std::error::Error for RunFailure {}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
pub(super) struct StartParams {
pub(super) session_id: String,
pub(super) prompt: String,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
pub(super) struct RunParams {
pub(super) session_id: String,
pub(super) run_id: String,
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
pub(super) struct AnswerParams {
pub(super) session_id: String,
pub(super) run_id: String,
pub(super) approval_id: String,
pub(super) allow: bool,
}
pub(super) fn parse<T: serde::de::DeserializeOwned>(
params: &serde_json::value::RawValue,
) -> Result<T, &'static str> {
if !params.get().starts_with('{') {
return Err("invalid_params");
}
serde_json::from_str(params.get()).map_err(|_| "invalid_params")
}
pub(super) struct WireEvent {
pub(super) name: &'static str,
pub(super) data: Value,
}
pub(super) struct TurnWorker {
pub(super) session_id: String,
pub(super) run_id: String,
receiver: Receiver<WireEvent>,
events: sink::Events,
approvals: Arc<approvals::Approvals>,
worker: Option<JoinHandle<(RuntimeContext, Value)>>,
}
impl TurnWorker {
pub(super) fn start(
runtime: RuntimeContext,
prompt: String,
) -> Result<Self, (Box<RuntimeContext>, &'static str)> {
let session_id = runtime
.attachment
.as_ref()
.expect("admitted attachment")
.session
.id()
.to_owned();
let run_id = uuid::Uuid::new_v4().to_string();
let (sender, receiver) = crossbeam_channel::bounded(32);
let cancel = Arc::new(AtomicBool::new(false));
let events = sink::Events::new(
sender,
Arc::clone(&cancel),
session_id.clone(),
run_id.clone(),
);
let approvals = Arc::new(approvals::Approvals::default());
let owned_events = events.clone();
let owned_approvals = Arc::clone(&approvals);
let slot = Arc::new(Mutex::new(Some(runtime)));
let owned = Arc::clone(&slot);
let worker = std::thread::Builder::new()
.name("local-agent-turn".into())
.spawn(move || {
let mut runtime = owned
.lock()
.unwrap_or_else(|e| e.into_inner())
.take()
.expect("one turn worker");
let session = runtime.attachment.as_ref().expect("admitted attachment").session.clone();
let mut sink = sink::Sink::new(
owned_events.clone(), session.clone(), runtime.launch.cwd.clone(), owned_approvals,
);
let output = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
runtime.attachment.as_mut().expect("admitted attachment")
.validate_for_turn().map_err(RunFailure)?;
let options = runtime.run_options(&prompt, Some(&session), Some(&mut sink), cancel);
crate::agent::runner::run_provider_once_streaming_with_steering(
runtime.effective_config(), &runtime.launch.instructions, &runtime.policy.skills,
options, crate::agent::steering::AgentSteering::default(),
)
})).unwrap_or_else(|_| Err(RunFailure("execution_failed").into()));
let mut error_code = output.as_ref().err().map(|error| {
if let Some(error) = error.downcast_ref::<RunFailure>() {
error.0
} else if error.downcast_ref::<crate::agent::RequiredUserInputPersistenceError>().is_some() {
"persistence_failed"
} else {
"execution_failed"
}
});
let mut outcome = if output.is_ok() {
"completed"
} else if output.as_ref().err().is_some_and(crate::cancellation::is_run_canceled) {
error_code = None;
"cancelled"
} else {
"failed"
};
if output.as_ref().is_ok_and(|output| output.persistence_degraded) {
sink.degraded.store(true, Ordering::SeqCst);
}
if let Err(error) = sink.finish(outcome != "completed") {
outcome = "failed";
error_code = Some(error.downcast_ref::<RunFailure>()
.map(|e| e.0).unwrap_or("execution_failed"));
}
let degraded = sink.degraded.load(Ordering::SeqCst)
|| (sink.persisted && !terminal_recorded(&session));
if degraded {
let _ = owned_events.diagnostic(
"persistence_failed", "History incomplete; do not automatically resubmit this prompt.",
);
}
let attachment = runtime.attachment.as_mut().expect("admitted attachment");
let revision = if !sink.persisted {
None
} else {
match attachment.refresh_history() {
Ok(()) => Some(attachment.revision().to_owned()),
Err(code) => {
let _ = owned_events.diagnostic(code, "History unavailable; reopen explicitly.");
None
}
}
};
if owned_events.failed.load(Ordering::SeqCst) {
outcome = "failed";
error_code = Some("output_unavailable");
}
owned_events.stopped.store(true, Ordering::SeqCst);
let persistence = if !sink.persisted {
"not_started"
} else if degraded {
"degraded"
} else {
"durable"
};
(runtime, json!({"outcome":outcome,"persistence":persistence,"revision":revision,"error_code":error_code}))
});
match worker {
Ok(worker) => Ok(Self {
session_id,
run_id,
receiver,
events,
approvals,
worker: Some(worker),
}),
Err(_) => Err((
Box::new(
slot.lock()
.unwrap_or_else(|e| e.into_inner())
.take()
.expect("unstarted worker"),
),
"execution_failed",
)),
}
}
pub(super) fn next_event(&self) -> Option<WireEvent> {
self.receiver.try_recv().ok()
}
pub(super) fn poll(&mut self) -> Option<(RuntimeContext, Value)> {
if !self.receiver.is_empty() || !self.worker.as_ref()?.is_finished() {
return None;
}
self.worker.take()?.join().ok()
}
pub(super) fn cancel(&self) {
self.events.cancel();
}
pub(super) fn answer(&self, id: &str, allow: bool) -> Result<(), &'static str> {
self.approvals.answer(id, allow, &self.events.cancellation)
}
pub(super) fn finish(mut self) {
self.cancel();
if let Some(worker) = self.worker.take() {
drop(worker.join());
}
}
}
fn terminal_recorded(session: &Session) -> bool {
let Ok(read) = session.read_events_tolerant_bounded(
crate::context::REPLAY_JSONL_MAX_LINES,
crate::context::REPLAY_JSONL_MAX_BYTES,
) else {
return false;
};
if !read.diagnostics.is_empty() {
return false;
}
read.events
.iter()
.rev()
.take_while(|e| e.kind() != Some(crate::sessions::SessionEventKind::UserInput))
.any(|e| {
e.kind() == Some(crate::sessions::SessionEventKind::AssistantOutput)
|| e.kind() == Some(crate::sessions::SessionEventKind::TurnStatus)
&& e.payload
.get("status")
.and_then(Value::as_str)
.is_some_and(|status| matches!(status, "complete" | "cancelled" | "failed"))
})
}