board_watch/
trace.rs

1//! The low level: events exactly as the agent sent them.
2//!
3//! [`Session`](ag_ui::client::Session) assembles; this does not. A proxy, a
4//! recorder, a bridge to another protocol and a person debugging a stream all
5//! want the events unassembled, and that is what
6//! [`RemoteAgent`](ag_ui::client::RemoteAgent) is for.
7//!
8//! Including the human-in-the-loop round trip: [`interrupts_of`] reads what a
9//! `RUN_FINISHED` paused on and [`resume_run`] builds the request that answers
10//! it, so pausing and resuming needs no session at all — only the previous
11//! request, which the caller already has.
12
13use std::io::{self, Write};
14
15use ag_ui::client::{HttpAgent, InterruptExt as _, RunParams, interrupts_of, resume_run};
16use ag_ui::{Event, EventType, ResumeEntry, RunAgentInput, Tool};
17use futures_util::StreamExt as _;
18use serde_json::{Value, json};
19
20/// Streams one run and prints every event, then answers any pause and streams
21/// the resumed run too.
22///
23/// `approve` decides what a pause is answered with, and `tools` is what the
24/// agent is offered — an agent that needs one it was not sent fails the run.
25/// Returns how many events were printed across every run it drove.
26pub async fn trace(
27    agent: &HttpAgent,
28    thread: &str,
29    said: &str,
30    tools: Vec<Tool>,
31    approve: bool,
32    out: &mut impl Write,
33) -> io::Result<usize> {
34    let mut input: RunAgentInput = RunParams::new(thread, format!("{thread}-run-1"))
35        .user(format!("{thread}-msg-1"), said)
36        .tools(tools)
37        .into();
38
39    let mut printed = 0;
40    let mut round = 1;
41
42    loop {
43        writeln!(out, "--- run {round} · {}", input.run_id)?;
44        let (count, paused) = stream_once(agent, input.clone(), approve, out).await?;
45        printed += count;
46
47        let Some(entries) = paused else {
48            return Ok(printed);
49        };
50        // A resumed run is a run of its own: same conversation, same state, new
51        // id. `resume_run` carries the rest over, so a caller cannot forget a
52        // field the agent needs.
53        round += 1;
54        input = resume_run(&input, format!("{thread}-run-{round}"), entries);
55    }
56}
57
58/// Prints one run's events, reporting the answers its pause needs.
59async fn stream_once(
60    agent: &HttpAgent,
61    input: RunAgentInput,
62    approve: bool,
63    out: &mut impl Write,
64) -> io::Result<(usize, Option<Vec<ResumeEntry>>)> {
65    let mut events = agent.run(input);
66    let mut printed = 0;
67    let mut resume = None;
68
69    while let Some(event) = events.next().await {
70        let event = match event {
71            Ok(event) => event,
72            // A transport that cannot reach the agent yields one error item and
73            // ends, so this is the whole of the failure path.
74            Err(error) => {
75                writeln!(out, "  !! {error}")?;
76                break;
77            }
78        };
79        printed += 1;
80        writeln!(
81            out,
82            "  {:<26} {}",
83            name(event.event_type()),
84            payload(&event)
85        )?;
86
87        let interrupts = interrupts_of(&event);
88        if !interrupts.is_empty() {
89            resume = Some(
90                interrupts
91                    .iter()
92                    .map(|interrupt| {
93                        if approve {
94                            interrupt.resolve(json!({"confirm": true}))
95                        } else {
96                            interrupt.cancel()
97                        }
98                    })
99                    .collect(),
100            );
101        }
102    }
103    Ok((printed, resume))
104}
105
106/// The wire name of an event type.
107fn name(kind: EventType) -> String {
108    serde_json::to_value(kind)
109        .ok()
110        .and_then(|value| value.as_str().map(str::to_owned))
111        .unwrap_or_else(|| format!("{kind:?}"))
112}
113
114/// The event's fields, minus the two every event carries.
115fn payload(event: &Event) -> String {
116    let Ok(Value::Object(mut fields)) = serde_json::to_value(event) else {
117        return String::new();
118    };
119    fields.remove("type");
120    fields.remove("timestamp");
121    serde_json::to_string(&fields).unwrap_or_default()
122}
board_watch/trace.rs

board_watch/
trace.rs
