clawft-core 0.8.3

//! Conversation sink seam.
//!
//! [`ConversationSink`] is the substrate-bound write hook used by
//! [`AgentLoop::handle_turn`](super::loop_core::AgentLoop::handle_turn)
//! to persist per-turn JSONL and publish status heartbeats. The shape
//! of [`Turn`] matches the substrate JSONL layout from
//! `chat-agent-v1.md` §11.5 (one record per role event:
//! user / assistant / tool).
//!
//! Phase C3 will land a substrate-backed implementation in
//! `clawft-service-agent`. Today the in-memory [`InMemorySink`] is
//! the only impl — it satisfies the trait contract for tests and is
//! the default attached to [`AgentLoop`](super::loop_core::AgentLoop).
//!
//! ## Lock semantics
//!
//! [`ConversationSink::lock_conversation`] is the entry point for the
//! per-conv mutex used to serialise turns inside the same conversation
//! (see `chat-agent-v1.md` §10). The real implementation lives in the
//! `clawft-service-agent::AgentService` (Phase C1's
//! `DashMap<ConvId, Mutex<()>>`); the sink just exposes it so the
//! loop has a single seam. [`InMemorySink`] treats the call as a
//! no-op because tests run sequentially anyway.

use std::collections::HashMap;
use std::sync::Mutex;

use async_trait::async_trait;

/// One per-turn record. Mirrors the substrate JSONL line.
///
/// Tool-call intermediates and the final assistant response each get
/// their own [`Turn`]; the role discriminates. `tool_calls` is set on
/// assistant turns that invoked tools; `tool_call_id` is set on
/// `role == "tool"` turns to bind the result to its dispatch.
#[derive(Debug, Clone, PartialEq)]
pub struct Turn {
    /// Stable per-turn identifier. The sink is responsible for
    /// monotonic ULIDs in the substrate impl; in-memory tests can
    /// supply any unique string.
    pub turn_id: String,
    /// One of `"user" | "assistant" | "tool" | "system"`.
    pub role: String,
    /// Text payload — the message body (STT transcript for voice).
    pub content: String,
    /// JSON tool_calls array on assistant turns that invoked tools.
    pub tool_calls: Option<Vec<serde_json::Value>>,
    /// Tool call ID this turn is responding to (only `role == "tool"`).
    pub tool_call_id: Option<String>,
    /// Wall-clock millisecond timestamp the sink saw the turn.
    pub ts_ms: u64,
    /// Optional per-utterance voice decomposition (Wave 1 §W1.2). `None` for
    /// every text turn (agent.chat + tests). Set only by the voice recorder
    /// path (`agent.turn.record`), it rides the anchor→`index_turn` seam so the
    /// blob is stored verbatim on the turn node and its `emotion` overrides the
    /// classification emotion axis (`tier:"voice"`). An opaque `serde_json`
    /// value — the daemon never reshapes it (the flat-key parity contract).
    pub voice_analysis: Option<serde_json::Value>,
    /// Optional substrate-pointed audio for multimodal turns (WEFT-350).
    /// When set, the substrate sink persists
    /// [`clawft_types::TurnContent::Audio`] or `Mixed` instead of plain
    /// `Text`.
    pub audio: Option<clawft_types::AudioRef>,
}

impl Turn {
    /// Build a plain text turn (no tools, no voice analysis, no audio).
    pub fn plain(role: impl Into<String>, content: impl Into<String>, ts_ms: u64) -> Self {
        Self {
            turn_id: String::new(),
            role: role.into(),
            content: content.into(),
            tool_calls: None,
            tool_call_id: None,
            ts_ms,
            voice_analysis: None,
            audio: None,
        }
    }

    /// Attach a substrate audio ref (builder-style).
    pub fn with_audio(mut self, audio: clawft_types::AudioRef) -> Self {
        self.audio = Some(audio);
        self
    }
}

/// Per-conversation persistence seam.
///
/// Implementations:
/// - [`InMemorySink`] (default; HashMap-backed, test-only).
/// - [`LocalFileSink`](super::local_file_sink::LocalFileSink) (in-process JSONL).
/// - Substrate sink (Phase C3, in `clawft-service-agent`).
///
/// The `async_trait` `?Send` relaxation is applied for the `browser`
/// feature so WASM-resident sinks that await platform FS (browser
/// `FileSystem` futures are `!Send`) satisfy the trait. Native impls
/// keep the strict `Send` bound for tokio multi-threaded runtimes.
#[cfg_attr(not(feature = "browser"), async_trait)]
#[cfg_attr(feature = "browser", async_trait(?Send))]
pub trait ConversationSink: Send + Sync + 'static {
    /// Acquire the conversation lock. The substrate impl awaits the
    /// per-conv `Mutex<()>` from the `AgentService` DashMap so two
    /// turns in the same conversation can never interleave. The
    /// in-memory impl is a no-op.
    async fn lock_conversation(&self, conv_id: &str);

    /// Append a [`Turn`] to the conversation log. Errors return a
    /// `String` so the trait stays cheap to implement; callers map
    /// to richer errors at the boundary.
    async fn append_turn(&self, conv_id: &str, turn: Turn) -> Result<(), String>;

    /// Publish a status event (heartbeat, "thinking", "tool_running",
    /// "done", …) to the substrate `derived/chat/<conv>/status`
    /// topic. The in-memory impl drops the payload.
    async fn publish_status(
        &self,
        conv_id: &str,
        status: &str,
        payload: serde_json::Value,
    ) -> Result<(), String>;

    /// Load the full conversation history. Substrate impl reads from
    /// `derived/chat/<conv>/turns/`; in-memory returns the HashMap
    /// entry.
    async fn load_history(&self, conv_id: &str) -> Result<Vec<Turn>, String>;

    /// Read back conversation history for **hydration** (M3 store
    /// collapse, design §D2). Returns the sink's turn log in
    /// chronological order, truncated to the most recent `window`
    /// turns; `window == 0` returns every turn.
    ///
    /// ## Superset semantics (load-bearing for P3)
    ///
    /// The returned turns are the **superset** — they include the
    /// tool-loop intermediates (`assistant`-with-`tool_calls` turns and
    /// `role == "tool"` result turns) that the retired SessionManager
    /// store never held. P3 hydration reconstructs the assembly
    /// `Session` by dropping those intermediates (keep `user` turns +
    /// the final `assistant` of each exchange). Because the window is
    /// applied to the **superset**, a superset window of `N` is *not*
    /// equivalent to a store-1 window of `N`: the intermediates inflate
    /// the count. P3 must therefore either pass `window == 0` (read all,
    /// filter, and let the assembly stage re-window via
    /// `Session::get_history(memory_window)` exactly as today), or pass
    /// a bound comfortably larger than `memory_window`. The
    /// `prompt_identity_golden` gate proves the filter reproduces store
    /// 1 only when the full superset is available.
    ///
    /// ## Failure mode
    ///
    /// Infallible by contract: a substrate/read error degrades to an
    /// empty vec with a warn at the impl site (design §6: "degrade,
    /// don't crash"). Hydration must never fail a turn because the read
    /// path hiccuped.
    ///
    /// Default impl returns `vec![]` so non-substrate sinks compile
    /// untouched.
    async fn history(&self, _conv_id: &str, _window: usize) -> Vec<Turn> {
        Vec::new()
    }

    /// Read the per-conversation metadata sidecar (M3 design §D3).
    ///
    /// Home of the hallucination score (and any future per-conv K/V)
    /// once SessionManager stops persisting session metadata. Returns
    /// [`serde_json::Value::Null`] when unset or on read error — the
    /// caller defaults the hallucination score to `0.0` on `Null`
    /// exactly as the current `loop_core` path does.
    ///
    /// Default impl returns `Null` so non-substrate sinks compile
    /// untouched.
    async fn meta(&self, _conv_id: &str) -> serde_json::Value {
        serde_json::Value::Null
    }

    /// Write the per-conversation metadata sidecar (M3 design §D3).
    ///
    /// Best-effort and infallible by contract: a write failure warns at
    /// the impl site and is swallowed, mirroring the turn-append path.
    /// Default impl is a no-op so non-substrate sinks compile untouched.
    async fn set_meta(&self, _conv_id: &str, _meta: serde_json::Value) {}
}

/// Test-only [`ConversationSink`] backed by a `Mutex<HashMap>`.
///
/// Mirrors the trait shape without any of the substrate machinery.
/// Used as the default sink for [`AgentLoop`] until Phase C3 lands
/// the real impl.
#[derive(Debug, Default)]
pub struct InMemorySink {
    turns: Mutex<HashMap<String, Vec<Turn>>>,
    /// Per-conv metadata sidecar (design §D3). Mirrors the substrate
    /// `_derived/chat/<conv>/meta` node so the hydration K/V is
    /// testable without the kernel.
    meta: Mutex<HashMap<String, serde_json::Value>>,
}

impl InMemorySink {
    /// Create an empty sink.
    pub fn new() -> Self {
        Self::default()
    }

    /// Number of turns recorded for `conv_id`. Useful for tests.
    pub fn turn_count(&self, conv_id: &str) -> usize {
        self.turns
            .lock()
            .map(|g| g.get(conv_id).map(|v| v.len()).unwrap_or(0))
            .unwrap_or(0)
    }
}

#[cfg_attr(not(feature = "browser"), async_trait)]
#[cfg_attr(feature = "browser", async_trait(?Send))]
impl ConversationSink for InMemorySink {
    async fn lock_conversation(&self, _conv_id: &str) {
        // No-op for the in-memory impl. The real per-conv mutex
        // lives in clawft-service-agent (Phase C1).
    }

    async fn append_turn(&self, conv_id: &str, turn: Turn) -> Result<(), String> {
        let mut guard = self
            .turns
            .lock()
            .map_err(|_| "InMemorySink mutex poisoned".to_string())?;
        guard.entry(conv_id.to_string()).or_default().push(turn);
        Ok(())
    }

    async fn publish_status(
        &self,
        _conv_id: &str,
        _status: &str,
        _payload: serde_json::Value,
    ) -> Result<(), String> {
        // Tests don't observe status; drop on the floor.
        Ok(())
    }

    async fn load_history(&self, conv_id: &str) -> Result<Vec<Turn>, String> {
        let guard = self
            .turns
            .lock()
            .map_err(|_| "InMemorySink mutex poisoned".to_string())?;
        Ok(guard.get(conv_id).cloned().unwrap_or_default())
    }

    async fn history(&self, conv_id: &str, window: usize) -> Vec<Turn> {
        let guard = match self.turns.lock() {
            Ok(g) => g,
            Err(_) => return Vec::new(), // poisoned — degrade to empty (design §6)
        };
        let all = guard.get(conv_id).cloned().unwrap_or_default();
        // Truncate to the most recent `window` turns; `0` means all.
        if window == 0 || all.len() <= window {
            all
        } else {
            all[all.len() - window..].to_vec()
        }
    }

    async fn meta(&self, conv_id: &str) -> serde_json::Value {
        self.meta
            .lock()
            .map(|g| g.get(conv_id).cloned().unwrap_or(serde_json::Value::Null))
            .unwrap_or(serde_json::Value::Null)
    }

    async fn set_meta(&self, conv_id: &str, meta: serde_json::Value) {
        if let Ok(mut g) = self.meta.lock() {
            g.insert(conv_id.to_string(), meta);
        }
    }
}

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

    #[allow(dead_code)]
    fn _coerce(_sink: &dyn ConversationSink) {}

    fn make_turn(role: &str, content: &str) -> Turn {
        Turn {
            turn_id: format!("test-{role}"),
            role: role.into(),
            content: content.into(),
            tool_calls: None,
            tool_call_id: None,
            ts_ms: 1_700_000_000_000,
            voice_analysis: None,
            audio: None,
        }
    }

    #[tokio::test]
    async fn round_trip_one_turn() {
        let sink = InMemorySink::new();
        sink.lock_conversation("c1").await;
        sink.append_turn("c1", make_turn("user", "hello"))
            .await
            .unwrap();

        let history = sink.load_history("c1").await.unwrap();
        assert_eq!(history.len(), 1);
        assert_eq!(history[0].role, "user");
        assert_eq!(history[0].content, "hello");
    }

    #[tokio::test]
    async fn empty_history_for_unknown_conv() {
        let sink = InMemorySink::new();
        let history = sink.load_history("nope").await.unwrap();
        assert!(history.is_empty());
    }

    #[tokio::test]
    async fn turns_are_isolated_per_conversation() {
        let sink = InMemorySink::new();
        sink.append_turn("a", make_turn("user", "from a"))
            .await
            .unwrap();
        sink.append_turn("b", make_turn("user", "from b"))
            .await
            .unwrap();

        let a = sink.load_history("a").await.unwrap();
        let b = sink.load_history("b").await.unwrap();
        assert_eq!(a.len(), 1);
        assert_eq!(b.len(), 1);
        assert_eq!(a[0].content, "from a");
        assert_eq!(b[0].content, "from b");
    }

    #[tokio::test]
    async fn append_preserves_order() {
        let sink = InMemorySink::new();
        sink.append_turn("c", make_turn("user", "1")).await.unwrap();
        sink.append_turn("c", make_turn("assistant", "2"))
            .await
            .unwrap();
        sink.append_turn("c", make_turn("tool", "3")).await.unwrap();

        let history = sink.load_history("c").await.unwrap();
        assert_eq!(history.len(), 3);
        assert_eq!(history[0].content, "1");
        assert_eq!(history[1].content, "2");
        assert_eq!(history[2].content, "3");
    }

    #[tokio::test]
    async fn publish_status_is_a_noop() {
        let sink = InMemorySink::new();
        let r = sink
            .publish_status("c1", "thinking", json!({"step": 1}))
            .await;
        assert!(r.is_ok());
    }

    #[tokio::test]
    async fn turn_count_helper_reports_length() {
        let sink = InMemorySink::new();
        assert_eq!(sink.turn_count("x"), 0);
        sink.append_turn("x", make_turn("user", "hi"))
            .await
            .unwrap();
        assert_eq!(sink.turn_count("x"), 1);
    }

    #[tokio::test]
    async fn history_windows_to_the_most_recent_turns() {
        let sink = InMemorySink::new();
        for i in 0..5 {
            sink.append_turn("c", make_turn("user", &i.to_string()))
                .await
                .unwrap();
        }

        // window == 0 → all turns, in append order.
        let all = sink.history("c", 0).await;
        assert_eq!(all.len(), 5);
        assert_eq!(all[0].content, "0");
        assert_eq!(all[4].content, "4");

        // window smaller than history → the tail, in order.
        let tail = sink.history("c", 2).await;
        assert_eq!(tail.len(), 2);
        assert_eq!(tail[0].content, "3");
        assert_eq!(tail[1].content, "4");

        // window larger than history → all of it, no panic.
        let over = sink.history("c", 100).await;
        assert_eq!(over.len(), 5);
    }

    #[tokio::test]
    async fn history_preserves_the_superset_intermediates() {
        // history returns the raw superset — tool intermediates are NOT
        // dropped here (that's P3's hydration filter). This locks the
        // contract the golden gate depends on.
        let sink = InMemorySink::new();
        sink.append_turn("c", make_turn("user", "q")).await.unwrap();
        let tool_call = Turn {
            tool_calls: Some(vec![json!({"name": "search"})]),
            ..make_turn("assistant", "calling")
        };
        sink.append_turn("c", tool_call).await.unwrap();
        let tool_result = Turn {
            tool_call_id: Some("call-1".into()),
            ..make_turn("tool", "result")
        };
        sink.append_turn("c", tool_result).await.unwrap();
        sink.append_turn("c", make_turn("assistant", "final"))
            .await
            .unwrap();

        let turns = sink.history("c", 0).await;
        assert_eq!(turns.len(), 4, "history must keep all superset turns");
        assert!(turns[1].tool_calls.is_some());
        assert_eq!(turns[2].tool_call_id.as_deref(), Some("call-1"));
    }

    #[tokio::test]
    async fn meta_round_trips_and_defaults_to_null() {
        let sink = InMemorySink::new();
        // Absent → Null (P3 defaults hallucination score to 0.0 on Null).
        assert_eq!(sink.meta("c").await, serde_json::Value::Null);

        sink.set_meta("c", json!({"hallucination_score": 0.42}))
            .await;
        assert_eq!(sink.meta("c").await["hallucination_score"], 0.42);

        // Overwrite in place.
        sink.set_meta("c", json!({"hallucination_score": 0.99}))
            .await;
        assert_eq!(sink.meta("c").await["hallucination_score"], 0.99);

        // Metadata is per-conv isolated.
        assert_eq!(sink.meta("other").await, serde_json::Value::Null);
    }

    #[tokio::test]
    async fn default_trait_impls_degrade_safely() {
        // A minimal sink that overrides nothing must inherit the safe
        // defaults: empty history, Null meta, no-op set_meta.
        struct BareSink;
        #[async_trait]
        impl ConversationSink for BareSink {
            async fn lock_conversation(&self, _c: &str) {}
            async fn append_turn(&self, _c: &str, _t: Turn) -> Result<(), String> {
                Ok(())
            }
            async fn publish_status(
                &self,
                _c: &str,
                _s: &str,
                _p: serde_json::Value,
            ) -> Result<(), String> {
                Ok(())
            }
            async fn load_history(&self, _c: &str) -> Result<Vec<Turn>, String> {
                Ok(vec![])
            }
        }
        let sink = BareSink;
        assert!(sink.history("c", 10).await.is_empty());
        assert_eq!(sink.meta("c").await, serde_json::Value::Null);
        sink.set_meta("c", json!({"x": 1})).await; // no panic
    }

    #[tokio::test]
    async fn turn_with_tool_calls_round_trips() {
        let sink = InMemorySink::new();
        let turn = Turn {
            turn_id: "t1".into(),
            role: "assistant".into(),
            content: "calling".into(),
            tool_calls: Some(vec![json!({"name": "echo"})]),
            tool_call_id: None,
            ts_ms: 1_700_000_000_000,
            voice_analysis: None,
            audio: None,
        };
        sink.append_turn("c1", turn.clone()).await.unwrap();
        let history = sink.load_history("c1").await.unwrap();
        assert_eq!(history[0], turn);
    }
}