diff --git a/docs/internal/events-strategy.md b/docs/internal/events-strategy.md index c57b60d04..b2e931bfb 100644 --- a/docs/internal/events-strategy.md +++ b/docs/internal/events-strategy.md @@ -158,7 +158,7 @@ For cache-backed lifecycle work, emit slow-path start events only when the opera Check: - CLI progress parsing -- `fabro logs` +- `fabro events` - retro duration extraction - store validation - tests or fixtures that inspect event names or fields diff --git a/docs/internal/events.md b/docs/internal/events.md index b2f0c30c5..580de2349 100644 --- a/docs/internal/events.md +++ b/docs/internal/events.md @@ -1,6 +1,6 @@ # Events -Every serialized run event envelope, whether streamed over SSE, returned by `fabro logs`, or written to a JSONL sink, uses this structure: +Every serialized run event envelope, whether streamed over SSE, returned by `fabro events`, or written to a JSONL sink, uses this structure: ```json { diff --git a/docs/internal/logging-strategy.md b/docs/internal/logging-strategy.md index 4968d2763..bc122e598 100644 --- a/docs/internal/logging-strategy.md +++ b/docs/internal/logging-strategy.md @@ -31,7 +31,7 @@ Fixed server and per-run logs use a per-event-buffered writer opened with `O_APP - Hot loops or per-token streaming events (use DEBUG only if truly needed for diagnosis) - Data that belongs in user-facing output (`eprintln!` for interactive CLI feedback, not tracing) -- Detached user-visible warnings or errors that need to survive `attach`/`logs` (`detach.log` is debug-only; emit an `Event` into the run event stream instead) +- Detached user-visible warnings or errors that need to survive `attach`/`events` (`detach.log` is debug-only; emit an `Event` into the run event stream instead) - Redundant information already captured by a parent event (if you logged "starting X", you don't need to log every sub-step at the same level) - Events that are already traced via `EventEnum::trace()` — the event enums (`AgentEvent`, `PipelineEvent`, `ExecutionEnvEvent`) each have a `trace()` method called automatically at their emit site; do not add manual `info!`/`debug!` calls that duplicate what `trace()` already emits - Wrapper/forwarding variants that re-emit an inner event — `PipelineEvent::Agent`, `PipelineEvent::ExecutionEnv`, and `AgentEvent::SubAgentEvent` are no-ops in `trace()` because the inner event is already traced at its origin @@ -192,7 +192,7 @@ The domain event enums (`AgentEvent`, `PipelineEvent`, `ExecutionEnvEvent`) each - **Add tracing for new variants** by adding a match arm in the enum's `trace()` method. Choose the level based on the guidelines above (INFO for lifecycle boundaries, DEBUG for individual steps, WARN/ERROR for failures). - **Do not add manual log calls at emit sites.** The `trace()` call in the emitter handles it. Adding `info!` or `debug!` next to an `emit()` call will double-log. -- **Detached UX belongs in events, not stderr.** If an attached user needs to see the message later via `fabro attach` or `fabro logs`, emit a workflow event (for example `RunNotice`) and let tracing capture the developer-oriented copy separately. +- **Detached UX belongs in events, not stderr.** If an attached user needs to see the message later via `fabro attach` or `fabro events`, emit a workflow event (for example `RunNotice`) and let tracing capture the developer-oriented copy separately. - **Wrapper variants are no-ops.** When one event enum wraps another (`PipelineEvent::Agent` wraps `AgentEvent`, `AgentEvent::SubAgentEvent` wraps a child `AgentEvent`), the wrapper's `trace()` arm is `{}` because the inner event was already traced at its origin. This prevents double-logging. - **Streaming noise variants are no-ops.** `TextDelta` and `ToolCallOutputDelta` produce no log output — per-token events would flood the logs even at DEBUG level. diff --git a/docs/public/agents/outputs.mdx b/docs/public/agents/outputs.mdx index 8af17dbf4..ac0bdffe7 100644 --- a/docs/public/agents/outputs.mdx +++ b/docs/public/agents/outputs.mdx @@ -119,7 +119,7 @@ The tracked paths are stored as `files_touched` on the stage outcome: | Location | How it's used | |---|---| -| `StageCompleted` event | Emitted with `files_touched` in the event stream and surfaced by `fabro logs` / exported event streams | +| `StageCompleted` event | Emitted with `files_touched` in the event stream and surfaced by `fabro events` / exported event streams | | Preambles | Listed under each completed stage so downstream agents know what changed | | Retros | Included per-stage and aggregated across the full run | | `status.json` | Written to the stage's logs directory after each node completes | diff --git a/docs/public/execution/observability.mdx b/docs/public/execution/observability.mdx index d10a5ec91..2e0926b73 100644 --- a/docs/public/execution/observability.mdx +++ b/docs/public/execution/observability.mdx @@ -71,14 +71,14 @@ Because event payload lives in `properties`, most shell queries should look ther ```bash # Count tool calls in a run -fabro logs 01JKXYZ... | jq -r 'select(.event == "agent.tool.started") | .properties.tool_name' | wc -l +fabro events 01JKXYZ... | jq -r 'select(.event == "agent.tool.started") | .properties.tool_name' | wc -l # Find stage failures -fabro logs 01JKXYZ... | jq 'select(.event == "stage.failed")' +fabro events 01JKXYZ... | jq 'select(.event == "stage.failed")' # See which edges were taken jq '{from: .properties.from_node, to: .properties.to_node, label: .properties.label}' \ - <(fabro logs 01JKXYZ...) | head + <(fabro events 01JKXYZ...) | head ``` If you need files on disk for offline analysis, `fabro dump` exports `events.jsonl` plus run-state projections. @@ -110,7 +110,7 @@ Lifecycle events such as `agent.sub.spawned` and `agent.sub.completed` are emitt ### API: Server-Sent Events -When running workflows through the API server, subscribe to the [run events endpoint](/api-reference/runs/stream-run-events). Each SSE payload is a serialized run event envelope in the same shape used by `fabro logs` and `events.jsonl` exports. +When running workflows through the API server, subscribe to the [run events endpoint](/api-reference/runs/stream-run-events). Each SSE payload is a serialized run event envelope in the same shape used by `fabro events` and `events.jsonl` exports. ### Web UI @@ -130,7 +130,8 @@ Post-run analysis surfaces include: | Surface | Description | |---|---| -| `fabro logs ` | Full event envelope stream as NDJSON | +| `fabro events ` | Full event envelope stream as NDJSON | +| `fabro logs ` | Raw per-run worker tracing log, when available | | `fabro inspect ` | Current durable run state, including run/start/checkpoint/conclusion records | | `fabro dump --output ` | Exported `events.jsonl` plus reconstructed JSON and node files | diff --git a/docs/public/reference/cli.mdx b/docs/public/reference/cli.mdx index 165f165bc..5abeaa31d 100644 --- a/docs/public/reference/cli.mdx +++ b/docs/public/reference/cli.mdx @@ -73,11 +73,12 @@ fabro [OPTIONS] [COMMAND] | `fabro docs` | Open the docs website in the browser | | `fabro doctor` | Check environment and integration health | | `fabro dump` | Export a run's durable state to a directory | +| `fabro events` | View the event log of a workflow run | | `fabro fork` | Fork a workflow run from an earlier checkpoint into a new run | | `fabro graph` | Render a workflow graph as SVG | | `fabro inspect` | Show detailed information about a workflow run | | `fabro install` | Set up the Fabro environment (LLMs, certs, GitHub) | -| `fabro logs` | View the event log of a workflow run | +| `fabro logs` | View the raw worker tracing log of a workflow run | | `fabro model` | List and test LLM models | | `fabro pr` | Pull request operations | | `fabro preflight` | Validate run configuration without executing | @@ -364,6 +365,30 @@ fabro dump [OPTIONS] --output | `-o, --output ` | Output directory (must not exist or be empty) | | `--server ` | Fabro server target: http(s) URL or absolute Unix socket path | +### `fabro events` + +View the event log of a workflow run + +```bash +fabro events [OPTIONS] +``` + +#### Arguments + +| Name | Description | +| --- | --- | +| `RUN` | Run ID prefix or workflow name (most recent run) | + +#### Options + +| Option | Description | +| --- | --- | +| `-f, --follow` | Follow event output | +| `-p, --pretty` | Formatted colored output with rendered assistant text | +| `--server ` | Fabro server target: http(s) URL or absolute Unix socket path | +| `--since ` | Events since timestamp or relative (e.g. "42m", "2h", "2026-01-02T13:00:00Z") | +| `-n, --tail ` | Lines from end (default: all) | + ### `fabro fork` Fork a workflow run from an earlier checkpoint into a new run @@ -467,7 +492,7 @@ fabro install github [OPTIONS] ### `fabro logs` -View the event log of a workflow run +View the raw worker tracing log of a workflow run ```bash fabro logs [OPTIONS] @@ -483,10 +508,7 @@ fabro logs [OPTIONS] | Option | Description | | --- | --- | -| `-f, --follow` | Follow log output | -| `-p, --pretty` | Formatted colored output with rendered assistant text | | `--server ` | Fabro server target: http(s) URL or absolute Unix socket path | -| `--since ` | Logs since timestamp or relative (e.g. "42m", "2h", "2026-01-02T13:00:00Z") | | `-n, --tail ` | Lines from end (default: all) | ### `fabro model` diff --git a/docs/public/reference/run-directory.mdx b/docs/public/reference/run-directory.mdx index 853652dfc..9ba03ee03 100644 --- a/docs/public/reference/run-directory.mdx +++ b/docs/public/reference/run-directory.mdx @@ -27,10 +27,10 @@ The naming format is `YYYYMMDD-{run_id}`, where `run_id` is the ULID assigned to These paths are local runtime state and caches, not the canonical run state. - **`worktree/`** — When running in worktree mode, Fabro creates a Git worktree here as the working directory for agents and commands. -- **`runtime/`** — Local runtime files. Today this is mainly materialized blob payloads under `runtime/blobs/`. +- **`runtime/`** — Local runtime files. Today this includes `runtime/server.log` for the raw per-run worker tracing log and materialized blob payloads under `runtime/blobs/`. - **`nodes/{manager_node}_{visit}/child/`** — Nested scratch directories for manager-loop child workflows. -Large durable values, event streams, checkpoints, diffs, conclusions, and retros are no longer projected into live scratch by default. Use `fabro logs`, `fabro inspect`, the API, or `fabro dump` for those surfaces. +Large durable values, event streams, checkpoints, diffs, conclusions, and retros are no longer projected into live scratch by default. Use `fabro events`, `fabro inspect`, the API, or `fabro dump` for those surfaces. Use `fabro logs` for the raw per-run worker tracing log when it is available. ## Reconstructed and export-only layouts diff --git a/lib/crates/fabro-cli/src/args.rs b/lib/crates/fabro-cli/src/args.rs index f605b166c..4204ebb88 100644 --- a/lib/crates/fabro-cli/src/args.rs +++ b/lib/crates/fabro-cli/src/args.rs @@ -365,16 +365,16 @@ pub(crate) struct RunsUnarchiveArgs { } #[derive(Args)] -pub(crate) struct LogsArgs { +pub(crate) struct EventsArgs { #[command(flatten)] pub(crate) server: ServerTargetArgs, /// Run ID prefix or workflow name (most recent run) pub(crate) run: String, - /// Follow log output + /// Follow event output #[arg(short, long)] pub(crate) follow: bool, - /// Logs since timestamp or relative (e.g. "42m", "2h", + /// Events since timestamp or relative (e.g. "42m", "2h", /// "2026-01-02T13:00:00Z") #[arg(long)] pub(crate) since: Option, @@ -386,6 +386,18 @@ pub(crate) struct LogsArgs { pub(crate) pretty: bool, } +#[derive(Args)] +pub(crate) struct LogsArgs { + #[command(flatten)] + pub(crate) server: ServerTargetArgs, + + /// Run ID prefix or workflow name (most recent run) + pub(crate) run: String, + /// Lines from end (default: all) + #[arg(short = 'n', long)] + pub(crate) tail: Option, +} + #[derive(Args)] pub(crate) struct ValidateArgs { /// Path to the .fabro workflow file @@ -974,6 +986,8 @@ pub(crate) enum RunCommands { #[command(hide = true)] Diff(DiffArgs), /// View the event log of a workflow run + Events(EventsArgs), + /// View the raw worker tracing log of a workflow run Logs(LogsArgs), /// Resume an interrupted workflow run Resume(ResumeArgs), @@ -996,6 +1010,7 @@ impl RunCommands { Self::Attach(_) => "attach", Self::RunWorker(_) => "__run-worker", Self::Diff(_) => "diff", + Self::Events(_) => "events", Self::Logs(_) => "logs", Self::Resume(_) => "resume", Self::Rewind(_) => "rewind", diff --git a/lib/crates/fabro-cli/src/commands/run/events.rs b/lib/crates/fabro-cli/src/commands/run/events.rs new file mode 100644 index 000000000..6196e6baa --- /dev/null +++ b/lib/crates/fabro-cli/src/commands/run/events.rs @@ -0,0 +1,1343 @@ +#![expect( + clippy::disallowed_types, + reason = "sync CLI `run events` command: blocking std::io::Write is the intended output mechanism" +)] +#![expect( + clippy::disallowed_methods, + reason = "sync CLI `run events` command: streams event lines to std::io::stdout directly" +)] + +use std::fmt::Write as _; +use std::io::{self, IsTerminal, Write}; +use std::time::Duration; + +use anyhow::{Context, Result, bail}; +use chrono::{DateTime, Utc}; +use fabro_redact::redact_jsonl_line; +use fabro_types::RunNoticeCode; +use fabro_util::json::normalize_json_value; +use fabro_util::terminal::Styles; +use tokio::time; +use tracing::{debug, info}; + +use crate::args::EventsArgs; +use crate::command_context::CommandContext; +use crate::server_client; +use crate::shared::format_usd_micros; + +const FOLLOW_TERMINAL_GRACE: Duration = Duration::from_millis(500); + +pub(crate) async fn run( + args: &EventsArgs, + styles: &Styles, + base_ctx: &CommandContext, +) -> Result<()> { + let ctx = base_ctx.with_target(&args.server)?; + let client = ctx.server().await?; + let run_id = client.resolve_run(&args.run).await?.run_id; + info!(run_id = %run_id, "Showing events"); + + let since_cutoff = match &args.since { + Some(value) => Some(parse_since(value)?), + None => None, + }; + + let events = client + .list_run_events(&run_id, None, None) + .await + .context("Failed to list server-backed run events")?; + let last_seq = events.last().map_or(0, |event| event.seq); + let all_lines = events + .iter() + .map(event_payload_line) + .collect::>>()?; + let filtered = apply_filters(&all_lines, since_cutoff.as_ref(), args.tail); + + let stdout = io::stdout(); + let is_tty = stdout.is_terminal(); + let mut out = stdout.lock(); + let pretty = args.pretty && !ctx.json_output(); + let mut pretty_state = PrettyEventState::default(); + + for line in &filtered { + if pretty { + if let Some(formatted) = format_event_pretty_streamed(line, styles, &mut pretty_state) { + writeln!(out, "{formatted}")?; + } + } else { + writeln!(out, "{line}")?; + } + } + + if args.follow { + follow_store_logs( + client.as_ref(), + &run_id, + if last_seq == 0 { 1 } else { last_seq + 1 }, + pretty, + styles, + is_tty, + pretty_state, + ) + .await?; + } + + Ok(()) +} + +fn event_name(event: &fabro_store::EventEnvelope) -> &str { + event.event.event_name() +} + +fn apply_filters( + lines: &[String], + since: Option<&DateTime>, + tail: Option, +) -> Vec { + let filtered: Vec = match since { + Some(cutoff) => lines + .iter() + .filter(|line| extract_timestamp(line).is_none_or(|ts| ts >= *cutoff)) + .cloned() + .collect(), + None => lines.to_vec(), + }; + + match tail { + Some(n) if n < filtered.len() => filtered[filtered.len() - n..].to_vec(), + _ => filtered, + } +} + +fn extract_timestamp(line: &str) -> Option> { + let value: serde_json::Value = serde_json::from_str(line).ok()?; + let ts_str = value.get("ts")?.as_str()?; + ts_str.parse::>().ok() +} + +pub(crate) fn parse_since(s: &str) -> Result> { + let s = s.trim(); + if s.is_empty() { + bail!("empty --since value"); + } + + if let Some(duration) = try_parse_relative_duration(s) { + return Ok(Utc::now() - duration); + } + + if let Ok(ts) = s.parse::>() { + return Ok(ts); + } + + bail!( + "invalid --since value '{s}' (expected relative like '42m', '2h', '7d' or ISO 8601 timestamp)" + ) +} + +fn try_parse_relative_duration(s: &str) -> Option { + if s.len() < 2 { + return None; + } + let (num_str, unit) = s.split_at(s.len() - 1); + let num = i64::try_from(num_str.parse::().ok()?).ok()?; + match unit { + "s" => Some(chrono::Duration::seconds(num)), + "m" => Some(chrono::Duration::minutes(num)), + "h" => Some(chrono::Duration::hours(num)), + "d" => Some(chrono::Duration::days(num)), + _ => None, + } +} + +async fn follow_store_logs( + client: &server_client::Client, + run_id: &fabro_types::RunId, + seq: u32, + pretty: bool, + styles: &Styles, + _is_tty: bool, + mut pretty_state: PrettyEventState, +) -> Result<()> { + let stdout = io::stdout(); + let mut out = stdout.lock(); + let mut next_seq = seq; + let mut terminal_deadline = None; + + loop { + match time::timeout( + Duration::from_millis(200), + client.list_run_events(run_id, Some(next_seq), None), + ) + .await + { + Ok(Ok(events)) => { + let had_events = !events.is_empty(); + let saw_terminal = events + .iter() + .any(|event| matches!(event_name(event), "run.completed" | "run.failed")); + for event in events { + let line = event_payload_line(&event)?; + if pretty { + if let Some(formatted) = + format_event_pretty_streamed(&line, styles, &mut pretty_state) + { + writeln!(out, "{formatted}")?; + } + } else { + writeln!(out, "{line}")?; + } + out.flush()?; + next_seq = event.seq.saturating_add(1); + } + if saw_terminal || (terminal_deadline.is_some() && had_events) { + terminal_deadline = Some(time::Instant::now() + FOLLOW_TERMINAL_GRACE); + } + } + Err(_) => { + if run_concluded(client, run_id).await? { + terminal_deadline + .get_or_insert_with(|| time::Instant::now() + FOLLOW_TERMINAL_GRACE); + } + } + Ok(Err(err)) => return Err(err), + } + + let Some(deadline) = terminal_deadline else { + continue; + }; + if time::Instant::now() < deadline { + continue; + } + + let flushed_next_seq = flush_remaining_store_events( + client, + run_id, + next_seq, + pretty, + styles, + &mut pretty_state, + &mut out, + ) + .await?; + if flushed_next_seq > next_seq { + next_seq = flushed_next_seq; + terminal_deadline = Some(time::Instant::now() + FOLLOW_TERMINAL_GRACE); + continue; + } + + debug!("Run reached terminal status and log tail is quiet, stopping follow"); + break; + } + + Ok(()) +} + +async fn run_concluded( + client: &server_client::Client, + run_id: &fabro_types::RunId, +) -> Result { + let state = client + .get_run_state(run_id) + .await + .context("Failed to read run state from server while following events")?; + Ok(state.conclusion.is_some() + || state + .status + .is_some_and(fabro_types::RunStatus::is_terminal)) +} + +async fn flush_remaining_store_events( + client: &server_client::Client, + run_id: &fabro_types::RunId, + next_seq: u32, + pretty: bool, + styles: &Styles, + pretty_state: &mut PrettyEventState, + out: &mut dyn Write, +) -> Result { + let events = client + .list_run_events(run_id, Some(next_seq), None) + .await + .context("Failed to list server-backed run events while finalizing follow")?; + + let mut next_seq = next_seq; + for event in events { + let line = event_payload_line(&event)?; + if pretty { + if let Some(formatted) = format_event_pretty_streamed(&line, styles, pretty_state) { + writeln!(out, "{formatted}")?; + } + } else { + writeln!(out, "{line}")?; + } + next_seq = event.seq.saturating_add(1); + } + out.flush()?; + Ok(next_seq) +} + +fn event_payload_line(event: &fabro_store::EventEnvelope) -> Result { + let mut value = normalize_json_value(event.event.to_value()?); + restore_empty_run_properties(&mut value); + let line = serde_json::to_string(&value)?; + Ok(redact_jsonl_line(&line)) +} + +fn restore_empty_run_properties(value: &mut serde_json::Value) { + let Some(object) = value.as_object_mut() else { + return; + }; + let Some(event_name) = object.get("event").and_then(serde_json::Value::as_str) else { + return; + }; + if matches!(event_name, "run.submitted" | "run.running") && !object.contains_key("properties") { + let run_id = object.remove("run_id"); + let ts = object.remove("ts"); + object.insert("properties".to_string(), serde_json::json!({})); + if let Some(run_id) = run_id { + object.insert("run_id".to_string(), run_id); + } + if let Some(ts) = ts { + object.insert("ts".to_string(), ts); + } + } +} + +fn render_indented_markdown(styles: &Styles, text: &str, indent: &str) -> String { + let term_width = Styles::terminal_width(); + let wrap_width = term_width.saturating_sub(indent.len()); + let rendered = styles.render_markdown_width(text, wrap_width); + rendered + .lines() + .map(|line| format!("{indent}{line}")) + .collect::>() + .join("\n") +} + +#[derive(Debug, Default)] +struct PrettyEventState { + saw_metadata_snapshot_failure: bool, +} + +fn format_event_pretty_streamed( + line: &str, + styles: &Styles, + state: &mut PrettyEventState, +) -> Option { + let envelope: serde_json::Value = serde_json::from_str(line).ok()?; + let event = envelope.get("event")?.as_str()?; + if event == "run.notice" + && state.saw_metadata_snapshot_failure + && is_metadata_snapshot_compat_notice(&envelope) + { + return None; + } + let formatted = format_event_pretty_value(&envelope, styles); + if event == "metadata.snapshot.failed" { + state.saw_metadata_snapshot_failure = true; + } + formatted +} + +#[cfg_attr( + not(test), + allow( + dead_code, + reason = "Production pretty events use the stateful stream formatter; unit tests exercise this single-line helper." + ) +)] +pub(crate) fn format_event_pretty(line: &str, styles: &Styles) -> Option { + let envelope: serde_json::Value = serde_json::from_str(line).ok()?; + format_event_pretty_value(&envelope, styles) +} + +fn format_event_pretty_value(envelope: &serde_json::Value, styles: &Styles) -> Option { + let event = envelope.get("event")?.as_str()?; + let ts = format_timestamp(envelope.get("ts")?.as_str()?); + + match event { + "run.started" => { + let name = prop_str_field(envelope, "name").unwrap_or("?"); + let run_id = str_field(envelope, "run_id").unwrap_or("?"); + let header = format!( + "{} {} {} {}", + styles.dim.apply_to(&ts), + styles.bold_cyan.apply_to("\u{25b6}"), + styles.bold.apply_to(name), + styles.dim.apply_to(run_id), + ); + match prop_str_field(envelope, "goal") { + Some(goal) if !goal.is_empty() => { + let body = render_indented_markdown(styles, goal, " "); + Some(format!("{header}\n{body}\n")) + } + _ => Some(header), + } + } + "run.completed" => { + let duration = format_duration_ms(prop_field(envelope, "duration_ms")); + let status_str = match prop_str_field(envelope, "status") { + Some(status) if !status.is_empty() => status, + _ => "succeeded", + }; + let status_upper = status_str.to_uppercase(); + let status_style = match status_str { + "succeeded" | "partially_succeeded" => &styles.bold_green, + _ => &styles.bold_red, + }; + let cost = format_cost( + prop_field(envelope, "total_usd_micros") + .or_else(|| prop_field(envelope, "total_cost")), + ); + + let mut summary = format!( + "{} {} {}", + styles.dim.apply_to(&ts), + status_style.apply_to(format!("\u{2713} {status_upper}")), + styles.bold.apply_to(&duration), + ); + if !cost.is_empty() { + write!(summary, " {}", styles.dim.apply_to(&cost)).expect("write to string"); + } + + let mut lines = vec![summary]; + + if let Some(billing) = + prop_field(envelope, "billing").or_else(|| prop_field(envelope, "usage")) + { + let total = billing + .get("total_tokens") + .and_then(serde_json::Value::as_u64) + .unwrap_or(0); + let pad = " ".repeat(ts.len() + 1); + if total > 0 { + lines.push(format!( + "{}{}", + pad, + styles + .dim + .apply_to(format!("Tokens: {}", format_tokens(total))) + )); + } + if let Some(cache_read) = billing + .get("cache_read_tokens") + .and_then(serde_json::Value::as_u64) + { + let cache_write = billing + .get("cache_write_tokens") + .and_then(serde_json::Value::as_u64) + .unwrap_or(0); + lines.push(format!( + "{}{}", + pad, + styles.dim.apply_to(format!( + "Cache: {} read, {} write", + format_tokens(cache_read), + format_tokens(cache_write) + )) + )); + } + if let Some(reasoning) = billing + .get("reasoning_tokens") + .and_then(serde_json::Value::as_u64) + { + if reasoning > 0 { + lines.push(format!( + "{}{}", + pad, + styles.dim.apply_to(format!( + "Reasoning: {} tokens", + format_tokens(reasoning) + )) + )); + } + } + } + + Some(lines.join("\n")) + } + "run.failed" => { + let error = prop_str_field(envelope, "error").unwrap_or("unknown error"); + Some(format!( + "{} {} {}", + styles.dim.apply_to(&ts), + styles.bold_red.apply_to("\u{2717} Failed"), + styles.red.apply_to(error), + )) + } + "run.notice" => { + let level = prop_str_field(envelope, "level").unwrap_or("info"); + let code = prop_str_field(envelope, "code").unwrap_or(""); + let message = prop_str_field(envelope, "message").unwrap_or(""); + let label = match level { + "warn" => styles.yellow.apply_to("Warning:").to_string(), + "error" => styles.bold_red.apply_to("Error:").to_string(), + _ => styles.bold.apply_to("Info:").to_string(), + }; + let code_suffix = if code.is_empty() { + String::new() + } else { + format!(" {}", styles.dim.apply_to(format!("[{code}]"))) + }; + Some(format!( + "{} {} {}{}", + styles.dim.apply_to(&ts), + label, + message, + code_suffix, + )) + } + "metadata.snapshot.completed" => { + let phase = prop_str_field(envelope, "phase").unwrap_or("?"); + let duration = format_duration_ms(prop_field(envelope, "duration_ms")); + Some(format!( + "{} Metadata {} {}", + styles.dim.apply_to(&ts), + phase, + styles.dim.apply_to(&duration), + )) + } + "metadata.snapshot.failed" => { + let phase = prop_str_field(envelope, "phase").unwrap_or("?"); + let failure_kind = prop_str_field(envelope, "failure_kind").unwrap_or(""); + let error = prop_str_field(envelope, "error").unwrap_or("unknown error"); + let kind_suffix = if failure_kind.is_empty() { + String::new() + } else { + format!(" {}", styles.dim.apply_to(format!("[{failure_kind}]"))) + }; + Some(format!( + "{} {} Metadata {} failed: {}{}", + styles.dim.apply_to(&ts), + styles.yellow.apply_to("Warning:"), + phase, + error, + kind_suffix, + )) + } + "stage.started" => { + let label = str_field(envelope, "node_label").unwrap_or("?"); + Some(format!( + "{} {} {}", + styles.dim.apply_to(&ts), + styles.bold_cyan.apply_to("\u{25b6}"), + styles.bold.apply_to(label), + )) + } + "stage.completed" => { + let label = str_field(envelope, "node_label").unwrap_or("?"); + let duration = format_duration_ms(prop_field(envelope, "duration_ms")); + let billing = prop_field(envelope, "billing").or_else(|| prop_field(envelope, "usage")); + let cost = format_cost( + billing + .and_then(|value| value.get("total_usd_micros")) + .or_else(|| billing.and_then(|value| value.get("cost"))), + ); + let input_tokens = billing + .and_then(|value| value.get("input_tokens")) + .and_then(serde_json::Value::as_u64) + .unwrap_or(0); + let output_tokens = billing + .and_then(|value| value.get("output_tokens")) + .and_then(serde_json::Value::as_u64) + .unwrap_or(0); + let token_total = input_tokens.saturating_add(output_tokens); + let mut line = format!( + "{} {} {} {} {}", + styles.dim.apply_to(&ts), + styles.green.apply_to("\u{2713}"), + styles.bold.apply_to(label), + cost, + duration, + ); + if token_total > 0 { + let _ = write!( + line, + " {}", + styles.dim.apply_to(format_tokens(token_total)) + ); + } + Some(line) + } + "stage.failed" => { + let label = str_field(envelope, "node_label").unwrap_or("?"); + let error = prop_str_field(envelope, "error").unwrap_or("unknown error"); + Some(format!( + "{} {} {} {}", + styles.dim.apply_to(&ts), + styles.red.apply_to("\u{2717}"), + styles.bold.apply_to(label), + styles.red.apply_to(error), + )) + } + "agent.message" => { + let stage = str_field(envelope, "node_id").unwrap_or("?"); + let model = prop_str_field(envelope, "model").unwrap_or("?"); + let text = prop_str_field(envelope, "text").unwrap_or(""); + let header = format!( + "{} {} {} {}{}{}", + styles.dim.apply_to(&ts), + "\u{1f4ac}", + styles.bold.apply_to(stage), + styles.dim.apply_to("["), + styles.dim.apply_to(model), + styles.dim.apply_to("]"), + ); + let body = render_indented_markdown(styles, text, " "); + Some(format!("{header}\n{body}\n")) + } + "agent.tool.started" => { + let tool = prop_str_field(envelope, "tool_name").unwrap_or("?"); + let detail = tool_detail(envelope); + let display = match detail { + Some(value) => format!("{tool}({value})"), + None => tool.to_string(), + }; + Some(format!( + "{} {} {}", + styles.dim.apply_to(&ts), + styles.dim.apply_to("\u{2699}"), + styles.dim.apply_to(&display), + )) + } + "agent.tool.completed" => { + let tool = prop_str_field(envelope, "tool_name").unwrap_or("?"); + let is_error = prop_field(envelope, "is_error") + .and_then(serde_json::Value::as_bool) + .unwrap_or(false); + let detail = tool_detail(envelope); + let display = match detail { + Some(value) => format!("{tool}({value})"), + None => tool.to_string(), + }; + let glyph = if is_error { "\u{2717}" } else { "\u{2713}" }; + let style = if is_error { &styles.red } else { &styles.green }; + Some(format!( + "{} {} {}", + styles.dim.apply_to(&ts), + style.apply_to(glyph), + display, + )) + } + "edge.selected" => { + let to = prop_str_field(envelope, "to_node").unwrap_or("?"); + let reason = prop_str_field(envelope, "reason").unwrap_or("?"); + let condition = prop_str_field(envelope, "condition"); + let detail = match condition { + Some(value) => format!(" [{value}]"), + None => String::new(), + }; + Some(format!( + "{} {} {} {}{}", + styles.dim.apply_to(&ts), + styles.dim.apply_to("\u{2192}"), + to, + styles.dim.apply_to(reason), + styles.dim.apply_to(&detail), + )) + } + "sandbox.ready" => { + let provider = prop_str_field(envelope, "provider").unwrap_or("?"); + let duration = format_duration_ms(prop_field(envelope, "duration_ms")); + Some(format!( + "{} Sandbox: {} {}", + styles.dim.apply_to(&ts), + provider, + styles.dim.apply_to(&duration), + )) + } + "sandbox.snapshot.pulling" => { + let name = prop_str_field(envelope, "name").unwrap_or("?"); + Some(format!( + "{} Sandbox: pulling {}", + styles.dim.apply_to(&ts), + name, + )) + } + "sandbox.snapshot.creating" => { + let name = prop_str_field(envelope, "name").unwrap_or("?"); + Some(format!( + "{} Sandbox: building {}", + styles.dim.apply_to(&ts), + name, + )) + } + "sandbox.snapshot.ready" => { + let name = prop_str_field(envelope, "name").unwrap_or("?"); + let duration = format_duration_ms(prop_field(envelope, "duration_ms")); + Some(format!( + "{} Sandbox snapshot: {} {}", + styles.dim.apply_to(&ts), + name, + styles.dim.apply_to(&duration), + )) + } + "sandbox.snapshot.failed" => { + let name = prop_str_field(envelope, "name").unwrap_or("?"); + let error = prop_str_field(envelope, "error").unwrap_or("unknown error"); + Some(format!( + "{} {} Sandbox snapshot {} failed: {}", + styles.dim.apply_to(&ts), + styles.bold_red.apply_to("\u{2717}"), + name, + styles.red.apply_to(error), + )) + } + "setup.completed" => { + let count = prop_field(envelope, "command_count").and_then(serde_json::Value::as_u64); + let duration = format_duration_ms(prop_field(envelope, "duration_ms")); + Some(match count { + Some(count) => format!( + "{} Setup: {} commands {}", + styles.dim.apply_to(&ts), + count, + styles.dim.apply_to(&duration), + ), + None => format!( + "{} Setup: {}", + styles.dim.apply_to(&ts), + styles.dim.apply_to(&duration), + ), + }) + } + "agent.compaction.completed" => { + let original = prop_field(envelope, "original_turn_count") + .and_then(serde_json::Value::as_u64) + .unwrap_or(0); + let preserved = prop_field(envelope, "preserved_turn_count") + .and_then(serde_json::Value::as_u64) + .unwrap_or(0); + Some(format!( + "{} {}", + styles.dim.apply_to(&ts), + styles + .dim + .apply_to(format!("compaction: {original}\u{2192}{preserved} turns")), + )) + } + "parallel.started" => { + let count = prop_field(envelope, "branch_count") + .and_then(serde_json::Value::as_u64) + .unwrap_or(0); + Some(format!( + "{} {} Parallel {} branches", + styles.dim.apply_to(&ts), + styles.bold_cyan.apply_to("\u{25b6}"), + count, + )) + } + "parallel.branch.started" => { + let label = str_field(envelope, "node_label").unwrap_or("?"); + Some(format!( + "{} {} {}", + styles.dim.apply_to(&ts), + styles.cyan.apply_to("\u{25b6}"), + label, + )) + } + "parallel.branch.completed" => { + let label = str_field(envelope, "node_label").unwrap_or("?"); + Some(format!( + "{} {} {}", + styles.dim.apply_to(&ts), + styles.green.apply_to("\u{2713}"), + label, + )) + } + "parallel.completed" => { + let duration = format_duration_ms(prop_field(envelope, "duration_ms")); + Some(format!( + "{} {} Parallel {}", + styles.dim.apply_to(&ts), + styles.green.apply_to("\u{2713}"), + duration, + )) + } + "pull_request.created" => { + let url = prop_str_field(envelope, "pr_url").unwrap_or("?"); + let draft = prop_field(envelope, "draft") + .and_then(serde_json::Value::as_bool) + .unwrap_or(false); + let label = if draft { "Draft PR:" } else { "PR:" }; + Some(format!( + "{} {} {}", + styles.dim.apply_to(&ts), + styles.bold.apply_to(label), + url, + )) + } + "pull_request.failed" => { + let error = prop_str_field(envelope, "error").unwrap_or("unknown error"); + Some(format!( + "{} {} {}", + styles.dim.apply_to(&ts), + styles.bold_red.apply_to("PR failed:"), + styles.red.apply_to(error), + )) + } + "retro.completed" => { + let duration = format_duration_ms(prop_field(envelope, "duration_ms")); + Some(format!( + "{} {} Retro {}", + styles.dim.apply_to(&ts), + styles.green.apply_to("\u{2713}"), + duration, + )) + } + "retro.failed" => { + let error = prop_str_field(envelope, "error").unwrap_or("unknown error"); + let duration = format_duration_ms(prop_field(envelope, "duration_ms")); + Some(format!( + "{} {} Retro {} {}", + styles.dim.apply_to(&ts), + styles.bold_red.apply_to("\u{2717}"), + duration, + styles.red.apply_to(error), + )) + } + "retro.started" => Some(format!( + "{} {} Retro", + styles.dim.apply_to(&ts), + styles.bold_cyan.apply_to("\u{25b6}"), + )), + _ => None, + } +} + +fn is_metadata_snapshot_compat_notice(envelope: &serde_json::Value) -> bool { + prop_str_field(envelope, "code") + .and_then(|code| code.parse::().ok()) + .is_some_and(RunNoticeCode::is_metadata_snapshot_compat) +} + +fn str_field<'a>(value: &'a serde_json::Value, key: &str) -> Option<&'a str> { + value.get(key)?.as_str() +} + +fn prop_field<'a>(value: &'a serde_json::Value, key: &str) -> Option<&'a serde_json::Value> { + value.get("properties")?.get(key) +} + +fn prop_str_field<'a>(value: &'a serde_json::Value, key: &str) -> Option<&'a str> { + prop_field(value, key)?.as_str() +} + +fn format_timestamp(ts: &str) -> String { + ts.parse::>() + .map_or_else(|_| ts.to_string(), |dt| dt.format("%H:%M:%S").to_string()) +} + +fn format_duration_ms(value: Option<&serde_json::Value>) -> String { + let ms = value.and_then(serde_json::Value::as_u64).unwrap_or(0); + if ms < 1000 { + format!("{ms}ms") + } else { + let secs = ms as f64 / 1000.0; + if secs < 60.0 { + format!("{secs:.0}s") + } else { + let mins = secs / 60.0; + format!("{mins:.1}m") + } + } +} + +fn format_cost(value: Option<&serde_json::Value>) -> String { + match value { + Some(value) => { + if let Some(usd_micros) = value.as_i64() { + if usd_micros > 0 { + return format_usd_micros(usd_micros); + } + } + let cost = value.as_f64().unwrap_or(0.0); + if cost > 0.0 { + format!("${cost:.2}") + } else { + String::new() + } + } + None => String::new(), + } +} + +fn format_tokens(tokens: u64) -> String { + if tokens >= 1000 { + format!("{:.1}k toks", tokens as f64 / 1000.0) + } else { + format!("{tokens} toks") + } +} + +fn tool_detail(envelope: &serde_json::Value) -> Option { + let tool_name = prop_str_field(envelope, "tool_name")?; + let arguments = prop_field(envelope, "arguments")?; + let arg = |key: &str| arguments.get(key).and_then(|v| v.as_str()); + + match tool_name { + "bash" | "shell" | "execute_command" => arg("command").map(|c| truncate(c, 60)), + "glob" => arg("pattern").map(String::from), + "grep" | "ripgrep" => arg("pattern").map(|p| truncate(p, 40)), + "read_file" | "read" => arg("path") + .or_else(|| arg("file_path")) + .map(|p| truncate(p, 60)), + "write_file" | "write" | "create_file" => arg("path") + .or_else(|| arg("file_path")) + .map(|p| truncate(p, 60)), + "edit_file" | "edit" => arg("path") + .or_else(|| arg("file_path")) + .map(|p| truncate(p, 60)), + "list_dir" => arg("path") + .or_else(|| arg("file_path")) + .map(|p| truncate(p, 60)), + "web_search" => arg("query").map(|q| truncate(q, 60)), + "web_fetch" => arg("url").map(|u| truncate(u, 60)), + "spawn_agent" => arg("task").map(|t| truncate(t, 60)), + "wait" | "send_input" | "close_agent" => arg("agent_id").map(String::from), + "use_skill" => arg("skill_name").map(String::from), + "apply_patch" => Some("…".into()), + "read_many_files" => arguments + .get("paths") + .and_then(|v| v.as_array()) + .map(|a| format!("{} files", a.len())), + _ => None, + } +} + +fn truncate(s: &str, max: usize) -> String { + if s.len() <= max { + s.to_string() + } else { + let boundary = s.floor_char_boundary(max.saturating_sub(1)); + format!("{}\u{2026}", &s[..boundary]) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn no_color_styles() -> Styles { + Styles::new(false) + } + + #[test] + fn parse_since_relative_minutes() { + let before = Utc::now(); + let result = parse_since("42m").unwrap(); + let after = Utc::now(); + let expected_lower = after - chrono::Duration::minutes(42) - chrono::Duration::seconds(1); + let expected_upper = before - chrono::Duration::minutes(42) + chrono::Duration::seconds(1); + assert!(result >= expected_lower && result <= expected_upper); + } + + #[test] + fn parse_since_relative_hours() { + let before = Utc::now(); + let result = parse_since("2h").unwrap(); + let expected = before - chrono::Duration::hours(2); + assert!((result - expected).num_seconds().abs() < 2); + } + + #[test] + fn parse_since_relative_days() { + let before = Utc::now(); + let result = parse_since("7d").unwrap(); + let expected = before - chrono::Duration::days(7); + assert!((result - expected).num_seconds().abs() < 2); + } + + #[test] + fn parse_since_iso8601() { + let result = parse_since("2026-01-01T12:00:00Z").unwrap(); + assert_eq!(result.to_rfc3339(), "2026-01-01T12:00:00+00:00"); + } + + #[test] + fn parse_since_invalid() { + assert!(parse_since("").is_err()); + assert!(parse_since("abc").is_err()); + assert!(parse_since("notadate").is_err()); + } + + #[test] + fn parse_since_overflow_is_invalid() { + assert!(parse_since("9223372036854775808s").is_err()); + } + + #[test] + fn tail_returns_last_n_lines() { + let lines: Vec = (0..10).map(|i| format!("line {i}")).collect(); + let result = apply_filters(&lines, None, Some(3)); + assert_eq!(result.len(), 3); + assert_eq!(result[0], "line 7"); + assert_eq!(result[2], "line 9"); + } + + #[test] + fn tail_all_when_n_exceeds_total() { + let lines: Vec = (0..3).map(|i| format!("line {i}")).collect(); + let result = apply_filters(&lines, None, Some(100)); + assert_eq!(result.len(), 3); + } + + #[test] + fn since_filters_by_timestamp() { + let cutoff = "2026-01-01T12:00:00Z".parse::>().unwrap(); + let lines = vec![ + r#"{"ts":"2026-01-01T11:00:00Z","event":"stage.started"}"#.to_string(), + r#"{"ts":"2026-01-01T12:30:00Z","event":"stage.completed"}"#.to_string(), + r#"{"ts":"2026-01-01T13:00:00Z","event":"run.completed"}"#.to_string(), + ]; + let result = apply_filters(&lines, Some(&cutoff), None); + assert_eq!(result.len(), 2); + } + + #[test] + fn raw_lines_pass_through_verbatim() { + let lines = vec![ + r#"{"ts":"2026-01-01T12:00:00Z","event":"stage.started","node_label":"plan"}"# + .to_string(), + ]; + let result = apply_filters(&lines, None, None); + assert_eq!(result, lines); + } + + #[test] + fn pretty_stage_started() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:23:09Z","event":"stage.started","node_label":"plan","node_id":"plan","properties":{"index":0}}"#; + let result = format_event_pretty(line, &styles).unwrap(); + assert!(result.contains("plan"), "got: {result}"); + assert!(result.contains("\u{25b6}"), "got: {result}"); + } + + #[test] + fn pretty_stage_completed() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:23:15Z","event":"stage.completed","node_label":"plan","properties":{"duration_ms":8000,"status":"succeeded","usage":{"cost":0.12,"input_tokens":10000,"output_tokens":5200}}}"#; + let result = format_event_pretty(line, &styles).unwrap(); + assert!(result.contains("plan"), "got: {result}"); + assert!(result.contains("$0.12"), "got: {result}"); + assert!(result.contains("8s"), "got: {result}"); + assert!(result.contains("15.2k toks"), "got: {result}"); + } + + #[test] + fn pretty_assistant_message() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:23:12Z","event":"agent.message","node_id":"plan","properties":{"model":"claude-opus-4-6","text":"I'll start by reading the code.","usage":{"input_tokens":100,"output_tokens":50},"tool_call_count":0}}"#; + let result = format_event_pretty(line, &styles).unwrap(); + assert!(result.contains("plan"), "got: {result}"); + assert!(result.contains("claude-opus-4-6"), "got: {result}"); + assert!(result.contains("reading the code"), "got: {result}"); + } + + #[test] + fn pretty_tool_call_started() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:23:12Z","event":"agent.tool.started","properties":{"tool_name":"read_file","tool_call_id":"tc_1","arguments":{"path":"src/main.rs"}}}"#; + let result = format_event_pretty(line, &styles).unwrap(); + assert!(result.contains("read_file"), "got: {result}"); + assert!(result.contains("src/main.rs"), "got: {result}"); + } + + #[test] + fn pretty_skips_noise_events() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:23:12Z","event":"agent.text.delta","properties":{"delta":"hello"}}"#; + assert!(format_event_pretty(line, &styles).is_none()); + } + + #[test] + fn pretty_skips_assistant_output_replace_noise_event() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:23:12Z","event":"agent.output.replace","properties":{"text":""}}"#; + assert!(format_event_pretty(line, &styles).is_none()); + } + + #[test] + fn pretty_unknown_events_return_none() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:23:12Z","event":"SomeFutureEvent","data":123}"#; + assert!(format_event_pretty(line, &styles).is_none()); + } + + #[test] + fn pretty_workflow_run_started() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:23:01Z","run_id":"abc123","event":"run.started","properties":{"name":"smoke"}}"#; + let result = format_event_pretty(line, &styles).unwrap(); + assert!(result.contains("smoke"), "got: {result}"); + assert!(result.contains("abc123"), "got: {result}"); + } + + #[test] + fn pretty_workflow_run_started_with_goal() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:23:01Z","run_id":"abc123","event":"run.started","properties":{"name":"smoke","goal":"Fix the bug"}}"#; + let result = format_event_pretty(line, &styles).unwrap(); + assert!(result.contains("smoke"), "got: {result}"); + assert!(result.contains("abc123"), "got: {result}"); + assert!(result.contains("Fix the bug"), "got: {result}"); + assert!(result.contains('\n'), "got: {result}"); + } + + #[test] + fn pretty_workflow_run_started_without_goal_no_extra_lines() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:23:01Z","run_id":"abc123","event":"run.started","properties":{"name":"smoke"}}"#; + let result = format_event_pretty(line, &styles).unwrap(); + assert!(!result.contains('\n'), "got: {result}"); + } + + #[test] + fn pretty_workflow_run_completed() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:23:32Z","run_id":"abc123","event":"run.completed","properties":{"duration_ms":25000,"status":"succeeded","total_usd_micros":570000,"billing":{"input_tokens":5000,"output_tokens":2000,"total_tokens":7000,"cache_read_tokens":3000,"cache_write_tokens":500,"reasoning_tokens":800}}}"#; + let result = format_event_pretty(line, &styles).unwrap(); + assert!(result.contains("SUCCEEDED"), "got: {result}"); + assert!(result.contains("25s"), "got: {result}"); + assert!(result.contains("$0.57"), "got: {result}"); + assert!(result.contains("7.0k toks"), "got: {result}"); + assert!(result.contains("Cache:"), "got: {result}"); + assert!(result.contains("3.0k toks read"), "got: {result}"); + assert!(result.contains("Reasoning:"), "got: {result}"); + } + + #[test] + fn pretty_workflow_run_completed_backward_compat() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:23:32Z","run_id":"abc123","event":"run.completed","properties":{"duration_ms":25000,"total_cost":0.57}}"#; + let result = format_event_pretty(line, &styles).unwrap(); + assert!(result.contains("SUCCEEDED"), "got: {result}"); + assert!(result.contains("25s"), "got: {result}"); + assert!(result.contains("$0.57"), "got: {result}"); + assert!(!result.contains("Tokens:"), "got: {result}"); + } + + #[test] + fn pretty_workflow_run_completed_fail_status() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:23:32Z","event":"run.completed","properties":{"duration_ms":25000,"status":"failed"}}"#; + let result = format_event_pretty(line, &styles).unwrap(); + assert!(result.contains("FAIL"), "got: {result}"); + } + + #[test] + fn pretty_pull_request_created() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"pull_request.created","properties":{"pr_url":"https://github.com/owner/repo/pull/42","pr_number":42,"draft":false}}"#; + let result = format_event_pretty(line, &styles).unwrap(); + assert!(result.contains("PR:"), "got: {result}"); + assert!( + result.contains("https://github.com/owner/repo/pull/42"), + "got: {result}" + ); + } + + #[test] + fn pretty_pull_request_created_draft() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"pull_request.created","properties":{"pr_url":"https://github.com/owner/repo/pull/42","pr_number":42,"draft":true}}"#; + let result = format_event_pretty(line, &styles).unwrap(); + assert!(result.contains("Draft PR:"), "got: {result}"); + } + + #[test] + fn pretty_pull_request_failed() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"pull_request.failed","properties":{"error":"auth token expired"}}"#; + let result = format_event_pretty(line, &styles).unwrap(); + assert!(result.contains("PR failed:"), "got: {result}"); + assert!(result.contains("auth token expired"), "got: {result}"); + } + + #[test] + fn pretty_run_notice_warn() { + let styles = no_color_styles(); + let code = RunNoticeCode::SandboxCleanupFailed.to_string(); + let line = serde_json::json!({ + "ts": "2026-01-01T14:25:00Z", + "event": "run.notice", + "properties": { + "level": "warn", + "code": code, + "message": "sandbox cleanup failed: boom", + }, + }) + .to_string(); + let result = format_event_pretty(&line, &styles).unwrap(); + assert!(result.contains("Warning:"), "got: {result}"); + assert!( + result.contains("sandbox cleanup failed: boom"), + "got: {result}" + ); + assert!( + result.contains(&format!("[{}]", RunNoticeCode::SandboxCleanupFailed)), + "got: {result}" + ); + } + + #[test] + fn pretty_run_notice_error() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"run.notice","properties":{"level":"error","code":"launch_failed","message":"failed to start engine"}}"#; + let result = format_event_pretty(line, &styles).unwrap(); + assert!(result.contains("Error:"), "got: {result}"); + assert!(result.contains("failed to start engine"), "got: {result}"); + assert!(result.contains("[launch_failed]"), "got: {result}"); + } + + #[test] + fn pretty_metadata_snapshot_completed() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"metadata.snapshot.completed","properties":{"phase":"checkpoint","branch":"fabro/meta","duration_ms":2800,"entry_count":2,"bytes":42,"commit_sha":"abc123"}}"#; + let result = format_event_pretty(line, &styles).unwrap(); + assert!(result.contains("Metadata checkpoint"), "got: {result}"); + assert!(result.contains("3s"), "got: {result}"); + } + + #[test] + fn pretty_metadata_snapshot_failed() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"metadata.snapshot.failed","properties":{"phase":"finalize","branch":"fabro/meta","duration_ms":900,"failure_kind":"push","error":"push rejected","commit_sha":"abc123","entry_count":2,"bytes":42}}"#; + let result = format_event_pretty(line, &styles).unwrap(); + assert!(result.contains("Warning:"), "got: {result}"); + assert!( + result.contains("Metadata finalize failed: push rejected"), + "got: {result}" + ); + assert!(result.contains("[push]"), "got: {result}"); + } + + #[test] + fn pretty_sandbox_snapshot_pulling() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"sandbox.snapshot.pulling","properties":{"name":"buildpack-deps:noble"}}"#; + let result = format_event_pretty(line, &styles).unwrap(); + assert!(result.contains("Sandbox: pulling"), "got: {result}"); + assert!(result.contains("buildpack-deps:noble"), "got: {result}"); + } + + #[test] + fn pretty_sandbox_snapshot_creating() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"sandbox.snapshot.creating","properties":{"name":"fabro-v9-test"}}"#; + let result = format_event_pretty(line, &styles).unwrap(); + assert!(result.contains("Sandbox: building"), "got: {result}"); + assert!(result.contains("fabro-v9-test"), "got: {result}"); + } + + #[test] + fn pretty_sandbox_snapshot_ready() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"sandbox.snapshot.ready","properties":{"name":"buildpack-deps:noble","duration_ms":8200}}"#; + let result = format_event_pretty(line, &styles).unwrap(); + assert!(result.contains("Sandbox snapshot:"), "got: {result}"); + assert!(result.contains("buildpack-deps:noble"), "got: {result}"); + assert!(result.contains("8s"), "got: {result}"); + } + + #[test] + fn pretty_sandbox_snapshot_failed() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"sandbox.snapshot.failed","properties":{"name":"buildpack-deps:noble","error":"pull failed"}}"#; + let result = format_event_pretty(line, &styles).unwrap(); + assert!( + result.contains("Sandbox snapshot buildpack-deps:noble failed: pull failed"), + "got: {result}" + ); + } + + #[test] + fn pretty_stream_suppresses_metadata_compat_notice_only() { + let styles = no_color_styles(); + let failed = r#"{"ts":"2026-01-01T14:25:00Z","event":"metadata.snapshot.failed","properties":{"phase":"checkpoint","branch":"fabro/meta","duration_ms":900,"failure_kind":"write","error":"write failed"}}"#; + let compat_notice = serde_json::json!({ + "ts": "2026-01-01T14:25:01Z", + "event": "run.notice", + "properties": { + "level": "warn", + "code": RunNoticeCode::CheckpointMetadataWriteFailed, + "message": "legacy metadata warning", + }, + }) + .to_string(); + let degraded_notice = serde_json::json!({ + "ts": "2026-01-01T14:25:02Z", + "event": "run.notice", + "properties": { + "level": "warn", + "code": RunNoticeCode::CheckpointMetadataDegraded, + "message": "metadata snapshots disabled", + }, + }) + .to_string(); + let mut state = PrettyEventState::default(); + + assert!(format_event_pretty_streamed(failed, &styles, &mut state).is_some()); + assert!(format_event_pretty_streamed(&compat_notice, &styles, &mut state).is_none()); + let degraded = format_event_pretty_streamed(°raded_notice, &styles, &mut state).unwrap(); + assert!( + degraded.contains("metadata snapshots disabled"), + "got: {degraded}" + ); + } + + #[test] + fn pretty_workflow_run_failed() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:23:32Z","run_id":"abc123","event":"run.failed","properties":{"error":"sandbox timeout"}}"#; + let result = format_event_pretty(line, &styles).unwrap(); + assert!(result.contains("Failed"), "got: {result}"); + assert!(result.contains("sandbox timeout"), "got: {result}"); + } + + #[test] + fn pretty_setup_completed_without_command_count() { + let styles = no_color_styles(); + let line = r#"{"ts":"2026-01-01T14:23:32Z","event":"setup.completed","properties":{"duration_ms":800}}"#; + let result = format_event_pretty(line, &styles).unwrap(); + assert!(result.contains("Setup:"), "got: {result}"); + assert!(result.contains("800ms"), "got: {result}"); + assert!(!result.contains("0 commands"), "got: {result}"); + } + + #[test] + fn format_duration_ms_subsecond() { + assert_eq!(format_duration_ms(Some(&serde_json::json!(500))), "500ms"); + } + + #[test] + fn format_duration_ms_seconds() { + assert_eq!(format_duration_ms(Some(&serde_json::json!(8000))), "8s"); + } + + #[test] + fn format_duration_ms_minutes() { + assert_eq!(format_duration_ms(Some(&serde_json::json!(90000))), "1.5m"); + } + + #[test] + fn format_tokens_small() { + assert_eq!(format_tokens(500), "500 toks"); + } + + #[test] + fn format_tokens_thousands() { + assert_eq!(format_tokens(15200), "15.2k toks"); + } + + #[test] + fn truncate_short_string() { + assert_eq!(truncate("hello", 10), "hello"); + } + + #[test] + fn truncate_long_string() { + let result = truncate("a very long command string here", 15); + assert!(result.chars().count() <= 15, "got: {result}"); + assert!(result.ends_with('\u{2026}')); + } +} diff --git a/lib/crates/fabro-cli/src/commands/run/logs.rs b/lib/crates/fabro-cli/src/commands/run/logs.rs index 7d7e9063b..130b3c4f9 100644 --- a/lib/crates/fabro-cli/src/commands/run/logs.rs +++ b/lib/crates/fabro-cli/src/commands/run/logs.rs @@ -4,1336 +4,50 @@ )] #![expect( clippy::disallowed_methods, - reason = "sync CLI `run logs` command: streams log lines to std::io::stdout directly" + reason = "sync CLI `run logs` command: writes raw log bytes to std::io::stdout directly" )] -use std::fmt::Write as _; -use std::io::{self, IsTerminal, Write}; -use std::time::Duration; +use std::io::{self, Write}; -use anyhow::{Context, Result, bail}; -use chrono::{DateTime, Utc}; -use fabro_redact::redact_jsonl_line; -use fabro_types::RunNoticeCode; -use fabro_util::json::normalize_json_value; -use fabro_util::terminal::Styles; -use tokio::time; -use tracing::{debug, info}; +use anyhow::{Context as _, Result}; +use tracing::info; use crate::args::LogsArgs; use crate::command_context::CommandContext; -use crate::server_client; -use crate::shared::format_usd_micros; -const FOLLOW_TERMINAL_GRACE: Duration = Duration::from_millis(500); +pub(crate) async fn run(args: &LogsArgs, base_ctx: &CommandContext) -> Result<()> { + base_ctx.require_no_json_override()?; -pub(crate) async fn run(args: &LogsArgs, styles: &Styles, base_ctx: &CommandContext) -> Result<()> { let ctx = base_ctx.with_target(&args.server)?; let client = ctx.server().await?; let run_id = client.resolve_run(&args.run).await?.run_id; - info!(run_id = %run_id, "Showing logs"); + info!(run_id = %run_id, "Showing raw run log"); - let since_cutoff = match &args.since { - Some(value) => Some(parse_since(value)?), - None => None, - }; - - let events = client - .list_run_events(&run_id, None, None) + let bytes = client + .get_run_logs(&run_id) .await - .context("Failed to list server-backed run events")?; - let last_seq = events.last().map_or(0, |event| event.seq); - let all_lines = events - .iter() - .map(event_payload_line) - .collect::>>()?; - let filtered = apply_filters(&all_lines, since_cutoff.as_ref(), args.tail); + .context("Failed to fetch run log from server")? + .ok_or_else(|| anyhow::anyhow!("Run log not available"))?; let stdout = io::stdout(); - let is_tty = stdout.is_terminal(); let mut out = stdout.lock(); - let pretty = args.pretty && !ctx.json_output(); - let mut pretty_state = PrettyEventState::default(); - for line in &filtered { - if pretty { - if let Some(formatted) = format_event_pretty_streamed(line, styles, &mut pretty_state) { - writeln!(out, "{formatted}")?; - } - } else { - writeln!(out, "{line}")?; - } - } - - if args.follow { - follow_store_logs( - client.as_ref(), - &run_id, - if last_seq == 0 { 1 } else { last_seq + 1 }, - pretty, - styles, - is_tty, - pretty_state, - ) - .await?; + match args.tail { + None => out.write_all(&bytes)?, + Some(0) => {} + Some(tail) => write_tail(&bytes, tail, &mut out)?, } Ok(()) } -fn event_name(event: &fabro_store::EventEnvelope) -> &str { - event.event.event_name() -} - -fn apply_filters( - lines: &[String], - since: Option<&DateTime>, - tail: Option, -) -> Vec { - let filtered: Vec = match since { - Some(cutoff) => lines - .iter() - .filter(|line| extract_timestamp(line).is_none_or(|ts| ts >= *cutoff)) - .cloned() - .collect(), - None => lines.to_vec(), - }; - - match tail { - Some(n) if n < filtered.len() => filtered[filtered.len() - n..].to_vec(), - _ => filtered, +fn write_tail(bytes: &[u8], tail: usize, out: &mut dyn Write) -> Result<()> { + let text = std::str::from_utf8(bytes) + .context("Run log is not valid UTF-8; omit --tail to print raw bytes")?; + let lines = text.lines().collect::>(); + let start = lines.len().saturating_sub(tail); + for line in &lines[start..] { + writeln!(out, "{line}")?; } -} - -fn extract_timestamp(line: &str) -> Option> { - let value: serde_json::Value = serde_json::from_str(line).ok()?; - let ts_str = value.get("ts")?.as_str()?; - ts_str.parse::>().ok() -} - -pub(crate) fn parse_since(s: &str) -> Result> { - let s = s.trim(); - if s.is_empty() { - bail!("empty --since value"); - } - - if let Some(duration) = try_parse_relative_duration(s) { - return Ok(Utc::now() - duration); - } - - if let Ok(ts) = s.parse::>() { - return Ok(ts); - } - - bail!( - "invalid --since value '{s}' (expected relative like '42m', '2h', '7d' or ISO 8601 timestamp)" - ) -} - -fn try_parse_relative_duration(s: &str) -> Option { - if s.len() < 2 { - return None; - } - let (num_str, unit) = s.split_at(s.len() - 1); - let num = i64::try_from(num_str.parse::().ok()?).ok()?; - match unit { - "s" => Some(chrono::Duration::seconds(num)), - "m" => Some(chrono::Duration::minutes(num)), - "h" => Some(chrono::Duration::hours(num)), - "d" => Some(chrono::Duration::days(num)), - _ => None, - } -} - -async fn follow_store_logs( - client: &server_client::Client, - run_id: &fabro_types::RunId, - seq: u32, - pretty: bool, - styles: &Styles, - _is_tty: bool, - mut pretty_state: PrettyEventState, -) -> Result<()> { - let stdout = io::stdout(); - let mut out = stdout.lock(); - let mut next_seq = seq; - let mut terminal_deadline = None; - - loop { - match time::timeout( - Duration::from_millis(200), - client.list_run_events(run_id, Some(next_seq), None), - ) - .await - { - Ok(Ok(events)) => { - let had_events = !events.is_empty(); - let saw_terminal = events - .iter() - .any(|event| matches!(event_name(event), "run.completed" | "run.failed")); - for event in events { - let line = event_payload_line(&event)?; - if pretty { - if let Some(formatted) = - format_event_pretty_streamed(&line, styles, &mut pretty_state) - { - writeln!(out, "{formatted}")?; - } - } else { - writeln!(out, "{line}")?; - } - out.flush()?; - next_seq = event.seq.saturating_add(1); - } - if saw_terminal || (terminal_deadline.is_some() && had_events) { - terminal_deadline = Some(time::Instant::now() + FOLLOW_TERMINAL_GRACE); - } - } - Err(_) => { - if run_concluded(client, run_id).await? { - terminal_deadline - .get_or_insert_with(|| time::Instant::now() + FOLLOW_TERMINAL_GRACE); - } - } - Ok(Err(err)) => return Err(err), - } - - let Some(deadline) = terminal_deadline else { - continue; - }; - if time::Instant::now() < deadline { - continue; - } - - let flushed_next_seq = flush_remaining_store_events( - client, - run_id, - next_seq, - pretty, - styles, - &mut pretty_state, - &mut out, - ) - .await?; - if flushed_next_seq > next_seq { - next_seq = flushed_next_seq; - terminal_deadline = Some(time::Instant::now() + FOLLOW_TERMINAL_GRACE); - continue; - } - - debug!("Run reached terminal status and log tail is quiet, stopping follow"); - break; - } - Ok(()) } - -async fn run_concluded( - client: &server_client::Client, - run_id: &fabro_types::RunId, -) -> Result { - let state = client - .get_run_state(run_id) - .await - .context("Failed to read run state from server while following logs")?; - Ok(state.conclusion.is_some() - || state - .status - .is_some_and(fabro_types::RunStatus::is_terminal)) -} - -async fn flush_remaining_store_events( - client: &server_client::Client, - run_id: &fabro_types::RunId, - next_seq: u32, - pretty: bool, - styles: &Styles, - pretty_state: &mut PrettyEventState, - out: &mut dyn Write, -) -> Result { - let events = client - .list_run_events(run_id, Some(next_seq), None) - .await - .context("Failed to list server-backed run events while finalizing follow")?; - - let mut next_seq = next_seq; - for event in events { - let line = event_payload_line(&event)?; - if pretty { - if let Some(formatted) = format_event_pretty_streamed(&line, styles, pretty_state) { - writeln!(out, "{formatted}")?; - } - } else { - writeln!(out, "{line}")?; - } - next_seq = event.seq.saturating_add(1); - } - out.flush()?; - Ok(next_seq) -} - -fn event_payload_line(event: &fabro_store::EventEnvelope) -> Result { - let mut value = normalize_json_value(event.event.to_value()?); - restore_empty_run_properties(&mut value); - let line = serde_json::to_string(&value)?; - Ok(redact_jsonl_line(&line)) -} - -fn restore_empty_run_properties(value: &mut serde_json::Value) { - let Some(object) = value.as_object_mut() else { - return; - }; - let Some(event_name) = object.get("event").and_then(serde_json::Value::as_str) else { - return; - }; - if matches!(event_name, "run.submitted" | "run.running") && !object.contains_key("properties") { - let run_id = object.remove("run_id"); - let ts = object.remove("ts"); - object.insert("properties".to_string(), serde_json::json!({})); - if let Some(run_id) = run_id { - object.insert("run_id".to_string(), run_id); - } - if let Some(ts) = ts { - object.insert("ts".to_string(), ts); - } - } -} - -fn render_indented_markdown(styles: &Styles, text: &str, indent: &str) -> String { - let term_width = Styles::terminal_width(); - let wrap_width = term_width.saturating_sub(indent.len()); - let rendered = styles.render_markdown_width(text, wrap_width); - rendered - .lines() - .map(|line| format!("{indent}{line}")) - .collect::>() - .join("\n") -} - -#[derive(Debug, Default)] -struct PrettyEventState { - saw_metadata_snapshot_failure: bool, -} - -fn format_event_pretty_streamed( - line: &str, - styles: &Styles, - state: &mut PrettyEventState, -) -> Option { - let envelope: serde_json::Value = serde_json::from_str(line).ok()?; - let event = envelope.get("event")?.as_str()?; - if event == "run.notice" - && state.saw_metadata_snapshot_failure - && is_metadata_snapshot_compat_notice(&envelope) - { - return None; - } - let formatted = format_event_pretty_value(&envelope, styles); - if event == "metadata.snapshot.failed" { - state.saw_metadata_snapshot_failure = true; - } - formatted -} - -#[cfg_attr( - not(test), - allow( - dead_code, - reason = "Production pretty logs use the stateful stream formatter; unit tests exercise this single-line helper." - ) -)] -pub(crate) fn format_event_pretty(line: &str, styles: &Styles) -> Option { - let envelope: serde_json::Value = serde_json::from_str(line).ok()?; - format_event_pretty_value(&envelope, styles) -} - -fn format_event_pretty_value(envelope: &serde_json::Value, styles: &Styles) -> Option { - let event = envelope.get("event")?.as_str()?; - let ts = format_timestamp(envelope.get("ts")?.as_str()?); - - match event { - "run.started" => { - let name = prop_str_field(envelope, "name").unwrap_or("?"); - let run_id = str_field(envelope, "run_id").unwrap_or("?"); - let header = format!( - "{} {} {} {}", - styles.dim.apply_to(&ts), - styles.bold_cyan.apply_to("\u{25b6}"), - styles.bold.apply_to(name), - styles.dim.apply_to(run_id), - ); - match prop_str_field(envelope, "goal") { - Some(goal) if !goal.is_empty() => { - let body = render_indented_markdown(styles, goal, " "); - Some(format!("{header}\n{body}\n")) - } - _ => Some(header), - } - } - "run.completed" => { - let duration = format_duration_ms(prop_field(envelope, "duration_ms")); - let status_str = match prop_str_field(envelope, "status") { - Some(status) if !status.is_empty() => status, - _ => "succeeded", - }; - let status_upper = status_str.to_uppercase(); - let status_style = match status_str { - "succeeded" | "partially_succeeded" => &styles.bold_green, - _ => &styles.bold_red, - }; - let cost = format_cost( - prop_field(envelope, "total_usd_micros") - .or_else(|| prop_field(envelope, "total_cost")), - ); - - let mut summary = format!( - "{} {} {}", - styles.dim.apply_to(&ts), - status_style.apply_to(format!("\u{2713} {status_upper}")), - styles.bold.apply_to(&duration), - ); - if !cost.is_empty() { - write!(summary, " {}", styles.dim.apply_to(&cost)).expect("write to string"); - } - - let mut lines = vec![summary]; - - if let Some(billing) = - prop_field(envelope, "billing").or_else(|| prop_field(envelope, "usage")) - { - let total = billing - .get("total_tokens") - .and_then(serde_json::Value::as_u64) - .unwrap_or(0); - let pad = " ".repeat(ts.len() + 1); - if total > 0 { - lines.push(format!( - "{}{}", - pad, - styles - .dim - .apply_to(format!("Tokens: {}", format_tokens(total))) - )); - } - if let Some(cache_read) = billing - .get("cache_read_tokens") - .and_then(serde_json::Value::as_u64) - { - let cache_write = billing - .get("cache_write_tokens") - .and_then(serde_json::Value::as_u64) - .unwrap_or(0); - lines.push(format!( - "{}{}", - pad, - styles.dim.apply_to(format!( - "Cache: {} read, {} write", - format_tokens(cache_read), - format_tokens(cache_write) - )) - )); - } - if let Some(reasoning) = billing - .get("reasoning_tokens") - .and_then(serde_json::Value::as_u64) - { - if reasoning > 0 { - lines.push(format!( - "{}{}", - pad, - styles.dim.apply_to(format!( - "Reasoning: {} tokens", - format_tokens(reasoning) - )) - )); - } - } - } - - Some(lines.join("\n")) - } - "run.failed" => { - let error = prop_str_field(envelope, "error").unwrap_or("unknown error"); - Some(format!( - "{} {} {}", - styles.dim.apply_to(&ts), - styles.bold_red.apply_to("\u{2717} Failed"), - styles.red.apply_to(error), - )) - } - "run.notice" => { - let level = prop_str_field(envelope, "level").unwrap_or("info"); - let code = prop_str_field(envelope, "code").unwrap_or(""); - let message = prop_str_field(envelope, "message").unwrap_or(""); - let label = match level { - "warn" => styles.yellow.apply_to("Warning:").to_string(), - "error" => styles.bold_red.apply_to("Error:").to_string(), - _ => styles.bold.apply_to("Info:").to_string(), - }; - let code_suffix = if code.is_empty() { - String::new() - } else { - format!(" {}", styles.dim.apply_to(format!("[{code}]"))) - }; - Some(format!( - "{} {} {}{}", - styles.dim.apply_to(&ts), - label, - message, - code_suffix, - )) - } - "metadata.snapshot.completed" => { - let phase = prop_str_field(envelope, "phase").unwrap_or("?"); - let duration = format_duration_ms(prop_field(envelope, "duration_ms")); - Some(format!( - "{} Metadata {} {}", - styles.dim.apply_to(&ts), - phase, - styles.dim.apply_to(&duration), - )) - } - "metadata.snapshot.failed" => { - let phase = prop_str_field(envelope, "phase").unwrap_or("?"); - let failure_kind = prop_str_field(envelope, "failure_kind").unwrap_or(""); - let error = prop_str_field(envelope, "error").unwrap_or("unknown error"); - let kind_suffix = if failure_kind.is_empty() { - String::new() - } else { - format!(" {}", styles.dim.apply_to(format!("[{failure_kind}]"))) - }; - Some(format!( - "{} {} Metadata {} failed: {}{}", - styles.dim.apply_to(&ts), - styles.yellow.apply_to("Warning:"), - phase, - error, - kind_suffix, - )) - } - "stage.started" => { - let label = str_field(envelope, "node_label").unwrap_or("?"); - Some(format!( - "{} {} {}", - styles.dim.apply_to(&ts), - styles.bold_cyan.apply_to("\u{25b6}"), - styles.bold.apply_to(label), - )) - } - "stage.completed" => { - let label = str_field(envelope, "node_label").unwrap_or("?"); - let duration = format_duration_ms(prop_field(envelope, "duration_ms")); - let billing = prop_field(envelope, "billing").or_else(|| prop_field(envelope, "usage")); - let cost = format_cost( - billing - .and_then(|value| value.get("total_usd_micros")) - .or_else(|| billing.and_then(|value| value.get("cost"))), - ); - let input_tokens = billing - .and_then(|value| value.get("input_tokens")) - .and_then(serde_json::Value::as_u64) - .unwrap_or(0); - let output_tokens = billing - .and_then(|value| value.get("output_tokens")) - .and_then(serde_json::Value::as_u64) - .unwrap_or(0); - let token_total = input_tokens.saturating_add(output_tokens); - let mut line = format!( - "{} {} {} {} {}", - styles.dim.apply_to(&ts), - styles.green.apply_to("\u{2713}"), - styles.bold.apply_to(label), - cost, - duration, - ); - if token_total > 0 { - let _ = write!( - line, - " {}", - styles.dim.apply_to(format_tokens(token_total)) - ); - } - Some(line) - } - "stage.failed" => { - let label = str_field(envelope, "node_label").unwrap_or("?"); - let error = prop_str_field(envelope, "error").unwrap_or("unknown error"); - Some(format!( - "{} {} {} {}", - styles.dim.apply_to(&ts), - styles.red.apply_to("\u{2717}"), - styles.bold.apply_to(label), - styles.red.apply_to(error), - )) - } - "agent.message" => { - let stage = str_field(envelope, "node_id").unwrap_or("?"); - let model = prop_str_field(envelope, "model").unwrap_or("?"); - let text = prop_str_field(envelope, "text").unwrap_or(""); - let header = format!( - "{} {} {} {}{}{}", - styles.dim.apply_to(&ts), - "\u{1f4ac}", - styles.bold.apply_to(stage), - styles.dim.apply_to("["), - styles.dim.apply_to(model), - styles.dim.apply_to("]"), - ); - let body = render_indented_markdown(styles, text, " "); - Some(format!("{header}\n{body}\n")) - } - "agent.tool.started" => { - let tool = prop_str_field(envelope, "tool_name").unwrap_or("?"); - let detail = tool_detail(envelope); - let display = match detail { - Some(value) => format!("{tool}({value})"), - None => tool.to_string(), - }; - Some(format!( - "{} {} {}", - styles.dim.apply_to(&ts), - styles.dim.apply_to("\u{2699}"), - styles.dim.apply_to(&display), - )) - } - "agent.tool.completed" => { - let tool = prop_str_field(envelope, "tool_name").unwrap_or("?"); - let is_error = prop_field(envelope, "is_error") - .and_then(serde_json::Value::as_bool) - .unwrap_or(false); - let detail = tool_detail(envelope); - let display = match detail { - Some(value) => format!("{tool}({value})"), - None => tool.to_string(), - }; - let glyph = if is_error { "\u{2717}" } else { "\u{2713}" }; - let style = if is_error { &styles.red } else { &styles.green }; - Some(format!( - "{} {} {}", - styles.dim.apply_to(&ts), - style.apply_to(glyph), - display, - )) - } - "edge.selected" => { - let to = prop_str_field(envelope, "to_node").unwrap_or("?"); - let reason = prop_str_field(envelope, "reason").unwrap_or("?"); - let condition = prop_str_field(envelope, "condition"); - let detail = match condition { - Some(value) => format!(" [{value}]"), - None => String::new(), - }; - Some(format!( - "{} {} {} {}{}", - styles.dim.apply_to(&ts), - styles.dim.apply_to("\u{2192}"), - to, - styles.dim.apply_to(reason), - styles.dim.apply_to(&detail), - )) - } - "sandbox.ready" => { - let provider = prop_str_field(envelope, "provider").unwrap_or("?"); - let duration = format_duration_ms(prop_field(envelope, "duration_ms")); - Some(format!( - "{} Sandbox: {} {}", - styles.dim.apply_to(&ts), - provider, - styles.dim.apply_to(&duration), - )) - } - "sandbox.snapshot.pulling" => { - let name = prop_str_field(envelope, "name").unwrap_or("?"); - Some(format!( - "{} Sandbox: pulling {}", - styles.dim.apply_to(&ts), - name, - )) - } - "sandbox.snapshot.creating" => { - let name = prop_str_field(envelope, "name").unwrap_or("?"); - Some(format!( - "{} Sandbox: building {}", - styles.dim.apply_to(&ts), - name, - )) - } - "sandbox.snapshot.ready" => { - let name = prop_str_field(envelope, "name").unwrap_or("?"); - let duration = format_duration_ms(prop_field(envelope, "duration_ms")); - Some(format!( - "{} Sandbox snapshot: {} {}", - styles.dim.apply_to(&ts), - name, - styles.dim.apply_to(&duration), - )) - } - "sandbox.snapshot.failed" => { - let name = prop_str_field(envelope, "name").unwrap_or("?"); - let error = prop_str_field(envelope, "error").unwrap_or("unknown error"); - Some(format!( - "{} {} Sandbox snapshot {} failed: {}", - styles.dim.apply_to(&ts), - styles.bold_red.apply_to("\u{2717}"), - name, - styles.red.apply_to(error), - )) - } - "setup.completed" => { - let count = prop_field(envelope, "command_count").and_then(serde_json::Value::as_u64); - let duration = format_duration_ms(prop_field(envelope, "duration_ms")); - Some(match count { - Some(count) => format!( - "{} Setup: {} commands {}", - styles.dim.apply_to(&ts), - count, - styles.dim.apply_to(&duration), - ), - None => format!( - "{} Setup: {}", - styles.dim.apply_to(&ts), - styles.dim.apply_to(&duration), - ), - }) - } - "agent.compaction.completed" => { - let original = prop_field(envelope, "original_turn_count") - .and_then(serde_json::Value::as_u64) - .unwrap_or(0); - let preserved = prop_field(envelope, "preserved_turn_count") - .and_then(serde_json::Value::as_u64) - .unwrap_or(0); - Some(format!( - "{} {}", - styles.dim.apply_to(&ts), - styles - .dim - .apply_to(format!("compaction: {original}\u{2192}{preserved} turns")), - )) - } - "parallel.started" => { - let count = prop_field(envelope, "branch_count") - .and_then(serde_json::Value::as_u64) - .unwrap_or(0); - Some(format!( - "{} {} Parallel {} branches", - styles.dim.apply_to(&ts), - styles.bold_cyan.apply_to("\u{25b6}"), - count, - )) - } - "parallel.branch.started" => { - let label = str_field(envelope, "node_label").unwrap_or("?"); - Some(format!( - "{} {} {}", - styles.dim.apply_to(&ts), - styles.cyan.apply_to("\u{25b6}"), - label, - )) - } - "parallel.branch.completed" => { - let label = str_field(envelope, "node_label").unwrap_or("?"); - Some(format!( - "{} {} {}", - styles.dim.apply_to(&ts), - styles.green.apply_to("\u{2713}"), - label, - )) - } - "parallel.completed" => { - let duration = format_duration_ms(prop_field(envelope, "duration_ms")); - Some(format!( - "{} {} Parallel {}", - styles.dim.apply_to(&ts), - styles.green.apply_to("\u{2713}"), - duration, - )) - } - "pull_request.created" => { - let url = prop_str_field(envelope, "pr_url").unwrap_or("?"); - let draft = prop_field(envelope, "draft") - .and_then(serde_json::Value::as_bool) - .unwrap_or(false); - let label = if draft { "Draft PR:" } else { "PR:" }; - Some(format!( - "{} {} {}", - styles.dim.apply_to(&ts), - styles.bold.apply_to(label), - url, - )) - } - "pull_request.failed" => { - let error = prop_str_field(envelope, "error").unwrap_or("unknown error"); - Some(format!( - "{} {} {}", - styles.dim.apply_to(&ts), - styles.bold_red.apply_to("PR failed:"), - styles.red.apply_to(error), - )) - } - "retro.completed" => { - let duration = format_duration_ms(prop_field(envelope, "duration_ms")); - Some(format!( - "{} {} Retro {}", - styles.dim.apply_to(&ts), - styles.green.apply_to("\u{2713}"), - duration, - )) - } - "retro.failed" => { - let error = prop_str_field(envelope, "error").unwrap_or("unknown error"); - let duration = format_duration_ms(prop_field(envelope, "duration_ms")); - Some(format!( - "{} {} Retro {} {}", - styles.dim.apply_to(&ts), - styles.bold_red.apply_to("\u{2717}"), - duration, - styles.red.apply_to(error), - )) - } - "retro.started" => Some(format!( - "{} {} Retro", - styles.dim.apply_to(&ts), - styles.bold_cyan.apply_to("\u{25b6}"), - )), - _ => None, - } -} - -fn is_metadata_snapshot_compat_notice(envelope: &serde_json::Value) -> bool { - prop_str_field(envelope, "code") - .and_then(|code| code.parse::().ok()) - .is_some_and(RunNoticeCode::is_metadata_snapshot_compat) -} - -fn str_field<'a>(value: &'a serde_json::Value, key: &str) -> Option<&'a str> { - value.get(key)?.as_str() -} - -fn prop_field<'a>(value: &'a serde_json::Value, key: &str) -> Option<&'a serde_json::Value> { - value.get("properties")?.get(key) -} - -fn prop_str_field<'a>(value: &'a serde_json::Value, key: &str) -> Option<&'a str> { - prop_field(value, key)?.as_str() -} - -fn format_timestamp(ts: &str) -> String { - ts.parse::>() - .map_or_else(|_| ts.to_string(), |dt| dt.format("%H:%M:%S").to_string()) -} - -fn format_duration_ms(value: Option<&serde_json::Value>) -> String { - let ms = value.and_then(serde_json::Value::as_u64).unwrap_or(0); - if ms < 1000 { - format!("{ms}ms") - } else { - let secs = ms as f64 / 1000.0; - if secs < 60.0 { - format!("{secs:.0}s") - } else { - let mins = secs / 60.0; - format!("{mins:.1}m") - } - } -} - -fn format_cost(value: Option<&serde_json::Value>) -> String { - match value { - Some(value) => { - if let Some(usd_micros) = value.as_i64() { - if usd_micros > 0 { - return format_usd_micros(usd_micros); - } - } - let cost = value.as_f64().unwrap_or(0.0); - if cost > 0.0 { - format!("${cost:.2}") - } else { - String::new() - } - } - None => String::new(), - } -} - -fn format_tokens(tokens: u64) -> String { - if tokens >= 1000 { - format!("{:.1}k toks", tokens as f64 / 1000.0) - } else { - format!("{tokens} toks") - } -} - -fn tool_detail(envelope: &serde_json::Value) -> Option { - let tool_name = prop_str_field(envelope, "tool_name")?; - let arguments = prop_field(envelope, "arguments")?; - let arg = |key: &str| arguments.get(key).and_then(|v| v.as_str()); - - match tool_name { - "bash" | "shell" | "execute_command" => arg("command").map(|c| truncate(c, 60)), - "glob" => arg("pattern").map(String::from), - "grep" | "ripgrep" => arg("pattern").map(|p| truncate(p, 40)), - "read_file" | "read" => arg("path") - .or_else(|| arg("file_path")) - .map(|p| truncate(p, 60)), - "write_file" | "write" | "create_file" => arg("path") - .or_else(|| arg("file_path")) - .map(|p| truncate(p, 60)), - "edit_file" | "edit" => arg("path") - .or_else(|| arg("file_path")) - .map(|p| truncate(p, 60)), - "list_dir" => arg("path") - .or_else(|| arg("file_path")) - .map(|p| truncate(p, 60)), - "web_search" => arg("query").map(|q| truncate(q, 60)), - "web_fetch" => arg("url").map(|u| truncate(u, 60)), - "spawn_agent" => arg("task").map(|t| truncate(t, 60)), - "wait" | "send_input" | "close_agent" => arg("agent_id").map(String::from), - "use_skill" => arg("skill_name").map(String::from), - "apply_patch" => Some("…".into()), - "read_many_files" => arguments - .get("paths") - .and_then(|v| v.as_array()) - .map(|a| format!("{} files", a.len())), - _ => None, - } -} - -fn truncate(s: &str, max: usize) -> String { - if s.len() <= max { - s.to_string() - } else { - let boundary = s.floor_char_boundary(max.saturating_sub(1)); - format!("{}\u{2026}", &s[..boundary]) - } -} - -#[cfg(test)] -mod tests { - use super::*; - - fn no_color_styles() -> Styles { - Styles::new(false) - } - - #[test] - fn parse_since_relative_minutes() { - let before = Utc::now(); - let result = parse_since("42m").unwrap(); - let after = Utc::now(); - let expected_lower = after - chrono::Duration::minutes(42) - chrono::Duration::seconds(1); - let expected_upper = before - chrono::Duration::minutes(42) + chrono::Duration::seconds(1); - assert!(result >= expected_lower && result <= expected_upper); - } - - #[test] - fn parse_since_relative_hours() { - let before = Utc::now(); - let result = parse_since("2h").unwrap(); - let expected = before - chrono::Duration::hours(2); - assert!((result - expected).num_seconds().abs() < 2); - } - - #[test] - fn parse_since_relative_days() { - let before = Utc::now(); - let result = parse_since("7d").unwrap(); - let expected = before - chrono::Duration::days(7); - assert!((result - expected).num_seconds().abs() < 2); - } - - #[test] - fn parse_since_iso8601() { - let result = parse_since("2026-01-01T12:00:00Z").unwrap(); - assert_eq!(result.to_rfc3339(), "2026-01-01T12:00:00+00:00"); - } - - #[test] - fn parse_since_invalid() { - assert!(parse_since("").is_err()); - assert!(parse_since("abc").is_err()); - assert!(parse_since("notadate").is_err()); - } - - #[test] - fn parse_since_overflow_is_invalid() { - assert!(parse_since("9223372036854775808s").is_err()); - } - - #[test] - fn tail_returns_last_n_lines() { - let lines: Vec = (0..10).map(|i| format!("line {i}")).collect(); - let result = apply_filters(&lines, None, Some(3)); - assert_eq!(result.len(), 3); - assert_eq!(result[0], "line 7"); - assert_eq!(result[2], "line 9"); - } - - #[test] - fn tail_all_when_n_exceeds_total() { - let lines: Vec = (0..3).map(|i| format!("line {i}")).collect(); - let result = apply_filters(&lines, None, Some(100)); - assert_eq!(result.len(), 3); - } - - #[test] - fn since_filters_by_timestamp() { - let cutoff = "2026-01-01T12:00:00Z".parse::>().unwrap(); - let lines = vec![ - r#"{"ts":"2026-01-01T11:00:00Z","event":"stage.started"}"#.to_string(), - r#"{"ts":"2026-01-01T12:30:00Z","event":"stage.completed"}"#.to_string(), - r#"{"ts":"2026-01-01T13:00:00Z","event":"run.completed"}"#.to_string(), - ]; - let result = apply_filters(&lines, Some(&cutoff), None); - assert_eq!(result.len(), 2); - } - - #[test] - fn raw_lines_pass_through_verbatim() { - let lines = vec![ - r#"{"ts":"2026-01-01T12:00:00Z","event":"stage.started","node_label":"plan"}"# - .to_string(), - ]; - let result = apply_filters(&lines, None, None); - assert_eq!(result, lines); - } - - #[test] - fn pretty_stage_started() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:23:09Z","event":"stage.started","node_label":"plan","node_id":"plan","properties":{"index":0}}"#; - let result = format_event_pretty(line, &styles).unwrap(); - assert!(result.contains("plan"), "got: {result}"); - assert!(result.contains("\u{25b6}"), "got: {result}"); - } - - #[test] - fn pretty_stage_completed() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:23:15Z","event":"stage.completed","node_label":"plan","properties":{"duration_ms":8000,"status":"succeeded","usage":{"cost":0.12,"input_tokens":10000,"output_tokens":5200}}}"#; - let result = format_event_pretty(line, &styles).unwrap(); - assert!(result.contains("plan"), "got: {result}"); - assert!(result.contains("$0.12"), "got: {result}"); - assert!(result.contains("8s"), "got: {result}"); - assert!(result.contains("15.2k toks"), "got: {result}"); - } - - #[test] - fn pretty_assistant_message() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:23:12Z","event":"agent.message","node_id":"plan","properties":{"model":"claude-opus-4-6","text":"I'll start by reading the code.","usage":{"input_tokens":100,"output_tokens":50},"tool_call_count":0}}"#; - let result = format_event_pretty(line, &styles).unwrap(); - assert!(result.contains("plan"), "got: {result}"); - assert!(result.contains("claude-opus-4-6"), "got: {result}"); - assert!(result.contains("reading the code"), "got: {result}"); - } - - #[test] - fn pretty_tool_call_started() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:23:12Z","event":"agent.tool.started","properties":{"tool_name":"read_file","tool_call_id":"tc_1","arguments":{"path":"src/main.rs"}}}"#; - let result = format_event_pretty(line, &styles).unwrap(); - assert!(result.contains("read_file"), "got: {result}"); - assert!(result.contains("src/main.rs"), "got: {result}"); - } - - #[test] - fn pretty_skips_noise_events() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:23:12Z","event":"agent.text.delta","properties":{"delta":"hello"}}"#; - assert!(format_event_pretty(line, &styles).is_none()); - } - - #[test] - fn pretty_skips_assistant_output_replace_noise_event() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:23:12Z","event":"agent.output.replace","properties":{"text":""}}"#; - assert!(format_event_pretty(line, &styles).is_none()); - } - - #[test] - fn pretty_unknown_events_return_none() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:23:12Z","event":"SomeFutureEvent","data":123}"#; - assert!(format_event_pretty(line, &styles).is_none()); - } - - #[test] - fn pretty_workflow_run_started() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:23:01Z","run_id":"abc123","event":"run.started","properties":{"name":"smoke"}}"#; - let result = format_event_pretty(line, &styles).unwrap(); - assert!(result.contains("smoke"), "got: {result}"); - assert!(result.contains("abc123"), "got: {result}"); - } - - #[test] - fn pretty_workflow_run_started_with_goal() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:23:01Z","run_id":"abc123","event":"run.started","properties":{"name":"smoke","goal":"Fix the bug"}}"#; - let result = format_event_pretty(line, &styles).unwrap(); - assert!(result.contains("smoke"), "got: {result}"); - assert!(result.contains("abc123"), "got: {result}"); - assert!(result.contains("Fix the bug"), "got: {result}"); - assert!(result.contains('\n'), "got: {result}"); - } - - #[test] - fn pretty_workflow_run_started_without_goal_no_extra_lines() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:23:01Z","run_id":"abc123","event":"run.started","properties":{"name":"smoke"}}"#; - let result = format_event_pretty(line, &styles).unwrap(); - assert!(!result.contains('\n'), "got: {result}"); - } - - #[test] - fn pretty_workflow_run_completed() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:23:32Z","run_id":"abc123","event":"run.completed","properties":{"duration_ms":25000,"status":"succeeded","total_usd_micros":570000,"billing":{"input_tokens":5000,"output_tokens":2000,"total_tokens":7000,"cache_read_tokens":3000,"cache_write_tokens":500,"reasoning_tokens":800}}}"#; - let result = format_event_pretty(line, &styles).unwrap(); - assert!(result.contains("SUCCEEDED"), "got: {result}"); - assert!(result.contains("25s"), "got: {result}"); - assert!(result.contains("$0.57"), "got: {result}"); - assert!(result.contains("7.0k toks"), "got: {result}"); - assert!(result.contains("Cache:"), "got: {result}"); - assert!(result.contains("3.0k toks read"), "got: {result}"); - assert!(result.contains("Reasoning:"), "got: {result}"); - } - - #[test] - fn pretty_workflow_run_completed_backward_compat() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:23:32Z","run_id":"abc123","event":"run.completed","properties":{"duration_ms":25000,"total_cost":0.57}}"#; - let result = format_event_pretty(line, &styles).unwrap(); - assert!(result.contains("SUCCEEDED"), "got: {result}"); - assert!(result.contains("25s"), "got: {result}"); - assert!(result.contains("$0.57"), "got: {result}"); - assert!(!result.contains("Tokens:"), "got: {result}"); - } - - #[test] - fn pretty_workflow_run_completed_fail_status() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:23:32Z","event":"run.completed","properties":{"duration_ms":25000,"status":"failed"}}"#; - let result = format_event_pretty(line, &styles).unwrap(); - assert!(result.contains("FAIL"), "got: {result}"); - } - - #[test] - fn pretty_pull_request_created() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"pull_request.created","properties":{"pr_url":"https://github.com/owner/repo/pull/42","pr_number":42,"draft":false}}"#; - let result = format_event_pretty(line, &styles).unwrap(); - assert!(result.contains("PR:"), "got: {result}"); - assert!( - result.contains("https://github.com/owner/repo/pull/42"), - "got: {result}" - ); - } - - #[test] - fn pretty_pull_request_created_draft() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"pull_request.created","properties":{"pr_url":"https://github.com/owner/repo/pull/42","pr_number":42,"draft":true}}"#; - let result = format_event_pretty(line, &styles).unwrap(); - assert!(result.contains("Draft PR:"), "got: {result}"); - } - - #[test] - fn pretty_pull_request_failed() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"pull_request.failed","properties":{"error":"auth token expired"}}"#; - let result = format_event_pretty(line, &styles).unwrap(); - assert!(result.contains("PR failed:"), "got: {result}"); - assert!(result.contains("auth token expired"), "got: {result}"); - } - - #[test] - fn pretty_run_notice_warn() { - let styles = no_color_styles(); - let code = RunNoticeCode::SandboxCleanupFailed.to_string(); - let line = serde_json::json!({ - "ts": "2026-01-01T14:25:00Z", - "event": "run.notice", - "properties": { - "level": "warn", - "code": code, - "message": "sandbox cleanup failed: boom", - }, - }) - .to_string(); - let result = format_event_pretty(&line, &styles).unwrap(); - assert!(result.contains("Warning:"), "got: {result}"); - assert!( - result.contains("sandbox cleanup failed: boom"), - "got: {result}" - ); - assert!( - result.contains(&format!("[{}]", RunNoticeCode::SandboxCleanupFailed)), - "got: {result}" - ); - } - - #[test] - fn pretty_run_notice_error() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"run.notice","properties":{"level":"error","code":"launch_failed","message":"failed to start engine"}}"#; - let result = format_event_pretty(line, &styles).unwrap(); - assert!(result.contains("Error:"), "got: {result}"); - assert!(result.contains("failed to start engine"), "got: {result}"); - assert!(result.contains("[launch_failed]"), "got: {result}"); - } - - #[test] - fn pretty_metadata_snapshot_completed() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"metadata.snapshot.completed","properties":{"phase":"checkpoint","branch":"fabro/meta","duration_ms":2800,"entry_count":2,"bytes":42,"commit_sha":"abc123"}}"#; - let result = format_event_pretty(line, &styles).unwrap(); - assert!(result.contains("Metadata checkpoint"), "got: {result}"); - assert!(result.contains("3s"), "got: {result}"); - } - - #[test] - fn pretty_metadata_snapshot_failed() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"metadata.snapshot.failed","properties":{"phase":"finalize","branch":"fabro/meta","duration_ms":900,"failure_kind":"push","error":"push rejected","commit_sha":"abc123","entry_count":2,"bytes":42}}"#; - let result = format_event_pretty(line, &styles).unwrap(); - assert!(result.contains("Warning:"), "got: {result}"); - assert!( - result.contains("Metadata finalize failed: push rejected"), - "got: {result}" - ); - assert!(result.contains("[push]"), "got: {result}"); - } - - #[test] - fn pretty_sandbox_snapshot_pulling() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"sandbox.snapshot.pulling","properties":{"name":"buildpack-deps:noble"}}"#; - let result = format_event_pretty(line, &styles).unwrap(); - assert!(result.contains("Sandbox: pulling"), "got: {result}"); - assert!(result.contains("buildpack-deps:noble"), "got: {result}"); - } - - #[test] - fn pretty_sandbox_snapshot_creating() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"sandbox.snapshot.creating","properties":{"name":"fabro-v9-test"}}"#; - let result = format_event_pretty(line, &styles).unwrap(); - assert!(result.contains("Sandbox: building"), "got: {result}"); - assert!(result.contains("fabro-v9-test"), "got: {result}"); - } - - #[test] - fn pretty_sandbox_snapshot_ready() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"sandbox.snapshot.ready","properties":{"name":"buildpack-deps:noble","duration_ms":8200}}"#; - let result = format_event_pretty(line, &styles).unwrap(); - assert!(result.contains("Sandbox snapshot:"), "got: {result}"); - assert!(result.contains("buildpack-deps:noble"), "got: {result}"); - assert!(result.contains("8s"), "got: {result}"); - } - - #[test] - fn pretty_sandbox_snapshot_failed() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:25:00Z","event":"sandbox.snapshot.failed","properties":{"name":"buildpack-deps:noble","error":"pull failed"}}"#; - let result = format_event_pretty(line, &styles).unwrap(); - assert!( - result.contains("Sandbox snapshot buildpack-deps:noble failed: pull failed"), - "got: {result}" - ); - } - - #[test] - fn pretty_stream_suppresses_metadata_compat_notice_only() { - let styles = no_color_styles(); - let failed = r#"{"ts":"2026-01-01T14:25:00Z","event":"metadata.snapshot.failed","properties":{"phase":"checkpoint","branch":"fabro/meta","duration_ms":900,"failure_kind":"write","error":"write failed"}}"#; - let compat_notice = serde_json::json!({ - "ts": "2026-01-01T14:25:01Z", - "event": "run.notice", - "properties": { - "level": "warn", - "code": RunNoticeCode::CheckpointMetadataWriteFailed, - "message": "legacy metadata warning", - }, - }) - .to_string(); - let degraded_notice = serde_json::json!({ - "ts": "2026-01-01T14:25:02Z", - "event": "run.notice", - "properties": { - "level": "warn", - "code": RunNoticeCode::CheckpointMetadataDegraded, - "message": "metadata snapshots disabled", - }, - }) - .to_string(); - let mut state = PrettyEventState::default(); - - assert!(format_event_pretty_streamed(failed, &styles, &mut state).is_some()); - assert!(format_event_pretty_streamed(&compat_notice, &styles, &mut state).is_none()); - let degraded = format_event_pretty_streamed(°raded_notice, &styles, &mut state).unwrap(); - assert!( - degraded.contains("metadata snapshots disabled"), - "got: {degraded}" - ); - } - - #[test] - fn pretty_workflow_run_failed() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:23:32Z","run_id":"abc123","event":"run.failed","properties":{"error":"sandbox timeout"}}"#; - let result = format_event_pretty(line, &styles).unwrap(); - assert!(result.contains("Failed"), "got: {result}"); - assert!(result.contains("sandbox timeout"), "got: {result}"); - } - - #[test] - fn pretty_setup_completed_without_command_count() { - let styles = no_color_styles(); - let line = r#"{"ts":"2026-01-01T14:23:32Z","event":"setup.completed","properties":{"duration_ms":800}}"#; - let result = format_event_pretty(line, &styles).unwrap(); - assert!(result.contains("Setup:"), "got: {result}"); - assert!(result.contains("800ms"), "got: {result}"); - assert!(!result.contains("0 commands"), "got: {result}"); - } - - #[test] - fn format_duration_ms_subsecond() { - assert_eq!(format_duration_ms(Some(&serde_json::json!(500))), "500ms"); - } - - #[test] - fn format_duration_ms_seconds() { - assert_eq!(format_duration_ms(Some(&serde_json::json!(8000))), "8s"); - } - - #[test] - fn format_duration_ms_minutes() { - assert_eq!(format_duration_ms(Some(&serde_json::json!(90000))), "1.5m"); - } - - #[test] - fn format_tokens_small() { - assert_eq!(format_tokens(500), "500 toks"); - } - - #[test] - fn format_tokens_thousands() { - assert_eq!(format_tokens(15200), "15.2k toks"); - } - - #[test] - fn truncate_short_string() { - assert_eq!(truncate("hello", 10), "hello"); - } - - #[test] - fn truncate_long_string() { - let result = truncate("a very long command string here", 15); - assert!(result.chars().count() <= 15, "got: {result}"); - assert!(result.ends_with('\u{2026}')); - } -} diff --git a/lib/crates/fabro-cli/src/commands/run/mod.rs b/lib/crates/fabro-cli/src/commands/run/mod.rs index f28062ab0..165e6f256 100644 --- a/lib/crates/fabro-cli/src/commands/run/mod.rs +++ b/lib/crates/fabro-cli/src/commands/run/mod.rs @@ -14,6 +14,7 @@ pub(crate) mod command; pub(crate) mod cp; pub(crate) mod create; pub(crate) mod diff; +pub(crate) mod events; pub(crate) mod fork; pub(crate) mod logs; pub(crate) mod output; @@ -99,10 +100,11 @@ pub(crate) async fn dispatch( .await } RunCommands::Diff(args) => diff::run(args, base_ctx).await, - RunCommands::Logs(args) => { + RunCommands::Events(args) => { let styles = Styles::detect_stdout(); - logs::run(&args, &styles, base_ctx).await + events::run(&args, &styles, base_ctx).await } + RunCommands::Logs(args) => logs::run(&args, base_ctx).await, RunCommands::Resume(args) => { let styles: &'static Styles = Box::leak(Box::new(Styles::detect_stderr())); #[cfg(feature = "sleep_inhibitor")] diff --git a/lib/crates/fabro-cli/src/landing.rs b/lib/crates/fabro-cli/src/landing.rs index 72483efad..ef6f87d9c 100644 --- a/lib/crates/fabro-cli/src/landing.rs +++ b/lib/crates/fabro-cli/src/landing.rs @@ -54,7 +54,8 @@ pub(crate) fn print() { section( "Inspect runs", &[ - ("logs", "View the event log of a workflow run"), + ("events", "View the event log of a workflow run"), + ("logs", "View the raw worker tracing log of a workflow run"), ("sandbox ssh", "SSH into a run's sandbox"), ], cmd_width, diff --git a/lib/crates/fabro-cli/tests/it/cmd/attach.rs b/lib/crates/fabro-cli/tests/it/cmd/attach.rs index 75d3d2950..6b244188a 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/attach.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/attach.rs @@ -279,7 +279,7 @@ fn attach_before_completion_streams_to_finished_state() { #[test] #[expect( clippy::disallowed_methods, - reason = "This sync integration test polls logs for a human gate without creating a Tokio runtime." + reason = "This sync integration test polls events for a human gate without creating a Tokio runtime." )] fn attach_json_errors_without_prompting_for_human_input() { let context = test_context!(); @@ -331,13 +331,13 @@ fn attach_json_errors_without_prompting_for_human_input() { } let deadline = std::time::Instant::now() + SHARED_DAEMON_TIMEOUT; loop { - let logs_output = context + let events_output = context .command() - .args(["logs", &run_id, "--json"]) + .args(["events", &run_id, "--json"]) .output() - .expect("logs should execute"); - assert!(logs_output.status.success(), "logs should succeed"); - let log_events: Vec = String::from_utf8(logs_output.stdout) + .expect("events should execute"); + assert!(events_output.status.success(), "events should succeed"); + let log_events: Vec = String::from_utf8(events_output.stdout) .expect("stdout should be UTF-8") .lines() .filter(|line| !line.trim().is_empty()) @@ -371,13 +371,13 @@ fn attach_json_errors_without_prompting_for_human_input() { !stderr.contains("Approve?"), "attach should not prompt on stderr" ); - let logs_output = context + let events_output = context .command() - .args(["logs", &run_id, "--json"]) + .args(["events", &run_id, "--json"]) .output() - .expect("logs should execute"); - assert!(logs_output.status.success(), "logs should succeed"); - let log_events: Vec = String::from_utf8(logs_output.stdout) + .expect("events should execute"); + assert!(events_output.status.success(), "events should succeed"); + let log_events: Vec = String::from_utf8(events_output.stdout) .expect("stdout should be UTF-8") .lines() .filter(|line| !line.trim().is_empty()) diff --git a/lib/crates/fabro-cli/tests/it/cmd/events.rs b/lib/crates/fabro-cli/tests/it/cmd/events.rs new file mode 100644 index 000000000..7e664d325 --- /dev/null +++ b/lib/crates/fabro-cli/tests/it/cmd/events.rs @@ -0,0 +1,250 @@ +use fabro_test::{fabro_snapshot, test_context}; +use serde_json::Value; + +use super::support::{setup_detached_dry_run, setup_seeded_completed_dry_run}; + +fn parse_ndjson(stdout: &[u8]) -> Vec { + String::from_utf8(stdout.to_vec()) + .expect("stdout should be valid UTF-8") + .lines() + .filter(|line| !line.trim().is_empty()) + .map(|line| { + serde_json::from_str::(line).expect("events output should be valid NDJSON") + }) + .collect() +} + +fn assert_event_sequence_contains(events: &[Value], expected: &[&str]) { + let event_names: Vec<&str> = events + .iter() + .filter_map(|event| event["event"].as_str()) + .collect(); + + let mut cursor = 0; + for expected_name in expected { + let Some(found_at) = event_names[cursor..] + .iter() + .position(|name| name == expected_name) + else { + panic!("missing event {expected_name} in sequence: {event_names:?}"); + }; + cursor += found_at + 1; + } +} + +fn assert_events_belong_to_run(events: &[Value], run_id: &str) { + assert!(!events.is_empty(), "expected at least one event"); + for event in events { + assert_eq!( + event["run_id"].as_str(), + Some(run_id), + "event should belong to requested run: {event}" + ); + } +} + +#[test] +fn help() { + let context = test_context!(); + let mut cmd = context.command(); + cmd.args(["events", "--help"]); + fabro_snapshot!(context.filters(), cmd, @r#" + success: true + exit_code: 0 + ----- stdout ----- + View the event log of a workflow run + + Usage: fabro events [OPTIONS] + + Arguments: + Run ID prefix or workflow name (most recent run) + + Options: + --json Output as JSON [env: FABRO_JSON=] + --server Fabro server target: http(s) URL or absolute Unix socket path [env: FABRO_SERVER=] + --debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=] + -f, --follow Follow event output + --no-upgrade-check Disable automatic upgrade check [env: FABRO_NO_UPGRADE_CHECK=true] + --since Events since timestamp or relative (e.g. "42m", "2h", "2026-01-02T13:00:00Z") + -n, --tail Lines from end (default: all) + --quiet Suppress non-essential output [env: FABRO_QUIET=] + -p, --pretty Formatted colored output with rendered assistant text + --verbose Enable verbose output [env: FABRO_VERBOSE=] + -h, --help Print help + ----- stderr ----- + "#); +} + +#[test] +fn events_completed_run_outputs_raw_ndjson() { + let context = test_context!(); + let run = setup_seeded_completed_dry_run(&context); + let mut cmd = context.command(); + cmd.args(["events", &run.run_id]); + let output = cmd.output().expect("command should execute"); + assert!( + output.status.success(), + "events should succeed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + + let events = parse_ndjson(&output.stdout); + assert_events_belong_to_run(&events, &run.run_id); + assert_event_sequence_contains(&events, &[ + "run.created", + "run.running", + "stage.started", + "stage.completed", + "run.completed", + "sandbox.cleanup.completed", + ]); +} + +#[test] +fn events_completed_run_reads_store_without_progress_jsonl() { + let context = test_context!(); + let run = setup_seeded_completed_dry_run(&context); + + let mut filters = context.filters(); + filters.push(( + r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z".to_string(), + "[TIMESTAMP]".to_string(), + )); + filters.push(( + r#""duration_ms":\s*\d+"#.to_string(), + r#""duration_ms": [DURATION_MS]"#.to_string(), + )); + filters.push(( + r#""id":"[0-9a-f-]+""#.to_string(), + r#""id":"[EVENT_ID]""#.to_string(), + )); + filters.push(( + r#""run_dir":"(?:\[DRY_RUN_DIR\]|\[STORAGE_DIR\]/scratch/REDACTED)""#.to_string(), + r#""run_dir":"[RUN_DIR]""#.to_string(), + )); + + let mut cmd = context.command(); + cmd.args(["events", "--tail", "2", &run.run_id]); + + fabro_snapshot!(filters, cmd, @r#" + success: true + exit_code: 0 + ----- stdout ----- + {"actor":{"kind":"worker","run_id":"[ULID]"},"event":"sandbox.cleanup.started","id":"[EVENT_ID]","properties":{"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} + {"actor":{"kind":"worker","run_id":"[ULID]"},"event":"sandbox.cleanup.completed","id":"[EVENT_ID]","properties":{"duration_ms": [DURATION_MS],"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} + ----- stderr ----- + "#); +} + +#[test] +fn events_tail_limits_output() { + let context = test_context!(); + let run = setup_seeded_completed_dry_run(&context); + let mut filters = context.filters(); + filters.push(( + r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z".to_string(), + "[TIMESTAMP]".to_string(), + )); + filters.push(( + r#""duration_ms":\s*\d+"#.to_string(), + r#""duration_ms": [DURATION_MS]"#.to_string(), + )); + filters.push(( + r#""id":"[0-9a-f-]+""#.to_string(), + r#""id":"[EVENT_ID]""#.to_string(), + )); + filters.push(( + r#""run_dir":"(?:\[DRY_RUN_DIR\]|\[STORAGE_DIR\]/scratch/REDACTED)""#.to_string(), + r#""run_dir":"[RUN_DIR]""#.to_string(), + )); + let mut cmd = context.command(); + cmd.args(["events", "--tail", "2", &run.run_id]); + + fabro_snapshot!(filters, cmd, @r#" + success: true + exit_code: 0 + ----- stdout ----- + {"actor":{"kind":"worker","run_id":"[ULID]"},"event":"sandbox.cleanup.started","id":"[EVENT_ID]","properties":{"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} + {"actor":{"kind":"worker","run_id":"[ULID]"},"event":"sandbox.cleanup.completed","id":"[EVENT_ID]","properties":{"duration_ms": [DURATION_MS],"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} + ----- stderr ----- + "#); +} + +#[test] +fn events_since_filters_output() { + let context = test_context!(); + let run = setup_seeded_completed_dry_run(&context); + let mut cmd = context.command(); + cmd.args(["events", "--since", "2999-01-01T00:00:00Z", &run.run_id]); + + fabro_snapshot!(context.filters(), cmd, @r#" + success: true + exit_code: 0 + ----- stdout ----- + ----- stderr ----- + "#); +} + +#[test] +fn events_pretty_formats_small_run() { + let context = test_context!(); + let run = setup_seeded_completed_dry_run(&context); + let mut filters = context.filters(); + filters.push((r"\b\d{2}:\d{2}:\d{2}\b".to_string(), "[CLOCK]".to_string())); + filters.push(( + r"\b\d+(\.\d+)?(ms|s)\b".to_string(), + "[DURATION]".to_string(), + )); + let mut cmd = context.command(); + cmd.args(["events", "--pretty", &run.run_id]); + + fabro_snapshot!(filters, cmd, @" + success: true + exit_code: 0 + ----- stdout ----- + [CLOCK] Sandbox: local [DURATION] + [CLOCK] ▶ Simple [ULID] + Run tests and report results + + [CLOCK] ▶ Start + [CLOCK] ✓ Start [DURATION] + [CLOCK] → run_tests unconditional + [CLOCK] ▶ Run Tests + [CLOCK] ✓ Run Tests [DURATION] + [CLOCK] → report unconditional + [CLOCK] ▶ Report + [CLOCK] ✓ Report [DURATION] + [CLOCK] → exit unconditional + [CLOCK] ▶ Exit + [CLOCK] ✓ Exit [DURATION] + [CLOCK] ✓ SUCCEEDED [DURATION] + ----- stderr ----- + "); +} + +#[test] +fn events_follow_detached_run_streams_until_completion() { + let context = test_context!(); + let run = setup_detached_dry_run(&context); + let mut cmd = context.command(); + cmd.args(["events", "--follow", &run.run_id]); + let output = cmd.output().expect("command should execute"); + assert!( + output.status.success(), + "events --follow should succeed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + + let events = parse_ndjson(&output.stdout); + assert_events_belong_to_run(&events, &run.run_id); + assert_event_sequence_contains(&events, &[ + "run.created", + "run.running", + "stage.started", + "stage.completed", + "run.completed", + "sandbox.cleanup.completed", + ]); +} diff --git a/lib/crates/fabro-cli/tests/it/cmd/fabro.rs b/lib/crates/fabro-cli/tests/it/cmd/fabro.rs index 5fff5e6d1..ff151778f 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/fabro.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/fabro.rs @@ -16,7 +16,8 @@ fn help() { create Create a workflow run (allocate run dir, persist spec) start Start a created workflow run on the server attach Attach to a running or finished workflow run - logs View the event log of a workflow run + events View the event log of a workflow run + logs View the raw worker tracing log of a workflow run resume Resume an interrupted workflow run rewind Rewind a workflow run to an earlier checkpoint fork Fork a workflow run from an earlier checkpoint into a new run @@ -93,7 +94,8 @@ fn no_args_prints_curated_landing() { Inspect runs - fabro logs View the event log of a workflow run + fabro events View the event log of a workflow run + fabro logs View the raw worker tracing log of a workflow run fabro sandbox ssh SSH into a run's sandbox If you need help along the way: diff --git a/lib/crates/fabro-cli/tests/it/cmd/json_global.rs b/lib/crates/fabro-cli/tests/it/cmd/json_global.rs index 1303a451b..de9d07400 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/json_global.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/json_global.rs @@ -172,20 +172,20 @@ fn ps_supports_global_flag_and_env_var() { } #[test] -fn logs_json_wins_over_pretty() { +fn events_json_wins_over_pretty() { let context = test_context!(); let run = setup_seeded_completed_dry_run(&context); let output = context .command() - .args(["--json", "logs", "--pretty", &run.run_id]) + .args(["--json", "events", "--pretty", &run.run_id]) .output() .expect("command should run"); assert!(output.status.success()); let stdout = String::from_utf8(output.stdout).unwrap(); let first_line = stdout.lines().find(|line| !line.is_empty()).unwrap(); - let value: Value = serde_json::from_str(first_line).expect("logs output should remain JSONL"); + let value: Value = serde_json::from_str(first_line).expect("events output should remain JSONL"); assert!(value.get("event").is_some()); } diff --git a/lib/crates/fabro-cli/tests/it/cmd/logs.rs b/lib/crates/fabro-cli/tests/it/cmd/logs.rs index 87d740bab..6b331651f 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/logs.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/logs.rs @@ -1,46 +1,16 @@ +use std::path::Path; + use fabro_test::{fabro_snapshot, test_context}; -use serde_json::Value; -use super::support::{setup_detached_dry_run, setup_seeded_completed_dry_run}; +use super::support::{output_stderr, setup_seeded_completed_dry_run}; -fn parse_ndjson(stdout: &[u8]) -> Vec { - String::from_utf8(stdout.to_vec()) - .expect("stdout should be valid UTF-8") - .lines() - .filter(|line| !line.trim().is_empty()) - .map(|line| { - serde_json::from_str::(line).expect("logs output should be valid NDJSON") - }) - .collect() -} - -fn assert_event_sequence_contains(events: &[Value], expected: &[&str]) { - let event_names: Vec<&str> = events - .iter() - .filter_map(|event| event["event"].as_str()) - .collect(); - - let mut cursor = 0; - for expected_name in expected { - let Some(found_at) = event_names[cursor..] - .iter() - .position(|name| name == expected_name) - else { - panic!("missing event {expected_name} in sequence: {event_names:?}"); - }; - cursor += found_at + 1; - } -} - -fn assert_events_belong_to_run(events: &[Value], run_id: &str) { - assert!(!events.is_empty(), "expected at least one log event"); - for event in events { - assert_eq!( - event["run_id"].as_str(), - Some(run_id), - "event should belong to requested run: {event}" - ); - } +fn seed_run_log(run_dir: &Path, contents: &[u8]) { + // Raw logs are the CLI surface for this runtime-owned file; no public + // command creates deterministic contents suitable for exact assertions. + let log_path = run_dir.join("runtime/server.log"); + std::fs::create_dir_all(log_path.parent().expect("log path should have parent")) + .expect("runtime log directory should be created"); + std::fs::write(&log_path, contents).expect("runtime log should be seeded"); } #[test] @@ -52,7 +22,7 @@ fn help() { success: true exit_code: 0 ----- stdout ----- - View the event log of a workflow run + View the raw worker tracing log of a workflow run Usage: fabro logs [OPTIONS] @@ -63,12 +33,9 @@ fn help() { --json Output as JSON [env: FABRO_JSON=] --server Fabro server target: http(s) URL or absolute Unix socket path [env: FABRO_SERVER=] --debug Enable DEBUG-level logging (default is INFO) [env: FABRO_DEBUG=] - -f, --follow Follow log output - --no-upgrade-check Disable automatic upgrade check [env: FABRO_NO_UPGRADE_CHECK=true] - --since Logs since timestamp or relative (e.g. "42m", "2h", "2026-01-02T13:00:00Z") -n, --tail Lines from end (default: all) + --no-upgrade-check Disable automatic upgrade check [env: FABRO_NO_UPGRADE_CHECK=true] --quiet Suppress non-essential output [env: FABRO_QUIET=] - -p, --pretty Formatted colored output with rendered assistant text --verbose Enable verbose output [env: FABRO_VERBOSE=] -h, --help Print help ----- stderr ----- @@ -76,160 +43,109 @@ fn help() { } #[test] -fn logs_completed_run_outputs_raw_ndjson() { +fn logs_run_outputs_seeded_runtime_log_exactly() { let context = test_context!(); let run = setup_seeded_completed_dry_run(&context); - let mut cmd = context.command(); - cmd.args(["logs", &run.run_id]); - let output = cmd.output().expect("command should execute"); + seed_run_log(&run.run_dir, b"worker started\nworker finished"); + + let output = context + .command() + .args(["logs", &run.run_id]) + .output() + .expect("command should run"); + assert!( output.status.success(), "logs should succeed\nstdout:\n{}\nstderr:\n{}", String::from_utf8_lossy(&output.stdout), String::from_utf8_lossy(&output.stderr) ); - - let events = parse_ndjson(&output.stdout); - assert_events_belong_to_run(&events, &run.run_id); - assert_event_sequence_contains(&events, &[ - "run.created", - "run.running", - "stage.started", - "stage.completed", - "run.completed", - "sandbox.cleanup.completed", - ]); + assert_eq!(output.stdout, b"worker started\nworker finished"); } #[test] -fn logs_completed_run_reads_store_without_progress_jsonl() { +fn logs_tail_one_outputs_final_line() { let context = test_context!(); let run = setup_seeded_completed_dry_run(&context); + seed_run_log(&run.run_dir, b"worker started\nworker finished"); - let mut filters = context.filters(); - filters.push(( - r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z".to_string(), - "[TIMESTAMP]".to_string(), - )); - filters.push(( - r#""duration_ms":\s*\d+"#.to_string(), - r#""duration_ms": [DURATION_MS]"#.to_string(), - )); - filters.push(( - r#""id":"[0-9a-f-]+""#.to_string(), - r#""id":"[EVENT_ID]""#.to_string(), - )); - filters.push(( - r#""run_dir":"(?:\[DRY_RUN_DIR\]|\[STORAGE_DIR\]/scratch/REDACTED)""#.to_string(), - r#""run_dir":"[RUN_DIR]""#.to_string(), - )); + let output = context + .command() + .args(["logs", "--tail", "1", &run.run_id]) + .output() + .expect("command should run"); + assert!(output.status.success()); + assert_eq!(output.stdout, b"worker finished\n"); +} + +#[test] +fn logs_tail_zero_outputs_nothing() { + let context = test_context!(); + let run = setup_seeded_completed_dry_run(&context); + seed_run_log(&run.run_dir, b"worker started\nworker finished"); + + let output = context + .command() + .args(["logs", "--tail", "0", &run.run_id]) + .output() + .expect("command should run"); + + assert!(output.status.success()); + assert!(output.stdout.is_empty()); +} + +#[test] +fn logs_follow_is_unknown_argument() { + let context = test_context!(); let mut cmd = context.command(); - cmd.args(["logs", "--tail", "2", &run.run_id]); + cmd.args(["logs", "--follow", "01K00000000000000000000000"]); - fabro_snapshot!(filters, cmd, @r#" - success: true - exit_code: 0 + fabro_snapshot!(context.filters(), cmd, @r#" + success: false + exit_code: 2 ----- stdout ----- - {"actor":{"kind":"worker","run_id":"[ULID]"},"event":"sandbox.cleanup.started","id":"[EVENT_ID]","properties":{"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} - {"actor":{"kind":"worker","run_id":"[ULID]"},"event":"sandbox.cleanup.completed","id":"[EVENT_ID]","properties":{"duration_ms": [DURATION_MS],"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} ----- stderr ----- + error: unexpected argument '--follow' found + + tip: to pass '--follow' as a value, use '-- --follow' + + Usage: fabro logs [OPTIONS] + + For more information, try '--help'. "#); } #[test] -fn logs_tail_limits_output() { +fn logs_json_is_rejected() { let context = test_context!(); let run = setup_seeded_completed_dry_run(&context); - let mut filters = context.filters(); - filters.push(( - r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z".to_string(), - "[TIMESTAMP]".to_string(), - )); - filters.push(( - r#""duration_ms":\s*\d+"#.to_string(), - r#""duration_ms": [DURATION_MS]"#.to_string(), - )); - filters.push(( - r#""id":"[0-9a-f-]+""#.to_string(), - r#""id":"[EVENT_ID]""#.to_string(), - )); - filters.push(( - r#""run_dir":"(?:\[DRY_RUN_DIR\]|\[STORAGE_DIR\]/scratch/REDACTED)""#.to_string(), - r#""run_dir":"[RUN_DIR]""#.to_string(), - )); - let mut cmd = context.command(); - cmd.args(["logs", "--tail", "2", &run.run_id]); + seed_run_log(&run.run_dir, b"worker started\n"); - fabro_snapshot!(filters, cmd, @r#" - success: true - exit_code: 0 - ----- stdout ----- - {"actor":{"kind":"worker","run_id":"[ULID]"},"event":"sandbox.cleanup.started","id":"[EVENT_ID]","properties":{"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} - {"actor":{"kind":"worker","run_id":"[ULID]"},"event":"sandbox.cleanup.completed","id":"[EVENT_ID]","properties":{"duration_ms": [DURATION_MS],"provider":"local"},"run_id":"[ULID]","ts":"[TIMESTAMP]"} - ----- stderr ----- - "#); + let output = context + .command() + .args(["--json", "logs", &run.run_id]) + .output() + .expect("command should run"); + + assert!(!output.status.success()); + let stderr = output_stderr(&output); + assert!(stderr.contains("--json is not supported for this command")); } #[test] -fn logs_pretty_formats_small_run() { +fn logs_missing_runtime_log_exits_nonzero() { let context = test_context!(); let run = setup_seeded_completed_dry_run(&context); - let mut filters = context.filters(); - filters.push((r"\b\d{2}:\d{2}:\d{2}\b".to_string(), "[CLOCK]".to_string())); - filters.push(( - r"\b\d+(\.\d+)?(ms|s)\b".to_string(), - "[DURATION]".to_string(), - )); - let mut cmd = context.command(); - cmd.args(["logs", "--pretty", &run.run_id]); + let _ = std::fs::remove_file(run.run_dir.join("runtime/server.log")); - fabro_snapshot!(filters, cmd, @" - success: true - exit_code: 0 - ----- stdout ----- - [CLOCK] Sandbox: local [DURATION] - [CLOCK] ▶ Simple [ULID] - Run tests and report results + let output = context + .command() + .args(["logs", &run.run_id]) + .output() + .expect("command should run"); - [CLOCK] ▶ Start - [CLOCK] ✓ Start [DURATION] - [CLOCK] → run_tests unconditional - [CLOCK] ▶ Run Tests - [CLOCK] ✓ Run Tests [DURATION] - [CLOCK] → report unconditional - [CLOCK] ▶ Report - [CLOCK] ✓ Report [DURATION] - [CLOCK] → exit unconditional - [CLOCK] ▶ Exit - [CLOCK] ✓ Exit [DURATION] - [CLOCK] ✓ SUCCEEDED [DURATION] - ----- stderr ----- - "); -} - -#[test] -fn logs_follow_detached_run_streams_until_completion() { - let context = test_context!(); - let run = setup_detached_dry_run(&context); - let mut cmd = context.command(); - cmd.args(["logs", "--follow", &run.run_id]); - let output = cmd.output().expect("command should execute"); - assert!( - output.status.success(), - "logs --follow should succeed\nstdout:\n{}\nstderr:\n{}", - String::from_utf8_lossy(&output.stdout), - String::from_utf8_lossy(&output.stderr) - ); - - let events = parse_ndjson(&output.stdout); - assert_events_belong_to_run(&events, &run.run_id); - assert_event_sequence_contains(&events, &[ - "run.created", - "run.running", - "stage.started", - "stage.completed", - "run.completed", - "sandbox.cleanup.completed", - ]); + assert!(!output.status.success()); + let stderr = output_stderr(&output); + assert!(stderr.contains("Run log not available")); } diff --git a/lib/crates/fabro-cli/tests/it/cmd/mod.rs b/lib/crates/fabro-cli/tests/it/cmd/mod.rs index fed4a406a..58a98b5e3 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/mod.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/mod.rs @@ -11,6 +11,7 @@ mod discord; mod docs; mod doctor; mod dump; +mod events; mod exec; mod fabro; mod fork; diff --git a/lib/crates/fabro-cli/tests/it/cmd/run.rs b/lib/crates/fabro-cli/tests/it/cmd/run.rs index 2aa614979..cccd1eb21 100644 --- a/lib/crates/fabro-cli/tests/it/cmd/run.rs +++ b/lib/crates/fabro-cli/tests/it/cmd/run.rs @@ -785,12 +785,12 @@ fn dry_run_persists_event_history_in_store() { wait_for_event_names(&run_dir, &["run.completed", "sandbox.cleanup.completed"]); let output = context .command() - .args(["logs", &run_id]) + .args(["events", &run_id]) .output() - .expect("logs command should execute"); + .expect("events command should execute"); assert!( output.status.success(), - "logs failed:\nstdout:\n{}\nstderr:\n{}", + "events failed:\nstdout:\n{}\nstderr:\n{}", String::from_utf8_lossy(&output.stdout), String::from_utf8_lossy(&output.stderr) ); @@ -798,7 +798,7 @@ fn dry_run_persists_event_history_in_store() { .expect("stdout should be UTF-8") .lines() .filter(|line| !line.trim().is_empty()) - .map(|line| serde_json::from_str(line).expect("logs output should be JSONL")) + .map(|line| serde_json::from_str(line).expect("events output should be JSONL")) .collect(); assert!( !progress.is_empty(), @@ -828,12 +828,12 @@ fn dry_run_persists_event_history_in_store() { let tail_output = context .command() - .args(["logs", "--tail", "1", &run_id]) + .args(["events", "--tail", "1", &run_id]) .output() - .expect("tail logs command should execute"); + .expect("tail events command should execute"); assert!( tail_output.status.success(), - "tail logs failed:\nstdout:\n{}\nstderr:\n{}", + "tail events failed:\nstdout:\n{}\nstderr:\n{}", String::from_utf8_lossy(&tail_output.stdout), String::from_utf8_lossy(&tail_output.stderr) ); @@ -841,8 +841,8 @@ fn dry_run_persists_event_history_in_store() { .expect("stdout should be UTF-8") .lines() .find(|line| !line.trim().is_empty()) - .map(|line| serde_json::from_str(line).expect("tail logs output should be JSON")) - .expect("tail logs should include the latest event"); + .map(|line| serde_json::from_str(line).expect("tail events output should be JSON")) + .expect("tail events should include the latest event"); fabro_json_snapshot!(context, &live_content, @r#" { "actor": { diff --git a/lib/crates/fabro-cli/tests/it/scenario/lifecycle.rs b/lib/crates/fabro-cli/tests/it/scenario/lifecycle.rs index 2b7f1a078..c91a1664d 100644 --- a/lib/crates/fabro-cli/tests/it/scenario/lifecycle.rs +++ b/lib/crates/fabro-cli/tests/it/scenario/lifecycle.rs @@ -67,13 +67,13 @@ fn local_run_lifecycle() { items[0]["conclusion"].is_object(), "inspect should include conclusion" ); - // 4. logs — non-empty, first line is valid JSONL with event field - let logs_out = cmd(&["logs", &run_id]).success(); - let logs_stdout = String::from_utf8(logs_out.get_output().stdout.clone()).unwrap(); - assert!(!logs_stdout.is_empty(), "logs should not be empty"); - let first_line = logs_stdout.lines().next().unwrap(); + // 4. events — non-empty, first line is valid JSONL with event field + let events_out = cmd(&["events", &run_id]).success(); + let events_stdout = String::from_utf8(events_out.get_output().stdout.clone()).unwrap(); + assert!(!events_stdout.is_empty(), "events should not be empty"); + let first_line = events_stdout.lines().next().unwrap(); let log_entry: Value = - serde_json::from_str(first_line).expect("first log line should be valid JSON"); + serde_json::from_str(first_line).expect("first event line should be valid JSON"); assert!( log_entry["event"].is_string(), "first log line should have an event field"