plana-text 0.4.2

mod parsing;
mod rendering;

use serde::{Deserialize, Serialize};
use serde_json::Value;
use uuid::Uuid;

#[derive(Default, Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum StreamChunkKind {
    #[default]
    Text,
    Thinking,
    DeepThinking,
    /// A prompt-change NOTICE chunk (user direction 2026-10-22): the
    /// live-stream twin of `StreamSegment::PromptNotice`'s provenance row.
    /// Carries the notice's SOURCE as the chunk text — never model prose,
    /// never thinking. Unlike the segment-level arm (which is
    /// caller-recorded into run records only), this kind exists precisely
    /// to stream live.
    Notice,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum StreamToolEvent {
    ToolCall {
        tool_name: String,
        call_id: Uuid,
        params_summary: Option<String>,
        agent_type: Option<String>,
    },
    ToolResult {
        tool_name: String,
        call_id: Uuid,
        result: String,
        success: bool,
        duration_ms: Option<u64>,
        agent_type: Option<String>,
    },
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct InputMessage {
    /// LLM wire role: `system` / `user` / … — one row per composed layer.
    pub role: String,
    /// The layer's verbatim content (never truncated here; the run
    /// record's stream budget governs total size).
    pub content: String,
    /// The layer's SOURCE label — which composed section this row is
    /// ("env context", "core system doc", "actor", "soul", "tools", the
    /// skill's own section name, …), so the panel can render provenance
    /// instead of a bare "system" row (user direction 2026-10-22).
    ///
    /// `None` on legacy rows and on layers whose caller has no name:
    /// consumers must render such rows without a source suffix — never
    /// fabricate one.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub label: Option<String>,
}

impl InputMessage {
    /// A layer with no source label (the pre-label shape).
    pub fn new(role: impl Into<String>, content: impl Into<String>) -> Self {
        Self {
            role: role.into(),
            content: content.into(),
            label: None,
        }
    }

    /// Attach the layer's source label. A blank label is dropped (no
    /// provenance is better than an empty suffix), matching the notice's
    /// blank-source contract.
    pub fn with_label(mut self, label: impl Into<String>) -> Self {
        let label = label.into();
        let trimmed = label.trim();
        self.label = (!trimmed.is_empty()).then(|| trimmed.to_string());
        self
    }
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub enum StreamSegment {
    /// The composed INPUT side of an LLM call (user direction
    /// 2026-10-10): every other segment records the model's OUTPUT
    /// (thinking / text / tool traffic), so a run record without an
    /// Input segment reads as if the run had consumed no prompt
    /// composition at all. Recorded by the CALLER (scepter's pipeline,
    /// where the composed messages live) — never streamed live, hence
    /// never produced by `LlmStreamBuilder`.
    Input {
        messages: Vec<InputMessage>,
        #[serde(default)]
        message_id: Option<Uuid>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        ts: Option<u64>,
    },
    /// A mid-run PROMPT NOTICE (user direction 2026-10-22): the composed
    /// system stack changed between two LLM invocations of the same run,
    /// so the panel draws ONE provenance line at the point of change
    /// instead of letting the prompt silently drift. `source` names what
    /// injected/changed (the changed layer labels, or the emitting
    /// subsystem); `detail` optionally carries a human sentence.
    ///
    /// Recorded by the CALLER (scepter's pipeline — the only place that
    /// knows a recomposition happened); never streamed live by
    /// `LlmStreamBuilder`, hence never produced from an LLM chunk.
    PromptNotice {
        source: String,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        detail: Option<String>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        ts: Option<u64>,
    },
    Text {
        text: String,
        #[serde(default)]
        message_id: Option<Uuid>,
        /// Wall-clock ms epoch at the segment's LAST activity (creation
        /// or latest append) — stamped by `LlmStreamBuilder` at push
        /// time. Absent on legacy rows (serde default) and on segments
        /// built by hand; consumers must treat `None` as "no clock"
        /// exactly like a pre-field wire row.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        ts: Option<u64>,
    },
    Thinking {
        text: String,
        #[serde(default)]
        message_id: Option<Uuid>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        ts: Option<u64>,
    },
    DeepThinking {
        text: String,
        #[serde(default)]
        message_id: Option<Uuid>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        ts: Option<u64>,
    },
    ToolCall {
        tool_name: String,
        call_id: Uuid,
        params: Value,
        #[serde(default)]
        agent_type: Option<String>,
        #[serde(default)]
        message_id: Option<Uuid>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        ts: Option<u64>,
    },
    ToolResult {
        tool_name: String,
        call_id: Uuid,
        success: bool,
        data: Value,
        #[serde(default)]
        duration_ms: Option<u64>,
        #[serde(default)]
        agent_type: Option<String>,
        #[serde(default)]
        message_id: Option<Uuid>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        ts: Option<u64>,
    },
}

impl StreamSegment {
    pub fn text(&self) -> &str {
        match self {
            Self::Text { text, .. }
            | Self::Thinking { text, .. }
            | Self::DeepThinking { text, .. } => text,
            // Input / PromptNotice are machine-side metadata, never
            // message text: they must not bleed into raw_text()/Display.
            Self::Input { .. } | Self::PromptNotice { .. } => "",
            Self::ToolCall { params, .. } => match params {
                Value::String(s) => s,
                _ => "",
            },
            Self::ToolResult { data, .. } => match data {
                Value::String(s) => s,
                Value::Null => "",
                _ => "",
            },
        }
    }

    pub fn text_or_json(&self) -> String {
        match self {
            Self::Text { text, .. }
            | Self::Thinking { text, .. }
            | Self::DeepThinking { text, .. } => text.clone(),
            Self::Input { .. } | Self::PromptNotice { .. } => String::new(),
            Self::ToolCall { params, .. } | Self::ToolResult { data: params, .. } => match params {
                Value::String(s) => s.clone(),
                Value::Null => String::new(),
                other => other.to_string(),
            },
        }
    }

    pub fn is_tool(&self) -> bool {
        matches!(self, Self::ToolCall { .. } | Self::ToolResult { .. })
    }

    /// A mid-run prompt-change notice (see [`Self::PromptNotice`]).
    pub fn is_prompt_notice(&self) -> bool {
        matches!(self, Self::PromptNotice { .. })
    }

    /// The segment's wall-clock stamp, when the builder recorded one.
    /// `None` = legacy/hand-built segment — consumers must fall back to
    /// their own sequencing, never to a fake clock.
    pub fn wall_ts(&self) -> Option<u64> {
        match self {
            Self::Input { ts, .. }
            | Self::Text { ts, .. }
            | Self::Thinking { ts, .. }
            | Self::DeepThinking { ts, .. }
            | Self::ToolCall { ts, .. }
            | Self::ToolResult { ts, .. }
            | Self::PromptNotice { ts, .. } => *ts,
        }
    }

    pub fn is_thinking(&self) -> bool {
        matches!(self, Self::Thinking { .. } | Self::DeepThinking { .. })
    }

    pub fn is_text(&self) -> bool {
        matches!(self, Self::Text { .. })
    }

    pub fn chunk_kind(&self) -> Option<StreamChunkKind> {
        match self {
            Self::Text { .. } => Some(StreamChunkKind::Text),
            Self::Thinking { .. } => Some(StreamChunkKind::Thinking),
            Self::DeepThinking { .. } => Some(StreamChunkKind::DeepThinking),
            // PromptNotice reports NO chunk kind: `text_segments()` filters
            // on this, and a (Notice, "") pair there would be noise for
            // every kind-matching consumer. The live-stream emission maps
            // the segment to StreamChunkKind::Notice itself.
            _ => None,
        }
    }

    pub fn tool_name(&self) -> Option<&str> {
        match self {
            Self::ToolCall { tool_name, .. } => Some(tool_name),
            Self::ToolResult { tool_name, .. } => Some(tool_name),
            _ => None,
        }
    }

    pub fn call_id(&self) -> Option<Uuid> {
        match self {
            Self::ToolCall { call_id, .. } => Some(*call_id),
            Self::ToolResult { call_id, .. } => Some(*call_id),
            _ => None,
        }
    }

    pub fn agent_type(&self) -> Option<&str> {
        match self {
            Self::ToolCall { agent_type, .. } => agent_type.as_deref(),
            Self::ToolResult { agent_type, .. } => agent_type.as_deref(),
            _ => None,
        }
    }

    pub fn tool_params(&self) -> Option<&Value> {
        match self {
            Self::ToolCall { params, .. } => Some(params),
            _ => None,
        }
    }

    pub fn tool_result_data(&self) -> Option<&Value> {
        match self {
            Self::ToolResult { data, .. } => Some(data),
            _ => None,
        }
    }

    pub fn tool_result_success(&self) -> Option<bool> {
        match self {
            Self::ToolResult { success, .. } => Some(*success),
            _ => None,
        }
    }
}

pub use parsing::LlmStreamBuilder;
pub use rendering::LlmStream;

#[cfg(test)]
mod tests {
    use super::*;
    use anyhow::{Context, Result};

    #[derive(serde::Serialize)]
    struct PathParam<'a> {
        path: &'a str,
    }

    fn empty() -> serde_json::Value {
        serde_json::Value::Object(Default::default())
    }

    /// The Input segment (user direction 2026-10-10): serde round-trips
    /// its message layers, `prepend_input` leads the stream, and the
    /// legacy helpers treat it as payload-bearing but output-free.
    #[test]
    fn test_input_segment_round_trip_and_prepend() -> Result<()> {
        let mut stream = LlmStreamBuilder::new();
        stream.push_chunk("answer", StreamChunkKind::Text);
        let mut stream = stream.seal_coalesced();

        stream.prepend_input(vec![
            InputMessage::new("system", "standing law + soul + skill").with_label("soul + skill"),
            InputMessage::new("user", "ingest the workspace knowledge"),
        ]);

        // Leads the stream — the prompts precede the output they shaped.
        assert!(matches!(
            stream.segments().first(),
            Some(StreamSegment::Input { .. })
        ));
        // Output-side helpers stay Output-side: no text bleeds in, but
        // the payload counts toward size and emptiness.
        assert_eq!(stream.raw_text(), "answer");
        assert!(!stream.is_empty());

        // Wire round-trip keeps the layers (the run record column holds
        // this shape; the webui adapter switches on the tag).
        let value = serde_json::to_value(&stream)?;
        let back: LlmStream = serde_json::from_value(value)?;
        match back.segments().first() {
            Some(StreamSegment::Input { messages, .. }) => {
                assert_eq!(messages.len(), 2);
                assert_eq!(messages[0].role, "system");
                assert_eq!(messages[1].content, "ingest the workspace knowledge");
            }
            other => panic!("expected leading Input segment, got {other:?}"),
        }
        Ok(())
    }

    /// Prepending nothing is a no-op — callers without a composed input
    /// (nudges, tests) keep their stream untouched.
    #[test]
    fn test_prepend_input_empty_is_noop() -> Result<()> {
        let mut stream = LlmStream::from_plain_text("solo");
        stream.prepend_input(Vec::new());
        assert_eq!(stream.segments().len(), 1);
        Ok(())
    }

    /// An Input-ONLY stream is not empty — the composition alone is
    /// payload (R1 gap: a stream that also holds Text masked this arm).
    #[test]
    fn test_input_only_stream_is_not_empty() {
        let mut stream = LlmStream::empty();
        stream.prepend_input(vec![InputMessage::new("system", "law")]);
        assert!(!stream.is_empty());
    }

    /// The layer SOURCE label round-trips, and a legacy row (written
    /// before the label field existed) deserializes to `None` — the wire
    /// stays backward compatible, so an old run record never fabricates a
    /// provenance it does not have.
    #[test]
    fn test_input_message_label_round_trips_and_legacy_rows_stay_labelless() -> Result<()> {
        let labelled =
            serde_json::to_value(InputMessage::new("system", "soul").with_label("soul"))?;
        assert_eq!(labelled["label"], serde_json::json!("soul"));
        let back: InputMessage = serde_json::from_value(labelled)?;
        assert_eq!(back.label.as_deref(), Some("soul"));

        // Legacy row: `{role, content}` only — no `label` key on the wire.
        let legacy: InputMessage =
            serde_json::from_value(serde_json::json!({ "role": "system", "content": "old" }))?;
        assert_eq!(legacy.label, None);
        // An unlabelled row serializes WITHOUT the key (skip_serializing_if).
        let unlabelled = serde_json::to_value(InputMessage::new("user", "turn"))?;
        assert!(unlabelled.get("label").is_none());
        Ok(())
    }

    /// The mid-run prompt notice (user direction 2026-10-22): appended at
    /// the TAIL (the change happened here, not at the run's head), kept
    /// out of every text surface, and non-empty on its own.
    #[test]
    fn test_push_notice_appends_tail_and_stays_out_of_text() -> Result<()> {
        let mut stream = LlmStream::empty();
        stream.push_notice(
            "soul, lessons",
            Some("system prompt recomposed"),
            Some(1_790_000_000_000),
        );
        assert!(
            stream
                .segments()
                .last()
                .is_some_and(|s| s.is_prompt_notice())
        );
        // A notice is an event, not message text: every text surface stays
        // empty so no consumer mistakes it for output.
        assert_eq!(stream.raw_text(), "");
        assert_eq!(stream.display_text(), "");
        assert_eq!(stream.thinking_text(), "");
        assert!(!stream.is_empty());

        // Tail position with existing output ahead of it.
        let mut stream = LlmStream::from_plain_text("answer");
        stream.push_notice("tools", None, None);
        let segs = stream.segments();
        assert_eq!(segs.len(), 2);
        assert!(matches!(segs[0], StreamSegment::Text { .. }));
        assert!(matches!(segs[1], StreamSegment::PromptNotice { .. }));
        assert_eq!(segs[1].wall_ts(), None);
        // The plain-text surface must stay empty too: plana_timeline feeds
        // `seg.text()` into the body block list, so a leak here would
        // render the provenance row as assistant prose (R1 M2b).
        assert_eq!(segs[1].text(), "");
        match &segs[1] {
            StreamSegment::PromptNotice { source, detail, .. } => {
                assert_eq!(source, "tools");
                assert_eq!(*detail, None);
            }
            other => panic!("expected PromptNotice, got {other:?}"),
        }
        Ok(())
    }

    /// The notice's position contract: a notice that describes THIS
    /// stream's prompt rides directly after the leading Input segment (the
    /// output under it is what that prompt produced); without an Input it
    /// leads the stream. A tail notice (the other slot) describes a change
    /// observed after the output it did not shape.
    #[test]
    fn test_push_notice_after_input_rides_the_prompt_boundary() -> Result<()> {
        // With a leading Input: [Input, Notice, output…].
        let mut stream = LlmStream::empty();
        stream.prepend_input(vec![InputMessage::new("system", "law")]);
        stream.push_notice_after_input("soul", None, Some(1));
        let segs = stream.segments();
        assert_eq!(segs.len(), 2);
        assert!(matches!(segs[0], StreamSegment::Input { .. }));
        assert!(matches!(segs[1], StreamSegment::PromptNotice { .. }));
        // The same call on a stream that ALREADY has output still places
        // the notice at the prompt boundary, not after the output.
        let mut stream = LlmStream::from_plain_text("answer");
        stream.prepend_input(vec![InputMessage::new("system", "law")]);
        stream.push_notice_after_input("tools", None, None);
        let segs = stream.segments();
        assert_eq!(segs.len(), 3);
        assert!(matches!(segs[0], StreamSegment::Input { .. }));
        assert!(matches!(segs[1], StreamSegment::PromptNotice { .. }));
        assert!(matches!(segs[2], StreamSegment::Text { .. }));

        // No leading Input: the notice leads the stream.
        let mut stream = LlmStream::from_plain_text("answer");
        stream.push_notice_after_input("env context", None, None);
        let segs = stream.segments();
        assert_eq!(segs.len(), 2);
        assert!(matches!(segs[0], StreamSegment::PromptNotice { .. }));
        assert!(matches!(segs[1], StreamSegment::Text { .. }));

        // Same blank-source contract as the tail variant.
        let mut stream = LlmStream::from_plain_text("answer");
        assert!(!stream.push_notice_after_input("   ", None, None));
        assert_eq!(stream.segments().len(), 1);
        Ok(())
    }

    /// A blank detail is normalized away (an empty second line is noise),
    /// and a blank source label never produces an empty provenance suffix.
    #[test]
    fn test_notice_detail_and_label_blanks_are_normalized() -> Result<()> {
        let mut stream = LlmStream::from_plain_text("answer");
        stream.push_notice("tools", Some("   "), None);
        match stream.segments().last() {
            Some(StreamSegment::PromptNotice { detail, .. }) => assert_eq!(*detail, None),
            other => panic!("expected PromptNotice, got {other:?}"),
        }
        let labelled = InputMessage::new("system", "x").with_label("  ");
        assert_eq!(labelled.label, None, "a blank label must not survive");
        let labelled = InputMessage::new("system", "x").with_label("  soul  ");
        assert_eq!(labelled.label.as_deref(), Some("soul"));
        Ok(())
    }

    /// An anonymous notice is a call-site bug: the row exists ONLY to name
    /// the origin, so a blank source is dropped rather than recorded.
    #[test]
    fn test_push_notice_rejects_blank_source() {
        let mut stream = LlmStream::from_plain_text("answer");
        assert!(!stream.push_notice("   ", Some("detail"), None));
        assert_eq!(stream.segments().len(), 1);
        assert!(!stream.push_notice("", None, None));
        assert_eq!(stream.segments().len(), 1);
    }

    /// The BUILDER's live notice (user direction 2026-10-22): pushed the
    /// moment a mid-run prompt change is observed, sealed with the stream,
    /// and never visible on any text surface. A blank source is refused.
    #[test]
    fn test_builder_push_notice_streams_and_seals() -> Result<()> {
        let mut builder = LlmStreamBuilder::new();
        builder.push_chunk("answer", StreamChunkKind::Text);
        assert!(builder.push_notice("soul, tools", Some("recomposed")));
        assert!(
            !builder.push_notice("  ", None),
            "blank sources are refused"
        );

        let stream = builder.seal();
        let segs = stream.segments();
        // Order: the notice trails the text it follows (append semantics).
        assert!(matches!(segs[0], StreamSegment::Text { .. }));
        match &segs[1] {
            StreamSegment::PromptNotice { source, detail, ts } => {
                assert_eq!(source, "soul, tools");
                assert_eq!(detail.as_deref(), Some("recomposed"));
                assert!(ts.is_some(), "the live notice carries its wall clock");
            }
            other => panic!("expected PromptNotice, got {other:?}"),
        }
        // No text surface sees it, and it never reports a chunk kind.
        assert_eq!(stream.raw_text(), "answer");
        assert_eq!(segs[1].chunk_kind(), None);
        assert!(segs[1].is_prompt_notice());
        Ok(())
    }

    /// A Notice KIND handed to push_chunk is a warned no-op: the source
    /// must not smear into the prose segments.
    #[test]
    fn test_push_chunk_rejects_the_notice_kind() -> Result<()> {
        let mut builder = LlmStreamBuilder::new();
        builder.push_chunk("prose", StreamChunkKind::Text);
        builder.push_chunk("soul, tools", StreamChunkKind::Notice);
        let stream = builder.seal();
        assert_eq!(
            stream.segments().len(),
            1,
            "the notice-kind chunk is dropped"
        );
        assert_eq!(stream.raw_text(), "prose");
        Ok(())
    }

    /// The notice survives the wire round-trip (the run record column
    /// holds this shape and the webui adapter switches on the tag).
    #[test]
    fn test_prompt_notice_wire_round_trip() -> Result<()> {
        let mut stream = LlmStream::from_plain_text("answer");
        stream.push_notice(
            "env context",
            Some("language changed"),
            Some(1_790_000_000_123),
        );
        let value = serde_json::to_value(&stream)?;
        let back: LlmStream = serde_json::from_value(value)?;
        let last = back.segments().last().expect("notice kept");
        assert!(last.is_prompt_notice());
        assert_eq!(last.wall_ts(), Some(1_790_000_000_123));
        match last {
            StreamSegment::PromptNotice { source, detail, .. } => {
                assert_eq!(source, "env context");
                assert_eq!(detail.as_deref(), Some("language changed"));
            }
            other => panic!("expected PromptNotice, got {other:?}"),
        }
        // Pin the exact wire tag the webui adapter switches on.
        let raw = serde_json::to_string(&stream)?;
        assert!(raw.contains("\"PromptNotice\""), "wire tag drifted: {raw}");
        Ok(())
    }

    #[test]
    fn test_builder_text_segments() -> Result<()> {
        let mut builder = LlmStreamBuilder::new();
        builder.push_chunk("hello ", StreamChunkKind::Text);
        builder.push_chunk("world", StreamChunkKind::Text);
        assert_eq!(builder.as_str(), "hello world");
        assert_eq!(builder.segments().len(), 1);

        let stream = builder.seal_coalesced();
        assert_eq!(stream.segments().len(), 1);
        assert_eq!(stream.raw_text(), "hello world");
        Ok(())
    }

    #[test]
    fn test_builder_mixed_segments() -> Result<()> {
        let mut builder = LlmStreamBuilder::new();
        builder.push_chunk("thinking...", StreamChunkKind::Thinking);
        let call_id = Uuid::now_v7();
        builder.push_tool_call(
            "read_file".to_string(),
            call_id,
            Some(serde_json::to_value(PathParam { path: "/test" }).unwrap_or_default()),
            None,
        );
        builder.push_tool_result(
            "read_file".to_string(),
            call_id,
            Value::String("file contents".to_string()),
            true,
            Some(150),
            None,
        );
        builder.push_chunk("Here is the file:", StreamChunkKind::Text);

        assert_eq!(builder.segments().len(), 4);
        assert_eq!(builder.as_str(), "Here is the file:");

        let stream = builder.seal();
        let tools: Vec<_> = stream.tool_segments().collect();
        assert_eq!(tools.len(), 2);
        assert!(matches!(tools[0], StreamSegment::ToolCall { .. }));
        assert!(matches!(tools[1], StreamSegment::ToolResult { .. }));
        assert_eq!(tools[0].text(), "");
        assert_eq!(tools[1].text(), "file contents");
        assert_eq!(tools[0].call_id(), Some(call_id));
        assert_eq!(tools[1].call_id(), Some(call_id));
        Ok(())
    }

    #[test]
    fn test_llm_stream_from_plain_text() -> Result<()> {
        let stream = LlmStream::from_plain_text("test content");
        assert_eq!(stream.raw_text(), "test content");
        assert_eq!(stream.segments().len(), 1);
        Ok(())
    }

    #[test]
    fn test_display_text_vs_thinking() -> Result<()> {
        let mut builder = LlmStreamBuilder::new();
        builder.push_chunk("let me think", StreamChunkKind::Thinking);
        builder.push_chunk("Here is my answer", StreamChunkKind::Text);

        let stream = builder.seal();
        assert_eq!(stream.thinking_text(), "let me think");
        assert_eq!(stream.display_text(), "Here is my answer");
        Ok(())
    }

    #[test]
    fn test_every_segment_has_text() -> Result<()> {
        let mut builder = LlmStreamBuilder::new();
        builder.push_chunk("report intro", StreamChunkKind::Text);
        let call_id = Uuid::now_v7();
        builder.push_tool_call("report".to_string(), call_id, Some(empty()), None);
        builder.push_tool_result(
            "report".to_string(),
            call_id,
            Value::String("This is the actual report content".to_string()),
            true,
            None,
            None,
        );
        builder.push_chunk("conclusion", StreamChunkKind::Text);

        let stream = builder.seal();
        let full_text = stream.raw_text();
        assert!(full_text.contains("report intro"));
        assert!(full_text.contains("This is the actual report content"));
        assert!(full_text.contains("conclusion"));
        Ok(())
    }

    #[test]
    fn test_strip_tool_call_text_removes_tool_descriptions() -> Result<()> {
        let mut builder = LlmStreamBuilder::new();
        builder.push_chunk(
            "Some intro text\nreport({\"key\": \"value\"})\nMore text",
            StreamChunkKind::Text,
        );
        let call_id = Uuid::now_v7();
        builder.push_tool_call(
            "report".to_string(),
            call_id,
            Some(serde_json::json!({"key": "value"})),
            None,
        );
        builder.push_tool_result(
            "report".to_string(),
            call_id,
            Value::String("result".to_string()),
            true,
            None,
            None,
        );
        builder.push_chunk(
            "report.content(\"final output\")\nEnding",
            StreamChunkKind::Text,
        );

        let stream = builder.seal_coalesced();
        let display = stream.display_text();
        assert!(display.contains("Some intro text"), "should keep intro");
        assert!(display.contains("More text"), "should keep middle text");
        assert!(display.contains("Ending"), "should keep ending");
        assert!(!display.contains("report("), "should strip report(...)");
        assert!(
            !display.contains("report.content"),
            "should strip report.content(...)"
        );
        Ok(())
    }

    #[test]
    fn test_strip_tool_call_text_preserves_normal_text() -> Result<()> {
        let mut builder = LlmStreamBuilder::new();
        builder.push_chunk(
            "Hello world\nThis is a report about things\nGoodbye",
            StreamChunkKind::Text,
        );

        let stream = builder.seal_coalesced();
        let display = stream.display_text();
        assert!(display.contains("Hello world"));
        assert!(display.contains("This is a report about things"));
        assert!(display.contains("Goodbye"));
        Ok(())
    }

    #[test]
    fn test_strip_tool_call_text_no_tool_calls_fallback() -> Result<()> {
        let mut builder = LlmStreamBuilder::new();
        builder.push_chunk(
            "exec({\"code\": \"import { report } from 'hubris'; report()\"})",
            StreamChunkKind::Text,
        );

        let stream = builder.seal_coalesced();
        assert!(
            !stream.display_text().contains("exec("),
            "fallback should strip exec(...) even without tool call segments"
        );
        Ok(())
    }

    #[test]
    fn test_strip_tool_call_text_whitespace_and_json_variants() -> Result<()> {
        let mut builder = LlmStreamBuilder::new();
        builder.push_chunk(
            "  exec({ \"code\": \"import { report } from 'hubris'; report({text: 'hello'})\" })\n\
             exec({\n  \"code\": \"const x = 1\"\n})\n\
             Some normal text here\n\
             Also normal",
            StreamChunkKind::Text,
        );

        let stream = builder.seal_coalesced();
        let display = stream.display_text();
        assert!(!display.contains("exec("), "should strip exec({{...)");
        assert!(
            display.contains("Some normal text here"),
            "should preserve normal text"
        );
        assert!(
            display.contains("Also normal"),
            "should preserve normal text"
        );
        assert!(
            display.contains("Also normal"),
            "should preserve trailing normal text"
        );
        Ok(())
    }

    #[test]
    fn test_seal_coalesced_closes_pending_tool_calls() -> Result<()> {
        let mut builder = LlmStreamBuilder::new();
        let call_id_1 = Uuid::now_v7();
        let call_id_2 = Uuid::now_v7();

        builder.push_tool_call(
            "read_file".to_string(),
            call_id_1,
            Some(serde_json::to_value(PathParam { path: "/a" }).unwrap_or_default()),
            None,
        );
        builder.push_tool_result(
            "read_file".to_string(),
            call_id_1,
            Value::String("content A".to_string()),
            true,
            None,
            None,
        );
        builder.push_tool_call(
            "write_file".to_string(),
            call_id_2,
            Some(serde_json::to_value(PathParam { path: "/b" }).unwrap_or_default()),
            None,
        );

        assert!(builder.has_pending_tool_calls());

        let stream = builder.seal_coalesced();

        let tools: Vec<_> = stream.tool_segments().collect();
        assert_eq!(
            tools.len(),
            4,
            "should have 2 calls + 1 real result + 1 synthetic result"
        );

        let synthetic = tools.iter().find(|s| {
            matches!(s, StreamSegment::ToolResult { call_id, success, .. } if *call_id == call_id_2 && !success)
        });
        assert!(
            synthetic.is_some(),
            "should have synthetic result for pending call_id_2 with success=false"
        );
        let synth_data = synthetic
            .context("missing synthetic result")?
            .tool_result_data()
            .context("test precondition")?;
        assert_eq!(
            synth_data,
            &Value::String("[stream sealed before result arrived]".to_string())
        );
        Ok(())
    }

    #[test]
    fn test_seal_coalesced_no_synthetic_when_all_resolved() -> Result<()> {
        let mut builder = LlmStreamBuilder::new();
        let call_id = Uuid::now_v7();
        builder.push_tool_call("exec".to_string(), call_id, Some(empty()), None);
        builder.push_tool_result(
            "exec".to_string(),
            call_id,
            Value::String("ok".to_string()),
            true,
            None,
            None,
        );

        assert!(!builder.has_pending_tool_calls());

        let stream = builder.seal_coalesced();
        let tools: Vec<_> = stream.tool_segments().collect();
        assert_eq!(
            tools.len(),
            2,
            "should have exactly call + result, no synthetic"
        );
        Ok(())
    }

    #[test]
    fn builder_stamps_a_wall_clock_on_every_push() {
        use super::super::LlmStreamBuilder;
        let mut b = LlmStreamBuilder::new();
        b.push_chunk("hel", StreamChunkKind::Text);
        let first = b.segments()[0].wall_ts().expect("creation stamps");
        // A real stamp: past-2023 epoch (kills a miswired clock returning
        // constants) — 2ms sleep so the refresh check below is strictly
        // deterministic despite millisecond truncation.
        assert!(
            first > 1_700_000_000_000,
            "the stamp must be a real wall clock, got {first}"
        );
        std::thread::sleep(std::time::Duration::from_millis(2));
        b.push_chunk("lo", StreamChunkKind::Text);
        let refreshed = b.segments()[0].wall_ts().expect("append keeps the stamp");
        assert!(refreshed > first, "the stamp rides the LAST activity");
        assert_eq!(b.segments().len(), 1, "appends coalesce");
        let call = Uuid::now_v7();
        b.push_tool_call("exec".to_string(), call, None, None);
        b.push_tool_result(
            "exec".to_string(),
            call,
            serde_json::json!("ok"),
            true,
            Some(3),
            None,
        );
        assert!(b.segments()[1].wall_ts().is_some(), "tool calls stamp");
        assert!(b.segments()[2].wall_ts().is_some(), "tool results stamp");
    }

    #[test]
    fn ts_is_off_the_wire_when_absent_and_a_number_when_present() {
        // Legacy rows (pre-field) deserialize to None and re-serialize
        // WITHOUT the key — the old wire shape is preserved exactly.
        let legacy: StreamSegment =
            serde_json::from_str(r#"{"Text":{"text":"hi"}}"#).expect("legacy wire");
        assert_eq!(legacy.wall_ts(), None);
        let wire = serde_json::to_value(&legacy).expect("serialize");
        assert!(
            wire["Text"].get("ts").is_none(),
            "absent stays off the wire"
        );

        let stamped: StreamSegment =
            serde_json::from_str(r#"{"Text":{"text":"hi","ts":1790000000123}}"#)
                .expect("stamped wire");
        assert_eq!(stamped.wall_ts(), Some(1_790_000_000_123));
        let wire = serde_json::to_value(&stamped).expect("serialize");
        assert_eq!(wire["Text"]["ts"].as_u64(), Some(1_790_000_000_123));
    }

    #[test]
    fn test_text_segment_returns_string_directly() -> Result<()> {
        let seg = StreamSegment::Text {
            text: "hello".to_string(),
            message_id: None,
            ts: None,
        };
        assert_eq!(seg.text(), "hello");
        Ok(())
    }

    #[test]
    fn test_tool_call_params_structured() -> Result<()> {
        let seg = StreamSegment::ToolCall {
            tool_name: "exec".to_string(),
            call_id: Uuid::now_v7(),
            params: serde_json::json!({"code": "import { report } from 'hubris'; report({text: 'hi'})"}),
            agent_type: None,
            message_id: None,
            ts: None,
        };
        let params = seg.tool_params().context("expected tool params")?;
        assert_eq!(
            params["code"],
            "import { report } from 'hubris'; report({text: 'hi'})"
        );
        Ok(())
    }

    #[test]
    fn test_tool_result_data_structured() -> Result<()> {
        let seg = StreamSegment::ToolResult {
            tool_name: "report_human".to_string(),
            call_id: Uuid::now_v7(),
            success: true,
            data: serde_json::json!({"ok": true, "data": "Hello!", "error": null}),
            duration_ms: None,
            agent_type: None,
            message_id: None,
            ts: None,
        };
        let data = seg
            .tool_result_data()
            .context("expected tool result data")?;
        assert_eq!(data["ok"], true);
        assert_eq!(data["data"], "Hello!");
        Ok(())
    }
}