mod.rs

//! Stream accumulation: raw sidecar JSON events → IntermediateMessage snapshots.
//!
//! Responsibilities (split across submodules):
//! - `streaming` — Claude block-level streaming (content_block_*) +
//!   `build_partial_*` snapshot constructors + Claude text extractors.
//! - `codex` — Codex App Server event handling: delta accumulation,
//!   camelCase→snake_case normalization, Claude-format synthesis.
//! - This file — struct definition, public API, top-level `push_event`
//!   dispatch, lifecycle, Claude full-message handlers, and the shared
//!   collection helpers used by both submodules.

mod codex;
mod cursor;
mod kimi;
mod opencode;
mod streaming;

use std::collections::{BTreeMap, HashMap, HashSet};
use std::time::{SystemTime, UNIX_EPOCH};

use serde_json::Value;

fn now_ms() -> f64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs_f64()
        * 1000.0
}

use super::types::{
    AgentUsage, CollectedTurn, IntermediateMessage, MessageRole, ParsedAgentOutput,
};
use streaming::StreamingBlock;

fn is_claude_compact_lifecycle_event(value: &Value) -> bool {
    let subtype = value.get("subtype").and_then(Value::as_str);
    matches!(subtype, Some("compact_boundary"))
        || (subtype == Some("status")
            && value.get("status").and_then(Value::as_str) == Some("compacting"))
}

#[cfg(test)]
mod tests;

// ---------------------------------------------------------------------------
// PushOutcome
// ---------------------------------------------------------------------------

/// Classifies the effect a single `push_event` call had on accumulator
/// state. The pipeline uses this to decide whether the next emit should be
/// a `Full` snapshot, a `Partial` streaming update, or `None`.
///
/// Keeping the classification next to the handlers makes it impossible
/// for a new SDK event type to land without an explicit rendering-tier
/// decision: the dispatch in `push_event` MUST return a variant, which
/// forces the author to think about it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PushOutcome {
    /// A fully-formed message landed in `collected[]` (or one was replaced
    /// in place by `collect_or_replace`). The pipeline must run the full
    /// adapter + collapse pipeline because the change can affect any part
    /// of the rendered output, not just the trailing partial.
    Finalized,
    /// Only the streaming buffer (`blocks` / `fallback_text` /
    /// `fallback_thinking`) changed. The pipeline can build just the
    /// trailing partial message and emit a `Partial`, without re-running
    /// the full thread render.
    StreamingDelta,
    /// Control event with no rendering effect (sidecar framing markers,
    /// SDK turn-lifecycle pings, etc.). The pipeline emits nothing.
    NoOp,
}

// ---------------------------------------------------------------------------
// StreamAccumulator
// ---------------------------------------------------------------------------

/// Unified stream accumulator for both Claude and Codex providers.
///
/// Tracks block-level streaming state for real-time rendering and collects
/// persistence data (turns, usage, model) for the DB layer in `agents.rs`.
///
/// Fields are module-private; the `streaming` and `codex` submodules see
/// them as descendants and mutate state through free functions that take
/// `&mut StreamAccumulator`. External code goes through the methods below.
pub struct StreamAccumulator {
    provider: String,

    // ── Rendering state (replaces TS StreamAccumulator) ──────────────
    /// Finalized full messages ready for the adapter.
    collected: Vec<IntermediateMessage>,
    /// Block-level tracking for Claude structured streaming.
    blocks: BTreeMap<usize, StreamingBlock>,
    /// Whether we've seen at least one content_block_start event.
    has_block_structure: bool,
    /// Fallback flat delta text (legacy backends without block structure).
    fallback_text: String,
    /// Fallback flat delta thinking text.
    fallback_thinking: String,
    /// Stable timestamp for the current streaming partial.
    partial_created_at: Option<String>,
    /// DB UUID for the currently in-flight assistant turn. Minted on first
    /// need (first streaming partial OR first `handle_assistant` of a new
    /// turn), reused as `IntermediateMessage.id` for every partial / full
    /// snapshot of that turn AND as `CollectedTurn.id` when the turn flushes.
    /// Consumed via `take()` by `flush_assistant` / `materialize_partial`.
    active_turn_id: Option<String>,
    line_count: u64,

    // ── Persistence state (replaces Rust ClaudeOutputAccumulator) ────
    /// Completed turns for DB persistence.
    turns: Vec<CollectedTurn>,
    /// Provider session ID (Claude session_id or Codex thread_id).
    session_id: Option<String>,
    /// Resolved model name.
    resolved_model: String,
    /// Token usage counters.
    usage: AgentUsage,
    /// Raw result JSON line.
    result_json: Option<String>,
    /// Pre-assigned id for the result (Claude `result` / Codex
    /// `turn.completed`) row. `handle_result` / codex's `handle_turn_completed`
    /// mint this up-front and pass it as the `collected[]` id so the DB
    /// insert done by `persist_result_and_finalize` can reuse the same
    /// UUID — no post-hoc id sync needed.
    result_id: Option<String>,
    /// Concatenated assistant text (for persistence finalization).
    assistant_text: String,
    /// Concatenated thinking text (Claude only).
    thinking_text: String,
    saw_text_delta: bool,
    saw_thinking_delta: bool,

    // ── Claude-specific accumulation ─────────────────────────────────
    /// Current assistant message ID being built (for turn batching).
    cur_asst_id: Option<String>,
    /// Content blocks from the current assistant message.
    cur_asst_blocks: Vec<Value>,
    /// Template of the current assistant message (for rebuilding).
    cur_asst_template: Option<Value>,
    /// Running count of content blocks accumulated across all `assistant`
    /// events of the current turn. Used to assign globally-unique
    /// `__part_id` indices when the SDK delivers finalized blocks in
    /// separate per-block `assistant` events (delta-style).
    cur_asst_block_count: usize,
    /// `parent_tool_use_id` of the turn currently streaming into `blocks`.
    /// Set from the `stream_event` envelope so `build_partial_from_blocks`
    /// can tag a subagent's mid-stream partial as `child:<pt>:<turn>` —
    /// matching the finalized render so the live partial nests under its
    /// parent Task/Agent tool call instead of flashing as a top-level bubble.
    pub(super) cur_streaming_parent_id: Option<String>,
    local_bash_task_refs: HashSet<String>,
    compact_lifecycle_id: Option<String>,
    // ── Codex state ──────────────────────────────────────────────────
    /// Per-item delta accumulation for Codex App Server streaming.
    codex_items: HashMap<String, codex::CodexItemState>,
    /// Index into `collected[]` of the entry most recently written by
    /// `collect_or_replace`. Used by `build_codex_partial` to render
    /// only the last-touched entry as a streaming partial.
    codex_partial_idx: Option<usize>,
    /// Timestamp (ms since epoch) when the current Codex turn started.
    /// Used to compute turn duration since the App Server doesn't provide it.
    pub(super) codex_turn_started_at: Option<f64>,

    // ── Cursor state ─────────────────────────────────────────────────
    /// Per-run cursor state; see `cursor.rs`.
    cursor_state: cursor::CursorRunState,

    // ── opencode state ───────────────────────────────────────────────
    /// Per-turn opencode part accumulation; see `opencode.rs`.
    opencode_state: opencode::OpencodeRunState,
    /// Index into `collected[]` driving `build_opencode_partial`.
    opencode_partial_idx: Option<usize>,

    // ── kimi (ACP) state ─────────────────────────────────────────────
    /// Per-turn kimi part accumulation; see `kimi.rs`.
    kimi_state: kimi::KimiRunState,
    /// Index into `collected[]` driving `build_kimi_partial`.
    kimi_partial_idx: Option<usize>,

    // ── Coverage guard ───────────────────────────────────────────────
    /// Top-level event types that fell through `push_event`'s match
    /// without a handler. Tested as a hard-zero invariant in
    /// `pipeline_streams.rs` so any new SDK type silently dropped here
    /// fails the build immediately.
    dropped_event_types: Vec<String>,
}

/// Map an `SDKAssistantMessageError` category string to a human-readable
/// label. Both Claude `assistant.error` and (in the future) any other
/// turn-level failure surface route through this — the rendered SystemNotice
/// is the same shape across providers, so the frontend never branches.
fn assistant_error_fallback_text(value: &Value) -> Option<String> {
    value
        .get("message")
        .and_then(|message| message.get("content"))
        .and_then(Value::as_array)
        .and_then(|blocks| {
            blocks.iter().find_map(|block| {
                block
                    .get("type")
                    .and_then(Value::as_str)
                    .filter(|ty| *ty == "text")
                    .and_then(|_| block.get("text"))
                    .and_then(Value::as_str)
                    .map(str::trim)
                    .filter(|text| !text.is_empty())
                    .map(str::to_string)
            })
        })
        .or_else(|| {
            value
                .get("text")
                .and_then(Value::as_str)
                .map(str::trim)
                .filter(|text| !text.is_empty())
                .map(str::to_string)
        })
}

fn assistant_error_message(category: &str, value: &Value) -> String {
    match category {
        "rate_limit" => "Rate limited by Anthropic API",
        "max_output_tokens" => "Output truncated: max output tokens reached",
        "billing_error" => "Billing error: check your Anthropic account",
        "authentication_failed" => "Authentication failed: please re-authenticate",
        "invalid_request" => "Invalid request: malformed payload",
        "server_error" => "Anthropic API server error (5xx)",
        "unknown" => {
            if let Some(message) = assistant_error_fallback_text(value) {
                return message;
            }
            "Assistant turn ended in an unknown error state"
        }
        other => return format!("Assistant error: {other}"),
    }
    .to_string()
}

fn patch_tool_use_block(block: &mut Value, resolved: &HashSet<String>) -> bool {
    let Some(obj) = block.as_object_mut() else {
        return false;
    };
    let block_type = obj.get("type").and_then(Value::as_str);
    if !matches!(
        block_type,
        Some("tool_use") | Some("server_tool_use") | Some("mcp_tool_use")
    ) {
        return false;
    }
    let Some(id) = obj.get("id").and_then(Value::as_str) else {
        return false;
    };
    if resolved.contains(id) {
        return false;
    }
    let current = obj.get("__streaming_status").and_then(Value::as_str);
    if matches!(current, Some("done" | "error")) {
        return false;
    }
    obj.insert(
        "__streaming_status".to_string(),
        Value::String("error".to_string()),
    );
    true
}

fn assistant_block_type(block: &Value) -> Option<&str> {
    block.get("type").and_then(Value::as_str)
}

/// `__is_streaming` is a live-only marker consumed by the frontend to keep
/// finalized reasoning blocks expanded in the current session. Persistence
/// paths (`flush_assistant`, `materialize_partial`) strip it so historical
/// reloads don't resurrect every old thinking block as "just completed".
fn strip_is_streaming_markers(blocks: &mut [Value]) {
    for block in blocks.iter_mut() {
        if let Some(obj) = block.as_object_mut() {
            obj.remove("__is_streaming");
        }
    }
}

/// Whether `next` re-sends the SAME block as `prev` (cumulative snapshot)
/// rather than a NEW block of the same type (delta-style). Type alone is
/// not enough: omitted-thinking turns deliver several distinct thinking
/// blocks per message — judging those cumulative reuses the first block's
/// `__part_id` and inherits its `__duration_ms`, rendering N identical
/// "Thought for Ns" chips.
fn assistant_block_is_same(prev: &Value, next: &Value) -> bool {
    let ty = assistant_block_type(prev);
    if ty != assistant_block_type(next) {
        return false;
    }
    match ty {
        Some("tool_use" | "server_tool_use" | "mcp_tool_use") => {
            prev.get("id").and_then(Value::as_str) == next.get("id").and_then(Value::as_str)
        }
        Some("thinking") => {
            prev.get("signature").and_then(Value::as_str)
                == next.get("signature").and_then(Value::as_str)
                && prev.get("thinking").and_then(Value::as_str)
                    == next.get("thinking").and_then(Value::as_str)
        }
        Some("text") => {
            prev.get("text").and_then(Value::as_str) == next.get("text").and_then(Value::as_str)
        }
        _ => true,
    }
}

fn cumulative_assistant_snapshot_prefix_matches(prev: &[Value], next: &[Value]) -> bool {
    if next.len() < prev.len() {
        return false;
    }

    prev.iter()
        .zip(next.iter())
        .all(|(prev_block, next_block)| assistant_block_is_same(prev_block, next_block))
}

fn collect_resolved_id(block: &Value, resolved: &mut HashSet<String>) {
    let Some(obj) = block.as_object() else {
        return;
    };
    let block_type = obj.get("type").and_then(Value::as_str);
    if !matches!(block_type, Some("tool_result") | Some("mcp_tool_result")) {
        return;
    }
    if let Some(id) = obj.get("tool_use_id").and_then(Value::as_str) {
        resolved.insert(id.to_string());
    }
}

impl StreamAccumulator {
    pub fn new(provider: &str, fallback_model: &str) -> Self {
        Self {
            provider: provider.to_string(),
            collected: Vec::new(),
            blocks: BTreeMap::new(),
            has_block_structure: false,
            fallback_text: String::new(),
            fallback_thinking: String::new(),
            partial_created_at: None,
            active_turn_id: None,
            line_count: 0,
            turns: Vec::new(),
            session_id: None,
            resolved_model: fallback_model.to_string(),
            usage: AgentUsage::default(),
            result_json: None,
            result_id: None,
            assistant_text: String::new(),
            thinking_text: String::new(),
            saw_text_delta: false,
            saw_thinking_delta: false,
            cur_asst_id: None,
            cur_asst_blocks: Vec::new(),
            cur_asst_template: None,
            cur_asst_block_count: 0,
            cur_streaming_parent_id: None,
            local_bash_task_refs: HashSet::new(),
            compact_lifecycle_id: None,
            codex_items: codex::new_item_states(),
            codex_partial_idx: None,
            codex_turn_started_at: None,
            cursor_state: cursor::new_run_state(),
            opencode_state: opencode::new_run_state(),
            opencode_partial_idx: None,
            kimi_state: kimi::new_run_state(),
            kimi_partial_idx: None,
            dropped_event_types: Vec::new(),
        }
    }

    // =====================================================================
    // Public API
    // =====================================================================

    /// Feed a raw sidecar JSON event into the accumulator and report what
    /// kind of state change it produced. The caller (`MessagePipeline`)
    /// uses the returned `PushOutcome` to decide between a full re-render,
    /// a partial render, or skipping emission entirely.
    pub fn push_event(&mut self, value: &Value, raw_line: &str) -> PushOutcome {
        self.line_count += 1;

        // Extract session ID
        if let Some(sid) = value
            .get("session_id")
            .and_then(Value::as_str)
            .or_else(|| value.get("thread_id").and_then(Value::as_str))
        {
            self.session_id = Some(sid.to_string());
        }

        // Extract resolved model (Claude only)
        if self.provider != "codex" {
            if let Some(model) = streaming::extract_claude_model_name(value) {
                self.resolved_model = model;
            }
        }

        let event_type = value.get("type").and_then(Value::as_str);

        // Top-level noise filter — types listed in
        // `pipeline::event_filter::SUPPRESSED_EVENT_TYPES` get dropped
        // before any handler runs. Edit that file to toggle.
        if let Some(t) = event_type {
            if crate::pipeline::event_filter::is_suppressed_event_type(t) {
                return PushOutcome::NoOp;
            }
        }

        match event_type {
            // ── Claude streaming deltas ────────────────────────────────
            // These mutate `blocks`/`fallback_text`/`fallback_thinking`
            // and the pipeline can render them via `build_partial`.
            Some("stream_event") => {
                streaming::handle_stream_event(self, value);
                PushOutcome::StreamingDelta
            }
            Some("tool_progress") => {
                streaming::handle_tool_progress(self, value);
                PushOutcome::StreamingDelta
            }

            // ── Claude finalized full messages ─────────────────────────
            Some("assistant") => {
                self.handle_assistant(value, raw_line);
                PushOutcome::Finalized
            }
            Some("user") => {
                self.handle_user(raw_line, value);
                PushOutcome::Finalized
            }
            // Mid-turn steer injection — same semantics as a regular user
            // turn boundary (flush in-flight assistant, push a user turn,
            // subsequent assistant events start a fresh message). The
            // inner JSON shape matches what `persist_user_message` writes
            // for initial prompts, so streaming persistence and reload
            // both go through the adapter's existing `user_prompt` branch
            // without any special-case handling.
            Some("user_prompt") => {
                self.handle_user(raw_line, value);
                PushOutcome::Finalized
            }
            Some("result") => {
                self.handle_result(value, raw_line);
                PushOutcome::Finalized
            }
            Some("error") => {
                self.handle_error(raw_line, value);
                PushOutcome::Finalized
            }
            Some("rate_limit_event") => {
                self.handle_rate_limit_event(raw_line, value);
                PushOutcome::Finalized
            }
            // `prompt_suggestion` and `auth_status` are normally caught
            // by the noise filter above. The arms stay so uncommenting
            // either entry in `event_filter.rs` reaches a real handler
            // (or NoOp for auth_status, which has no body to render).
            Some("prompt_suggestion") => {
                self.handle_prompt_suggestion(raw_line, value);
                PushOutcome::Finalized
            }
            Some("auth_status") => PushOutcome::NoOp,
            // Resolved Codex/OpenCode user-input question — the sidecar
            // emits this at answer time so the Q&A lands in the transcript
            // at its natural stream position. Claude AskUserQuestion skips
            // this path (its tool_use already lives in the assistant turn).
            Some("user_question") => {
                self.handle_user_question(value);
                PushOutcome::Finalized
            }
            Some("system") => {
                self.handle_claude_system(raw_line, value);
                PushOutcome::Finalized
            }

            // SDKToolUseSummaryMessage — the SDK summarizes a long
            // tool result so the model's context doesn't blow up. We
            // surface it as a SystemNotice so the user knows the raw
            // output was truncated; reshape into a Claude-style
            // `{type: system, subtype: tool_use_summary}` envelope so
            // the existing `convert_system_msg` path renders it.
            Some("tool_use_summary") => {
                self.handle_tool_use_summary(raw_line, value);
                PushOutcome::Finalized
            }

            // ── Codex App Server item events ──────────────────────────
            Some("item/completed") => {
                codex::handle_item_completed(self, raw_line, value);
                self.codex_partial_idx = None;
                PushOutcome::Finalized
            }
            Some("item/started") => {
                codex::handle_item_started(self, raw_line, value);
                PushOutcome::StreamingDelta
            }

            // ── Codex App Server delta streaming ─────────────────────
            Some("item/agentMessage/delta") => {
                codex::handle_text_delta(self, value);
                PushOutcome::StreamingDelta
            }
            Some("item/commandExecution/outputDelta") => {
                codex::handle_cmd_output_delta(self, value);
                PushOutcome::StreamingDelta
            }
            Some("item/reasoning/textDelta") | Some("item/reasoning/summaryTextDelta") => {
                codex::handle_reasoning_delta(self, value);
                PushOutcome::StreamingDelta
            }
            Some("item/fileChange/outputDelta") => {
                codex::handle_file_change_delta(self, value);
                PushOutcome::StreamingDelta
            }
            Some("item/plan/delta") => {
                codex::handle_plan_delta(self, value);
                PushOutcome::StreamingDelta
            }
            Some("turn/plan/updated") => {
                codex::handle_turn_plan_updated(self, raw_line, value);
                PushOutcome::Finalized
            }

            // ── Codex App Server turn/thread lifecycle ───────────────
            Some("turn/completed") => {
                codex::handle_turn_completed(self, raw_line, value);
                PushOutcome::Finalized
            }
            Some("turn/started") => {
                self.codex_turn_started_at = Some(now_ms());
                PushOutcome::NoOp
            }
            Some("thread/compacted") => {
                codex::handle_thread_compacted(self, raw_line, value);
                PushOutcome::Finalized
            }
            Some("thread/started") => {
                if let Some(tid) = value
                    .get("thread")
                    .and_then(|t| t.get("id"))
                    .and_then(Value::as_str)
                {
                    self.session_id = Some(tid.to_string());
                }
                PushOutcome::NoOp
            }

            // ── Cursor SDK events (namespaced by sidecar manager) ─────
            // Synthetic — session_id already lifted by push_event extractor.
            Some("cursor/agent_init") => PushOutcome::NoOp,
            Some("cursor/status") => cursor::handle_status(self, value),
            Some("cursor/thinking") => cursor::handle_thinking(self, value),
            Some("cursor/assistant") => cursor::handle_assistant_delta(self, value),
            Some("cursor/tool_call_start") => cursor::handle_tool_call_start(self, value),
            Some("cursor/tool_call_end") => cursor::handle_tool_call_end(self, value),

            // ── opencode events (namespaced by the sidecar manager) ───
            Some("opencode/session_init") => PushOutcome::NoOp,
            Some("opencode/message.updated") => opencode::handle_message_updated(self, value),
            Some("opencode/message.part.updated") => opencode::handle_part_updated(self, value),
            // Token-by-token text/reasoning deltas (parallel to part.updated snapshots).
            Some("opencode/message.part.delta") => opencode::handle_part_delta(self, value),
            // Subagent (`task` tool) parts, tagged with the parent `callID`.
            Some("opencode/subtask.message.updated") => {
                opencode::handle_subtask_message_updated(self, value)
            }
            Some("opencode/subtask.message.part.updated") => {
                opencode::handle_subtask_part_updated(self, value)
            }
            Some("opencode/subtask.message.part.delta") => {
                opencode::handle_subtask_part_delta(self, value)
            }
            // A turn finalizes when its session goes idle.
            Some("opencode/session.idle") => opencode::handle_session_idle(self),
            Some("opencode/session.status") => opencode::handle_session_status(self, value),
            // Redundant/informational forms — handled as NoOps for the coverage guard.
            Some("opencode/session.error") => opencode::handle_session_error(self, value),
            Some("opencode/session.created")
            | Some("opencode/session.updated")
            | Some("opencode/session.diff")
            | Some("opencode/todo.updated")
            | Some("opencode/message.removed")
            | Some("opencode/message.part.removed") => PushOutcome::NoOp,

            // ── kimi (ACP) events (namespaced by the sidecar manager) ─
            // session_id already lifted by push_event; nothing to render.
            Some("kimi/session_init") => PushOutcome::NoOp,
            Some("kimi/agent_message_chunk") => kimi::handle_message_chunk(self, value),
            Some("kimi/agent_thought_chunk") => kimi::handle_thought_chunk(self, value),
            // tool_call + tool_call_update both merge by tool_call_id.
            Some("kimi/tool_call") | Some("kimi/tool_call_update") => {
                kimi::handle_tool_call(self, value)
            }
            Some("kimi/plan") => kimi::handle_plan(self, value),
            // The sidecar's `session/prompt` response → finalize the turn.
            Some("kimi/turn_complete") => kimi::handle_turn_complete(self, value),

            // ── Codex informational notifications (no render) ────────
            Some("thread/status/changed")
            | Some("thread/tokenUsage/updated")
            | Some("thread/name/updated")
            | Some("thread/goal/updated")
            | Some("thread/goal/cleared")
            | Some("account/rateLimits/updated")
            | Some("account/updated")
            | Some("mcpServer/startupStatus/updated")
            | Some("mcpServer/oauthLogin/completed")
            | Some("model/rerouted")
            | Some("configWarning") => PushOutcome::NoOp,

            // Sidecar protocol control events.
            Some("end")
            | Some("aborted")
            | Some("ready")
            | Some("pong")
            | Some("stopped")
            | Some("titleGenerated") => PushOutcome::NoOp,

            other => {
                // Coverage guard: any unhandled top-level event type is
                // recorded so `pipeline_streams.rs` can fail the test if a
                // fixture exercises a type we don't yet parse. Adding a
                // handler above must clear the corresponding entry here.
                let label = other.unwrap_or("<missing-type>").to_string();
                if !self.dropped_event_types.contains(&label) {
                    self.dropped_event_types.push(label);
                }
                PushOutcome::NoOp
            }
        }
    }

    /// Top-level event types seen during this run that no handler matched.
    /// Empty in steady state — `pipeline_streams.rs` asserts on this.
    pub fn dropped_event_types(&self) -> &[String] {
        &self.dropped_event_types
    }

    /// Borrow the collected (finalized) messages — no allocation.
    pub fn collected(&self) -> &[IntermediateMessage] {
        &self.collected
    }

    /// Build only the trailing partial message (if any streaming content exists).
    /// Returns `None` if there is no active streaming content.
    /// This is the only allocation needed per render cycle.
    ///
    /// `_context_key` is retained for API compatibility; stable DB UUIDs
    /// no longer need a disambiguating context prefix.
    pub fn build_partial(
        &mut self,
        _context_key: &str,
        session_id: &str,
    ) -> Option<IntermediateMessage> {
        if !self.blocks.is_empty() {
            let (partial_id, created_at) = self.get_or_create_turn_identity();
            streaming::build_partial_from_blocks(self, session_id, partial_id, created_at)
        } else {
            let text = self.fallback_text.trim();
            let thinking = self.fallback_thinking.trim();
            if !text.is_empty() || !thinking.is_empty() {
                let (partial_id, created_at) = self.get_or_create_turn_identity();
                Some(streaming::build_partial_fallback(
                    self, session_id, partial_id, created_at,
                ))
            } else {
                None
            }
        }
    }

    /// Convenience: build full snapshot (collected + partial) as one Vec.
    /// Used by tests. Production code uses `collected()` + `build_partial()`
    /// to avoid cloning the collected vec.
    #[cfg(test)]
    pub fn snapshot(&mut self, context_key: &str, session_id: &str) -> Vec<IntermediateMessage> {
        let mut messages = self.collected.clone();
        if let Some(partial) = self.build_partial(context_key, session_id) {
            messages.push(partial);
        }
        messages
    }

    /// Build a streaming partial from the last Codex `collected[]` entry
    /// touched by `collect_or_replace`. Returns `None` if no entry was
    /// recently touched. This is the Codex counterpart to the Claude
    /// `build_partial` path: Codex items land directly in `collected[]`
    /// (full snapshots, not block-level deltas), so the partial is a
    /// clone of the last-touched entry with `is_streaming = true`.
    pub fn build_codex_partial(&mut self) -> Option<IntermediateMessage> {
        let idx = self.codex_partial_idx.take()?;
        let entry = self.collected.get(idx)?;
        Some(IntermediateMessage {
            id: entry.id.clone(),
            role: entry.role,
            raw_json: entry.raw_json.clone(),
            parsed: entry.parsed.clone(),
            created_at: entry.created_at.clone(),
            is_streaming: true,
        })
    }

    /// Streaming partial = clone of the last opencode `collected[]` snapshot.
    pub fn build_opencode_partial(&mut self) -> Option<IntermediateMessage> {
        let idx = self.opencode_partial_idx.take()?;
        let entry = self.collected.get(idx)?;
        Some(IntermediateMessage {
            id: entry.id.clone(),
            role: entry.role,
            raw_json: entry.raw_json.clone(),
            parsed: entry.parsed.clone(),
            created_at: entry.created_at.clone(),
            is_streaming: true,
        })
    }

    /// Streaming partial = clone of the last kimi `collected[]` snapshot.
    pub fn build_kimi_partial(&mut self) -> Option<IntermediateMessage> {
        let idx = self.kimi_partial_idx.take()?;
        let entry = self.collected.get(idx)?;
        Some(IntermediateMessage {
            id: entry.id.clone(),
            role: entry.role,
            raw_json: entry.raw_json.clone(),
            parsed: entry.parsed.clone(),
            created_at: entry.created_at.clone(),
            is_streaming: true,
        })
    }

    /// Whether the accumulator has an active streaming partial.
    pub fn has_active_partial(&self) -> bool {
        !self.blocks.is_empty()
            || !self.fallback_text.trim().is_empty()
            || !self.fallback_thinking.trim().is_empty()
            || self.codex_partial_idx.is_some()
            || self.opencode_partial_idx.is_some()
            || self.kimi_partial_idx.is_some()
    }

    // ── Persistence accessors ───────────────────────────────────────

    pub fn turns_len(&self) -> usize {
        self.turns.len()
    }

    pub fn turn_at(&self, index: usize) -> &CollectedTurn {
        &self.turns[index]
    }

    pub fn session_id(&self) -> Option<&str> {
        self.session_id.as_deref()
    }

    pub fn resolved_model(&self) -> &str {
        &self.resolved_model
    }

    pub fn usage(&self) -> &AgentUsage {
        &self.usage
    }

    pub fn result_json(&self) -> Option<&str> {
        self.result_json.as_deref()
    }

    /// The id the accumulator used for the current result row — returned
    /// so `persist_result_and_finalize` can reuse it as the DB row id,
    /// keeping the live-rendered id and the persisted id identical.
    pub fn take_result_id(&mut self) -> Option<String> {
        self.result_id.take()
    }

    /// Reclassify any in-flight tool_use as `error` so the adapter's
    /// settle pass can fill `result = "aborted by user"` and the
    /// frontend stops the spinner. Walks live Claude blocks, staged
    /// blocks, and collected[] (where Codex synthetic items live).
    /// Tool_uses with a matching tool_result are left alone.
    pub fn mark_pending_tools_aborted(&mut self) {
        let resolved_ids = self.collect_resolved_tool_use_ids();

        for block in self.blocks.values_mut() {
            if let StreamingBlock::ToolUse {
                tool_use_id,
                status,
                ..
            } = block
            {
                if resolved_ids.contains(tool_use_id.as_str()) {
                    continue;
                }
                if matches!(*status, "pending" | "streaming_input" | "running") {
                    *status = "error";
                }
            }
        }

        for block in self.cur_asst_blocks.iter_mut() {
            patch_tool_use_block(block, &resolved_ids);
        }

        for msg in self.collected.iter_mut() {
            if msg.role != MessageRole::Assistant {
                continue;
            }
            let Some(parsed) = msg.parsed.as_mut() else {
                continue;
            };
            let Some(blocks) = parsed
                .get_mut("message")
                .and_then(|m| m.get_mut("content"))
                .and_then(Value::as_array_mut)
            else {
                continue;
            };
            let mut changed = false;
            for block in blocks.iter_mut() {
                if patch_tool_use_block(block, &resolved_ids) {
                    changed = true;
                }
            }
            if changed {
                msg.raw_json =
                    serde_json::to_string(parsed).unwrap_or_else(|_| msg.raw_json.clone());
            }
        }
    }

    fn collect_resolved_tool_use_ids(&self) -> HashSet<String> {
        let mut ids = HashSet::new();
        for msg in &self.collected {
            let Some(parsed) = msg.parsed.as_ref() else {
                continue;
            };
            let Some(blocks) = parsed
                .get("message")
                .and_then(|m| m.get("content"))
                .and_then(Value::as_array)
            else {
                continue;
            };
            for block in blocks {
                collect_resolved_id(block, &mut ids);
            }
        }
        // mcp_tool_result can live inline in the still-staged assistant message
        for block in &self.cur_asst_blocks {
            collect_resolved_id(block, &mut ids);
        }
        ids
    }

    /// Flush staged Claude assistant blocks into `self.turns`. Idempotent.
    pub fn flush_pending(&mut self) {
        self.flush_assistant();
    }

    /// Flush all in-progress Codex items as completed so they get persisted
    /// on abort. No-op when no items are in flight.
    pub fn flush_codex_in_progress(&mut self) {
        codex::flush_in_progress(self);
    }

    /// Drain in-flight cursor state on abort (no FINISHED will arrive).
    /// Idempotent.
    pub fn flush_cursor_in_progress(&mut self) {
        cursor::flush_in_progress(self);
    }

    /// Finalize the in-flight opencode message on abort. Idempotent.
    pub fn flush_opencode_in_progress(&mut self) {
        opencode::flush_in_progress(self);
    }

    /// Finalize the in-flight kimi message on abort or error termination
    /// (in-flight tool parts settle to `failed`). Idempotent.
    pub fn flush_kimi_in_progress(&mut self) {
        kimi::flush_in_progress(self);
    }

    /// Convert any active streaming partial into a finalized assistant
    /// message so terminal notices can land after it in the rendered thread.
    /// Used by abort handling, where no final provider `assistant` event will
    /// arrive to consume the partial naturally.
    pub(super) fn materialize_partial(&mut self, _context_key: &str, _session_id: &str) {
        if !self.has_active_partial() {
            return;
        }

        let (partial_id, created_at) = self.get_or_create_turn_identity();
        let partial = if !self.blocks.is_empty() {
            streaming::build_materialized_partial_from_blocks(self, partial_id, created_at)
        } else {
            streaming::build_materialized_partial_fallback(self, partial_id, created_at)
        };

        if let Some(message) = partial {
            // The live copy keeps `__is_streaming: false` so the aborted
            // reasoning block still renders open. The persisted copy has
            // it stripped — mirror of what `flush_assistant` does on the
            // happy path.
            let content_json_for_persist = match serde_json::from_str::<Value>(&message.raw_json) {
                Ok(mut parsed) => {
                    if let Some(blocks) = parsed
                        .get_mut("message")
                        .and_then(|m| m.get_mut("content"))
                        .and_then(Value::as_array_mut)
                    {
                        strip_is_streaming_markers(blocks);
                    }
                    serde_json::to_string(&parsed).unwrap_or_else(|_| message.raw_json.clone())
                }
                Err(_) => message.raw_json.clone(),
            };
            self.turns.push(CollectedTurn {
                id: message.id.clone(),
                role: MessageRole::Assistant,
                content_json: content_json_for_persist,
            });
            self.collected.push(message);
        }

        // Turn UUID has been consumed into turns/collected — drop it here
        // so the next turn starts fresh. `finalize_blocks` no longer owns
        // the turn UUID lifecycle.
        self.active_turn_id = None;
        self.finalize_blocks();
    }

    /// Project accumulator state into `ParsedAgentOutput`. Always succeeds —
    /// empty input yields empty output. Drains owned fields, single-call.
    pub fn drain_output(&mut self, fallback_session_id: Option<&str>) -> ParsedAgentOutput {
        let assistant_text = self.assistant_text.trim().to_string();
        let thinking_text = {
            let t = self.thinking_text.trim().to_string();
            if t.is_empty() {
                None
            } else {
                Some(t)
            }
        };

        ParsedAgentOutput {
            assistant_text,
            thinking_text,
            session_id: self
                .session_id
                .take()
                .or_else(|| fallback_session_id.map(str::to_string)),
            resolved_model: self.resolved_model.clone(),
            usage: std::mem::take(&mut self.usage),
            result_json: self.result_json.take(),
        }
    }

    /// Push an `{type:"error",content:"aborted by user"}` row into both
    /// `collected[]` (live render) and `turns` (persistence). Same shape
    /// imported sessions use, so the existing adapter renders it.
    pub fn append_aborted_notice(&mut self) {
        const NOTICE_JSON: &str = r#"{"type":"error","content":"aborted by user"}"#;
        let parsed = serde_json::json!({
            "type": "error",
            "content": "aborted by user",
        });

        self.line_count += 1;
        let notice_id = uuid::Uuid::new_v4().to_string();
        self.collected.push(IntermediateMessage {
            id: notice_id.clone(),
            role: MessageRole::Error,
            raw_json: NOTICE_JSON.to_string(),
            parsed: Some(parsed),
            created_at: chrono::Utc::now().to_rfc3339(),
            is_streaming: false,
        });
        self.turns.push(CollectedTurn {
            id: notice_id,
            role: MessageRole::Error,
            content_json: NOTICE_JSON.to_string(),
        });
    }

    // =====================================================================
    // Claude full-message handlers (small enough to live alongside dispatch)
    // =====================================================================

    fn handle_assistant(&mut self, value: &Value, raw_line: &str) {
        // === Persistence ===
        if !self.saw_text_delta {
            if let Some(text) = streaming::extract_claude_assistant_text(value) {
                self.assistant_text.push_str(&text);
            }
        }
        if !self.saw_thinking_delta {
            if let Some(thinking) = streaming::extract_claude_thinking_text(value) {
                self.thinking_text.push_str(&thinking);
            }
        }

        // Turn batching for persistence: group content blocks by message ID.
        let msg_id = value
            .get("message")
            .and_then(|m| m.get("id"))
            .and_then(Value::as_str);

        // If we're already batching a different msg_id, flush it first.
        // `flush_assistant` consumes `active_turn_id` for the OLD turn so
        // the next mint below produces a fresh UUID for the NEW turn.
        let same_msg_id = self
            .cur_asst_id
            .as_deref()
            .is_some_and(|current| Some(current) == msg_id);
        if self.cur_asst_id.is_some() && !same_msg_id {
            self.flush_assistant();
        }

        // Ensure a turn UUID exists for this batch. If streaming partials
        // already minted one for this turn it's still in `active_turn_id`
        // and we reuse it; otherwise we mint here so `collect_message` and
        // a later `flush_assistant` share a single identifier.
        let turn_id = self.get_or_create_turn_identity().0;

        self.cur_asst_id = msg_id.map(str::to_string);
        self.cur_asst_template = Some(value.clone());

        // Snapshot the per-block start times of any in-flight thinking
        // blocks BEFORE finalize_blocks clears `self.blocks`. Keyed by
        // part_id (the SDK's block.index-derived id) so the stamping pass
        // below can line each assistant thinking block up with the
        // streaming block it came from and compute the live duration.
        let thinking_durations: HashMap<String, u64> = self
            .blocks
            .values()
            .filter_map(|block| match block {
                streaming::StreamingBlock::Thinking {
                    id, started_at_ms, ..
                } => {
                    let elapsed = streaming::now_ms().saturating_sub(*started_at_ms);
                    Some((id.clone(), elapsed))
                }
                _ => None,
            })
            .collect();

        // The Claude SDK sends each finalized content block as its own
        // `assistant` event with the SAME msg_id — i.e., delta-style,
        // not cumulative snapshot. Stamp every block with a globally-
        // unique `__part_id` derived from the turn UUID + a running
        // counter so the adapter never falls back to positional synthesis
        // (which would collide across events that each start at idx 0).
        let mut stamped_value = value.clone();
        if let Some(blocks) = value
            .get("message")
            .and_then(|m| m.get("content"))
            .and_then(Value::as_array)
        {
            let cumulative_snapshot = same_msg_id
                && cumulative_assistant_snapshot_prefix_matches(&self.cur_asst_blocks, blocks);

            // Snapshot the previous blocks BEFORE clearing so the stamping
            // loop below can reach back to the prior cumulative snapshot
            // (thinking duration carry-over, stable `__part_id` reuse).
            let prev_blocks: Vec<Value> = if cumulative_snapshot {
                std::mem::take(&mut self.cur_asst_blocks)
            } else {
                Vec::new()
            };

            // Cumulative: reset count so reused indices produce stable part_ids.
            // Non-cumulative new-msg: clear buffer but keep count monotonic so
            // part_ids don't collide with the previous message when two events
            // share the same active turn_id (e.g. no explicit `message.id`).
            if cumulative_snapshot {
                self.cur_asst_block_count = 0;
            } else if !same_msg_id {
                self.cur_asst_blocks.clear();
            }
            let mut stamped_blocks = blocks.clone();
            for (i, block) in stamped_blocks.iter_mut().enumerate() {
                let is_thinking = block.get("type").and_then(Value::as_str) == Some("thinking");
                let prev_block = if cumulative_snapshot {
                    prev_blocks.get(i).and_then(Value::as_object)
                } else {
                    None
                };
                if let Some(obj) = block.as_object_mut() {
                    let part_id = if cumulative_snapshot {
                        prev_block
                            .and_then(|prev| prev.get("__part_id"))
                            .and_then(Value::as_str)
                            .map(str::to_string)
                            .unwrap_or_else(|| format!("{turn_id}:blk:{i}"))
                    } else {
                        format!("{turn_id}:blk:{}", self.cur_asst_block_count + i)
                    };
                    obj.insert("__part_id".to_string(), Value::String(part_id.clone()));
                    if is_thinking {
                        // Mark finalized live thinking blocks so the
                        // frontend keeps them open + shows "Thought for Ns"
                        // even when the streaming partial window was too
                        // narrow for React to ever observe streaming=true.
                        obj.insert("__is_streaming".to_string(), Value::Bool(false));
                        let duration = thinking_durations.get(&part_id).copied().or_else(|| {
                            prev_block
                                .and_then(|prev| prev.get("__duration_ms"))
                                .and_then(Value::as_u64)
                        });
                        if let Some(ms) = duration {
                            obj.insert("__duration_ms".to_string(), Value::from(ms));
                        }
                    }
                }
            }
            if cumulative_snapshot {
                self.cur_asst_block_count = stamped_blocks.len();
                self.cur_asst_blocks = stamped_blocks.clone();
            } else {
                self.cur_asst_block_count += stamped_blocks.len();
                self.cur_asst_blocks.extend(stamped_blocks.iter().cloned());
            }

            // Also patch the value that goes into collected[] for rendering.
            if let Some(msg) = stamped_value.get_mut("message") {
                msg["content"] = Value::Array(stamped_blocks);
            }
        }

        // === Rendering ===
        // Finalize streaming blocks and push the full message to collected.
        // Matches TS behavior: always push, never replace.
        // The adapter's merge_adjacent_assistants handles merging.
        self.finalize_blocks();
        let stamped_raw =
            serde_json::to_string(&stamped_value).unwrap_or_else(|_| raw_line.to_string());
        self.collect_message(
            &stamped_raw,
            &stamped_value,
            MessageRole::Assistant,
            Some(&turn_id),
        );

        // Turn-level failure category → error envelope.
        if let Some(category) = value.get("error").and_then(Value::as_str) {
            let label = assistant_error_message(category, value);
            let synthetic = serde_json::json!({
                "type": "error",
                "message": label,
            });
            let s = serde_json::to_string(&synthetic).unwrap_or_default();
            self.collect_message(&s, &synthetic, MessageRole::Error, None);
        }
    }

    fn handle_user(&mut self, raw_line: &str, value: &Value) {
        // Persistence: flush any pending assistant turn
        self.flush_assistant();
        // Pre-mint the UUID so `collected[].id` and `CollectedTurn.id` are
        // the same string — no more post-hoc `sync_persisted_ids`.
        let turn_id = uuid::Uuid::new_v4().to_string();
        self.turns.push(CollectedTurn {
            id: turn_id.clone(),
            role: MessageRole::User,
            content_json: raw_line.to_string(),
        });

        // Rendering
        self.collect_message(raw_line, value, MessageRole::User, Some(&turn_id));
    }

    fn handle_result(&mut self, value: &Value, raw_line: &str) {
        // Persistence
        if self.assistant_text.trim().is_empty() {
            if let Some(text) = value.get("result").and_then(Value::as_str) {
                self.assistant_text.push_str(text);
            }
        }
        if let Some(parsed_usage) = value.get("usage") {
            self.usage.input_tokens = parsed_usage.get("input_tokens").and_then(Value::as_i64);
            self.usage.output_tokens = parsed_usage.get("output_tokens").and_then(Value::as_i64);
        }
        self.result_json = Some(raw_line.to_string());

        // Rendering — pre-mint the id so DB and live-render agree.
        let id = uuid::Uuid::new_v4().to_string();
        self.result_id = Some(id.clone());
        self.collect_message(raw_line, value, MessageRole::Assistant, Some(&id));
    }

    fn handle_error(&mut self, raw_line: &str, value: &Value) {
        self.collect_message(raw_line, value, MessageRole::Error, None);
    }

    /// Sidecar `user_question` event — a Codex/OpenCode question the user
    /// just answered (or declined). Normalizes the provider-raw questions
    /// into the canonical persisted shape and pushes a standalone turn so
    /// the Q&A card sits at its natural position between stream items.
    fn handle_user_question(&mut self, value: &Value) {
        let questions = crate::pipeline::user_question::normalize_questions(
            &self.provider,
            value.get("questions").unwrap_or(&Value::Null),
        );
        if questions.is_empty() {
            return;
        }
        let action = value
            .get("action")
            .and_then(Value::as_str)
            .unwrap_or("submit");
        let status = if action == "decline" {
            "declined"
        } else {
            "answered"
        };
        let mut synthetic = serde_json::json!({
            "type": "user_question",
            "userInputId": value.get("userInputId").and_then(Value::as_str).unwrap_or_default(),
            "source": value.get("source").and_then(Value::as_str).unwrap_or_default(),
            "questions": questions,
            "status": status,
        });
        if action == "submit" {
            if let Some(answers) = value.get("answers").filter(|v| v.is_object()) {
                synthetic["answers"] = answers.clone();
            }
        }
        let s = serde_json::to_string(&synthetic).unwrap_or_default();

        self.flush_assistant();
        let turn_id = uuid::Uuid::new_v4().to_string();
        self.turns.push(CollectedTurn {
            id: turn_id.clone(),
            role: MessageRole::Assistant,
            content_json: s.clone(),
        });
        self.collect_message(&s, &synthetic, MessageRole::Assistant, Some(&turn_id));
    }

    fn handle_rate_limit_event(&mut self, raw_line: &str, value: &Value) {
        // Collected as a "system" intermediate message; the adapter
        // recognizes type=rate_limit_event and emits a SystemNotice part.
        self.collect_message(raw_line, value, MessageRole::System, None);
    }

    fn handle_prompt_suggestion(&mut self, raw_line: &str, value: &Value) {
        // Collected as system role; the adapter recognizes
        // type=prompt_suggestion and emits a PromptSuggestion part.
        self.collect_message(raw_line, value, MessageRole::System, None);
    }

    fn handle_claude_system(&mut self, raw_line: &str, value: &Value) {
        // Suppressed system events never enter `collected[]` — they don't cross
        // IPC, don't render, don't waste downstream work. Edit
        // `event_filter` to toggle.
        if crate::pipeline::event_filter::is_suppressed_system_event(value) {
            return;
        }
        if is_claude_compact_lifecycle_event(value) {
            self.collect_compact_lifecycle(raw_line, value);
            return;
        }
        // `local_bash` task_* events duplicate the accompanying Bash tool.
        // Claude omits `task_type` on the later notification, so remember
        // refs from the start event and apply them to the completion event.
        if crate::pipeline::event_filter::is_explicit_local_bash_task(value) {
            crate::pipeline::event_filter::remember_task_refs(
                value,
                &mut self.local_bash_task_refs,
            );
            return;
        }
        if crate::pipeline::event_filter::is_local_bash_task_ref(value, &self.local_bash_task_refs)
        {
            return;
        }
        self.collect_message(raw_line, value, MessageRole::System, None);
    }

    /// Reshape an `SDKToolUseSummaryMessage` into a synthetic
    /// `{type: system, subtype: tool_use_summary}` envelope so the
    /// existing system-message conversion path can render it. The
    /// summary text and the count of preceding tool uses both flow
    /// through `convert_system_msg`.
    fn handle_tool_use_summary(&mut self, _raw_line: &str, value: &Value) {
        let summary = value
            .get("summary")
            .and_then(Value::as_str)
            .unwrap_or("")
            .to_string();
        if summary.is_empty() {
            return;
        }
        let count = value
            .get("preceding_tool_use_ids")
            .and_then(Value::as_array)
            .map(|a| a.len())
            .unwrap_or(0);
        let synthetic = serde_json::json!({
            "type": "system",
            "subtype": "tool_use_summary",
            "summary": summary,
            "tool_use_count": count,
        });
        let s = serde_json::to_string(&synthetic).unwrap_or_default();
        self.collect_message(&s, &synthetic, MessageRole::System, None);
    }

    // =====================================================================
    // Internal helpers — called by submodules and the handlers above.
    // =====================================================================

    pub(super) fn finalize_blocks(&mut self) {
        self.blocks.clear();
        self.has_block_structure = false;
        self.fallback_text.clear();
        self.fallback_thinking.clear();
        self.partial_created_at = None;
        // NOTE: `active_turn_id` is NOT cleared here — a single assistant
        // turn can span multiple content-block cycles (each emitting its
        // own `finalize_blocks` + `assistant` event batch), and the turn
        // UUID must stay stable across all of them. It's consumed only by
        // `flush_assistant` / `materialize_partial` at true turn boundaries.
    }

    fn flush_assistant(&mut self) {
        if self.cur_asst_blocks.is_empty() {
            self.cur_asst_id = None;
            self.active_turn_id = None;
            self.cur_asst_block_count = 0;
            return;
        }

        // Consume the turn UUID that every `collect_message` for this turn
        // already used — so the persisted row and the live-rendered row
        // end up with byte-identical ids. Fall back to a fresh UUID only
        // for the vanishingly rare case of a flush with no prior partial
        // / assistant event (defensive: shouldn't happen in practice).
        let turn_id = self
            .active_turn_id
            .take()
            .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());

        if let Some(mut template) = self.cur_asst_template.take() {
            let mut blocks_for_persist = std::mem::take(&mut self.cur_asst_blocks);
            strip_is_streaming_markers(&mut blocks_for_persist);
            if let Some(message) = template.get_mut("message") {
                message["content"] = Value::Array(blocks_for_persist);
            }
            self.turns.push(CollectedTurn {
                id: turn_id,
                role: MessageRole::Assistant,
                content_json: template.to_string(),
            });
        }

        self.cur_asst_id = None;
        self.cur_asst_block_count = 0;
    }

    fn collect_compact_lifecycle(&mut self, raw: &str, parsed: &Value) {
        let subtype = parsed.get("subtype").and_then(Value::as_str);
        let id = match subtype {
            Some("status") => {
                let id = parsed
                    .get("uuid")
                    .and_then(Value::as_str)
                    .map(str::to_string)
                    .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
                self.compact_lifecycle_id = Some(id.clone());
                id
            }
            Some("compact_boundary") => self
                .compact_lifecycle_id
                .take()
                .or_else(|| {
                    parsed
                        .get("uuid")
                        .and_then(Value::as_str)
                        .map(str::to_string)
                })
                .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()),
            _ => uuid::Uuid::new_v4().to_string(),
        };

        self.collect_or_replace(raw, parsed, MessageRole::System, Some(id.clone()));

        if subtype == Some("compact_boundary") {
            if let Some(pos) = self.turns.iter().rposition(|turn| turn.id == id) {
                self.turns[pos].content_json = raw.to_string();
            } else {
                self.turns.push(CollectedTurn {
                    id,
                    role: MessageRole::System,
                    content_json: raw.to_string(),
                });
            }
        }
    }

    pub(super) fn collect_message(
        &mut self,
        raw: &str,
        parsed: &Value,
        role: MessageRole,
        override_id: Option<&str>,
    ) {
        let id = override_id
            .map(|s| s.to_string())
            .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
        let created_at = self.get_partial_created_at();

        self.collected.push(IntermediateMessage {
            id,
            role,
            raw_json: raw.to_string(),
            parsed: Some(parsed.clone()),
            created_at,
            is_streaming: false,
        });
    }

    /// Collect a message OR replace the most recent existing entry whose
    /// id matches `override_id`. Used by Codex items so a later snapshot
    /// overwrites the earlier one for the same logical item.
    pub(super) fn collect_or_replace(
        &mut self,
        raw: &str,
        parsed: &Value,
        role: MessageRole,
        override_id: Option<String>,
    ) {
        if let Some(id) = override_id.as_deref() {
            if let Some(pos) = self.collected.iter().rposition(|m| m.id == id) {
                self.collected[pos].raw_json = raw.to_string();
                self.collected[pos].parsed = Some(parsed.clone());
                self.codex_partial_idx = Some(pos);
                return;
            }
        }
        let idx = self.collected.len();
        self.collect_message(raw, parsed, role, override_id.as_deref());
        self.codex_partial_idx = Some(idx);
    }

    fn get_partial_created_at(&mut self) -> String {
        if self.partial_created_at.is_none() {
            self.partial_created_at = Some(chrono::Utc::now().to_rfc3339());
        }
        self.partial_created_at.clone().unwrap()
    }

    /// Mint the turn-wide UUID if none is set yet, and return it alongside
    /// the turn's stable timestamp. The same pair feeds every streaming
    /// partial, every `collect_message` for the turn, and ultimately
    /// `CollectedTurn.id` — so the frontend sees one id from the first
    /// partial emit through DB commit, without any intermediate swap.
    ///
    /// `pub(super)` so the `streaming` submodule can stamp fresh block
    /// ids derived from the turn UUID at `content_block_start` time.
    pub(super) fn get_or_create_turn_identity(&mut self) -> (String, String) {
        let created_at = self.get_partial_created_at();
        if self.active_turn_id.is_none() {
            self.active_turn_id = Some(uuid::Uuid::new_v4().to_string());
        }
        (self.active_turn_id.clone().unwrap(), created_at)
    }
}