atman_runtime/

session.rs

use std::path::{Path, PathBuf};
use std::sync::Mutex;

use tokio::sync::{broadcast, watch};
use tokio_util::sync::CancellationToken;
use uuid::Uuid;

use crate::event::{Event, EventSink, FlowRunId, TurnId};
use crate::event_log::reader::{find_last_seq, replay_context_snapshot_from};
use crate::event_writer::EventWriter;
use crate::injection::{Injection, InjectionId, InjectionState};
use crate::message::{Message, MessageRole};
use crate::projection::message_window::{
    TranscriptEntry, replay_all_messages_with_seq, replay_messages_from, replay_messages_with_seq,
    replay_transcript_from,
};
use crate::stream::StreamFrame;

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct SessionId(pub Uuid);

fn is_auto_name_threshold(mut count: u64) -> bool {
    if count < 3 {
        return false;
    }
    while count % 3 == 0 {
        count /= 3;
    }
    count == 1
}

impl SessionId {
    pub fn now() -> Self {
        Self(Uuid::new_v4())
    }

    pub fn parse(s: &str) -> Result<Self, uuid::Error> {
        Uuid::parse_str(s).map(Self)
    }
}

impl std::fmt::Display for SessionId {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        self.0.fmt(f)
    }
}

type WatchKeepalive = (
    watch::Receiver<ContextSnapshot>,
    watch::Receiver<Option<String>>,
    watch::Receiver<usize>,
    watch::Receiver<Vec<crate::memory::todo::Todo>>,
    watch::Receiver<Vec<crate::memory::plan::Plan>>,
);

#[derive(Debug)]
pub struct TurnState {
    pub current_turn: Mutex<Option<TurnId>>,
    pub flow_cancel: Mutex<CancellationToken>,
    pub streamed: std::sync::atomic::AtomicBool,
}

impl TurnState {
    fn new() -> Self {
        Self {
            current_turn: Mutex::new(None),
            flow_cancel: Mutex::new(CancellationToken::new()),
            streamed: std::sync::atomic::AtomicBool::new(false),
        }
    }
}

pub struct WatchHub {
    pub stream_tx: broadcast::Sender<StreamFrame>,
    pub context: watch::Sender<ContextSnapshot>,
    pub goal: watch::Sender<Option<String>>,
    pub attach: watch::Sender<usize>,
    pub todos: watch::Sender<Vec<crate::memory::todo::Todo>>,
    pub plans: watch::Sender<Vec<crate::memory::plan::Plan>>,
    _keepalive: WatchKeepalive,
}

pub struct CompactionState {
    pub manual_pending: std::sync::atomic::AtomicBool,
    pub last_input_tokens: std::sync::atomic::AtomicU64,
    pub review_mode: Mutex<CompactReviewMode>,
    pub lock: std::sync::Arc<tokio::sync::Mutex<()>>,
}

impl CompactionState {
    fn new() -> Self {
        Self {
            manual_pending: std::sync::atomic::AtomicBool::new(false),
            last_input_tokens: std::sync::atomic::AtomicU64::new(0),
            review_mode: Mutex::new(CompactReviewMode::default()),
            lock: std::sync::Arc::new(tokio::sync::Mutex::new(())),
        }
    }
}

pub struct InteractionServices {
    pub approval: std::sync::Arc<ApprovalRegistry>,
    pub compact_reviews: std::sync::Arc<CompactReviewRegistry>,
    pub forms: std::sync::Arc<FormRegistry>,
}

impl InteractionServices {
    fn new() -> Self {
        Self {
            approval: std::sync::Arc::new(ApprovalRegistry::new()),
            compact_reviews: std::sync::Arc::new(CompactReviewRegistry::new()),
            forms: std::sync::Arc::new(FormRegistry::new()),
        }
    }
}

pub struct Session {
    id: SessionId,
    dir: PathBuf,
    writer: std::sync::Mutex<Option<EventWriter>>,
    sink: EventSink,
    message_stream: crate::message_stream::MessageStream,
    messages: std::sync::Arc<std::sync::Mutex<Vec<Message>>>,
    pub turn: TurnState,
    pub watch: WatchHub,
    pub watch_hub: std::sync::Arc<crate::watch::WatchHub>,
    pub flow_registry: std::sync::Arc<crate::tools::agent_ctrl::FlowRegistry>,
    /// Handle of the current root FlowRun; set per turn.
    current_root: std::sync::Mutex<Option<String>>,
    successful_flow_count: std::sync::atomic::AtomicU64,
    pub compaction: CompactionState,
    pub interactions: InteractionServices,
    injection_queue: Mutex<Vec<Injection>>,
    injection_tx: broadcast::Sender<Injection>,
    last_image_user_msg: Mutex<Option<LastImageUserMsg>>,
    read_files: std::sync::Arc<std::sync::Mutex<std::collections::HashSet<std::path::PathBuf>>>,
    output_store: std::sync::Arc<crate::tools::tool_output::OutputStore>,
    fs_access_mode: Mutex<Option<crate::fs_access::FsAccessMode>>,
    project_index: Option<std::sync::Arc<crate::index::AnchorIndex>>,
}

#[derive(Debug, Clone)]
pub struct PendingCompactReview {
    pub review_id: String,
    pub summary: String,
    pub slice_preview: String,
    pub slice_count: usize,
    pub range_start: usize,
    pub range_end: usize,
    pub tokens_before: u64,
    pub emitted_at: chrono::DateTime<chrono::Utc>,
}

#[derive(Debug, Clone)]
pub enum CompactReviewDecision {
    AcceptAsIs,
    AcceptEdited { summary: String },
    Reject,
}

pub struct CompactReviewRegistry {
    entry: std::sync::Mutex<Option<CompactReviewEntry>>,
    watch_tx: watch::Sender<Option<PendingCompactReview>>,
}

struct CompactReviewEntry {
    pending: PendingCompactReview,
    responder: tokio::sync::oneshot::Sender<CompactReviewDecision>,
}

impl Default for CompactReviewRegistry {
    fn default() -> Self {
        Self::new()
    }
}

impl CompactReviewRegistry {
    pub fn new() -> Self {
        let (watch_tx, _) = watch::channel(None);
        Self {
            entry: std::sync::Mutex::new(None),
            watch_tx,
        }
    }

    pub fn subscribe(&self) -> watch::Receiver<Option<PendingCompactReview>> {
        self.watch_tx.subscribe()
    }

    pub fn list_pending(&self) -> Option<PendingCompactReview> {
        self.entry
            .lock()
            .unwrap()
            .as_ref()
            .map(|e| e.pending.clone())
    }

    pub fn subscriber_count(&self) -> usize {
        self.watch_tx.receiver_count()
    }

    pub fn request(
        &self,
        pending: PendingCompactReview,
    ) -> tokio::sync::oneshot::Receiver<CompactReviewDecision> {
        let (tx, rx) = tokio::sync::oneshot::channel();
        if self.watch_tx.receiver_count() == 0 {
            let _ = tx.send(CompactReviewDecision::AcceptAsIs);
            return rx;
        }
        {
            let mut slot = self.entry.lock().unwrap();
            if let Some(prev) = slot.take() {
                let _ = prev.responder.send(CompactReviewDecision::Reject);
            }
            *slot = Some(CompactReviewEntry {
                pending: pending.clone(),
                responder: tx,
            });
        }
        let _ = self.watch_tx.send(Some(pending));
        rx
    }

    pub fn decide(&self, review_id: &str, decision: CompactReviewDecision) -> bool {
        let entry = {
            let mut slot = self.entry.lock().unwrap();
            match slot.as_ref() {
                Some(e) if e.pending.review_id == review_id => slot.take(),
                _ => None,
            }
        };
        match entry {
            Some(e) => {
                let _ = e.responder.send(decision);
                let _ = self.watch_tx.send(None);
                true
            }
            None => false,
        }
    }
}

#[derive(Debug, Clone)]
pub struct PendingApproval {
    pub tool_use_id: String,
    pub tool_name: String,
    pub args_preview: String,
    pub preview: Option<String>,
    pub level: crate::tool::ApprovalLevel,
    pub run_id: FlowRunId,
    pub emitted_at: chrono::DateTime<chrono::Utc>,
    pub bypass_auto_ceiling: bool,
}

#[derive(Debug, Clone)]
pub enum ApprovalDecision {
    Approve,
    Deny { reason: String },
}

pub struct FormRegistry {
    entries: std::sync::Mutex<Vec<FormEntry>>,
    watch_tx: watch::Sender<Vec<crate::form::PendingForm>>,
}

struct FormEntry {
    pending: crate::form::PendingForm,
    responder: tokio::sync::oneshot::Sender<crate::form::FormAnswer>,
}

impl Default for FormRegistry {
    fn default() -> Self {
        Self::new()
    }
}

impl FormRegistry {
    pub fn new() -> Self {
        let (watch_tx, _) = watch::channel(Vec::new());
        Self {
            entries: std::sync::Mutex::new(Vec::new()),
            watch_tx,
        }
    }

    pub fn subscribe(&self) -> watch::Receiver<Vec<crate::form::PendingForm>> {
        self.watch_tx.subscribe()
    }

    pub fn list_pending(&self) -> Vec<crate::form::PendingForm> {
        self.entries
            .lock()
            .unwrap()
            .iter()
            .map(|e| e.pending.clone())
            .collect()
    }

    pub fn subscriber_count(&self) -> usize {
        self.watch_tx.receiver_count()
    }

    // No TUI attached → auto-cancel so flows don't hang forever. Otherwise
    // enqueue and hand a receiver back to the caller.
    pub fn request(
        &self,
        pending: crate::form::PendingForm,
    ) -> tokio::sync::oneshot::Receiver<crate::form::FormAnswer> {
        let (tx, rx) = tokio::sync::oneshot::channel();
        if self.watch_tx.receiver_count() == 0 {
            let _ = tx.send(crate::form::FormAnswer::Cancelled);
            return rx;
        }
        {
            let mut entries = self.entries.lock().unwrap();
            entries.push(FormEntry {
                pending: pending.clone(),
                responder: tx,
            });
        }
        self.broadcast_snapshot();
        rx
    }

    pub fn submit(&self, form_id: &str, answer: crate::form::FormAnswer) -> bool {
        let entry = {
            let mut entries = self.entries.lock().unwrap();
            let pos = entries.iter().position(|e| e.pending.form_id == form_id);
            pos.map(|p| entries.remove(p))
        };
        match entry {
            Some(e) => {
                let _ = e.responder.send(answer);
                self.broadcast_snapshot();
                true
            }
            None => false,
        }
    }

    pub fn cancel_all(&self) {
        let drained: Vec<FormEntry> = {
            let mut entries = self.entries.lock().unwrap();
            std::mem::take(&mut *entries)
        };
        for e in drained {
            let _ = e.responder.send(crate::form::FormAnswer::Cancelled);
        }
        self.broadcast_snapshot();
    }

    pub fn promote(&self, form_id: &str) {
        let mut entries = self.entries.lock().unwrap();
        if let Some(pos) = entries.iter().position(|e| e.pending.form_id == form_id) {
            if pos == 0 {
                return;
            }
            let entry = entries.remove(pos);
            entries.insert(0, entry);
        }
        drop(entries);
        self.broadcast_snapshot();
    }

    fn broadcast_snapshot(&self) {
        let snap = self
            .entries
            .lock()
            .unwrap()
            .iter()
            .map(|e| e.pending.clone())
            .collect();
        let _ = self.watch_tx.send(snap);
    }
}

pub struct ApprovalRegistry {
    entries: std::sync::Mutex<Vec<ApprovalEntry>>,
    auto_ceiling: std::sync::Mutex<crate::tool::ApprovalLevel>,
    watch_tx: watch::Sender<Vec<PendingApproval>>,
}

struct ApprovalEntry {
    pending: PendingApproval,
    responder: tokio::sync::oneshot::Sender<ApprovalDecision>,
}

impl Default for ApprovalRegistry {
    fn default() -> Self {
        Self::new()
    }
}

impl ApprovalRegistry {
    pub fn new() -> Self {
        let (watch_tx, _) = watch::channel(Vec::new());
        Self {
            entries: std::sync::Mutex::new(Vec::new()),
            auto_ceiling: std::sync::Mutex::new(crate::tool::ApprovalLevel::Approve),
            watch_tx,
        }
    }

    pub fn subscribe(&self) -> watch::Receiver<Vec<PendingApproval>> {
        self.watch_tx.subscribe()
    }

    pub fn list_pending(&self) -> Vec<PendingApproval> {
        self.entries
            .lock()
            .unwrap()
            .iter()
            .map(|e| e.pending.clone())
            .collect()
    }

    pub fn set_auto_ceiling(&self, level: crate::tool::ApprovalLevel) {
        *self.auto_ceiling.lock().unwrap() = level;
    }

    pub fn request(
        &self,
        pending: PendingApproval,
    ) -> tokio::sync::oneshot::Receiver<ApprovalDecision> {
        let (tx, rx) = tokio::sync::oneshot::channel();
        if !pending.bypass_auto_ceiling && pending.level <= *self.auto_ceiling.lock().unwrap() {
            let _ = tx.send(ApprovalDecision::Approve);
            return rx;
        }
        {
            let mut entries = self.entries.lock().unwrap();
            entries.push(ApprovalEntry {
                pending,
                responder: tx,
            });
        }
        self.broadcast_snapshot();
        rx
    }

    pub fn decide(&self, tool_use_id: &str, decision: ApprovalDecision) -> bool {
        let mut entries = self.entries.lock().unwrap();
        if let Some(pos) = entries
            .iter()
            .position(|e| e.pending.tool_use_id == tool_use_id)
        {
            let entry = entries.remove(pos);
            let _ = entry.responder.send(decision);
            drop(entries);
            self.broadcast_snapshot();
            true
        } else {
            false
        }
    }

    pub fn decide_all(&self, decision: ApprovalDecision) -> usize {
        let mut entries = self.entries.lock().unwrap();
        let count = entries.len();
        for entry in entries.drain(..) {
            let _ = entry.responder.send(decision.clone());
        }
        drop(entries);
        self.broadcast_snapshot();
        count
    }

    fn broadcast_snapshot(&self) {
        let snapshot = self
            .entries
            .lock()
            .unwrap()
            .iter()
            .map(|e| e.pending.clone())
            .collect();
        let _ = self.watch_tx.send(snapshot);
    }
}
type ImagePart = (usize, String);

#[derive(Debug, Clone)]
struct LastImageUserMsg {
    message_seq: u64,
    images: Vec<ImagePart>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum CompactReviewMode {
    Always,
    #[default]
    ManualOnly,
    Never,
}

impl CompactReviewMode {
    pub fn parse(s: &str) -> Option<Self> {
        match s.trim() {
            "always" => Some(Self::Always),
            "manual-only" | "manual_only" => Some(Self::ManualOnly),
            "never" => Some(Self::Never),
            _ => None,
        }
    }

    pub fn should_review(self, forced: bool) -> bool {
        match self {
            Self::Always => true,
            Self::ManualOnly => forced,
            Self::Never => false,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CompactResult {
    pub before_tokens: u64,
    pub after_tokens: u64,
    pub compacted_start: usize,
    pub compacted_end: usize,
}

#[derive(Debug, Clone, Default, PartialEq)]
pub struct ContextSnapshot {
    pub model: String,
    pub tokens_in: u64,
    pub tokens_out: u64,
    pub cost_usd: f64,
    pub mcp_servers: Vec<crate::mcp::McpServerStatus>,
    pub memory_recent_count: u16,
    pub window_tokens: u64,
    pub window_budget: u64,
    pub cache_read: u64,
    pub cache_write: u64,
    pub last_ttft_ms: u64,
    pub last_tokens_per_sec: f64,
}

#[derive(Debug, thiserror::Error)]
pub enum SessionOpenError {
    #[error("invalid session id `{sid}` (want a UUID)")]
    InvalidId { sid: String },
    #[error("session `{sid}` not found at {}", dir.display())]
    NotFound { sid: String, dir: PathBuf },
    #[error("session writer init: {0}")]
    WriterInit(#[source] std::io::Error),
    #[error("replay {}: {source}", path.display())]
    Replay {
        path: PathBuf,
        #[source]
        source: std::io::Error,
    },
}

fn load_goal(dir: &Path) -> Option<String> {
    if dir.as_os_str().is_empty() {
        return None;
    }
    let store = crate::memory::goal::GoalStore::at(dir);
    match store.get() {
        Ok(s) if !s.is_empty() => Some(s),
        _ => None,
    }
}

#[derive(serde::Serialize, serde::Deserialize, Default)]
struct PersistedContextState {
    #[serde(default)]
    model: String,
    #[serde(default)]
    window_tokens: u64,
    #[serde(default)]
    window_budget: u64,
}

impl PersistedContextState {
    fn path(dir: &Path) -> PathBuf {
        dir.join("context_state.json")
    }

    fn load(dir: &Path) -> Self {
        match std::fs::read_to_string(Self::path(dir)) {
            Ok(text) => serde_json::from_str(&text).unwrap_or_default(),
            Err(_) => Self::default(),
        }
    }

    fn save(&self, dir: &Path) {
        if dir.as_os_str().is_empty() {
            return;
        }
        if let Ok(json) = serde_json::to_string_pretty(self) {
            let _ = std::fs::write(Self::path(dir), &json);
        }
    }
}

fn default_project_index(root: &Path) -> Option<std::sync::Arc<crate::index::AnchorIndex>> {
    match crate::index::AnchorIndex::open_project(root) {
        Ok(idx) => Some(std::sync::Arc::new(idx)),
        Err(e) => {
            crate::notify!(
                warn,
                "project index unavailable at {} — history search disabled: {e}",
                root.display()
            );
            None
        }
    }
}

impl Session {
    pub fn open(root: impl AsRef<Path>) -> std::io::Result<Self> {
        Self::open_with_redactor(root, None)
    }

    pub fn open_with_redactor(
        root: impl AsRef<Path>,
        redactor: Option<std::sync::Arc<crate::redact::Redactor>>,
    ) -> std::io::Result<Self> {
        let root_ref = root.as_ref();
        let project_index = default_project_index(root_ref);
        Self::open_with_context(root_ref, redactor, project_index)
    }

    pub fn open_with_context(
        root: impl AsRef<Path>,
        redactor: Option<std::sync::Arc<crate::redact::Redactor>>,
        project_index: Option<std::sync::Arc<crate::index::AnchorIndex>>,
    ) -> std::io::Result<Self> {
        let id = SessionId::now();
        let dir = root.as_ref().join("sessions").join(id.to_string());
        if let Some(ls) = crate::notify::log_sink() {
            ls.set_session_id(Some(id.to_string()));
        }
        let writer = EventWriter::spawn_full(
            &dir,
            redactor.clone(),
            project_index.clone(),
            Some(id.to_string()),
        )?;
        if let Err(e) = crate::session_meta::SessionMeta::from_cwd().save(&dir) {
            crate::notify!(error, "session meta write failed: {e}");
        }
        let mut sink = EventSink::new().with_forwarder(writer.sender());
        if let Some(r) = redactor {
            sink = sink.with_redactor(r);
        }
        let (injection_tx, _) = broadcast::channel(32);
        let (stream_tx, _) = broadcast::channel(2048);
        let (context_watch, context_rx) = watch::channel(ContextSnapshot::default());
        let (goal_watch, goal_rx) = watch::channel(None);
        let (attach_watch, attach_rx) = watch::channel(0);
        let (todos_watch, todos_rx) = watch::channel(Vec::new());
        let (plans_watch, plans_rx) = watch::channel(Vec::new());
        let events_handle = sink.events_handle();
        let output_store = std::sync::Arc::new(crate::tools::tool_output::OutputStore::at(&dir));
        Ok(Self {
            id,
            dir,
            writer: std::sync::Mutex::new(Some(writer)),
            sink,
            message_stream: crate::message_stream::MessageStream::new(events_handle),
            messages: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
            output_store: output_store.clone(),
            turn: TurnState::new(),
            watch: WatchHub {
                stream_tx,
                context: context_watch,
                goal: goal_watch,
                attach: attach_watch,
                todos: todos_watch,
                plans: plans_watch,
                _keepalive: (context_rx, goal_rx, attach_rx, todos_rx, plans_rx),
            },
            watch_hub: std::sync::Arc::new(crate::watch::WatchHub::new()),
            flow_registry: std::sync::Arc::new(crate::tools::agent_ctrl::FlowRegistry::new()),
            current_root: std::sync::Mutex::new(None),
            successful_flow_count: std::sync::atomic::AtomicU64::new(0),
            compaction: CompactionState::new(),
            interactions: InteractionServices::new(),
            injection_queue: Mutex::new(Vec::new()),
            injection_tx,
            last_image_user_msg: Mutex::new(None),
            read_files: std::sync::Arc::new(
                std::sync::Mutex::new(std::collections::HashSet::new()),
            ),
            fs_access_mode: Mutex::new(None),
            project_index,
        })
    }

    pub fn open_existing(root: impl AsRef<Path>, sid: &str) -> Result<Self, SessionOpenError> {
        Self::open_existing_with_redactor(root, sid, None)
    }

    pub fn open_existing_with_redactor(
        root: impl AsRef<Path>,
        sid: &str,
        redactor: Option<std::sync::Arc<crate::redact::Redactor>>,
    ) -> Result<Self, SessionOpenError> {
        let project_index = default_project_index(root.as_ref());
        Self::open_existing_with_context(root, sid, redactor, project_index)
    }

    pub fn open_existing_with_context(
        root: impl AsRef<Path>,
        sid: &str,
        redactor: Option<std::sync::Arc<crate::redact::Redactor>>,
        project_index: Option<std::sync::Arc<crate::index::AnchorIndex>>,
    ) -> Result<Self, SessionOpenError> {
        let id = SessionId::parse(sid).map_err(|_| SessionOpenError::InvalidId {
            sid: sid.to_string(),
        })?;
        let dir = root.as_ref().join("sessions").join(id.to_string());
        if let Some(ls) = crate::notify::log_sink() {
            ls.set_session_id(Some(id.to_string()));
        }
        if !dir.exists() {
            return Err(SessionOpenError::NotFound {
                sid: sid.to_string(),
                dir: dir.clone(),
            });
        }
        let writer = EventWriter::spawn_full(
            &dir,
            redactor.clone(),
            project_index.clone(),
            Some(id.to_string()),
        )
        .map_err(SessionOpenError::WriterInit)?;
        let mut sink = EventSink::new().with_forwarder(writer.sender());
        if let Some(r) = redactor {
            sink = sink.with_redactor(r);
        }
        let events_path = dir.join("events.jsonl");
        let messages = replay_messages_from(&events_path)?;
        let initial_msgs = replay_messages_with_seq(&events_path)?;
        let all_msgs = replay_all_messages_with_seq(&events_path)?;
        if let Some(last_seq) = find_last_seq(&events_path)? {
            sink.restore_seq(last_seq);
        }
        let mut initial_context = replay_context_snapshot_from(&events_path);
        let persisted = PersistedContextState::load(&dir);
        if !persisted.model.is_empty() {
            initial_context.model = persisted.model;
        }
        initial_context.window_tokens = persisted.window_tokens;
        initial_context.window_budget = persisted.window_budget;
        let initial_goal = load_goal(&dir);
        let (injection_tx, _) = broadcast::channel(32);
        let (stream_tx, _) = broadcast::channel(2048);
        let (context_watch, context_rx) = watch::channel(initial_context);
        let (goal_watch, goal_rx) = watch::channel(initial_goal);
        let (attach_watch, attach_rx) = watch::channel(0);
        let (todos_watch, todos_rx) = watch::channel(Vec::new());
        let (plans_watch, plans_rx) = watch::channel(Vec::new());
        let events_handle = sink.events_handle();
        let output_store = std::sync::Arc::new(crate::tools::tool_output::OutputStore::at(&dir));
        Ok(Self {
            id,
            dir,
            writer: std::sync::Mutex::new(Some(writer)),
            sink,
            message_stream: crate::message_stream::MessageStream::with_initial(
                events_handle,
                initial_msgs,
                all_msgs,
            ),
            messages: std::sync::Arc::new(std::sync::Mutex::new(messages)),
            output_store: output_store.clone(),
            turn: TurnState::new(),
            watch: WatchHub {
                stream_tx,
                context: context_watch,
                goal: goal_watch,
                attach: attach_watch,
                todos: todos_watch,
                plans: plans_watch,
                _keepalive: (context_rx, goal_rx, attach_rx, todos_rx, plans_rx),
            },
            watch_hub: std::sync::Arc::new(crate::watch::WatchHub::new()),
            flow_registry: std::sync::Arc::new(crate::tools::agent_ctrl::FlowRegistry::new()),
            current_root: std::sync::Mutex::new(None),
            successful_flow_count: std::sync::atomic::AtomicU64::new(0),
            compaction: {
                let c = CompactionState::new();
                if persisted.window_tokens > 0 {
                    c.last_input_tokens.store(
                        persisted.window_tokens,
                        std::sync::atomic::Ordering::Relaxed,
                    );
                }
                c
            },
            interactions: InteractionServices::new(),
            injection_queue: Mutex::new(Vec::new()),
            injection_tx,
            last_image_user_msg: Mutex::new(None),
            read_files: std::sync::Arc::new(
                std::sync::Mutex::new(std::collections::HashSet::new()),
            ),
            fs_access_mode: Mutex::new(None),
            project_index,
        })
    }

    pub fn open_ephemeral() -> Self {
        let (injection_tx, _) = broadcast::channel(32);
        let (stream_tx, _) = broadcast::channel(2048);
        let (context_watch, context_rx) = watch::channel(ContextSnapshot::default());
        let (goal_watch, goal_rx) = watch::channel(None);
        let (attach_watch, attach_rx) = watch::channel(0);
        let (todos_watch, todos_rx) = watch::channel(Vec::new());
        let (plans_watch, plans_rx) = watch::channel(Vec::new());
        let sink = EventSink::new();
        let events_handle = sink.events_handle();
        let output_store = std::sync::Arc::new(crate::tools::tool_output::OutputStore::default());
        Self {
            id: SessionId::now(),
            dir: PathBuf::new(),
            writer: std::sync::Mutex::new(None),
            sink,
            message_stream: crate::message_stream::MessageStream::new(events_handle),
            messages: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
            output_store: output_store.clone(),
            turn: TurnState::new(),
            watch: WatchHub {
                stream_tx,
                context: context_watch,
                goal: goal_watch,
                attach: attach_watch,
                todos: todos_watch,
                plans: plans_watch,
                _keepalive: (context_rx, goal_rx, attach_rx, todos_rx, plans_rx),
            },
            watch_hub: std::sync::Arc::new(crate::watch::WatchHub::new()),
            flow_registry: std::sync::Arc::new(crate::tools::agent_ctrl::FlowRegistry::new()),
            current_root: std::sync::Mutex::new(None),
            successful_flow_count: std::sync::atomic::AtomicU64::new(0),
            compaction: CompactionState::new(),
            interactions: InteractionServices::new(),
            injection_queue: Mutex::new(Vec::new()),
            injection_tx,
            last_image_user_msg: Mutex::new(None),
            read_files: std::sync::Arc::new(
                std::sync::Mutex::new(std::collections::HashSet::new()),
            ),
            fs_access_mode: Mutex::new(None),
            project_index: None,
        }
    }

    pub fn project_index(&self) -> Option<std::sync::Arc<crate::index::AnchorIndex>> {
        self.project_index.clone()
    }

    pub fn approval(&self) -> std::sync::Arc<ApprovalRegistry> {
        self.interactions.approval.clone()
    }

    pub fn compact_reviews(&self) -> std::sync::Arc<CompactReviewRegistry> {
        self.interactions.compact_reviews.clone()
    }

    pub fn forms(&self) -> std::sync::Arc<FormRegistry> {
        self.interactions.forms.clone()
    }

    pub fn fs_access_mode(&self) -> Option<crate::fs_access::FsAccessMode> {
        *self.fs_access_mode.lock().unwrap()
    }

    pub fn set_fs_access_mode(&self, mode: crate::fs_access::FsAccessMode) {
        *self.fs_access_mode.lock().unwrap() = Some(mode);
    }

    pub fn compact_review_mode(&self) -> CompactReviewMode {
        *self.compaction.review_mode.lock().unwrap()
    }

    pub fn set_compact_review_mode(&self, mode: CompactReviewMode) {
        *self.compaction.review_mode.lock().unwrap() = mode;
    }

    pub fn read_files(
        &self,
    ) -> std::sync::Arc<std::sync::Mutex<std::collections::HashSet<std::path::PathBuf>>> {
        self.read_files.clone()
    }

    pub fn output_store(&self) -> std::sync::Arc<crate::tools::tool_output::OutputStore> {
        self.output_store.clone()
    }

    pub fn mark_file_read(&self, path: &std::path::Path) {
        if let Ok(mut set) = self.read_files.lock() {
            set.insert(path.to_path_buf());
            if let Ok(canonical) = std::fs::canonicalize(path) {
                set.insert(canonical);
            }
        }
    }

    pub fn stream_tx(&self) -> broadcast::Sender<StreamFrame> {
        self.watch.stream_tx.clone()
    }

    pub fn set_current_root(&self, handle: String) {
        *self.current_root.lock().unwrap() = Some(handle);
    }

    pub fn current_root(&self) -> Option<String> {
        self.current_root.lock().unwrap().clone()
    }

    pub fn clear_current_root(&self) {
        *self.current_root.lock().unwrap() = None;
    }

    pub fn record_successful_flow(&self) -> Option<u64> {
        let count = self
            .successful_flow_count
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed)
            + 1;
        is_auto_name_threshold(count).then_some(count)
    }

    pub fn successful_flow_count(&self) -> u64 {
        self.successful_flow_count
            .load(std::sync::atomic::Ordering::Relaxed)
    }

    pub fn stream_subscribe(&self) -> broadcast::Receiver<StreamFrame> {
        self.watch.stream_tx.subscribe()
    }

    pub fn id(&self) -> &SessionId {
        &self.id
    }

    pub fn dir(&self) -> &Path {
        &self.dir
    }

    pub fn transcript_replay(&self) -> Vec<TranscriptEntry> {
        let Some(path) = self.events_path() else {
            return Vec::new();
        };
        replay_transcript_from(&path).unwrap_or_default()
    }

    pub fn events_path(&self) -> Option<std::path::PathBuf> {
        self.writer
            .lock()
            .unwrap()
            .as_ref()
            .map(|w| w.events_path().to_path_buf())
    }

    pub async fn plan_system_prompt(&self) -> Option<String> {
        let store = crate::memory::plan::PlanStore::at(&self.dir);
        let plan = store.latest().await.ok().flatten()?;
        Some(crate::tools::plan::render_plan(&plan))
    }

    pub fn goal(&self) -> Option<String> {
        if let Some(cached) = self.watch.goal.borrow().clone() {
            return Some(cached);
        }
        load_goal(&self.dir)
    }

    pub fn subscribe_goal(&self) -> watch::Receiver<Option<String>> {
        self.watch.goal.subscribe()
    }

    pub fn goal_watch(&self) -> &watch::Sender<Option<String>> {
        &self.watch.goal
    }

    pub fn subscribe_context(&self) -> watch::Receiver<ContextSnapshot> {
        self.watch.context.subscribe()
    }

    pub fn subscribe_attach(&self) -> watch::Receiver<usize> {
        self.watch.attach.subscribe()
    }

    pub fn subscribe_pending_approvals(&self) -> watch::Receiver<Vec<PendingApproval>> {
        self.interactions.approval.subscribe()
    }

    pub fn meta(&self) -> Option<crate::session_meta::SessionMeta> {
        crate::session_meta::SessionMeta::load(&self.dir)
    }

    pub fn request_manual_compact(&self) {
        self.compaction
            .manual_pending
            .store(true, std::sync::atomic::Ordering::SeqCst);
    }

    pub fn take_manual_compact_request(&self) -> bool {
        self.compaction
            .manual_pending
            .swap(false, std::sync::atomic::Ordering::SeqCst)
    }

    pub fn set_goal(&self, goal: Option<String>) {
        let _ = self.watch.goal.send(goal);
    }

    pub fn set_attach_count(&self, count: usize) {
        let _ = self.watch.attach.send(count);
    }

    #[allow(clippy::too_many_arguments)]
    pub fn record_llm_call(
        &self,
        model: &str,
        tokens_in: u64,
        tokens_out: u64,
        cache_read: u64,
        cache_write: u64,
        ttft_ms: Option<u64>,
        tokens_per_sec: Option<f64>,
    ) {
        if tokens_in > 0 {
            self.compaction
                .last_input_tokens
                .store(tokens_in, std::sync::atomic::Ordering::Relaxed);
        }
        self.watch.context.send_modify(|snap| {
            snap.model = model.to_string();
            snap.tokens_in = snap.tokens_in.saturating_add(tokens_in);
            snap.tokens_out = snap.tokens_out.saturating_add(tokens_out);
            snap.cache_read = snap.cache_read.saturating_add(cache_read);
            snap.cache_write = snap.cache_write.saturating_add(cache_write);
            snap.last_ttft_ms = ttft_ms.unwrap_or(0);
            snap.last_tokens_per_sec = tokens_per_sec.unwrap_or(0.0);
        });
        self.refresh_window_snapshot();
    }

    pub fn last_input_tokens(&self) -> u64 {
        self.compaction
            .last_input_tokens
            .load(std::sync::atomic::Ordering::Relaxed)
    }

    pub async fn acquire_compact_lock(&self) -> tokio::sync::MutexGuard<'_, ()> {
        self.compaction.lock.lock().await
    }

    pub async fn acquire_compact_lock_owned(&self) -> tokio::sync::OwnedMutexGuard<()> {
        self.compaction.lock.clone().lock_owned().await
    }

    pub fn compact_lock_handle(&self) -> std::sync::Arc<tokio::sync::Mutex<()>> {
        self.compaction.lock.clone()
    }

    pub fn refresh_window_snapshot(&self) {
        let provider_tokens = self.last_input_tokens();
        let estimated = crate::compaction::estimate_tokens_for_messages(&self.messages());
        let window = if provider_tokens > 0 {
            provider_tokens
        } else {
            estimated
        };
        let model = self.last_model();
        let budget = crate::model_registry::model_info(&model).context_budget;
        self.watch.context.send_modify(|snap| {
            snap.window_tokens = window;
            if budget > 0 {
                snap.window_budget = budget;
            }
        });
        let snap = self.watch.context.borrow();
        PersistedContextState {
            model,
            window_tokens: snap.window_tokens,
            window_budget: snap.window_budget,
        }
        .save(&self.dir);
    }

    pub fn cumulative_input_tokens(&self) -> u64 {
        self.watch.context.borrow().tokens_in
    }

    pub fn reset_input_tokens_to(&self, tokens: u64) {
        self.watch.context.send_modify(|snap| {
            snap.tokens_in = tokens;
        });
    }

    pub fn last_model(&self) -> String {
        self.watch.context.borrow().model.clone()
    }

    pub fn set_current_model(&self, model: impl Into<String>) {
        let model = model.into();
        let budget = crate::model_registry::model_info(&model).context_budget;
        self.watch.context.send_modify(|snap| {
            snap.model = model.clone();
            if budget > 0 {
                snap.window_budget = budget;
            }
        });
        let snap = self.watch.context.borrow();
        PersistedContextState {
            model,
            window_tokens: snap.window_tokens,
            window_budget: snap.window_budget,
        }
        .save(&self.dir);
    }

    pub fn update_mcp_server(&self, status: crate::mcp::McpServerStatus) {
        self.watch.context.send_modify(|snap| {
            if let Some(existing) = snap.mcp_servers.iter_mut().find(|s| s.name == status.name) {
                *existing = status;
            } else {
                snap.mcp_servers.push(status);
            }
        });
    }

    pub fn set_memory_recent_count(&self, count: u16) {
        self.watch.context.send_modify(|snap| {
            snap.memory_recent_count = count;
        });
    }

    pub fn subscribe_todos(&self) -> watch::Receiver<Vec<crate::memory::todo::Todo>> {
        self.watch.todos.subscribe()
    }

    pub fn todos_watch(&self) -> &watch::Sender<Vec<crate::memory::todo::Todo>> {
        &self.watch.todos
    }

    pub fn subscribe_plans(&self) -> watch::Receiver<Vec<crate::memory::plan::Plan>> {
        self.watch.plans.subscribe()
    }

    pub fn plans_watch(&self) -> &watch::Sender<Vec<crate::memory::plan::Plan>> {
        &self.watch.plans
    }

    pub async fn refresh_plans_from_store_async(&self) {
        if self.dir.as_os_str().is_empty() {
            return;
        }
        let store = crate::memory::plan::PlanStore::at(&self.dir);
        match store.list().await {
            Ok(list) => {
                let _ = self.watch.plans.send(list);
            }
            Err(e) => {
                crate::notify!(
                    warn,
                    location = Log,
                    stack = dedupe("memory.refresh_plans_async", 60_000),
                    "refresh_plans_from_store_async: {e}"
                );
            }
        }
    }

    pub fn refresh_todos_from_store(&self) {
        if self.dir.as_os_str().is_empty() {
            return;
        }
        let store = crate::memory::todo::TodoStore::at(&self.dir);
        match tokio::task::block_in_place(|| {
            tokio::runtime::Handle::try_current()
                .ok()
                .map(|h| h.block_on(store.list()))
        }) {
            Some(Ok(list)) => {
                let _ = self.watch.todos.send(list);
            }
            Some(Err(e)) => {
                crate::notify!(
                    warn,
                    location = Log,
                    stack = dedupe("memory.refresh_todos", 60_000),
                    "refresh_todos_from_store: {e}"
                );
            }
            None => {}
        }
    }

    pub async fn refresh_todos_from_store_async(&self) {
        if self.dir.as_os_str().is_empty() {
            return;
        }
        let store = crate::memory::todo::TodoStore::at(&self.dir);
        match store.list().await {
            Ok(list) => {
                let _ = self.watch.todos.send(list);
            }
            Err(e) => {
                crate::notify!(
                    warn,
                    location = Log,
                    stack = dedupe("memory.refresh_todos_async", 60_000),
                    "refresh_todos_from_store_async: {e}"
                );
            }
        }
    }

    pub fn sink(&self) -> &EventSink {
        &self.sink
    }

    /// Single-writer append. Emits the matching event before the in-memory push
    /// so events.jsonl remains the authority (§I5).
    pub fn append_message(&self, msg: Message, flow_run_id: Option<FlowRunId>) {
        AppendMessageCommand { msg, flow_run_id }.execute(self);
    }

    pub fn emit_attachment_degrade(
        &self,
        message_seq: u64,
        part_index: usize,
        file_basename: String,
        reason: String,
    ) {
        self.sink.emit(Event::AttachmentDegraded {
            turn_id: None,
            flow_run_id: None,
            message_seq,
            part_index,
            file_basename,
            reason,
        });
    }

    pub fn record_attachment_degrade(&self, reason: &str) -> usize {
        let target = self.last_image_user_msg.lock().unwrap().take();
        let Some(entry) = target else {
            return 0;
        };
        let turn_id = self.turn.current_turn.lock().unwrap().clone();
        for (part_index, basename) in &entry.images {
            self.sink.emit(Event::AttachmentDegraded {
                turn_id: turn_id.clone(),
                flow_run_id: None,
                message_seq: entry.message_seq,
                part_index: *part_index,
                file_basename: basename.clone(),
                reason: reason.into(),
            });
        }
        if let Ok(mut msgs) = self.messages.lock() {
            for m in msgs.iter_mut() {
                for (part_index, basename) in &entry.images {
                    if let Some(part) = m.parts.get_mut(*part_index)
                        && matches!(part, crate::message::MessagePart::Image { .. })
                    {
                        *part = crate::message::MessagePart::Text {
                            text: format!("[attachment unavailable: {basename} — {reason}]"),
                        };
                    }
                }
            }
        }
        entry.images.len()
    }

    pub fn messages(&self) -> crate::message_stream::MessageWindow {
        self.message_stream.window()
    }

    pub fn messages_full(&self) -> std::sync::Arc<Vec<Message>> {
        self.message_stream.full_messages()
    }

    pub fn messages_handle(&self) -> std::sync::Arc<std::sync::Mutex<Vec<Message>>> {
        self.messages.clone()
    }

    pub fn message_count(&self) -> usize {
        self.messages().len()
    }

    pub fn user_message_count(&self) -> usize {
        self.messages()
            .iter()
            .filter(|m| matches!(m.role, MessageRole::User))
            .count()
    }

    pub fn push_system_note(&self, text: String) {
        let _ = self
            .watch
            .stream_tx
            .send(crate::stream::StreamFrame::Note(text));
    }

    pub fn approval_cooldown_ok_for_compact(&self) -> bool {
        self.sink.last_compact_ago_seconds().is_none_or(|s| s >= 60)
    }

    pub fn emit_compact_warning(
        &self,
        model: &str,
        current_tokens: u64,
        threshold: u64,
        budget: u64,
        reason: &str,
    ) {
        let message = format!(
            "context {current_tokens} > threshold {threshold} (budget {budget}, model {model}); skipping compaction: {reason}"
        );
        self.sink.emit(Event::WatchWarn {
            turn_id: self.turn.current_turn.lock().unwrap().clone(),
            flow_run_id: None,
            target: "context.compaction".into(),
            trigger: "auto_compact".into(),
            message,
        });
        self.push_system_note(format!("[warn] compaction skipped: {reason}"));
    }

    /// Convenience wrapper that computes the compact range and token count
    /// from the current message window. Used by tests and internal callers
    /// that don't already have a pre-computed range.
    pub fn compact_messages_auto(&self, summary: String) -> Option<CompactResult> {
        let msgs = self.messages();
        let tokens = crate::compaction::estimate_tokens_for_messages(&msgs);
        let info = crate::model_registry::model_info(&self.last_model());
        let target = info.compaction_target_after();
        let range = crate::compaction::find_compact_range(&msgs, target)?;
        self.compact_messages(summary, range, tokens)
    }

    pub fn commit_rewritten_window(
        &self,
        replacement: Vec<Message>,
        before_tokens: u64,
        before_window_tokens: u64,
        rewritten_count: usize,
    ) -> Option<CompactResult> {
        let after_tokens = crate::compaction::estimate_tokens_for_messages(&replacement);
        if rewritten_count == 0 || after_tokens >= before_window_tokens {
            return None;
        }
        let summary = format!(
            "[atman: persistently compacted output from {rewritten_count} retained messages]"
        );
        self.sink.mark_compacted();
        self.sink.emit(Event::ContextCompact {
            session_id: self.id.to_string(),
            before_tokens,
            after_tokens,
            compacted_range_start: 0,
            compacted_range_end: 0,
            summary_text: Some(summary.clone()),
            replacement_msg_seq: None,
        });
        self.sink.emit(Event::CompactionSummary {
            session_id: self.id.to_string(),
            range_start: 0,
            range_end: 0,
            compacted_count: rewritten_count,
            before_tokens,
            after_tokens,
            summary: summary.clone(),
        });
        let _ = self
            .watch
            .stream_tx
            .send(crate::stream::StreamFrame::CompactionSummary {
                phase: crate::stream::CompactionPhase::Finished,
                range_start: 0,
                range_end: 0,
                summary,
                before_tokens,
                after_tokens,
                compacted_count: rewritten_count,
            });
        self.compaction
            .last_input_tokens
            .store(after_tokens, std::sync::atomic::Ordering::Relaxed);
        if let Ok(mut messages) = self.messages.lock() {
            *messages = replacement.clone();
        }
        self.sink.emit(Event::Checkpoint {
            session_id: self.id.to_string(),
            messages: replacement,
            window_tokens: after_tokens,
        });
        self.refresh_window_snapshot();
        Some(CompactResult {
            before_tokens,
            after_tokens,
            compacted_start: 0,
            compacted_end: 0,
        })
    }

    pub fn commit_compacted_window(
        &self,
        summary: String,
        replacement: Vec<Message>,
        range: crate::compaction::CompactRange,
        before_tokens: u64,
        before_window_tokens: u64,
    ) -> Option<CompactResult> {
        let after_tokens = crate::compaction::estimate_tokens_for_messages(&replacement);
        if after_tokens >= before_window_tokens {
            self.push_system_note(format!(
                "compaction skipped: replacement would not shrink transcript ({} >= {} tokens)",
                after_tokens, before_window_tokens
            ));
            return None;
        }
        self.sink.mark_compacted();
        self.sink.emit(Event::ContextCompact {
            session_id: self.id.to_string(),
            before_tokens,
            after_tokens,
            compacted_range_start: range.start as u64,
            compacted_range_end: range.end.saturating_sub(1) as u64,
            summary_text: Some(summary.clone()),
            replacement_msg_seq: None,
        });
        self.sink.emit(Event::CompactionSummary {
            session_id: self.id.to_string(),
            range_start: range.start as u64,
            range_end: range.end.saturating_sub(1) as u64,
            compacted_count: range.end - range.start,
            before_tokens,
            after_tokens,
            summary: summary.clone(),
        });
        let _ = self
            .watch
            .stream_tx
            .send(crate::stream::StreamFrame::CompactionSummary {
                phase: crate::stream::CompactionPhase::Finished,
                range_start: range.start,
                range_end: range.end.saturating_sub(1),
                summary,
                before_tokens,
                after_tokens,
                compacted_count: range.end - range.start,
            });
        self.compaction
            .last_input_tokens
            .store(after_tokens, std::sync::atomic::Ordering::Relaxed);
        if let Ok(mut messages) = self.messages.lock() {
            *messages = replacement.clone();
        }
        self.sink.emit(Event::Checkpoint {
            session_id: self.id.to_string(),
            messages: replacement,
            window_tokens: after_tokens,
        });
        self.refresh_window_snapshot();
        Some(CompactResult {
            before_tokens,
            after_tokens,
            compacted_start: range.start,
            compacted_end: range.end,
        })
    }

    pub fn compact_messages(
        &self,
        summary: String,
        range: crate::compaction::CompactRange,
        before_tokens: u64,
    ) -> Option<CompactResult> {
        use crate::compaction::{estimate_tokens_for_messages, replace_range_with_summary};
        let msgs = self.messages();
        let turn_id = msgs
            .get(range.start)
            .map(|m| m.turn_id.clone())
            .unwrap_or_else(TurnId::now);
        let after = replace_range_with_summary(&msgs, &range, summary.clone(), turn_id.clone());
        let after_tokens = estimate_tokens_for_messages(&after);
        if after_tokens >= before_tokens {
            self.push_system_note(format!(
                "compaction skipped: summary would not shrink transcript ({} >= {} tokens)",
                after_tokens, before_tokens
            ));
            return None;
        }
        let replacement_msg = after.first().cloned().unwrap_or_else(|| {
            Message::system_compact_summary(
                turn_id.clone(),
                summary.clone(),
                range.start as u64,
                range.end.saturating_sub(1) as u64,
                range.end - range.start,
            )
        });
        self.sink.mark_compacted();
        let replacement_seq = self.sink.next_seq_peek();
        self.sink.emit(Event::SystemMsg {
            turn_id: turn_id.clone(),
            message: replacement_msg,
        });
        self.sink.emit(Event::ContextCompact {
            session_id: self.id.to_string(),
            before_tokens,
            after_tokens,
            compacted_range_start: range.start as u64,
            compacted_range_end: range.end.saturating_sub(1) as u64,
            summary_text: Some(summary.clone()),
            replacement_msg_seq: Some(replacement_seq),
        });
        self.sink.emit(Event::CompactionSummary {
            session_id: self.id.to_string(),
            range_start: range.start as u64,
            range_end: range.end.saturating_sub(1) as u64,
            compacted_count: range.end - range.start,
            before_tokens,
            after_tokens,
            summary: summary.clone(),
        });
        let _ = self
            .watch
            .stream_tx
            .send(crate::stream::StreamFrame::CompactionSummary {
                phase: crate::stream::CompactionPhase::Finished,
                range_start: range.start,
                range_end: range.end.saturating_sub(1),
                summary,
                before_tokens,
                after_tokens,
                compacted_count: range.end - range.start,
            });
        let checkpoint_messages = self.messages();
        let window_tokens = estimate_tokens_for_messages(&checkpoint_messages);
        self.compaction
            .last_input_tokens
            .store(window_tokens, std::sync::atomic::Ordering::Relaxed);
        // Sync the messages Vec (root's messages_handle) with the compacted
        // windowed view so root's llm context via messages_handle respects
        // compaction (branch2 retired → unified on messages_handle).
        let window_owned = checkpoint_messages.to_vec();
        if let Ok(mut vec) = self.messages.lock() {
            *vec = window_owned;
        }
        self.refresh_window_snapshot();
        self.sink.emit(Event::Checkpoint {
            session_id: self.id.to_string(),
            messages: checkpoint_messages.to_vec(),
            window_tokens,
        });
        Some(CompactResult {
            before_tokens,
            after_tokens,
            compacted_start: range.start,
            compacted_end: range.end,
        })
    }

    pub fn begin_turn(&self, user_msg: Message) -> TurnId {
        BeginTurnCommand { user_msg }.execute(self)
    }

    pub fn mark_streamed(&self) {
        self.turn
            .streamed
            .store(true, std::sync::atomic::Ordering::Relaxed);
    }

    pub fn take_streamed_flag(&self) -> bool {
        self.turn
            .streamed
            .swap(false, std::sync::atomic::Ordering::Relaxed)
    }

    pub fn end_turn(&self) {
        self.turn
            .streamed
            .store(false, std::sync::atomic::Ordering::Relaxed);
        let turn_id = self.turn.current_turn.lock().unwrap().take();
        if let Some(turn_id) = turn_id {
            let mut q = self.injection_queue.lock().unwrap();
            for inj in q.iter_mut() {
                if inj.state == InjectionState::Pending && inj.turn_id == turn_id {
                    inj.state = InjectionState::Cancelled;
                    let _ = self.injection_tx.send(inj.clone());
                }
            }
            drop(q);
            self.sink.emit(Event::TurnEnd { turn_id });
        }
    }

    pub fn current_turn(&self) -> Option<TurnId> {
        self.turn.current_turn.lock().unwrap().clone()
    }

    pub fn enqueue_injection(&self, text: impl Into<String>) -> Result<InjectionId, EnqueueError> {
        self.enqueue_injection_with_level(text, crate::injection::InjectionLevel::L1Nudge, None)
    }

    pub fn enqueue_injection_with_level(
        &self,
        text: impl Into<String>,
        level: crate::injection::InjectionLevel,
        redirect_target: Option<String>,
    ) -> Result<InjectionId, EnqueueError> {
        let turn_id = self
            .turn
            .current_turn
            .lock()
            .unwrap()
            .clone()
            .ok_or(EnqueueError::NoActiveTurn)?;
        let inj = Injection::with_level(turn_id.clone(), text, level, redirect_target);
        let id = inj.id.clone();
        self.sink.emit(Event::UserInject {
            turn_id,
            injection: inj.clone(),
        });
        self.injection_queue.lock().unwrap().push(inj.clone());
        let _ = self.injection_tx.send(inj);
        Ok(id)
    }

    pub fn subscribe_injections(&self) -> broadcast::Receiver<Injection> {
        self.injection_tx.subscribe()
    }

    pub fn mark_injection_consumed(&self, id: &InjectionId) {
        let mut q = self.injection_queue.lock().unwrap();
        for inj in q.iter_mut() {
            if inj.id == *id && inj.state == InjectionState::Pending {
                inj.state = InjectionState::Injected;
                let _ = self.injection_tx.send(inj.clone());
                return;
            }
        }
    }

    pub fn peek_pending_l2_or_higher(&self, turn_id: &TurnId) -> Option<Injection> {
        let q = self.injection_queue.lock().unwrap();
        q.iter()
            .find(|i| {
                i.state == InjectionState::Pending
                    && i.turn_id == *turn_id
                    && !matches!(i.level, crate::injection::InjectionLevel::L1Nudge)
            })
            .cloned()
    }

    /// Drain all Pending injections for `turn_id`. Marks them Injected.
    /// Returns them in creation order.
    pub fn drain_injections(&self, turn_id: &TurnId) -> Vec<Injection> {
        let mut q = self.injection_queue.lock().unwrap();
        let mut out = Vec::new();
        for inj in q.iter_mut() {
            if inj.state == InjectionState::Pending && inj.turn_id == *turn_id {
                inj.state = InjectionState::Injected;
                let _ = self.injection_tx.send(inj.clone());
                out.push(inj.clone());
            }
        }
        out
    }

    pub fn list_pending_injections(&self) -> Vec<Injection> {
        self.injection_queue
            .lock()
            .unwrap()
            .iter()
            .filter(|i| i.state == InjectionState::Pending)
            .cloned()
            .collect()
    }

    pub fn cancel_flow(&self) {
        self.turn.flow_cancel.lock().unwrap().cancel();
    }

    pub fn flow_cancel_token(&self) -> CancellationToken {
        self.turn.flow_cancel.lock().unwrap().clone()
    }

    pub async fn shutdown(&self) {
        let writer = self.writer.lock().unwrap().take();
        if let Some(writer) = writer {
            writer.shutdown().await;
        }
    }

    // Rides FIFO queue ordering: once flush's own barrier is written,
    // every earlier sink.emit is on disk too.
    #[allow(clippy::await_holding_lock)]
    pub async fn flush_writer(&self) {
        let guard = self.writer.lock().unwrap();
        let Some(ref writer) = *guard else {
            return;
        };
        writer.flush().await;
    }
}

#[derive(Debug, thiserror::Error)]
pub enum EnqueueError {
    #[error("enqueue_injection called with no active turn")]
    NoActiveTurn,
}

pub struct AppendMessageCommand {
    pub msg: Message,
    pub flow_run_id: Option<FlowRunId>,
}

impl AppendMessageCommand {
    pub fn execute(&self, session: &Session) -> u64 {
        let flow_run_id_str = self.flow_run_id.as_ref().map(|r| r.0.to_string());
        let msg = crate::tools::tool_output::maybe_truncate_tool_message(
            &self.msg,
            Some(&session.output_store),
        );
        let is_internal = msg.origin == crate::message::MessageOrigin::Internal;
        let event =
            match msg.role {
                MessageRole::User => Event::UserMsg {
                    turn_id: msg.turn_id.clone(),
                    flow_run_id: self.flow_run_id.clone(),
                    message: msg.clone(),
                },
                MessageRole::Assistant => {
                    if !is_internal {
                        let _ = session.watch.stream_tx.send(
                            crate::stream::StreamFrame::AssistantMsg {
                                flow_run_id: flow_run_id_str.clone(),
                                message: msg.clone(),
                            },
                        );
                        for source in extract_mermaid_blocks(&msg) {
                            let _ = session.watch.stream_tx.send(
                                crate::stream::StreamFrame::MermaidDiagram {
                                    source: source.clone(),
                                },
                            );
                            session
                                .sink
                                .emit(crate::event::Event::MermaidDiagram { source });
                        }
                    }
                    Event::AssistantMsg {
                        turn_id: msg.turn_id.clone(),
                        flow_run_id: self.flow_run_id.clone(),
                        message: msg.clone(),
                    }
                }
                MessageRole::Tool => {
                    if !is_internal {
                        let _ = session.watch.stream_tx.send(
                            crate::stream::StreamFrame::ToolResultMsg {
                                flow_run_id: flow_run_id_str.clone(),
                                message: msg.clone(),
                            },
                        );
                    }
                    Event::ToolResultMsg {
                        turn_id: msg.turn_id.clone(),
                        flow_run_id: self.flow_run_id.clone(),
                        message: msg.clone(),
                    }
                }
                MessageRole::System => Event::SystemMsg {
                    turn_id: msg.turn_id.clone(),
                    message: msg.clone(),
                },
            };
        let seq = session.sink.emit_returning_seq(event);
        if matches!(msg.role, MessageRole::User) {
            let images: Vec<(usize, String)> = msg
                .parts
                .iter()
                .enumerate()
                .filter_map(|(i, p)| match p {
                    crate::message::MessagePart::Image { source } => {
                        let basename = match &source.data {
                            crate::message::ImageData::Path { path } => path
                                .file_name()
                                .and_then(|n| n.to_str())
                                .unwrap_or("unknown")
                                .to_string(),
                            crate::message::ImageData::Base64 { .. } => "base64".into(),
                        };
                        Some((i, basename))
                    }
                    _ => None,
                })
                .collect();
            if !images.is_empty() {
                *session.last_image_user_msg.lock().unwrap() = Some(LastImageUserMsg {
                    message_seq: seq,
                    images,
                });
            }
        }
        session.messages.lock().unwrap().push(msg.clone());
        seq
    }
}

pub struct BeginTurnCommand {
    pub user_msg: Message,
}

impl BeginTurnCommand {
    pub fn execute(&self, session: &Session) -> TurnId {
        let turn_id = self.user_msg.turn_id.clone();
        *session.turn.current_turn.lock().unwrap() = Some(turn_id.clone());
        *session.turn.flow_cancel.lock().unwrap() = tokio_util::sync::CancellationToken::new();
        session.sink.emit(Event::TurnStart {
            turn_id: turn_id.clone(),
        });
        AppendMessageCommand {
            msg: self.user_msg.clone(),
            flow_run_id: None,
        }
        .execute(session);
        turn_id
    }
}

fn extract_mermaid_blocks(msg: &crate::message::Message) -> Vec<String> {
    let text = msg.text_concat();
    let mut blocks = Vec::new();
    let mut lines = text.lines().peekable();
    while let Some(line) = lines.next() {
        let trimmed = line.trim();
        if trimmed.starts_with("```") {
            let lang = trimmed.trim_start_matches("```").trim();
            if lang == "mermaid" {
                let mut source = String::new();
                for inner in lines.by_ref() {
                    if inner.trim() == "```" {
                        break;
                    }
                    if !source.is_empty() {
                        source.push('\n');
                    }
                    source.push_str(inner);
                }
                if !source.is_empty() {
                    blocks.push(source);
                }
            } else {
                for inner in lines.by_ref() {
                    if inner.trim() == "```" {
                        break;
                    }
                }
            }
        }
    }
    blocks
}

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

    fn write_events(dir: &Path, lines: &[&str]) {
        let path = dir.join("events.jsonl");
        std::fs::write(&path, lines.join("\n") + "\n").unwrap();
    }

    #[test]
    fn successful_flow_count_triggers_at_powers_of_three_and_resets_per_session() {
        let session = Session::open_ephemeral();
        let mut hits = Vec::new();
        for _ in 0..27 {
            if let Some(count) = session.record_successful_flow() {
                hits.push(count);
            }
        }
        assert_eq!(hits, vec![3, 9, 27]);
        assert_eq!(session.successful_flow_count(), 27);
        assert_eq!(Session::open_ephemeral().successful_flow_count(), 0);
    }

    #[test]
    fn commit_rewritten_window_uses_checkpoint_without_legacy_range_replay() {
        let session = Session::open_ephemeral();
        let original = vec![
            Message::user_text(TurnId::now(), "first user"),
            Message::assistant_text(TurnId::now(), "large output".repeat(2_000)),
            Message::user_text(TurnId::now(), "current user"),
        ];
        for message in original.clone() {
            session.append_message(message, None);
        }
        let replacement = vec![
            original[0].clone(),
            Message::assistant_text(TurnId::now(), "persisted omission"),
            original[2].clone(),
        ];
        let before_tokens = crate::compaction::estimate_tokens_for_messages(&original);

        session
            .commit_rewritten_window(replacement.clone(), before_tokens, before_tokens, 1)
            .expect("rewrite commit");

        assert_eq!(session.messages().as_ref(), replacement.as_slice());
        let events = session.sink().snapshot();
        assert!(events.iter().any(|event| matches!(
            event,
            Event::ContextCompact {
                replacement_msg_seq: None,
                ..
            }
        )));
        assert!(
            !events
                .iter()
                .any(|event| matches!(event, Event::SystemMsg { .. }))
        );
        let checkpoint = events
            .into_iter()
            .find(|event| matches!(event, Event::Checkpoint { .. }))
            .expect("checkpoint");
        let replay = crate::message_stream::MessageStream::new(std::sync::Arc::new(
            std::sync::Mutex::new(vec![crate::event::EventEnvelope::new(1, checkpoint)]),
        ));
        assert_eq!(&*replay.window(), replacement.as_slice());
    }

    #[test]
    fn commit_compacted_window_updates_live_handle_and_checkpoint_replay() {
        let session = Session::open_ephemeral();
        let old = vec![
            Message::user_text(TurnId::now(), "old user".repeat(2_000)),
            Message::assistant_text(TurnId::now(), "old assistant".repeat(2_000)),
            Message::user_text(TurnId::now(), "current user"),
        ];
        for message in old.clone() {
            session.append_message(message, None);
        }
        let replacement = vec![
            Message::system_compact_summary(TurnId::now(), "anchor", 0, 1, 2),
            old[2].clone(),
            Message::assistant_text(TurnId::now(), "persisted omission"),
        ];
        let before_tokens = crate::compaction::estimate_tokens_for_messages(&old);
        let range = crate::compaction::CompactRange {
            start: 0,
            end: 2,
            tokens_saved_estimate: before_tokens,
        };

        session
            .commit_compacted_window(
                "anchor".into(),
                replacement.clone(),
                range,
                before_tokens,
                before_tokens,
            )
            .expect("commit");

        assert_eq!(session.messages().as_ref(), replacement.as_slice());
        assert_eq!(
            session.messages_handle().lock().unwrap().as_slice(),
            replacement.as_slice()
        );
        let checkpoint = session
            .sink()
            .snapshot()
            .into_iter()
            .find(|event| matches!(event, Event::Checkpoint { .. }))
            .expect("checkpoint");
        let replay = crate::message_stream::MessageStream::new(std::sync::Arc::new(
            std::sync::Mutex::new(vec![crate::event::EventEnvelope::new(1, checkpoint)]),
        ));
        assert_eq!(&*replay.window(), replacement.as_slice());
    }

    #[test]
    fn replay_applies_attachment_degraded_patch() {
        let dir = TempDir::new().unwrap();
        let user_msg = r#"{"type":"user_msg","seq":5,"turn_id":"019f0000-0000-7000-0000-000000000001","message":{"role":"user","parts":[{"type":"image","source":{"media_type":"image/png","data":{"kind":"path","path":"/tmp/photo.png"}}},{"type":"text","text":"describe"}],"turn_id":"019f0000-0000-7000-0000-000000000001"},"ts":"2026-07-07T00:00:00Z"}"#;
        let degrade = r#"{"type":"attachment_degraded","seq":6,"turn_id":null,"flow_run_id":null,"message_seq":5,"part_index":0,"file_basename":"photo.png","reason":"image_too_large","ts":"2026-07-07T00:00:01Z"}"#;
        write_events(dir.path(), &[user_msg, degrade]);
        let entries = replay_transcript_from(&dir.path().join("events.jsonl")).unwrap();
        let msg = entries
            .into_iter()
            .find_map(|e| match e {
                TranscriptEntry::Message { message, .. } => Some(message),
                _ => None,
            })
            .unwrap();
        assert_eq!(msg.parts.len(), 2);
        match &msg.parts[0] {
            crate::message::MessagePart::Text { text } => {
                assert!(text.contains("photo.png"), "expected basename: {text}");
                assert!(text.contains("image_too_large"), "expected reason: {text}");
                assert!(text.starts_with("[attachment unavailable"));
            }
            other => panic!("expected Text stub, got {other:?}"),
        }
        assert!(matches!(
            msg.parts[1],
            crate::message::MessagePart::Text { .. }
        ));
    }

    #[test]
    fn approval_registry_auto_approves_when_level_leq_ceiling() {
        let reg = ApprovalRegistry::new();
        reg.set_auto_ceiling(crate::tool::ApprovalLevel::Approve);
        let pending = PendingApproval {
            tool_use_id: "tu1".into(),
            tool_name: "fs.read".into(),
            args_preview: "{}".into(),
            preview: None,
            level: crate::tool::ApprovalLevel::Auto,
            run_id: FlowRunId::now(),
            emitted_at: chrono::Utc::now(),
            bypass_auto_ceiling: false,
        };
        let rx = reg.request(pending);
        let got = rx.blocking_recv().unwrap();
        assert!(matches!(got, ApprovalDecision::Approve));
        assert!(reg.list_pending().is_empty());
    }

    #[test]
    fn approval_registry_queues_when_level_above_ceiling() {
        let reg = std::sync::Arc::new(ApprovalRegistry::new());
        reg.set_auto_ceiling(crate::tool::ApprovalLevel::Auto);
        let pending = PendingApproval {
            tool_use_id: "tu42".into(),
            tool_name: "fs.write".into(),
            args_preview: "{}".into(),
            preview: None,
            level: crate::tool::ApprovalLevel::Approve,
            run_id: FlowRunId::now(),
            emitted_at: chrono::Utc::now(),
            bypass_auto_ceiling: false,
        };
        let mut rx = reg.request(pending);
        assert_eq!(reg.list_pending().len(), 1);
        assert!(rx.try_recv().is_err(), "should still be queued");
        assert!(reg.decide("tu42", ApprovalDecision::Approve));
        let got = rx.blocking_recv().unwrap();
        assert!(matches!(got, ApprovalDecision::Approve));
        assert!(reg.list_pending().is_empty());
    }

    #[test]
    fn approval_registry_decide_all_flushes_queue() {
        let reg = ApprovalRegistry::new();
        reg.set_auto_ceiling(crate::tool::ApprovalLevel::Auto);
        let mut rxs = Vec::new();
        for i in 0..3 {
            rxs.push(reg.request(PendingApproval {
                tool_use_id: format!("tu{i}"),
                tool_name: "bash.exec".into(),
                args_preview: "{}".into(),
                preview: None,
                level: crate::tool::ApprovalLevel::Dangerous,
                run_id: FlowRunId::now(),
                emitted_at: chrono::Utc::now(),
                bypass_auto_ceiling: false,
            }));
        }
        assert_eq!(reg.list_pending().len(), 3);
        assert_eq!(
            reg.decide_all(ApprovalDecision::Deny {
                reason: "user cancelled".into()
            }),
            3
        );
        assert!(reg.list_pending().is_empty());
    }

    #[test]
    fn compact_review_registry_auto_accepts_when_no_subscriber() {
        let reg = CompactReviewRegistry::new();
        let pending = PendingCompactReview {
            review_id: "r1".into(),
            summary: "gist".into(),
            slice_preview: String::new(),
            slice_count: 0,
            range_start: 0,
            range_end: 0,
            tokens_before: 0,
            emitted_at: chrono::Utc::now(),
        };
        let rx = reg.request(pending);
        let got = rx.blocking_recv().unwrap();
        assert!(matches!(got, CompactReviewDecision::AcceptAsIs));
        assert!(reg.list_pending().is_none());
    }

    #[test]
    fn compact_review_registry_holds_pending_and_decides() {
        let reg = std::sync::Arc::new(CompactReviewRegistry::new());
        let _sub = reg.subscribe();
        let pending = PendingCompactReview {
            review_id: "r2".into(),
            summary: "old".into(),
            slice_preview: "slice".into(),
            slice_count: 3,
            range_start: 1,
            range_end: 4,
            tokens_before: 500,
            emitted_at: chrono::Utc::now(),
        };
        let mut rx = reg.request(pending);
        assert!(rx.try_recv().is_err(), "should be queued");
        assert!(reg.list_pending().is_some());
        assert!(reg.decide(
            "r2",
            CompactReviewDecision::AcceptEdited {
                summary: "new".into()
            }
        ));
        let got = rx.blocking_recv().unwrap();
        match got {
            CompactReviewDecision::AcceptEdited { summary } => assert_eq!(summary, "new"),
            other => panic!("unexpected decision: {other:?}"),
        }
        assert!(reg.list_pending().is_none());
    }

    #[test]
    fn compact_review_registry_reject_flushes() {
        let reg = std::sync::Arc::new(CompactReviewRegistry::new());
        let _sub = reg.subscribe();
        let rx = reg.request(PendingCompactReview {
            review_id: "r3".into(),
            summary: String::new(),
            slice_preview: String::new(),
            slice_count: 0,
            range_start: 0,
            range_end: 0,
            tokens_before: 0,
            emitted_at: chrono::Utc::now(),
        });
        assert!(reg.decide("r3", CompactReviewDecision::Reject));
        let got = rx.blocking_recv().unwrap();
        assert!(matches!(got, CompactReviewDecision::Reject));
    }

    #[test]
    fn replay_context_snapshot_accumulates_llm_call_usage() {
        let dir = TempDir::new().unwrap();
        let events = [
            r#"{"type":"llm_call","seq":1,"model":"anthropic/claude-4","provider":"anthropic","usage":{"input":100,"cached_input":10,"output":50,"cache_write":0},"wallclock_ms":1000,"status":{"kind":"ok"},"run_id":"019f0000-0000-7000-0000-000000000099","ts":"2026-07-08T00:00:00Z"}"#,
            r#"{"type":"user_msg","seq":2,"turn_id":"019f0000-0000-7000-0000-000000000002","message":{"role":"user","parts":[{"type":"text","text":"hi"}],"turn_id":"019f0000-0000-7000-0000-000000000002"},"ts":"2026-07-08T00:00:00Z"}"#,
            r#"{"type":"llm_call","seq":3,"model":"anthropic/claude-4","provider":"anthropic","usage":{"input":200,"cached_input":0,"output":80,"cache_write":0},"wallclock_ms":1000,"status":{"kind":"ok"},"run_id":"019f0000-0000-7000-0000-000000000099","ts":"2026-07-08T00:00:01Z"}"#,
        ];
        write_events(dir.path(), &events);
        let snap = replay_context_snapshot_from(&dir.path().join("events.jsonl"));
        assert_eq!(snap.model, "anthropic/claude-4");
        assert_eq!(snap.tokens_in, 310);
        assert_eq!(snap.tokens_out, 130);
    }

    #[test]
    fn replay_context_snapshot_skips_subagent_llm_calls() {
        let dir = TempDir::new().unwrap();
        let events = [
            r#"{"type":"llm_call","seq":1,"model":"zhipuai/glm-5.2","provider":"zhipu","usage":{"input":100,"cached_input":0,"output":50,"cache_write":0},"wallclock_ms":1000,"status":{"kind":"ok"},"run_id":"019f0000-0000-7000-0000-000000000099","ts":"2026-07-08T00:00:00Z"}"#,
            r#"{"type":"llm_call","seq":2,"model":"gpt-4o-mini","provider":"openai","usage":{"input":200,"cached_input":0,"output":80,"cache_write":0},"wallclock_ms":1000,"status":{"kind":"ok"},"run_id":null,"ts":"2026-07-08T00:00:01Z"}"#,
        ];
        write_events(dir.path(), &events);
        let snap = replay_context_snapshot_from(&dir.path().join("events.jsonl"));
        assert_eq!(snap.model, "zhipuai/glm-5.2");
        assert_eq!(snap.tokens_in, 100);
        assert_eq!(snap.tokens_out, 50);
    }

    #[test]
    fn compact_review_mode_parses_all_variants() {
        assert_eq!(
            CompactReviewMode::parse("always"),
            Some(CompactReviewMode::Always)
        );
        assert_eq!(
            CompactReviewMode::parse("manual-only"),
            Some(CompactReviewMode::ManualOnly)
        );
        assert_eq!(
            CompactReviewMode::parse("manual_only"),
            Some(CompactReviewMode::ManualOnly)
        );
        assert_eq!(
            CompactReviewMode::parse("never"),
            Some(CompactReviewMode::Never)
        );
        assert_eq!(CompactReviewMode::parse(" bogus "), None);
    }

    #[test]
    fn compact_review_mode_should_review_matrix() {
        assert!(CompactReviewMode::Always.should_review(false));
        assert!(CompactReviewMode::Always.should_review(true));
        assert!(!CompactReviewMode::ManualOnly.should_review(false));
        assert!(CompactReviewMode::ManualOnly.should_review(true));
        assert!(!CompactReviewMode::Never.should_review(false));
        assert!(!CompactReviewMode::Never.should_review(true));
    }

    #[test]
    fn compact_review_registry_new_request_rejects_previous() {
        let reg = std::sync::Arc::new(CompactReviewRegistry::new());
        let _sub = reg.subscribe();
        let rx_a = reg.request(PendingCompactReview {
            review_id: "rA".into(),
            summary: String::new(),
            slice_preview: String::new(),
            slice_count: 0,
            range_start: 0,
            range_end: 0,
            tokens_before: 0,
            emitted_at: chrono::Utc::now(),
        });
        let _rx_b = reg.request(PendingCompactReview {
            review_id: "rB".into(),
            summary: String::new(),
            slice_preview: String::new(),
            slice_count: 0,
            range_start: 0,
            range_end: 0,
            tokens_before: 0,
            emitted_at: chrono::Utc::now(),
        });
        let got = rx_a.blocking_recv().unwrap();
        assert!(matches!(got, CompactReviewDecision::Reject));
    }

    fn mk_form(form_id: &str, prompt: &str) -> crate::form::PendingForm {
        crate::form::PendingForm {
            form_id: form_id.into(),
            run_id: crate::event::FlowRunId::now(),
            tool_use_id: "tu".into(),
            kind: crate::form::FormKind::Confirm {
                prompt: prompt.into(),
            },
            emitted_at: chrono::Utc::now(),
        }
    }

    #[test]
    fn form_registry_auto_cancels_without_subscriber() {
        let reg = FormRegistry::new();
        let rx = reg.request(mk_form("f1", "sure?"));
        let got = rx.blocking_recv().unwrap();
        assert_eq!(got, crate::form::FormAnswer::Cancelled);
        assert!(reg.list_pending().is_empty());
    }

    #[test]
    fn form_registry_delivers_answer_by_form_id() {
        let reg = std::sync::Arc::new(FormRegistry::new());
        let _sub = reg.subscribe();
        let rx = reg.request(mk_form("fA", "?"));
        assert_eq!(reg.list_pending().len(), 1);
        let ok = reg.submit("fA", crate::form::FormAnswer::Confirmed { value: true });
        assert!(ok);
        let got = rx.blocking_recv().unwrap();
        assert_eq!(got, crate::form::FormAnswer::Confirmed { value: true });
        assert!(reg.list_pending().is_empty());
    }

    #[test]
    fn form_registry_submit_unknown_id_is_noop() {
        let reg = std::sync::Arc::new(FormRegistry::new());
        let _sub = reg.subscribe();
        let _rx = reg.request(mk_form("real", "?"));
        assert!(!reg.submit("ghost", crate::form::FormAnswer::Cancelled));
        assert_eq!(reg.list_pending().len(), 1);
    }

    #[test]
    fn form_registry_cancel_all_flushes_pending() {
        let reg = std::sync::Arc::new(FormRegistry::new());
        let _sub = reg.subscribe();
        let rx_a = reg.request(mk_form("a", "?"));
        let rx_b = reg.request(mk_form("b", "?"));
        reg.cancel_all();
        assert_eq!(
            rx_a.blocking_recv().unwrap(),
            crate::form::FormAnswer::Cancelled
        );
        assert_eq!(
            rx_b.blocking_recv().unwrap(),
            crate::form::FormAnswer::Cancelled
        );
        assert!(reg.list_pending().is_empty());
    }

    #[test]
    fn form_registry_queues_multiple_pending() {
        let reg = std::sync::Arc::new(FormRegistry::new());
        let _sub = reg.subscribe();
        let _rx1 = reg.request(mk_form("1", "?"));
        let _rx2 = reg.request(mk_form("2", "?"));
        let pending = reg.list_pending();
        assert_eq!(pending.len(), 2);
        assert_eq!(pending[0].form_id, "1");
        assert_eq!(pending[1].form_id, "2");
    }

    #[test]
    fn replay_without_degraded_events_preserves_image_parts() {
        let dir = TempDir::new().unwrap();
        let user_msg = r#"{"type":"user_msg","seq":1,"turn_id":"019f0000-0000-7000-0000-000000000002","message":{"role":"user","parts":[{"type":"image","source":{"media_type":"image/png","data":{"kind":"path","path":"/tmp/x.png"}}}],"turn_id":"019f0000-0000-7000-0000-000000000002"},"ts":"2026-07-07T00:00:00Z"}"#;
        write_events(dir.path(), &[user_msg]);
        let entries = replay_transcript_from(&dir.path().join("events.jsonl")).unwrap();
        let msg = entries
            .into_iter()
            .find_map(|e| match e {
                TranscriptEntry::Message { message, .. } => Some(message),
                _ => None,
            })
            .unwrap();
        assert!(matches!(
            msg.parts[0],
            crate::message::MessagePart::Image { .. }
        ));
    }

    #[test]
    fn replay_messages_from_old_format_no_seq_no_ts() {
        let dir = TempDir::new().unwrap();
        // Old-style JSONL: no seq, no ts on events
        let user_json = r#"{"type":"user_msg","turn_id":"019f0000-0000-7000-0000-000000000001","message":{"role":"user","parts":[{"type":"text","text":"hello"}],"turn_id":"019f0000-0000-7000-0000-000000000001"}}"#;
        let asst_json = r#"{"type":"assistant_msg","turn_id":"019f0000-0000-7000-0000-000000000001","message":{"role":"assistant","parts":[{"type":"text","text":"hi there"}],"turn_id":"019f0000-0000-7000-0000-000000000001"},"flow_run_id":null}"#;
        write_events(dir.path(), &[user_json, asst_json]);
        let msgs = replay_messages_from(&dir.path().join("events.jsonl")).unwrap();
        assert_eq!(msgs.len(), 2, "should load both messages from old format");
        assert_eq!(msgs[0].text_concat(), "hello");
        assert_eq!(msgs[1].text_concat(), "hi there");
    }

    #[test]
    fn replay_messages_from_old_format_with_null_fields() {
        let dir = TempDir::new().unwrap();
        // Old JSON with null turn_id / flow_run_id (graceful parse)
        let sys_json = r#"{"type":"system_msg","turn_id":"019f0000-0000-7000-0000-000000000003","message":{"role":"system","parts":[{"type":"text","text":"note"}],"turn_id":"019f0000-0000-7000-0000-000000000003"}}"#;
        write_events(dir.path(), &[sys_json]);
        let msgs = replay_messages_from(&dir.path().join("events.jsonl")).unwrap();
        assert_eq!(msgs.len(), 1);
        assert_eq!(msgs[0].text_concat(), "note");
    }

    #[test]
    fn replay_messages_from_applies_attachment_degrade() {
        let dir = TempDir::new().unwrap();
        let user_msg = r#"{"type":"user_msg","seq":5,"turn_id":"019f0000-0000-7000-0000-000000000001","message":{"role":"user","parts":[{"type":"image","source":{"media_type":"image/png","data":{"kind":"path","path":"/tmp/photo.png"}}},{"type":"text","text":"describe"}],"turn_id":"019f0000-0000-7000-0000-000000000001"},"ts":"2026-07-07T00:00:00Z"}"#;
        let degrade = r#"{"type":"attachment_degraded","seq":6,"turn_id":null,"flow_run_id":null,"message_seq":5,"part_index":0,"file_basename":"photo.png","reason":"image_too_large","ts":"2026-07-07T00:00:01Z"}"#;
        write_events(dir.path(), &[user_msg, degrade]);
        let msgs = replay_messages_from(&dir.path().join("events.jsonl")).unwrap();
        assert_eq!(msgs.len(), 1, "only the user message (patched)");
        assert_eq!(msgs[0].parts.len(), 2);
        match &msgs[0].parts[0] {
            crate::message::MessagePart::Text { text } => {
                assert!(text.contains("photo.png"), "expected basename: {text}");
                assert!(text.contains("image_too_large"), "expected reason: {text}");
                assert!(
                    text.starts_with("[attachment unavailable"),
                    "expected stub prefix: {text}"
                );
            }
            other => panic!("expected Text stub, got {other:?}"),
        }
        assert!(
            matches!(msgs[0].parts[1], crate::message::MessagePart::Text { .. }),
            "second part should remain text"
        );
    }

    #[test]
    fn replay_messages_from_degrade_before_message_is_noop() {
        let dir = TempDir::new().unwrap();
        // Degrade event appears BEFORE the message it references (should not crash)
        let degrade = r#"{"type":"attachment_degraded","seq":1,"turn_id":null,"flow_run_id":null,"message_seq":99,"part_index":0,"file_basename":"x.png","reason":"test","ts":"2026-07-07T00:00:00Z"}"#;
        let user_msg = r#"{"type":"user_msg","seq":2,"turn_id":"019f0000-0000-7000-0000-000000000001","message":{"role":"user","parts":[{"type":"image","source":{"media_type":"image/png","data":{"kind":"path","path":"/tmp/x.png"}}}],"turn_id":"019f0000-0000-7000-0000-000000000001"},"ts":"2026-07-07T00:00:01Z"}"#;
        write_events(dir.path(), &[degrade, user_msg]);
        let msgs = replay_messages_from(&dir.path().join("events.jsonl")).unwrap();
        assert_eq!(msgs.len(), 1);
        // Image part preserved — degrade referenced unknown message_seq
        assert!(
            matches!(msgs[0].parts[0], crate::message::MessagePart::Image { .. }),
            "image should remain when degrade targets unknown seq"
        );
    }

    #[test]
    fn replay_messages_from_degrade_wrong_seq_leaves_image() {
        let dir = TempDir::new().unwrap();
        let user_msg = r#"{"type":"user_msg","seq":1,"turn_id":"019f0000-0000-7000-0000-000000000001","message":{"role":"user","parts":[{"type":"image","source":{"media_type":"image/png","data":{"kind":"path","path":"/tmp/x.png"}}}],"turn_id":"019f0000-0000-7000-0000-000000000001"},"ts":"2026-07-07T00:00:00Z"}"#;
        // Degrade references wrong message_seq (2, but message has seq 1)
        let degrade = r#"{"type":"attachment_degraded","seq":2,"turn_id":null,"flow_run_id":null,"message_seq":2,"part_index":0,"file_basename":"x.png","reason":"test","ts":"2026-07-07T00:00:01Z"}"#;
        write_events(dir.path(), &[user_msg, degrade]);
        let msgs = replay_messages_from(&dir.path().join("events.jsonl")).unwrap();
        assert_eq!(msgs.len(), 1);
        assert!(
            matches!(msgs[0].parts[0], crate::message::MessagePart::Image { .. }),
            "image should remain when degrade targets wrong seq"
        );
    }

    #[test]
    fn replay_messages_from_applies_context_compact() {
        let dir = TempDir::new().unwrap();
        let user1 = r#"{"type":"user_msg","seq":1,"turn_id":"019f0000-0000-7000-0000-000000000001","message":{"role":"user","parts":[{"type":"text","text":"old u1"}],"turn_id":"019f0000-0000-7000-0000-000000000001"},"ts":"2026-07-07T00:00:00Z"}"#;
        let asst1 = r#"{"type":"assistant_msg","seq":2,"turn_id":"019f0000-0000-7000-0000-000000000001","message":{"role":"assistant","parts":[{"type":"text","text":"old a1"}],"turn_id":"019f0000-0000-7000-0000-000000000001"},"flow_run_id":null,"ts":"2026-07-07T00:00:01Z"}"#;
        let user2 = r#"{"type":"user_msg","seq":3,"turn_id":"019f0000-0000-7000-0000-000000000002","message":{"role":"user","parts":[{"type":"text","text":"old u2"}],"turn_id":"019f0000-0000-7000-0000-000000000002"},"ts":"2026-07-07T00:00:02Z"}"#;
        // replacement message (compact summary)
        let summary = r#"{"type":"system_msg","seq":4,"turn_id":"019f0000-0000-7000-0000-000000000002","message":{"role":"system","parts":[{"type":"compact_summary","summary":"two messages compacted","seq_start":0,"seq_end":1,"count":2}],"turn_id":"019f0000-0000-7000-0000-000000000002"},"ts":"2026-07-07T00:00:03Z"}"#;
        let compact = r#"{"type":"context_compact","seq":5,"session_id":"sess","before_tokens":200,"after_tokens":50,"compacted_range_start":0,"compacted_range_end":1,"summary_text":"two messages compacted","replacement_msg_seq":4,"ts":"2026-07-07T00:00:04Z"}"#;
        let after = r#"{"type":"user_msg","seq":6,"turn_id":"019f0000-0000-7000-0000-000000000003","message":{"role":"user","parts":[{"type":"text","text":"after compact"}],"turn_id":"019f0000-0000-7000-0000-000000000003"},"ts":"2026-07-07T00:00:05Z"}"#;
        write_events(dir.path(), &[user1, asst1, user2, summary, compact, after]);
        let msgs = replay_messages_from(&dir.path().join("events.jsonl")).unwrap();
        // compact range 0-1 removes user1+asst1; user2 (outside range) + summary + after = 3
        assert_eq!(msgs.len(), 3, "compact summary + user2 + after compact");
        assert!(
            matches!(
                msgs[0].parts[0],
                crate::message::MessagePart::CompactSummary { .. }
            ),
            "first should be compact summary"
        );
        if let crate::message::MessagePart::CompactSummary { summary, .. } = &msgs[0].parts[0] {
            assert_eq!(summary, "two messages compacted");
        }
        assert_eq!(msgs[1].text_concat(), "old u2");
        assert_eq!(msgs[2].text_concat(), "after compact");
    }

    #[test]
    fn replay_messages_from_no_replacement_seq_ignores_compact() {
        let dir = TempDir::new().unwrap();
        let user1 = r#"{"type":"user_msg","seq":1,"turn_id":"019f0000-0000-7000-0000-000000000001","message":{"role":"user","parts":[{"type":"text","text":"hello"}],"turn_id":"019f0000-0000-7000-0000-000000000001"},"ts":"2026-07-07T00:00:00Z"}"#;
        // context_compact with no replacement_msg_seq → should be ignored
        let compact = r#"{"type":"context_compact","seq":2,"session_id":"sess","before_tokens":200,"after_tokens":50,"compacted_range_start":0,"compacted_range_end":0,"summary_text":"ignored","replacement_msg_seq":null,"ts":"2026-07-07T00:00:01Z"}"#;
        write_events(dir.path(), &[user1, compact]);
        let msgs = replay_messages_from(&dir.path().join("events.jsonl")).unwrap();
        assert_eq!(msgs.len(), 1, "compact without replacement seq is ignored");
        assert_eq!(msgs[0].text_concat(), "hello");
    }

    #[test]
    fn replay_messages_from_compact_after_no_change_ignored() {
        let dir = TempDir::new().unwrap();
        let user1 = r#"{"type":"user_msg","seq":1,"turn_id":"019f0000-0000-7000-0000-000000000001","message":{"role":"user","parts":[{"type":"text","text":"hello"}],"turn_id":"019f0000-0000-7000-0000-000000000001"},"ts":"2026-07-07T00:00:00Z"}"#;
        let summary = r#"{"type":"system_msg","seq":2,"turn_id":"019f0000-0000-7000-0000-000000000001","message":{"role":"system","parts":[{"type":"compact_summary","summary":"no change","seq_start":0,"seq_end":0,"count":1}],"turn_id":"019f0000-0000-7000-0000-000000000001"},"ts":"2026-07-07T00:00:01Z"}"#;
        // after_tokens >= before_tokens → compaction is no-op
        let compact = r#"{"type":"context_compact","seq":3,"session_id":"sess","before_tokens":50,"after_tokens":100,"compacted_range_start":0,"compacted_range_end":0,"summary_text":"no change","replacement_msg_seq":2,"ts":"2026-07-07T00:00:02Z"}"#;
        write_events(dir.path(), &[user1, summary, compact]);
        let msgs = replay_messages_from(&dir.path().join("events.jsonl")).unwrap();
        assert_eq!(msgs.len(), 2, "compact with after>=before is ignored");
    }

    #[test]
    fn replay_messages_from_missing_file_returns_empty() {
        let dir = TempDir::new().unwrap();
        let msgs = replay_messages_from(&dir.path().join("events.jsonl")).unwrap();
        assert!(msgs.is_empty());
    }

    #[test]
    fn replay_messages_from_empty_file_returns_empty() {
        let dir = TempDir::new().unwrap();
        write_events(dir.path(), &[]);
        let msgs = replay_messages_from(&dir.path().join("events.jsonl")).unwrap();
        assert!(msgs.is_empty());
    }

    #[test]
    fn replay_all_messages_with_seq_includes_compacted() {
        let dir = TempDir::new().unwrap();
        let user1 = r#"{"type":"user_msg","seq":1,"turn_id":"019f0000-0000-7000-0000-000000000001","message":{"role":"user","parts":[{"type":"text","text":"old"}],"turn_id":"019f0000-0000-7000-0000-000000000001"},"ts":"2026-07-07T00:00:00Z"}"#;
        let summary = r#"{"type":"system_msg","seq":4,"turn_id":"019f0000-0000-7000-0000-000000000002","message":{"role":"system","parts":[{"type":"compact_summary","summary":"s","seq_start":0,"seq_end":0,"count":1}],"turn_id":"019f0000-0000-7000-0000-000000000002"},"ts":"2026-07-07T00:00:01Z"}"#;
        let compact = r#"{"type":"context_compact","seq":5,"session_id":"sess","before_tokens":200,"after_tokens":50,"compacted_range_start":0,"compacted_range_end":0,"summary_text":"s","replacement_msg_seq":4,"ts":"2026-07-07T00:00:02Z"}"#;
        write_events(dir.path(), &[user1, summary, compact]);
        let all = replay_all_messages_with_seq(&dir.path().join("events.jsonl")).unwrap();
        // all includes both user1 and summary — compaction NOT applied
        assert_eq!(all.len(), 2, "all messages preserved (no compaction)");
        assert_eq!(all[0].1.text_concat(), "old");
    }
}