car_server_core/assistant/

do_json.rs

//! Machine-readable output for `car do --json`.
//!
//! ## Why two streams
//!
//! A host that delegates to `car do` — a Claude Code subagent, a Codex skill,
//! any script — needs one thing it can parse and one thing it can narrate.
//! Interleaving them into a single stream forces every consumer to re-derive
//! the boundary, and consumers get it wrong.
//!
//! So: **stdout carries exactly one JSON document**, the final result, written
//! once at the end. **stderr carries JSONL progress events**, one object per
//! line, written as they happen. Nothing else is written to either stream in
//! `--json` mode — the human progress rendering is suppressed.
//!
//! Neither Claude Code nor Codex streams a shell call, so in practice the
//! events are read after the process exits. They still earn their place: they
//! are how a caller explains a slow run and reports what was rejected
//! mid-flight, rather than narrating from the final blob alone.
//!
//! ## Size is a correctness property, not a nicety
//!
//! Claude Code delegates into an isolated subagent context, so a large result
//! costs that subagent's window and nothing else. **Codex has no
//! plugin-authored subagent surface** — a `car do` delegation there runs as a
//! skill on the main thread, and every byte lands in the user's primary
//! context. An envelope sized for the Claude Code case poisons Codex sessions.
//!
//! Hence: the summary is capped, receipts are reported as counts plus a
//! bounded sample rather than a transcript, and every truncation states what
//! was elided instead of silently clipping (same reasoning as `cap()` in
//! `car_server_core::assistant::agent_loop`).
//!
//! ## Contract stability
//!
//! The moment a plugin parses this, it is a public wire contract. It carries a
//! version (`"schema": "car.do/1"`) and is documented in
//! `docs/car-do-json.md`. Additive fields are compatible; removing or
//! retyping one is not, and takes a new schema id.
//!
//! ## Why this lives in the runtime and not in the CLI
//!
//! It was `car-cli`'s until the daemon's MCP endpoint grew
//! `assistant_start`/`assistant_poll` (car#972 §6), which needs the exact same
//! envelope: an `assistant_poll` result IS this document, and a poll's event
//! list IS these JSONL events. A second envelope would have been a second
//! contract to version, and — more to the point — a second place for the
//! `SUMMARY_CAP` / `RECEIPT_SAMPLE` truncation to be forgotten, when the reason
//! that truncation exists (a caller pays for every byte in the user's context)
//! applies to an MCP tool result verbatim.
//!
//! So the *shape* is here and the *destination* is the caller's:
//! [`JsonEmitter`] builds values and hands them to an [`EventSink`], and
//! [`JsonEmitter::finish`] returns the terminal document rather than printing
//! it. `car do --json` supplies a stderr sink and prints the returned document
//! to stdout; the MCP run registry supplies a buffer sink and returns the
//! document as the tool result.

use std::collections::BTreeMap;
use std::sync::Arc;
use std::time::Instant;

use serde_json::{json, Value};

use super::{AssistantEvent, AssistantOutcome, AssistantToolReceipt, AuthRequiredReason};

/// Wire-contract version. Bump the major on any non-additive change.
pub const SCHEMA: &str = "car.do/1";

/// Cap on the `summary` field, in bytes.
///
/// Chosen for the Codex main-thread case above: a summary is a report, and a
/// report that does not fit in a few thousand bytes is a transcript wearing a
/// summary's name. Overflow states the elision rather than clipping silently.
const SUMMARY_CAP: usize = 4096;

/// Cap on a single `brief` string extracted from tool parameters.
const BRIEF_CAP: usize = 160;

/// How many receipts reach `receipts.sample`. Failures are selected first —
/// a caller diagnosing a run needs the failures, and the successes are already
/// summarized by `by_tool`.
const RECEIPT_SAMPLE: usize = 8;

/// Truncate to `cap` bytes on a char boundary, stating what was dropped.
fn cap_text(s: &str, cap: usize) -> String {
    if s.len() <= cap {
        return s.to_string();
    }
    let mut end = cap;
    while !s.is_char_boundary(end) {
        end -= 1;
    }
    let elided = s.len() - end;
    format!(
        "{}\n…[truncated: {} of {} bytes shown; {} elided]…",
        &s[..end],
        end,
        s.len(),
        elided
    )
}

/// The one-line gist of a tool call, for events and receipt samples.
///
/// Deliberately NOT the full parameter object. Parameters carry file contents,
/// request bodies, and whatever the model put in a shell command; forwarding
/// them wholesale would blow the size budget and widen what leaves the process.
/// The named keys are the ones that identify a call to a reader.
fn brief(params: &Value) -> String {
    // Ordered by how identifying the key is, and kept in sync with the
    // assistant's actual tool parameters — a key missing here renders as an
    // empty brief, which is what `calculate` did on the first end-to-end run.
    const IDENTIFYING: &[&str] = &[
        "command",
        "path",
        "url",
        "query",
        "goal",
        "expression",
        "subject",
        "name",
        "content",
    ];
    let raw = IDENTIFYING
        .iter()
        .find_map(|k| params.get(*k).and_then(Value::as_str))
        .unwrap_or_default()
        .replace('\n', " ");
    cap_text(&raw, BRIEF_CAP)
}

/// Sandbox posture, captured before the environment is consumed by the runtime.
///
/// Recorded rather than re-derived because it answers the question a reviewing
/// human actually asks about an autonomous run — what could this have touched?
#[derive(Clone)]
pub struct SandboxPosture {
    pub sandboxed: bool,
    pub image: Option<String>,
    pub tier: String,
    pub root: String,
    /// The sandbox mount when it is WIDER than `root` — a repository root the
    /// run was widened to so git works (car#1269). `None` when the mount is
    /// `root` itself, and on every local run.
    pub mount: Option<String>,
    pub fallback_notice: Option<String>,
}

impl SandboxPosture {
    /// The `sandbox` block of the `car.do/1` document. Public so a caller can
    /// report the bound posture before the run finishes — `assistant_start`
    /// does, so a caller that asked for a sandbox and got the local host
    /// learns it up front rather than in the terminal document.
    pub fn to_json(&self) -> Value {
        json!({
            "mode": if self.sandboxed { "docker" } else { "local" },
            "image": self.image,
            "network": if self.sandboxed { "none" } else { "host" },
            "tier": self.tier,
            "root": self.root,
            // Present only when the container can reach MORE than `root`. A
            // caller reporting on what an autonomous run could touch has to
            // read this, not `root`, which is only where the run stands.
            "mount": self.mount,
            // Present only when the sandbox was requested and unavailable. A
            // run that silently fell back to the local host is materially
            // different from one that chose it, and the caller must be able to
            // tell them apart.
            "fallback_notice": self.fallback_notice,
        })
    }
}

/// Goal-mode outcome, present only for `--until` / `--infer-until` runs.
pub struct GoalReport {
    pub check: String,
    pub passed: bool,
    /// Whether completion was established by the deterministic check rather
    /// than a model judge. A `passed: true, grounded: false` run completed on
    /// a judge's say-so and must not be read as verified.
    pub grounded: bool,
    pub iterations: u32,
    pub halt: Option<String>,
}

impl GoalReport {
    fn to_json(&self) -> Value {
        json!({
            "check": self.check,
            "passed": self.passed,
            "grounded": self.grounded,
            "iterations": self.iterations,
            "halt": self.halt,
        })
    }
}

/// Where a progress event goes once it has been built.
///
/// The two implementations are `car do --json`'s stderr writer and the MCP run
/// registry's replay buffer. Split out so the event *shape* has one definition
/// and the destination is the caller's business — an MCP poll returns the same
/// objects `car do --json` writes as JSONL lines because they are built by the
/// same code, not merely documented as equivalent.
///
/// `Send + Sync` because the MCP sink is written from a spawned run task.
pub trait EventSink: Send + Sync {
    /// Take one progress event: `{ type, phase, message, data }`.
    fn emit(&self, event: Value);
}

/// Builds the progress events and the terminal document for one run.
pub struct JsonEmitter {
    started: Instant,
    posture: SandboxPosture,
    sink: Arc<dyn EventSink>,
    /// `delegate` calls seen on this run. Counted from the parent's own
    /// `ToolCall` events — a child's events never reach this emitter — so the
    /// number is delegations issued, whatever each one then did.
    delegations: std::sync::atomic::AtomicU32,
}

impl JsonEmitter {
    pub fn new(posture: SandboxPosture, sink: Arc<dyn EventSink>) -> Self {
        Self {
            started: Instant::now(),
            posture,
            sink,
            delegations: std::sync::atomic::AtomicU32::new(0),
        }
    }

    /// Build one progress event and hand it to the sink.
    fn event(&self, ty: &str, phase: &str, message: impl Into<String>, data: Value) {
        self.sink.emit(json!({
            "type": ty,
            "phase": phase,
            "message": message.into(),
            "data": data,
        }));
    }

    pub fn started(&self, goal: &str, model: &str) {
        self.event(
            "started",
            "run",
            "run started",
            json!({
                "goal": cap_text(goal, BRIEF_CAP),
                "model": model,
                "sandbox": self.posture.to_json(),
            }),
        );
    }

    /// Adapter for the assistant loop's `emit` callback.
    ///
    /// `Done` and `Error` are deliberately NOT emitted here. Exactly one of
    /// `completed` / `failed` terminates the stream, and it is written by
    /// [`Self::finish`] / [`Self::fail_run`] alongside the stdout document, so the
    /// two can never disagree about how the run ended.
    pub fn on_assistant_event(&self, ev: &AssistantEvent) {
        match ev {
            AssistantEvent::InferenceStarted {
                model,
                attempt,
                turn,
            } => self.event(
                "inference_started",
                "inference",
                format!("{model} inference attempt {attempt} started"),
                json!({ "model": model, "attempt": attempt, "turn": turn }),
            ),
            AssistantEvent::InferenceRetry {
                model,
                attempt,
                reason,
                backoff_ms,
            } => self.event(
                "inference_retry",
                "inference",
                format!("{model} inference retry {attempt} after {reason}"),
                json!({
                    "model": model,
                    "attempt": attempt,
                    "reason": reason,
                    "backoff_ms": backoff_ms,
                }),
            ),
            AssistantEvent::ModelServed {
                model_id,
                local_last_resort,
            } => self.event(
                "model_served",
                "inference",
                if *local_last_resort {
                    format!("{model_id} served via on-device last-resort fallback")
                } else {
                    format!("{model_id} served")
                },
                json!({
                    "model_id": model_id,
                    "local_last_resort": local_last_resort,
                }),
            ),
            AssistantEvent::Text(t) if !t.trim().is_empty() => {
                self.event("text", "reasoning", cap_text(t, BRIEF_CAP * 4), json!({}))
            }
            AssistantEvent::Text(_) => {}
            AssistantEvent::ToolCall {
                call_id,
                sequence,
                name,
                params,
            } => {
                if name == super::agent_loop::DELEGATE_TOOL {
                    self.delegations
                        .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
                }
                self.event(
                    "tool_called",
                    "acting",
                    format!("{name}({})", brief(params)),
                    json!({
                        "call_id": call_id,
                        "sequence": sequence,
                        "tool": name,
                        "brief": brief(params),
                    }),
                )
            }
            AssistantEvent::ToolResult {
                call_id,
                sequence,
                name,
                ok,
                ..
            } => self.event(
                if *ok { "tool_result" } else { "tool_failed" },
                "acting",
                format!("{name} {}", if *ok { "ok" } else { "failed" }),
                json!({
                    "call_id": call_id,
                    "sequence": sequence,
                    "tool": name,
                    "ok": ok,
                }),
            ),
            AssistantEvent::GoalEvaluated {
                iteration,
                met,
                grounded,
                reason,
            } => self.event(
                "goal_evaluated",
                "verifying",
                cap_text(reason, BRIEF_CAP * 2),
                json!({
                    "iteration": iteration,
                    "met": met,
                    "grounded": grounded,
                }),
            ),
            // Terminal events are written by `finish` / `fail_run` alongside
            // the stdout document, for the reason in this method's doc comment:
            // exactly one terminal event, and it cannot disagree with the
            // document beside it.
            AssistantEvent::Done { .. }
            | AssistantEvent::Error(_)
            | AssistantEvent::AuthRequired { .. } => {}
        }
    }

    /// Roll receipts up into counts plus a bounded, failure-first sample.
    fn receipts_json(receipts: &[AssistantToolReceipt]) -> Value {
        let mut by_tool: BTreeMap<&str, u64> = BTreeMap::new();
        let mut failed = 0u64;
        for r in receipts {
            *by_tool.entry(r.tool.as_str()).or_default() += 1;
            if !r.ok {
                failed += 1;
            }
        }
        // Failures first: a caller diagnosing a run needs those, and `by_tool`
        // already accounts for the successes.
        let sample: Vec<Value> = receipts
            .iter()
            .filter(|r| !r.ok)
            .chain(receipts.iter().filter(|r| r.ok))
            .take(RECEIPT_SAMPLE)
            .map(|r| match &r.via {
                // A receipt a delegate child produced: say so, or the sample
                // reads as the parent's own call.
                Some(via) => {
                    json!({ "tool": r.tool, "ok": r.ok, "brief": brief(&r.params), "via": via })
                }
                None => json!({ "tool": r.tool, "ok": r.ok, "brief": brief(&r.params) }),
            })
            .collect();
        let omitted = receipts.len().saturating_sub(sample.len());
        json!({
            "total": receipts.len(),
            "failed": failed,
            "by_tool": by_tool,
            "sample": sample,
            // Stated rather than implied. A sample that silently drops 40 calls
            // reads as a complete list to anyone who does not check `total`.
            "sample_omitted": omitted,
        })
    }

    /// Emit the terminal event and return the document for a finished run.
    ///
    /// Returned rather than written: the caller decides whether it goes to
    /// stdout (`car do --json`) or into a tool result (`assistant_poll`).
    ///
    /// A run whose loop errored is routed to the error shape, NOT reported as
    /// a result with an apologetic summary. `AssistantOutcome` carries the
    /// error text in `summary`, and a consumer that reads `summary` without
    /// checking `status` would present a transport failure as the answer.
    pub fn finish(&self, outcome: &AssistantOutcome, goal: Option<&GoalReport>) -> Value {
        if outcome.status == "error" || outcome.auth_required.is_some() {
            return self.fail_run(outcome);
        }
        let ungrounded = super::ungrounded_summary_claims(&outcome.summary, &outcome.tool_receipts);
        let elapsed = self.started.elapsed().as_secs_f64();

        self.event(
            "completed",
            "run",
            "run finished",
            json!({
                "status": outcome.status,
                "turns": outcome.turns,
                "models_served": outcome.models_served,
                "elapsed_seconds": elapsed,
            }),
        );

        let mut doc = json!({
            "schema": SCHEMA,
            "status": outcome.status,
            "summary": cap_text(&outcome.summary, SUMMARY_CAP),
            // The PARENT's turns only. A `delegate` child's turns and model
            // calls are metered in the journal but not counted here; the
            // number of delegations issued is reported beside it.
            "turns": outcome.turns,
            "delegations": self.delegations.load(std::sync::atomic::Ordering::Relaxed),
            "model_used": outcome.model_used,
            "models_served": outcome.models_served,
            "receipts": Self::receipts_json(&outcome.tool_receipts),
            // The differentiator. Operational claims in the prose with no
            // matching receipt. An empty array means nothing was detected, not
            // that the summary is verified — see `ungrounded_summary_claims`.
            "ungrounded_claims": ungrounded,
            "sandbox": self.posture.to_json(),
            "elapsed_seconds": elapsed,
        });
        if let Some(g) = goal {
            doc["goal"] = g.to_json();
        }
        doc
    }

    /// Emit the terminal `failed` event and return the error document.
    ///
    /// The error document has a **different shape** — `message` rather than
    /// `summary` — so a consumer that reads `summary` without checking
    /// `status` gets a missing key instead of quietly presenting a transport
    /// failure as the run's answer.
    ///
    /// Receipts are still reported: a run that died on turn nine did real work
    /// first, and that work is what the caller needs to reason about before
    /// retrying.
    fn fail_run(&self, outcome: &AssistantOutcome) -> Value {
        let elapsed = self.started.elapsed().as_secs_f64();
        // An account refusal takes the SAME shape — `message`, not `summary` —
        // because it is equally not an answer, and a consumer that reads
        // `summary` without checking `status` must still get a missing key.
        // `status` and the additive `reason`/`message` are what tell it apart;
        // `error` stays a plain string, as every other terminal document's
        // does, so nothing has to learn a second type for that field.
        // Reason-specific, and free of shell commands. `--json` is read by
        // programs and by hosts that have their own sign-in entry; handing them
        // a CLI incantation is the wrong instruction in the wrong place. And
        // `no_workspace` is not repaired by signing in at all — the person
        // already did.
        let (status, error, suggestions) = match outcome.auth_required {
            Some(reason) => (
                "auth_required",
                "AuthRequired",
                vec![
                    match reason {
                        AuthRequiredReason::SignedOut => {
                            "Sign in to your Parslee account, then re-run."
                        }
                        AuthRequiredReason::Expired => {
                            "Your Parslee sign-in has expired. Sign in again, then re-run."
                        }
                        AuthRequiredReason::NoWorkspace => {
                            "Finish setting up your Parslee account at parslee.ai, then re-run."
                        }
                    },
                    "Or name a different model for this run with --model / CAR_DO_MODEL.",
                ],
            ),
            None => (
                "error",
                "AssistantLoopFailed",
                vec![
                    "Re-run the goal; the run failed mid-loop rather than completing with an \
                     answer.",
                    "Check `receipts` for what had already executed before the failure.",
                ],
            ),
        };
        let mut event_data = json!({
            "error": error,
            "turns": outcome.turns,
            "turns_completed": outcome.turns_completed,
            "models_served": outcome.models_served,
        });
        if let Some(cause) = outcome.failure_cause {
            event_data["failure"] = json!(cause);
        }
        self.event(
            "failed",
            "run",
            cap_text(&outcome.summary, BRIEF_CAP * 2),
            event_data,
        );
        let mut doc = json!({
            "schema": SCHEMA,
            "status": status,
            "error": error,
            "message": cap_text(&outcome.summary, SUMMARY_CAP),
            "turns": outcome.turns,
            "turns_completed": outcome.turns_completed,
            "model_used": outcome.model_used,
            "models_served": outcome.models_served,
            "receipts": Self::receipts_json(&outcome.tool_receipts),
            "sandbox": self.posture.to_json(),
            "elapsed_seconds": elapsed,
            "suggestions": suggestions,
        });
        if let Some(reason) = outcome.auth_required {
            doc["reason"] = json!(reason.as_str());
        }
        if let Some(cause) = outcome.failure_cause {
            doc["failure"] = json!(cause);
        }
        doc
    }
}

/// The error document for a failure with no run outcome to describe — a bad
/// flag combination, an unresolvable working directory, a refused MCP start,
/// or a run whose task died without producing an outcome at all.
///
/// Separate from [`JsonEmitter::finish`] because there is no run outcome to
/// time or describe, and inventing a `SandboxPosture` to report would be a lie.
/// It still carries `schema` and `status`, so the caller's invariant — a
/// `--json` invocation always produces exactly one document — holds on the
/// paths where no run ever existed.
pub fn startup_error_doc(error: &str, message: &str, suggestions: &[&str]) -> Value {
    json!({
        "schema": SCHEMA,
        "status": "error",
        "error": error,
        "message": message,
        "suggestions": suggestions,
    })
}

#[cfg(test)]
mod tests {
    use super::super::AssistantModelAttribution;
    use super::*;

    fn posture() -> SandboxPosture {
        SandboxPosture {
            sandboxed: true,
            image: Some("python:3.11".into()),
            tier: "SandboxEdit".into(),
            root: "/work".into(),
            mount: None,
            fallback_notice: None,
        }
    }

    #[test]
    fn cap_text_states_what_it_dropped() {
        let s = "x".repeat(100);
        let out = cap_text(&s, 10);
        assert!(out.starts_with(&"x".repeat(10)));
        assert!(out.contains("90 elided"), "{out}");
    }

    #[test]
    fn cap_text_leaves_short_input_untouched() {
        assert_eq!(cap_text("short", 100), "short");
    }

    #[test]
    fn cap_text_respects_char_boundaries() {
        // Multi-byte input truncated mid-character must not panic.
        let s = "é".repeat(50);
        let out = cap_text(&s, 11);
        assert!(out.contains("elided"), "{out}");
    }

    #[test]
    fn brief_prefers_identifying_keys_and_flattens_newlines() {
        let p = json!({ "command": "cargo test\n--quiet", "body": "…huge…" });
        assert_eq!(brief(&p), "cargo test --quiet");
    }

    #[test]
    fn brief_is_empty_when_no_identifying_key_is_present() {
        assert_eq!(brief(&json!({ "body": "opaque" })), "");
    }

    #[test]
    fn brief_covers_the_tools_that_do_not_take_a_command_or_path() {
        // Regression: the first end-to-end run rendered `calculate()` with an
        // empty brief because `expression` was not in the list.
        assert_eq!(brief(&json!({ "expression": "17 * 23" })), "17 * 23");
        assert_eq!(
            brief(&json!({ "query": "rust lifetimes" })),
            "rust lifetimes"
        );
        assert_eq!(
            brief(&json!({ "subject": "deploy cadence" })),
            "deploy cadence"
        );
    }

    #[test]
    fn receipts_roll_up_counts_and_put_failures_in_the_sample_first() {
        let mut receipts: Vec<AssistantToolReceipt> = (0..20)
            .map(|i| AssistantToolReceipt {
                tool: "shell".into(),
                call_id: None,
                sequence: None,
                ok: true,
                params: json!({ "command": format!("ok-{i}") }),
                result: None,
                via: None,
            })
            .collect();
        receipts.push(AssistantToolReceipt {
            tool: "write_file".into(),
            call_id: None,
            sequence: None,
            ok: false,
            params: json!({ "path": "/denied" }),
            result: None,
            via: None,
        });

        let v = JsonEmitter::receipts_json(&receipts);
        assert_eq!(v["total"], 21);
        assert_eq!(v["failed"], 1);
        assert_eq!(v["by_tool"]["shell"], 20);
        assert_eq!(v["by_tool"]["write_file"], 1);
        // The failure leads, so a caller reading only the sample sees it.
        assert_eq!(v["sample"][0]["tool"], "write_file");
        assert_eq!(v["sample"].as_array().unwrap().len(), RECEIPT_SAMPLE);
        // And the drop is stated rather than implied.
        assert_eq!(v["sample_omitted"], 21 - RECEIPT_SAMPLE);
    }

    #[test]
    fn sandbox_posture_distinguishes_a_fallback_from_a_choice() {
        let chosen = posture().to_json();
        assert_eq!(chosen["mode"], "docker");
        assert_eq!(chosen["network"], "none");
        assert!(chosen["fallback_notice"].is_null());

        let fell_back = SandboxPosture {
            sandboxed: false,
            image: None,
            fallback_notice: Some("Docker not running".into()),
            ..posture()
        }
        .to_json();
        assert_eq!(fell_back["mode"], "local");
        assert_eq!(fell_back["network"], "host");
        assert_eq!(fell_back["fallback_notice"], "Docker not running");
    }

    #[test]
    fn an_errored_run_is_not_reported_as_a_result() {
        // The guard that matters: `summary` must be absent on the error shape,
        // so a consumer reading it without checking `status` fails loudly
        // rather than presenting a transport failure as the answer.
        let outcome = AssistantOutcome {
            status: "error",
            summary: "connection reset by peer".into(),
            turns: 9,
            turns_completed: 8,
            tools_called: vec![],
            prior_receipts: 0,
            tool_receipts: vec![AssistantToolReceipt {
                tool: "shell".into(),
                call_id: None,
                sequence: None,
                ok: true,
                params: json!({ "command": "ls" }),
                result: None,
                via: None,
            }],
            models_served: vec![AssistantModelAttribution {
                model_id: "claude-opus-5".into(),
                local_last_resort: false,
            }],
            model_used: "claude-opus-5".into(),
            auth_required: None,
            failure_cause: Some(
                super::super::agent_loop::AssistantFailureCause::TransientInference {
                    status: None,
                },
            ),
        };
        let sink = Arc::new(Captured::default());
        let doc = JsonEmitter::new(posture(), sink.clone()).finish(&outcome, None);
        assert_eq!(doc["status"], "error");
        assert!(doc.get("summary").is_none(), "summary leaked: {doc}");
        assert_eq!(doc["message"], "connection reset by peer");
        assert_eq!(doc["turns"], 9);
        assert_eq!(doc["turns_completed"], 8);
        assert_eq!(doc["failure"]["cause"], "transient_inference");
        assert!(doc["failure"]["status"].is_null());
        // Work done before the failure is still reported.
        assert_eq!(doc["receipts"]["total"], 1);
        assert_eq!(doc["models_served"][0]["model_id"], "claude-opus-5");
        let events = sink.0.lock().unwrap();
        assert_eq!(events.len(), 1);
        assert_eq!(events[0]["type"], "failed");
        assert_eq!(events[0]["data"]["turns_completed"], 8);
        assert_eq!(events[0]["data"]["failure"], doc["failure"]);
        assert_eq!(events[0]["data"]["models_served"], doc["models_served"]);
    }

    #[test]
    fn an_account_refusal_is_a_failed_document_with_its_own_status() {
        let outcome = AssistantOutcome {
            status: "auth_required",
            summary: super::super::agent_loop::AUTH_REQUIRED_SIGNED_OUT_MESSAGE.into(),
            turns: 1,
            turns_completed: 0,
            tools_called: vec![],
            tool_receipts: vec![],
            prior_receipts: 0,
            models_served: vec![],
            model_used: String::new(),
            auth_required: Some(AuthRequiredReason::SignedOut),
            failure_cause: None,
        };
        let sink = Arc::new(Captured::default());
        let doc = JsonEmitter::new(posture(), sink.clone()).finish(&outcome, None);

        // The document is still `car.do/1`: every field here is additive, and
        // `error` is still a string, so nothing pinned to the schema breaks.
        assert_eq!(doc["schema"], "car.do/1");
        assert_eq!(doc["status"], "auth_required");
        assert_eq!(doc["error"], "AuthRequired");
        assert_eq!(doc["reason"], "signed_out");
        assert!(
            doc["error"].is_string(),
            "`error` must stay a string: {doc}"
        );
        // Not an answer, so the failure shape's guard still applies.
        assert!(doc.get("summary").is_none(), "summary leaked: {doc}");
        assert_eq!(
            doc["message"],
            json!(super::super::agent_loop::AUTH_REQUIRED_SIGNED_OUT_MESSAGE)
        );

        let events = sink.0.lock().unwrap();
        assert_eq!(events.len(), 1, "exactly one terminal event");
        assert_eq!(events[0]["type"], "failed");
        assert_eq!(events[0]["data"]["error"], "AuthRequired");
    }

    /// The suggestion has to fit the reason, and it must not be a shell
    /// command: this document is read by programs and by hosts with their own
    /// sign-in entry, and a `no_workspace` account is not repaired by signing
    /// in — the person already did.
    #[test]
    fn the_json_suggestions_are_reason_specific_and_command_free() {
        for (reason, expected) in [
            (
                AuthRequiredReason::SignedOut,
                "Sign in to your Parslee account, then re-run.",
            ),
            (
                AuthRequiredReason::Expired,
                "Your Parslee sign-in has expired. Sign in again, then re-run.",
            ),
            (
                AuthRequiredReason::NoWorkspace,
                "Finish setting up your Parslee account at parslee.ai, then re-run.",
            ),
        ] {
            let outcome = AssistantOutcome {
                status: "auth_required",
                summary: reason.remedy().to_string(),
                turns: 1,
                turns_completed: 0,
                tools_called: vec![],
                tool_receipts: vec![],
                prior_receipts: 0,
                models_served: vec![],
                model_used: String::new(),
                auth_required: Some(reason),
                failure_cause: None,
            };
            let doc =
                JsonEmitter::new(posture(), Arc::new(Captured::default())).finish(&outcome, None);
            assert_eq!(doc["reason"], reason.as_str());
            assert_eq!(
                doc["suggestions"][0],
                expected,
                "wrong first suggestion for {}",
                reason.as_str()
            );
            // `auth login` rather than the full command, so a reworded
            // prefix is caught too — and so a `git grep` for the command
            // across this crate's `--json` surface stays genuinely empty
            // rather than finding this assertion.
            assert!(
                !doc.to_string().contains("auth login"),
                "no shell command belongs in a document read by programs and \
                 by hosts with their own sign-in entry: {doc}"
            );
        }
    }

    /// `on_assistant_event` must stay silent for the refusal, for the same
    /// reason it is silent for `Done`/`Error`: exactly one terminal event, and
    /// `finish` owns it so the event and the document cannot disagree.
    #[test]
    fn the_auth_required_event_does_not_emit_a_second_terminal() {
        let sink = Arc::new(Captured::default());
        let emitter = JsonEmitter::new(posture(), sink.clone());
        emitter.on_assistant_event(&AssistantEvent::AuthRequired {
            reason: AuthRequiredReason::NoWorkspace,
            message: super::super::agent_loop::AUTH_REQUIRED_NO_WORKSPACE_MESSAGE.into(),
        });
        assert!(sink.0.lock().unwrap().is_empty());
    }

    /// Keeps every event the emitter produces, so a test can read the stream
    /// without a subprocess.
    #[derive(Default)]
    struct Captured(std::sync::Mutex<Vec<Value>>);

    impl EventSink for Captured {
        fn emit(&self, event: Value) {
            self.0.lock().unwrap().push(event);
        }
    }

    /// The split that lets one envelope serve two destinations: events go to
    /// the sink, and the terminal document is RETURNED rather than printed.
    ///
    /// Before car#972 §6 this wrote straight to stderr and stdout, so the only
    /// way to assert the document was to mirror it in the test — which is
    /// exactly what `error_doc_for` above still has to do for the failure
    /// shape, and why that mirror is a liability rather than a pattern to copy.
    #[test]
    fn events_go_to_the_sink_and_the_document_comes_back() {
        let sink = Arc::new(Captured::default());
        let emitter = JsonEmitter::new(posture(), sink.clone());
        emitter.started("do the thing", "claude-opus-5");
        emitter.on_assistant_event(&AssistantEvent::ModelServed {
            model_id: "mlx/qwen3-4b:4bit".into(),
            local_last_resort: true,
        });
        emitter.on_assistant_event(&AssistantEvent::ToolCall {
            call_id: "test-call-1".into(),
            sequence: 1,
            name: "shell".into(),
            params: json!({ "command": "ls" }),
        });

        let doc = emitter.finish(
            &AssistantOutcome {
                status: "success",
                summary: "did the thing".into(),
                turns: 2,
                turns_completed: 2,
                tools_called: vec!["shell".into()],
                prior_receipts: 0,
                tool_receipts: vec![AssistantToolReceipt {
                    tool: "shell".into(),
                    call_id: None,
                    sequence: None,
                    ok: true,
                    params: json!({ "command": "ls" }),
                    result: None,
                    via: None,
                }],
                models_served: vec![
                    AssistantModelAttribution {
                        model_id: "openai/gpt-5".into(),
                        local_last_resort: false,
                    },
                    AssistantModelAttribution {
                        model_id: "mlx/qwen3-4b:4bit".into(),
                        local_last_resort: true,
                    },
                ],
                model_used: "mlx/qwen3-4b:4bit".into(),
                auth_required: None,
                failure_cause: None,
            },
            None,
        );

        assert_eq!(doc["schema"], SCHEMA);
        assert_eq!(doc["status"], "success");
        assert_eq!(doc["summary"], "did the thing");
        assert_eq!(doc["receipts"]["total"], 1);
        assert_eq!(doc["sandbox"]["mode"], "docker");
        assert_eq!(doc["model_used"], "mlx/qwen3-4b:4bit");
        assert_eq!(doc["models_served"][0]["model_id"], "openai/gpt-5");
        assert_eq!(doc["models_served"][1]["model_id"], "mlx/qwen3-4b:4bit");
        assert_eq!(doc["models_served"][1]["local_last_resort"], true);

        let events = sink.0.lock().unwrap().clone();
        let types: Vec<&str> = events
            .iter()
            .map(|e| e["type"].as_str().unwrap_or_default())
            .collect();
        // Exactly one terminal event, written alongside the document so the two
        // can never disagree about how the run ended.
        assert_eq!(
            types,
            vec!["started", "model_served", "tool_called", "completed"]
        );
        assert_eq!(events[1]["data"]["model_id"], "mlx/qwen3-4b:4bit");
        assert_eq!(events[1]["data"]["local_last_resort"], true);
        assert_eq!(events[2]["data"]["tool"], "shell");
        assert_eq!(events[2]["data"]["brief"], "ls");
        assert_eq!(
            events[3]["data"]["models_served"], doc["models_served"],
            "the terminal event and stdout run receipt must agree"
        );
    }

    #[test]
    fn goal_report_carries_grounded_separately_from_passed() {
        // A judge-completed run: passed, but not grounded. Callers must be able
        // to tell it apart from a deterministic pass.
        let g = GoalReport {
            check: "cargo test -q".into(),
            passed: true,
            grounded: false,
            iterations: 3,
            halt: None,
        }
        .to_json();
        assert_eq!(g["passed"], true);
        assert_eq!(g["grounded"], false);
    }

    /// A `delegate` call is one `tool_called` event whose brief is the GOAL
    /// (the bug class the `IDENTIFYING` comment records: a key missing there
    /// renders an empty brief), and the document counts delegations issued
    /// beside the parent's turns.
    #[test]
    fn delegate_calls_are_briefed_by_goal_and_counted() {
        let sink = Arc::new(Captured::default());
        let emitter = JsonEmitter::new(posture(), sink.clone());
        emitter.on_assistant_event(&AssistantEvent::ToolCall {
            call_id: "test-call-2".into(),
            sequence: 2,
            name: super::super::agent_loop::DELEGATE_TOOL.into(),
            params: json!({ "goal": "survey the repo layout", "tools": ["read_file"] }),
        });
        emitter.on_assistant_event(&AssistantEvent::ToolCall {
            call_id: "test-call-3".into(),
            sequence: 3,
            name: super::super::agent_loop::DELEGATE_TOOL.into(),
            params: json!({ "goal": "count the tests" }),
        });
        emitter.on_assistant_event(&AssistantEvent::ToolCall {
            call_id: "test-call-4".into(),
            sequence: 4,
            name: "shell".into(),
            params: json!({ "command": "ls" }),
        });
        let events = sink.0.lock().unwrap();
        let first = events
            .iter()
            .find(|e| e["type"] == "tool_called")
            .expect("a tool_called event");
        assert_eq!(first["data"]["brief"], "survey the repo layout");
        drop(events);

        let doc = emitter.finish(
            &AssistantOutcome {
                status: "success",
                summary: "done".into(),
                turns: 3,
                turns_completed: 3,
                tools_called: vec![],
                tool_receipts: vec![],
                prior_receipts: 0,
                models_served: vec![],
                model_used: "m".into(),
                auth_required: None,
                failure_cause: None,
            },
            None,
        );
        assert_eq!(doc["turns"], 3);
        assert_eq!(doc["delegations"], 2);
    }
}