use std::{
io::{self, Write},
process::ExitCode,
sync::{
Arc,
atomic::{AtomicBool, Ordering},
},
};
use basis::{Event, RunUsage};
use serde::Deserialize;
use serde_json::Value;
use crate::exit::{EXIT_BOUNDED, EXIT_FAILED, EXIT_OK};
use super::error::ClientError;
const SUMMARY_BUDGET: usize = 120;
#[derive(Clone)]
pub(crate) struct Live {
shown: bool,
answered: Arc<AtomicBool>,
}
impl Live {
pub(crate) fn when(shown: bool) -> Self {
Self {
shown,
answered: Arc::new(AtomicBool::new(false)),
}
}
pub(crate) fn show(&self, event: &Value) -> io::Result<()> {
if !self.shown {
return Ok(());
}
self.show_to(event, &mut io::stdout().lock(), &mut io::stderr().lock())
}
fn show_to(&self, event: &Value, out: &mut impl Write, err: &mut impl Write) -> io::Result<()> {
if !self.shown {
return Ok(());
}
if write_event(event, self.answered(), out, err)? {
self.answered.store(true, Ordering::Relaxed);
}
Ok(())
}
pub(crate) fn answered(&self) -> bool {
self.answered.load(Ordering::Relaxed)
}
pub(crate) fn settled(
&self,
payload: &Value,
structured: bool,
) -> Result<ExitCode, ClientError> {
if !self.repeats(payload, structured) {
return render_result(payload, structured);
}
print_hint(payload);
flush_stdout()?;
Ok(ExitCode::from(result_code(payload)))
}
fn repeats(&self, payload: &Value, structured: bool) -> bool {
!structured && self.answered() && payload["state"] == "succeeded"
}
}
impl basis_tasks::LiveSink for Live {
fn on_event(&self, event: &Value) {
let _ = self.show(event);
}
}
fn write_event(
event: &Value,
answered: bool,
out: &mut impl Write,
err: &mut impl Write,
) -> io::Result<bool> {
let Some(event) = typed(event, err)? else {
return Ok(false);
};
match event {
Event::AssistantDelta { text } => {
write!(out, "{text}")?;
out.flush()?;
return Ok(!text.is_empty());
}
Event::RunFinished { usage, .. } => {
if answered {
writeln!(out)?;
}
if let Some(spent) = usage.and_then(spent) {
writeln!(err, "basis: {spent}")?;
}
}
event => {
if let Some(line) = progress_line(&event) {
writeln!(err, "{line}")?;
}
}
}
Ok(false)
}
fn typed(event: &Value, err: &mut impl Write) -> io::Result<Option<Event>> {
match Event::deserialize(event) {
Ok(event) => Ok(Some(event)),
Err(_) => {
writeln!(
err,
"basis: unrecognized event `{}`",
label(text(event, "type"), "untyped")
)?;
Ok(None)
}
}
}
fn progress_line(event: &Event) -> Option<String> {
Some(match event {
Event::RunStarted {
model,
context_files,
..
} => format!(
"basis: {}, {} context file(s)",
label(model, "unknown model"),
context_files.len()
),
Event::ToolQueued {
tool_name, summary, ..
} => format!(
" · {}",
one_line(label(summary, tool_name), SUMMARY_BUDGET)
),
Event::ToolCompleted {
tool_call_id,
tool_name,
summary,
is_error,
} => {
let name = label(tool_name, tool_call_id);
if *is_error {
format!(" ! {name}: {}", one_line(summary, SUMMARY_BUDGET))
} else {
format!(" ✓ {name}")
}
}
Event::CompactionStarted { .. } => "basis: compacting the conversation".to_string(),
Event::RequestToolResultsElided {
canonical_tool_result_content_bytes,
projected_tool_result_content_bytes,
results,
..
} => format!(
"basis: {}",
tool_result_elision_line(
*canonical_tool_result_content_bytes,
*projected_tool_result_content_bytes,
results.len(),
)
),
Event::Retry {
error,
attempt,
max_attempts,
..
} => format!(
"basis: {} (retry {attempt}/{max_attempts})",
one_line(error, SUMMARY_BUDGET)
),
Event::Notice { message, .. } | Event::Error { message, .. } => {
format!("basis: {}", label(message, "task event"))
}
_ => return None,
})
}
fn tool_result_elision_line(
canonical_bytes: usize,
projected_bytes: usize,
changed: usize,
) -> String {
let result = if changed == 1 { "result" } else { "results" };
format!(
"request tool results reduced: {canonical_bytes} -> {projected_bytes} bytes; \
{changed} {result} changed"
)
}
fn spent(usage: RunUsage) -> Option<String> {
let (input, output) = (usage.input_tokens, usage.output_tokens);
(input > 0 || output > 0).then(|| {
format!(
"{} in · {} out",
compact_count(input),
compact_count(output)
)
})
}
fn compact_count(count: u64) -> String {
for (unit, scale) in [("M", 1_000_000_u64), ("k", 1_000)] {
if count >= scale {
let whole = count / scale;
let tenth = (count % scale) * 10 / scale;
return if tenth == 0 {
format!("{whole}{unit}")
} else {
format!("{whole}.{tenth}{unit}")
};
}
}
count.to_string()
}
fn text<'a>(event: &'a Value, field: &str) -> &'a str {
event[field].as_str().unwrap_or_default()
}
fn label<'a>(value: &'a str, fallback: &'a str) -> &'a str {
if value.is_empty() { fallback } else { value }
}
fn one_line(text: &str, budget: usize) -> String {
let line = text.lines().next().unwrap_or_default();
match line.char_indices().nth(budget) {
Some((end, _)) => format!("{}…", &line[..end]),
None => line.to_string(),
}
}
pub(crate) fn decorate_terminal(task: &str, mut payload: Value) -> Value {
let object = payload
.as_object_mut()
.expect("terminal payload is an object");
let state = object
.get("state")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string();
object.insert("task".to_string(), serde_json::json!(task));
object.insert(
"next".to_string(),
serde_json::json!(next_step(&state, task)),
);
payload
}
fn next_step(state: &str, task: &str) -> String {
match state {
"succeeded" => format!("basis watch {task}"),
"failed" | "cancelled" => "basis spawn <PROMPT>".to_string(),
"resumable" => format!("basis wait {task}"),
_ => format!("basis watch {task} or basis inbox {task}"),
}
}
pub(crate) fn render_result(payload: &Value, structured: bool) -> Result<ExitCode, ClientError> {
if structured {
println!("{payload}");
return Ok(ExitCode::from(result_code(payload)));
}
match payload["state"].as_str().unwrap_or("unknown") {
"running" | "resumable" | "accepted" | "cancel_requested" => {
if let Some(task) = payload["task"].as_str() {
let state = payload["state"].as_str().unwrap_or("unknown");
if state == "accepted"
&& let Some(message) = payload["message"].as_str()
{
println!("task {task}: accepted message {message}");
} else {
println!("task {task}: {state}");
}
}
}
"succeeded" => {
if let Some(result) = payload["result"].as_str()
&& !result.is_empty()
{
print!("{result}");
if !result.ends_with('\n') {
println!();
}
}
}
"failed" => eprintln!(
"basis: task failed: {}",
payload["error"].as_str().unwrap_or("unknown failure")
),
"cancelled" => eprintln!("basis: task was cancelled"),
state => println!("task state: {state}"),
}
print_hint(payload);
flush_stdout()?;
Ok(ExitCode::from(result_code(payload)))
}
fn flush_stdout() -> Result<(), ClientError> {
io::stdout()
.flush()
.map_err(|error| ClientError::new(format!("flush task output: {error}")))
}
pub(crate) fn result_code(payload: &Value) -> u8 {
if !payload["stopped_by"].is_null() {
return EXIT_BOUNDED;
}
match payload["state"].as_str() {
Some("running" | "resumable" | "accepted" | "cancel_requested" | "succeeded") => EXIT_OK,
_ => EXIT_FAILED,
}
}
pub(crate) fn print_hint(payload: &Value) {
let _ = write_hint(payload, &mut io::stderr());
}
fn write_hint(payload: &Value, err: &mut impl Write) -> io::Result<()> {
match payload["next"].as_str() {
Some(next) => writeln!(err, "next: use `{next}`"),
None => Ok(()),
}
}
#[cfg(test)]
mod tests {
use super::*;
use basis::{
Mutability, RunOutcome,
event::{
ContextFile, ElidedToolResult, RequestToolResultElisionPolicy, ToolResultContentKind,
ToolResultElisionAction,
},
};
use serde_json::json;
fn value(event: Event) -> Value {
serde_json::to_value(&event).expect("event serializes")
}
fn started(model: &str) -> Value {
value(Event::RunStarted {
schema: 1,
basis: "0.0.0".to_string(),
session_id: "s".to_string(),
workspace: "/repo".into(),
model: model.to_string(),
provider: "test".to_string(),
context_files: vec![ContextFile {
path: "/repo/AGENTS.md".into(),
scope: "workspace".to_string(),
}],
skills_dirs: Vec::new(),
skills: Vec::new(),
templates_dirs: Vec::new(),
templates: Vec::new(),
mcp_files: Vec::new(),
mcp_servers: Vec::new(),
})
}
fn tool_queued(summary: &str) -> Value {
value(Event::ToolQueued {
tool_call_id: "c1".to_string(),
tool_name: "shell".to_string(),
summary: summary.to_string(),
mutability: Mutability::Unknown,
input: Value::Null,
})
}
fn tool_completed(summary: &str, is_error: bool) -> Value {
value(Event::ToolCompleted {
tool_call_id: "c1".to_string(),
tool_name: "shell".to_string(),
summary: summary.to_string(),
is_error,
})
}
fn delta(text: &str) -> Value {
value(Event::AssistantDelta {
text: text.to_string(),
})
}
fn finished(usage: Option<RunUsage>) -> Value {
value(Event::RunFinished {
outcome: RunOutcome::Ok,
stopped_by: None,
usage,
})
}
#[test]
fn terminal_codes_do_not_depend_on_rendering() {
assert_eq!(result_code(&json!({"state": "succeeded"})), EXIT_OK);
assert_eq!(result_code(&json!({"state": "failed"})), EXIT_FAILED);
assert_eq!(result_code(&json!({"state": "resumable"})), EXIT_OK);
assert_eq!(
result_code(&json!({"state": "failed", "stopped_by": "deadline"})),
EXIT_BOUNDED
);
}
#[test]
fn a_terminal_payload_carries_the_handle_it_settled_under() {
let payload = decorate_terminal("w/t", json!({"state": "succeeded", "result": "done"}));
assert_eq!(payload["task"], "w/t");
}
#[test]
fn each_state_is_told_the_follow_up_that_works_on_it() {
let hints = [
("succeeded", "basis watch w/t"),
("failed", "basis spawn <PROMPT>"),
("cancelled", "basis spawn <PROMPT>"),
("resumable", "basis wait w/t"),
("running", "basis watch w/t or basis inbox w/t"),
("accepted", "basis watch w/t or basis inbox w/t"),
("cancel_requested", "basis watch w/t or basis inbox w/t"),
];
for (state, expected) in hints {
assert_eq!(
decorate_terminal("w/t", json!({"state": state}))["next"],
expected,
"the follow-up offered for {state}"
);
}
}
#[test]
fn the_answer_streams_to_stdout_and_the_work_to_stderr() {
let live = Live::when(true);
let (mut out, mut err) = (Vec::new(), Vec::new());
for event in [
started("test-model"),
tool_queued("shell: cargo test"),
value(Event::ToolStarted {
tool_call_id: "c1".to_string(),
tool_name: "shell".to_string(),
}),
tool_completed("0 failed", false),
delta("the tests "),
delta("pass"),
finished(None),
] {
live.show_to(&event, &mut out, &mut err)
.expect("writing to a vector");
}
let (out, err) = (
String::from_utf8(out).expect("utf8"),
String::from_utf8(err).expect("utf8"),
);
assert_eq!(
out, "the tests pass\n",
"stdout is the answer, closed by the finish line"
);
assert!(err.contains("test-model"), "{err}");
assert!(
err.contains("shell: cargo test"),
"a tool call names itself while it runs: {err}"
);
assert!(err.contains("shell"), "and reports finishing: {err}");
assert!(
!err.contains("the tests"),
"the answer must never be duplicated onto stderr: {err}"
);
assert!(live.answered(), "the answer reached the terminal");
}
#[test]
fn a_finished_run_says_what_it_spent_beside_the_answer_rather_than_in_it() {
let live = Live::when(true);
let (mut out, mut err) = (Vec::new(), Vec::new());
for event in [
delta("done"),
finished(Some(RunUsage {
input_tokens: 12_300,
output_tokens: 1_200,
cache_read_tokens: 40,
cache_creation_tokens: 5,
..RunUsage::default()
})),
] {
live.show_to(&event, &mut out, &mut err)
.expect("writing to a vector");
}
assert_eq!(
String::from_utf8(out).expect("utf8"),
"done\n",
"stdout is the answer, and a token count is not part of it"
);
assert_eq!(
String::from_utf8(err).expect("utf8"),
"basis: 12.3k in · 1.2k out\n"
);
}
#[test]
fn a_run_that_reported_no_usage_prints_no_usage_line() {
let live = Live::when(true);
let (mut out, mut err) = (Vec::new(), Vec::new());
for event in [finished(None), finished(Some(RunUsage::default()))] {
live.show_to(&event, &mut out, &mut err)
.expect("writing to a vector");
}
assert!(out.is_empty(), "and no answer was streamed to close");
assert!(err.is_empty(), "{}", String::from_utf8_lossy(&err));
}
#[test]
fn request_tool_result_elision_is_progress_not_answer_text() {
let live = Live::when(true);
let (mut out, mut err) = (Vec::new(), Vec::new());
let event = value(Event::RequestToolResultsElided {
agent_id: "agent-1".to_string(),
policy: RequestToolResultElisionPolicy::KeepRecent {
configured_keep_recent_tool_results: 3,
},
canonical_tool_result_content_bytes: 8_192,
projected_tool_result_content_bytes: 512,
results: vec![ElidedToolResult {
tool_call_id: "call-1".to_string(),
tool_name: Some("read".to_string()),
is_error: false,
canonical_content_kind: ToolResultContentKind::Text,
action: ToolResultElisionAction::Marker,
canonical_content_bytes: 8_192,
projected_content_bytes: 32,
}],
});
live.show_to(&event, &mut out, &mut err)
.expect("writing to a vector");
assert!(out.is_empty(), "projection telemetry is not answer text");
assert_eq!(
String::from_utf8(err).expect("utf8"),
"basis: request tool results reduced: 8192 -> 512 bytes; 1 result changed\n"
);
assert!(!live.answered());
}
#[test]
fn counts_are_written_the_way_a_person_reads_them() {
assert_eq!(compact_count(0), "0");
assert_eq!(compact_count(980), "980");
assert_eq!(compact_count(1_200), "1.2k");
assert_eq!(compact_count(12_000), "12k", "a bare thousand keeps no .0");
assert_eq!(compact_count(1_250_000), "1.2M");
}
#[test]
fn a_failing_tool_call_says_what_it_said() {
let live = Live::when(true);
let (mut out, mut err) = (Vec::new(), Vec::new());
live.show_to(
&tool_completed("no such file\nand a second line", true),
&mut out,
&mut err,
)
.expect("writing to a vector");
let err = String::from_utf8(err).expect("utf8");
assert_eq!(err, " ! shell: no such file\n", "{err}");
assert!(out.is_empty(), "a tool failure is not an answer");
assert!(!live.answered());
}
#[test]
fn a_run_nobody_is_watching_renders_nothing() {
let live = Live::when(false);
let (mut out, mut err) = (Vec::new(), Vec::new());
for event in [
delta("an answer"),
value(Event::ToolStarted {
tool_call_id: "c1".to_string(),
tool_name: "shell".to_string(),
}),
finished(None),
] {
live.show_to(&event, &mut out, &mut err)
.expect("writing to a vector");
}
assert!(out.is_empty() && err.is_empty());
assert!(
!live.answered(),
"and the settled record is still the only place the answer comes from"
);
}
#[test]
fn a_streamed_answer_is_not_printed_again_underneath_itself() {
let live = Live::when(true);
let succeeded = json!({"state": "succeeded", "result": "done"});
assert!(!live.repeats(&succeeded, false), "nothing streamed yet");
live.show_to(&delta("done"), &mut Vec::new(), &mut Vec::new())
.expect("writing to a vector");
assert!(live.repeats(&succeeded, false));
assert!(
!live.repeats(&succeeded, true),
"`--json` prints the object it was asked for, whatever a terminal saw"
);
assert!(
!live.repeats(&json!({"state": "failed", "error": "boom"}), false),
"a failure was never on the stream, so it still has to be said"
);
}
#[test]
fn a_notice_without_a_severity_still_renders_its_message() {
let live = Live::when(true);
let (mut out, mut err) = (Vec::new(), Vec::new());
live.show_to(
&json!({"type": "notice", "message": "event omitted because it exceeded 32768 bytes", "seq": 4}),
&mut out,
&mut err,
)
.expect("writing to a vector");
assert!(out.is_empty(), "a notice is never the answer");
assert_eq!(
String::from_utf8(err).expect("utf8"),
"basis: event omitted because it exceeded 32768 bytes\n"
);
}
#[test]
fn an_event_this_build_cannot_name_is_said_not_swallowed() {
let live = Live::when(true);
let (mut out, mut err) = (Vec::new(), Vec::new());
live.show_to(
&json!({"type": "from_the_future", "seq": 3}),
&mut out,
&mut err,
)
.expect("writing to a vector");
assert!(out.is_empty(), "not the answer stream's business");
assert_eq!(
String::from_utf8(err).expect("utf8"),
"basis: unrecognized event `from_the_future`\n"
);
assert!(!live.answered());
}
#[test]
fn the_hint_goes_to_the_terminal_and_never_into_the_answer() {
let (mut out, mut err) = (Vec::new(), Vec::new());
write_hint(
&json!({"state": "succeeded", "next": "basis watch w/t"}),
&mut err,
)
.expect("writing to a vector");
write_hint(&json!({"state": "succeeded"}), &mut out).expect("writing to a vector");
assert_eq!(
String::from_utf8(err).unwrap(),
"next: use `basis watch w/t`\n"
);
assert!(out.is_empty(), "a payload without a next step says nothing");
}
}