klieo-core 3.16.0

//! Memory traits — short-term, long-term, episodic.

use crate::error::MemoryError;
use crate::ids::{FactId, RunId, ThreadId};
use crate::llm::Message;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};

/// Outcome of a tool invocation as recorded in the episodic event stream.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "outcome", rename_all = "snake_case")]
pub enum ToolResult {
    /// Tool returned a successful JSON result.
    Ok {
        /// Result payload.
        value: serde_json::Value,
    },
    /// Tool returned an error message.
    Err {
        /// Error string.
        message: String,
    },
}

impl ToolResult {
    /// Build a successful result. Convenience for the
    /// `ToolResult::Ok { value }` struct-variant ceremony.
    pub fn ok(value: serde_json::Value) -> Self {
        Self::Ok { value }
    }

    /// Build an error result. Convenience for the
    /// `ToolResult::Err { message }` struct-variant ceremony.
    pub fn err(message: impl Into<String>) -> Self {
        Self::Err {
            message: message.into(),
        }
    }
}

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

    #[test]
    fn ok_factory_builds_ok_variant() {
        let v = serde_json::json!({"hit": true});
        let r = ToolResult::ok(v.clone());
        match r {
            ToolResult::Ok { value } => assert_eq!(value, v),
            _ => panic!("expected ToolResult::Ok variant"),
        }
    }

    #[test]
    fn err_factory_builds_err_variant() {
        let r = ToolResult::err("boom");
        match r {
            ToolResult::Err { message } => assert_eq!(message, "boom"),
            _ => panic!("expected ToolResult::Err variant"),
        }
    }
}

/// Conversation buffer scoped to a single thread.
///
/// ```
/// # tokio_test::block_on(async {
/// use klieo_core::test_utils::InMemoryShortTerm;
/// use klieo_core::{ShortTermMemory, Message, Role, ThreadId};
///
/// let m = InMemoryShortTerm::default();
/// let thread = ThreadId::new("t1");
/// m.append(thread.clone(), Message {
///     role: Role::User, content: "hi".into(),
///     tool_calls: vec![], tool_call_id: None,
/// }).await.unwrap();
/// let loaded = m.load(thread, 1024).await.unwrap();
/// assert_eq!(loaded.len(), 1);
/// # });
/// ```
#[async_trait]
pub trait ShortTermMemory: Send + Sync {
    /// Append a message to the thread's history.
    async fn append(&self, thread: ThreadId, msg: Message) -> Result<(), MemoryError>;

    /// Append a batch of messages, oldest first. The default appends each in
    /// turn; backends with bulk-insert support should override to collapse the
    /// N writes into one round-trip (the resume path replays a full history).
    async fn append_batch(
        &self,
        thread: ThreadId,
        messages: Vec<Message>,
    ) -> Result<(), MemoryError> {
        for msg in messages {
            self.append(thread.clone(), msg).await?;
        }
        Ok(())
    }

    /// Load up to `max_tokens` of the most-recent messages, oldest first.
    /// Implementations approximate token counts (provider-specific).
    async fn load(&self, thread: ThreadId, max_tokens: usize) -> Result<Vec<Message>, MemoryError>;

    /// Drop all messages for `thread`.
    async fn clear(&self, thread: ThreadId) -> Result<(), MemoryError>;
}

/// Characters per token, the ratio klieo's provider crates use for pre-flight
/// estimates. [`ShortTermMemory::load`]'s contract only requires an
/// approximation.
pub const CHARS_PER_TOKEN: usize = 4;

/// Substituted for the messages a budget walk had to leave out, so a gap in
/// the middle of a thread is visible to the model rather than silent.
pub const ELIDED_MESSAGES_MARKER_PREFIX: &str = "[history budget: ";

fn elision_marker(dropped: usize) -> Message {
    Message::system(format!(
        "{ELIDED_MESSAGES_MARKER_PREFIX}{dropped} earlier message(s) omitted to fit \
         max_history_tokens; the first message and the most recent turns are intact]"
    ))
}

fn cost_of(msg: &Message) -> usize {
    msg.content.chars().count()
}

/// Keeps the thread's first message plus the newest messages that fit
/// `max_tokens`, returned oldest-first — a ready-made
/// [`ShortTermMemory::load`] body for stores that already hold the whole
/// thread in memory.
///
/// Counts Unicode scalar values, not UTF-8 bytes, matching the heuristic
/// `summarize_history` compares against. Counting bytes would read multibyte
/// content (CJK, emoji, umlauts) as roughly three times its real token cost and
/// truncate far too aggressively.
///
/// # The two invariants
///
/// **A non-empty thread never loads as an empty history**, and **the first
/// message is never dropped while later ones are kept.** Both exist because
/// this function decides what an agent can see, and losing either one leaves
/// the agent working with no task.
///
/// Measured 2026-08-18 against the previous behaviour, which returned early on
/// the first over-budget message and therefore returned *nothing* when that
/// message was the newest: a 38,226-character brief under the default 8,000
/// token budget (32,000 chars) reached the model as an empty history, and the
/// request carried the system prompt alone. The agent explored a repository at
/// random and invented its own task. Keeping only the newest message would
/// have fixed that first step and left every later one blind, because the
/// brief is the *oldest* message and the runtime re-loads on every step.
///
/// The ends are kept even when they alone exceed `max_tokens`: the budget
/// bounds the middle, and an over-budget request that a provider rejects
/// loudly is worth more than a blind one it answers.
///
/// # The gap
///
/// The newest messages kept are contiguous, so tool results stay next to the
/// turn that produced them. When the walk cannot reach the first message, the
/// omitted span is replaced by a single marker
/// ([`ELIDED_MESSAGES_MARKER_PREFIX`]) rather than closed silently — a thread
/// that jumps from its brief to a much later turn with no explanation reads as
/// a coherent conversation the model then reasons about wrongly.
///
/// Backends that can push the budget down into the query should fetch fewer
/// rows first and call this on the result — but they must fetch the thread's
/// head row too, since a `LIMIT` over the newest rows cannot honour the second
/// invariant on its own.
pub fn most_recent_within_budget(messages: Vec<Message>, max_tokens: usize) -> Vec<Message> {
    most_recent_within_budget_reporting(messages, max_tokens).kept
}

/// What a budget walk kept, and how much it had to leave out.
#[derive(Debug)]
#[non_exhaustive]
pub struct BudgetedHistory {
    /// The messages to send, oldest-first, including the elision marker.
    pub kept: Vec<Message>,
    /// How many messages the budget forced out. Zero means nothing was lost.
    pub dropped: usize,
}

/// [`most_recent_within_budget`], reporting what it dropped.
///
/// Backends use this to log the loss against the thread id they hold — the
/// pure walk cannot name the thread it truncated, and a truncation nobody can
/// attribute is how a blind agent goes unexplained.
pub fn most_recent_within_budget_reporting(
    messages: Vec<Message>,
    max_tokens: usize,
) -> BudgetedHistory {
    if messages.len() <= 1 {
        return BudgetedHistory {
            kept: messages,
            dropped: 0,
        };
    }
    let budget = max_tokens.saturating_mul(CHARS_PER_TOKEN);

    let mut tail = Vec::new();
    let mut remaining = budget;
    // Walk newest-first, never past the first message: it is claimed below and
    // must not be counted twice on a thread short enough for the walk to reach.
    for msg in messages[1..].iter().rev() {
        let cost = cost_of(msg);
        if cost > remaining && !tail.is_empty() {
            break;
        }
        remaining = remaining.saturating_sub(cost);
        tail.push(msg.clone());
    }
    tail.reverse();

    let dropped = messages.len() - 1 - tail.len();
    let mut kept = Vec::with_capacity(tail.len() + 2);
    kept.push(messages[0].clone());
    if dropped > 0 {
        kept.push(elision_marker(dropped));
    }
    kept.extend(tail);
    BudgetedHistory { kept, dropped }
}

/// Namespacing for long-term memory facts.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum Scope {
    /// Workspace-scoped (multi-agent shared).
    Workspace(String),
    /// Per-agent scoped.
    Agent(String),
    /// Process-global (use sparingly).
    Global,
}

/// One stored fact in long-term memory.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[non_exhaustive]
pub struct Fact {
    /// Plain-text body, embedded for retrieval.
    pub text: String,
    /// Caller-supplied metadata, opaque to the store.
    #[serde(default)]
    pub metadata: serde_json::Value,
}

impl Fact {
    /// Prefer this over struct literals outside `klieo-core` (`#[non_exhaustive]`);
    /// metadata defaults to JSON null — attach it with [`Fact::with_metadata`].
    pub fn new(text: impl Into<String>) -> Self {
        Self {
            text: text.into(),
            metadata: serde_json::Value::Null,
        }
    }

    /// Override the default JSON-null metadata with an opaque caller value.
    pub fn with_metadata(mut self, metadata: serde_json::Value) -> Self {
        self.metadata = metadata;
        self
    }
}

/// Machine-readable declaration of how a [`LongTermMemory`] matches a query.
///
/// [`LongTermMemory::recall`]'s matching rule is implementation-defined, and
/// until this enum existed the only way to learn a store's rule was to read
/// its prose docs. A pipeline that requires semantic retrieval can now assert
/// its substrate at startup instead of discovering the answer in a
/// measurement — see [`LongTermMemory::recall_semantics`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[non_exhaustive]
pub enum RecallSemantics {
    /// Literal substring containment. A paraphrase of a stored fact does not
    /// match. `test_utils::InMemoryLongTerm` is the workspace example.
    Substring,
    /// Nearest-neighbour over embeddings — a paraphrase matches when the
    /// embedding model places it nearby.
    Vector,
    /// Vector recall narrowed or re-ranked by a second signal (graph
    /// traversal, keyword filter, reranker).
    Hybrid,
    /// The store persists facts and returns `k` of them, but applies no
    /// relevance ranking whatsoever — a vector store wired with a
    /// zero-variance embedder, for instance. Retrieval order is arbitrary.
    NonRanking,
    /// The implementation has not declared its rule. Assume nothing.
    Unspecified,
}

impl RecallSemantics {
    /// Whether a paraphrase of a stored fact can retrieve it.
    ///
    /// [`Self::Unspecified`] answers `false`: a caller that needs semantic
    /// recall and cannot establish it does not have it.
    pub fn is_semantic(&self) -> bool {
        matches!(self, Self::Vector | Self::Hybrid)
    }
}

/// Long-term semantic memory.
///
/// ```
/// # tokio_test::block_on(async {
/// use klieo_core::test_utils::InMemoryLongTerm;
/// use klieo_core::{Fact, LongTermMemory, Scope};
///
/// let m = InMemoryLongTerm::default();
/// let scope = Scope::Workspace("ws".into());
/// m.remember(scope.clone(), Fact::new("the sky is blue")).await.unwrap();
/// let hits = m.recall(scope, "sky", 1).await.unwrap();
/// assert_eq!(hits.len(), 1);
/// # });
/// ```
#[async_trait]
pub trait LongTermMemory: Send + Sync {
    /// Store a fact under `scope`. Returns a stable id.
    async fn remember(&self, scope: Scope, fact: Fact) -> Result<FactId, MemoryError>;

    /// Top-`k` recall under `scope` for the supplied query.
    ///
    /// # Contract
    ///
    /// Recall is **best-effort, and both its matching rule and its ordering
    /// are implementation-defined.** Implementations in this workspace span
    /// literal substring matching (`test_utils::InMemoryLongTerm`) to real
    /// cosine similarity over embeddings (the qdrant, pgvector and graph-rag
    /// backends), so a caller that tests against one and deploys on another
    /// gets different results for the same query — in both directions.
    ///
    /// Callers must therefore not assume semantic matching, and must not
    /// depend on ordering unless the specific implementation documents it.
    /// Implementations should state their matching rule and ordering in their
    /// own type docs, and prove the claim by running
    /// [`crate::conformance::long_term_memory`].
    ///
    /// A backend with no score threshold returns its top `k` however
    /// unrelated the query — an empty result means "nothing stored in scope",
    /// not "nothing relevant".
    async fn recall(&self, scope: Scope, query: &str, k: usize) -> Result<Vec<Fact>, MemoryError>;

    /// Declare the matching rule [`Self::recall`] applies, so a caller can
    /// assert its substrate at startup rather than measure it later.
    ///
    /// Defaults to [`RecallSemantics::Unspecified`] — every implementation
    /// should override, and wrappers should delegate to what they wrap. A
    /// store whose ranking depends on an injected embedder must report what
    /// that embedder actually does (see `Embedder::is_ranking` in
    /// `klieo-embed-common`), not what the backend is capable of.
    fn recall_semantics(&self) -> RecallSemantics {
        RecallSemantics::Unspecified
    }

    /// Remove a stored fact.
    async fn forget(&self, id: FactId) -> Result<(), MemoryError>;
}

/// One event in the episodic event stream of a single agent run.
///
/// Marked `#[non_exhaustive]` so additive variants (e.g.
/// [`Self::SummaryCheckpoint`]) can be introduced without forcing a
/// SemVer-major bump. Match arms in downstream crates must include a
/// fallback `_ => …`.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[non_exhaustive]
pub enum Episode {
    /// Run started.
    Started {
        /// Agent name.
        agent: String,
    },
    /// LLM call completed.
    ///
    /// `provider`/`model`/`prompt_tokens`/`completion_tokens` are
    /// `Option` so legacy emit sites can leave them `None`; downstream
    /// projectors fall back to the `LlmIo` sidecar in `klieo-runlog`
    /// when the structured fields are absent.
    LlmCall {
        /// Total tokens reported by the provider (prompt + completion).
        tokens: u32,
        /// Wall-clock latency in milliseconds.
        latency_ms: u32,
        /// Provider identifier — e.g. `"ollama"`, `"openai"`,
        /// `"anthropic"`, `"gemini"`. `None` for older records or
        /// providers that don't expose a stable name.
        ///
        /// `#[serde(default)]` so legacy episodes serialised under
        /// klieo 0.6.x deserialise as `None`.
        #[serde(default)]
        provider: Option<String>,
        /// Model identifier — e.g. `"qwen2.5:14b"`, `"gpt-4o-mini"`.
        #[serde(default)]
        model: Option<String>,
        /// Prompt-side token count when the provider splits the
        /// breakdown; `None` falls back to `tokens` for total-only
        /// reports.
        #[serde(default)]
        prompt_tokens: Option<u32>,
        /// Completion-side token count when the provider splits the
        /// breakdown.
        #[serde(default)]
        completion_tokens: Option<u32>,
    },
    /// Tool call completed.
    ToolCall {
        /// Tool name.
        name: String,
        /// JSON arguments.
        args: serde_json::Value,
        /// Tool outcome.
        result: ToolResult,
    },
    /// Agent published a bus message.
    BusPublish {
        /// Subject.
        subject: String,
    },
    /// Agent received a bus message.
    BusReceive {
        /// Subject.
        subject: String,
    },
    /// Causal link: this run received a bus message caused by another run's
    /// publish. Recorded by `AgentContext::record_received` when the publisher
    /// threaded its run id via the `klieo-causation-run` bus header. Additive —
    /// absent on records from publishers that did not thread the causation header.
    BusCausalLink {
        /// Subject the causal handoff occurred on.
        subject: String,
        /// Run id of the publisher that caused this receive.
        caused_by_run: String,
    },
    /// Run completed successfully.
    Completed,
    /// Run failed.
    Failed {
        /// Error message.
        error: String,
    },
    /// Summarizer checkpoint completed.
    ///
    /// Emitted by [`crate::summarize::summarize_history`] in lieu of
    /// [`Self::LlmCall`] so the audit trail can distinguish summarizer
    /// overhead from substantive agent reasoning. Downstream
    /// observability (e.g. `klieo-runlog`) typically projects this as
    /// a separate step kind so cost / latency attribution stays
    /// faithful.
    SummaryCheckpoint {
        /// Number of older messages folded into the summary call.
        input_message_count: u32,
        /// Length of the resulting summary, in Unicode scalar values.
        summary_chars: u32,
        /// Wall-clock latency of the summarizer call.
        latency_ms: u32,
        /// Total tokens reported by the summarizer LLM (prompt +
        /// completion).
        tokens: u32,
    },
    /// Operational-layer event (klieo-ops). Body is an opaque
    /// `serde_json::Value` to keep klieo-core free of an ops dependency.
    /// klieo-ops provides typed serde conversion helpers via `OpsEvent`.
    Ops(serde_json::Value),
    /// Non-PII tenant attribution stamped at run entry when an external
    /// caller drives the run.
    ///
    /// `tenant_label` is a derived identifier (e.g. truncated SHA-256
    /// of the caller's `sub`) — never the raw principal, which lives
    /// only in server-side tracing/authorization. Emitted at most once
    /// per run, adjacent to [`Self::Started`], so the audit trail can
    /// attribute each run to its driving tenant without admitting PII
    /// into agent memory or LLM-visible context.
    RunAttributed {
        /// Derived non-PII attribution label for the driving caller.
        tenant_label: String,
    },
    /// Cross-hop provenance origin stamped at run entry when an
    /// authenticated external caller supplies a parent-chain anchor.
    ///
    /// `parent_anchor` is the caller's own provenance chain-entry id (or
    /// its run's episodic-root hash) — recorded **verbatim** so the value
    /// equals the caller's identifier and downstream tooling can stitch
    /// klieo→klieo lineage across deployments. It is a **caller-asserted,
    /// unverified** claim (klieo does not own or validate the caller's
    /// chain); it is co-recorded with [`Self::RunAttributed`] so the
    /// claim is attributable to the authenticated principal that made it.
    /// Emitted at most once per run, adjacent to [`Self::Started`]; never
    /// admitted into agent memory or LLM-visible context.
    RunOrigin {
        /// Verbatim caller-supplied cross-hop provenance anchor.
        parent_anchor: String,
    },
    /// This run was spawned by another run in the same process. Recorded at run
    /// entry when the context carries a parent (set by
    /// `AgentContext::child`, or explicitly for a stage-to-stage handoff), so a
    /// run view can draw the parent → child edge.
    ///
    /// Distinct from [`Self::BusCausalLink`], which means a message crossed the
    /// bus, and from [`Self::RunOrigin`], which is an unverified anchor asserted
    /// by an external caller. This one is a local `RunId` the process minted.
    SpawnedBy {
        /// Run that spawned this one. Carries only the id, like
        /// [`Self::BusCausalLink`]'s `caused_by_run`: a view that draws the edge
        /// already knows every loaded run's agent name, and a second copy here
        /// could only ever disagree with it.
        parent_run: String,
    },
    /// A graphRAG recall performed during the run. Recorded by the
    /// recall-recording wrapper so the run view can surface — and
    /// deep-link — retrieval calls. `query` is redacted + length-bounded
    /// at the recording boundary before it reaches this episode.
    MemoryRecall {
        /// Redacted, length-bounded recall query text.
        query: String,
        /// Requested top-k.
        #[serde(default)]
        k: u32,
        /// Fact ids the recall returned.
        #[serde(default)]
        returned_fact_ids: Vec<FactId>,
    },
    /// A human reviewer edited tool-call arguments during a HITL
    /// edit-and-resume (ADR-054): `resume_from_checkpoint` dispatched some
    /// pending tool calls with operator-supplied arguments and skipped
    /// others. Recorded **before** dispatch, so the provenance chain shows
    /// human substitution rather than model-proposed calls.
    ///
    /// Carries tool_call **ids only** — never args or values — so a
    /// potentially-PII operator edit never enters the durable audit chain.
    /// The executed arguments live in short-term memory / the `ToolCall`
    /// dispatch itself, not here.
    OperatorEdit {
        /// Ids of pending tool calls dispatched with operator-supplied
        /// arguments.
        edited: Vec<String>,
        /// Ids of pending tool calls the operator omitted, and which were
        /// therefore skipped.
        skipped: Vec<String>,
    },
}

impl Episode {
    /// Construct an [`Episode::LlmCall`] with the legacy two-field shape
    /// (`tokens` + `latency_ms`), leaving the 0.7-added
    /// `provider`/`model`/`prompt_tokens`/`completion_tokens` fields as
    /// `None`.
    ///
    /// Use this when the emit site does not have the enriched fields to
    /// hand — e.g. test fixtures, providers that only report total
    /// tokens, or call sites being migrated incrementally. For full
    /// 0.7 emit semantics, construct the struct variant directly.
    ///
    /// ```
    /// use klieo_core::Episode;
    /// let ep = Episode::llm_call(42, 17);
    /// match ep {
    ///     Episode::LlmCall { tokens, latency_ms, provider, .. } => {
    ///         assert_eq!(tokens, 42);
    ///         assert_eq!(latency_ms, 17);
    ///         assert!(provider.is_none());
    ///     }
    ///     _ => unreachable!(),
    /// }
    /// ```
    pub fn llm_call(tokens: u32, latency_ms: u32) -> Self {
        Episode::LlmCall {
            tokens,
            latency_ms,
            provider: None,
            model: None,
            prompt_tokens: None,
            completion_tokens: None,
        }
    }
}

/// Filter passed to `EpisodicMemory::list_runs`.
#[derive(Debug, Clone, Default)]
pub struct RunFilter {
    /// Filter by agent name (substring match).
    pub agent: Option<String>,
    /// Inclusive lower bound on `started_at`.
    pub since: Option<DateTime<Utc>>,
    /// Inclusive upper bound on `started_at`.
    pub until: Option<DateTime<Utc>>,
    /// Maximum rows returned.
    pub limit: Option<usize>,
}

/// Index-row summary returned by `list_runs`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RunSummary {
    /// Run id.
    pub run_id: RunId,
    /// Agent name.
    pub agent: String,
    /// First-event timestamp.
    pub started_at: DateTime<Utc>,
    /// Last-event timestamp, if completed/failed.
    pub finished_at: Option<DateTime<Utc>>,
    /// Number of episodes.
    pub episode_count: u32,
}

/// Append-only event log of agent runs.
///
/// ```
/// # tokio_test::block_on(async {
/// use klieo_core::test_utils::InMemoryEpisodic;
/// use klieo_core::{Episode, EpisodicMemory, RunId};
///
/// let m = InMemoryEpisodic::default();
/// let run = RunId::new();
/// m.record(run, Episode::Started { agent: "a".into() }).await.unwrap();
/// let events = m.replay(run).await.unwrap();
/// assert_eq!(events.len(), 1);
/// # });
/// ```
#[async_trait]
pub trait EpisodicMemory: Send + Sync {
    /// Record an episode for `run`.
    async fn record(&self, run: RunId, event: Episode) -> Result<(), MemoryError>;

    /// Replay all episodes for `run` in order.
    async fn replay(&self, run: RunId) -> Result<Vec<Episode>, MemoryError>;

    /// Replay all episodes for `run` in order, each paired with the time it was
    /// recorded when the store keeps one.
    ///
    /// Default: [`Self::replay`] with `None` for every timestamp, so the frozen
    /// trait gains no required method (`docs/SEMVER.md`). Stores that persist a
    /// per-event time **should** override this — `klieo-memory-sqlite` has had a
    /// `ts` column all along, and dropping it here is why every projected
    /// `RunLog` reports a zero duration.
    async fn replay_with_times(
        &self,
        run: RunId,
    ) -> Result<Vec<(Episode, Option<DateTime<Utc>>)>, MemoryError> {
        Ok(self
            .replay(run)
            .await?
            .into_iter()
            .map(|episode| (episode, None))
            .collect())
    }

    /// Replay episodes for many runs.
    ///
    /// Returns one entry per requested run, in the requested order. A run with
    /// no recorded episodes yields an empty `Vec` so the caller's node set
    /// stays complete.
    ///
    /// # Performance
    ///
    /// The default implementation issues one `replay` call per run (N+1).
    /// Stores with a batched read **must** override this method — the SQLite
    /// impl uses a single `WHERE run_id IN (…)` query.
    async fn replay_many(&self, runs: &[RunId]) -> Result<Vec<(RunId, Vec<Episode>)>, MemoryError> {
        let mut out = Vec::with_capacity(runs.len());
        for &run in runs {
            out.push((run, self.replay(run).await?));
        }
        Ok(out)
    }

    /// List run summaries matching `filter`.
    async fn list_runs(&self, filter: RunFilter) -> Result<Vec<RunSummary>, MemoryError>;
}

/// Resolved trio of memory handles ready to drop into an
/// [`crate::agent::AgentContext`] or an `App`.
///
/// Impl crates (`klieo-memory-sqlite`, `klieo-memory-neo4j`,
/// `klieo-memory-qdrant`) provide `From` conversions where they cover
/// the full trio. Crates that only carry a subset (Neo4j covers
/// short + episodic; Qdrant covers long) compose via the `App` builder's
/// per-trait setters rather than a direct `From`.
#[derive(Clone)]
pub struct MemoryHandles {
    /// Short-term conversation memory.
    pub short_term: std::sync::Arc<dyn ShortTermMemory>,
    /// Long-term semantic memory.
    pub long_term: std::sync::Arc<dyn LongTermMemory>,
    /// Episodic event log.
    pub episodic: std::sync::Arc<dyn EpisodicMemory>,
}

impl MemoryHandles {
    /// Build directly from three already-`Arc`-wrapped handles.
    /// Most callers go through an impl crate's `From` instead.
    pub fn new(
        short_term: std::sync::Arc<dyn ShortTermMemory>,
        long_term: std::sync::Arc<dyn LongTermMemory>,
        episodic: std::sync::Arc<dyn EpisodicMemory>,
    ) -> Self {
        Self {
            short_term,
            long_term,
            episodic,
        }
    }
}

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

    #[test]
    fn fact_new_defaults_metadata_null() {
        let f = Fact::new("alice likes tea");
        assert_eq!(f.text, "alice likes tea");
        assert_eq!(f.metadata, serde_json::Value::Null);
        let f2 = Fact::new("x").with_metadata(serde_json::json!({"k":"v"}));
        assert_eq!(f2.metadata, serde_json::json!({"k":"v"}));
    }
}

#[cfg(test)]
mod budget_tests {
    use super::*;
    use crate::llm::Role;

    fn msg(content: &str) -> Message {
        Message {
            role: Role::User,
            content: content.into(),
            tool_calls: vec![],
            tool_call_id: None,
        }
    }

    /// The budget is in Unicode scalar values, not UTF-8 bytes. Counting bytes
    /// would price this three-byte-per-char content at 3x and drop messages
    /// that genuinely fit.
    #[test]
    fn multibyte_content_is_not_charged_at_its_byte_length() {
        // 4 chars, 12 bytes each. A 3-token budget is 12 chars, so exactly
        // three of them fit by character count and only one by byte count.
        let history = vec![msg("日本語だ"), msg("日本語だ"), msg("日本語だ")];
        assert_eq!(history[0].content.len(), 12, "guard: content is multibyte");

        let kept = most_recent_within_budget(history, 3);

        assert_eq!(
            kept.len(),
            3,
            "all three fit at 4 chars each against a 12-char budget; \
             charging UTF-8 bytes would have kept only one"
        );
    }

    #[test]
    fn a_tighter_budget_drops_the_middle_and_keeps_both_ends() {
        let history = vec![msg("first"), msg("bbbb"), msg("cccc"), msg("dddd")];

        // 2 tokens = 8 chars, which the two newest messages fill exactly.
        let kept = most_recent_within_budget(history, 2);

        let contents: Vec<&str> = kept.iter().map(|m| m.content.as_str()).collect();
        assert_eq!(
            contents.first(),
            Some(&"first"),
            "the brief is never dropped"
        );
        assert_eq!(
            contents.last(),
            Some(&"dddd"),
            "the newest turn is never dropped"
        );
        assert!(
            contents[1].starts_with(ELIDED_MESSAGES_MARKER_PREFIX),
            "the omitted span is marked, not closed silently: {contents:?}"
        );
    }

    /// The defect this function exists to make impossible: a brief larger than
    /// the whole budget used to return early and hand the agent nothing, so the
    /// request carried a system prompt and no task.
    #[test]
    fn a_single_oversized_message_still_loads() {
        let history = vec![msg("far larger than the budget allows")];

        let kept = most_recent_within_budget(history, 1);

        assert_eq!(
            kept.len(),
            1,
            "a non-empty thread must never load as an empty history"
        );
    }

    /// The brief is the OLDEST message and the runtime re-loads every step, so
    /// protecting only the newest message leaves every later step blind.
    #[test]
    fn the_first_message_survives_when_later_messages_fill_the_budget() {
        let history = vec![
            msg("the brief"),
            msg("tool output that fills the budget on its own"),
            msg("more tool output that also fills it"),
        ];

        let kept = most_recent_within_budget(history, 1);

        assert_eq!(
            kept.first().map(|m| m.content.as_str()),
            Some("the brief"),
            "the first message outranks the budget"
        );
        assert_eq!(
            kept.last().map(|m| m.content.as_str()),
            Some("more tool output that also fills it"),
            "so does the newest"
        );
    }

    /// A thread of one message must not be counted as both ends and returned
    /// twice.
    #[test]
    fn a_one_message_thread_is_returned_once() {
        let kept = most_recent_within_budget(vec![msg("only")], 1_000);

        assert_eq!(kept.len(), 1);
    }

    #[test]
    fn a_history_that_fits_is_returned_unchanged_and_unmarked() {
        let history = vec![msg("a"), msg("b"), msg("c")];

        let kept = most_recent_within_budget(history, 1_000);

        let contents: Vec<&str> = kept.iter().map(|m| m.content.as_str()).collect();
        assert_eq!(
            contents,
            vec!["a", "b", "c"],
            "no marker when nothing is cut"
        );
    }
}

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

    #[allow(dead_code)]
    fn _assert_dyn_short(_: &dyn ShortTermMemory) {}
    #[allow(dead_code)]
    fn _assert_dyn_long(_: &dyn LongTermMemory) {}
    #[allow(dead_code)]
    fn _assert_dyn_episodic(_: &dyn EpisodicMemory) {}

    /// Maps each variant to its published snake_case wire discriminant.
    ///
    /// The exhaustive match is the drift guard: a new `Episode` variant
    /// fails to compile here until its discriminant is added, which is the
    /// signal to also publish a payload schema under `docs/schemas/runlog/`
    /// and a fixture in `tests/schema_drift.rs`. `#[non_exhaustive]` does not
    /// force a wildcard inside the defining crate, so this stays exhaustive.
    fn kind_discriminant(episode: &Episode) -> &'static str {
        match episode {
            Episode::Started { .. } => "started",
            Episode::LlmCall { .. } => "llm_call",
            Episode::ToolCall { .. } => "tool_call",
            Episode::BusPublish { .. } => "bus_publish",
            Episode::BusReceive { .. } => "bus_receive",
            Episode::BusCausalLink { .. } => "bus_causal_link",
            Episode::Completed => "completed",
            Episode::Failed { .. } => "failed",
            Episode::SummaryCheckpoint { .. } => "summary_checkpoint",
            Episode::Ops(_) => "ops",
            Episode::RunAttributed { .. } => "run_attributed",
            Episode::RunOrigin { .. } => "run_origin",
            Episode::SpawnedBy { .. } => "spawned_by",
            Episode::MemoryRecall { .. } => "memory_recall",
            Episode::OperatorEdit { .. } => "operator_edit",
        }
    }

    fn one_sample_per_variant() -> Vec<Episode> {
        vec![
            Episode::Started {
                agent: String::new(),
            },
            Episode::LlmCall {
                tokens: 0,
                latency_ms: 0,
                provider: None,
                model: None,
                prompt_tokens: None,
                completion_tokens: None,
            },
            Episode::ToolCall {
                name: String::new(),
                args: serde_json::Value::Null,
                result: ToolResult::Ok {
                    value: serde_json::Value::Null,
                },
            },
            Episode::BusPublish {
                subject: String::new(),
            },
            Episode::BusReceive {
                subject: String::new(),
            },
            Episode::BusCausalLink {
                subject: String::new(),
                caused_by_run: String::new(),
            },
            Episode::Completed,
            Episode::Failed {
                error: String::new(),
            },
            Episode::SummaryCheckpoint {
                input_message_count: 0,
                summary_chars: 0,
                latency_ms: 0,
                tokens: 0,
            },
            Episode::Ops(serde_json::Value::Null),
            Episode::RunAttributed {
                tenant_label: String::new(),
            },
            Episode::RunOrigin {
                parent_anchor: String::new(),
            },
            Episode::SpawnedBy {
                parent_run: String::new(),
            },
            Episode::MemoryRecall {
                query: String::new(),
                k: 0,
                returned_fact_ids: Vec::new(),
            },
            Episode::OperatorEdit {
                edited: Vec::new(),
                skipped: Vec::new(),
            },
        ]
    }

    /// The set of Rust `Episode` discriminants must equal the `kind` enum
    /// published in the envelope schema. Compile-time exhaustiveness of
    /// [`kind_discriminant`] plus this runtime equality close the drift loop
    /// from the type side; `tests/schema_drift.rs` validates payload shapes.
    #[test]
    fn episode_discriminants_match_published_envelope_schema() {
        let mut discriminants: Vec<&str> = one_sample_per_variant()
            .iter()
            .map(kind_discriminant)
            .collect();
        discriminants.sort_unstable();

        let schema_path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
            .join("../../docs/schemas/runlog/episode.schema.json");
        let text = std::fs::read_to_string(&schema_path)
            .unwrap_or_else(|err| panic!("read {}: {err}", schema_path.display()));
        let schema: serde_json::Value =
            serde_json::from_str(&text).expect("envelope schema parses");
        let mut published: Vec<&str> = schema["properties"]["kind"]["enum"]
            .as_array()
            .expect("envelope schema declares a kind enum")
            .iter()
            .map(|value| value.as_str().expect("kind enum is strings"))
            .collect();
        published.sort_unstable();

        assert_eq!(
            discriminants, published,
            "Episode variants have drifted from the published envelope kind enum",
        );
    }

    /// Episodes serialised under klieo 0.6.x carry only `tokens` and
    /// `latency_ms`. The four 0.7 fields must decode as `None` via
    /// `#[serde(default)]`. Without that attribute, every persisted
    /// 0.6 row breaks replay on upgrade.
    #[test]
    fn legacy_llm_call_json_deserialises_with_none_for_new_fields() {
        let legacy = serde_json::json!({
            "LlmCall": {
                "tokens": 42,
                "latency_ms": 17
            }
        });
        let ep: Episode = serde_json::from_value(legacy).expect("legacy LlmCall decodes");
        match ep {
            Episode::LlmCall {
                tokens,
                latency_ms,
                provider,
                model,
                prompt_tokens,
                completion_tokens,
            } => {
                assert_eq!(tokens, 42);
                assert_eq!(latency_ms, 17);
                assert!(provider.is_none());
                assert!(model.is_none());
                assert!(prompt_tokens.is_none());
                assert!(completion_tokens.is_none());
            }
            other => panic!("expected LlmCall, got {other:?}"),
        }
    }

    /// [`Episode::llm_call`] returns the struct variant with all four
    /// 0.7-added enrichment fields as `None`, preserving the legacy
    /// emit shape for callers that don't have provider/model split.
    #[test]
    fn llm_call_ctor_leaves_enrichment_fields_none() {
        match Episode::llm_call(42, 17) {
            Episode::LlmCall {
                tokens,
                latency_ms,
                provider,
                model,
                prompt_tokens,
                completion_tokens,
            } => {
                assert_eq!(tokens, 42);
                assert_eq!(latency_ms, 17);
                assert!(provider.is_none());
                assert!(model.is_none());
                assert!(prompt_tokens.is_none());
                assert!(completion_tokens.is_none());
            }
            other => panic!("expected LlmCall, got {other:?}"),
        }
    }

    /// 0.7 emit sites with the full field set round-trip through serde
    /// unchanged.
    #[test]
    fn enriched_llm_call_round_trips() {
        let original = Episode::LlmCall {
            tokens: 60,
            latency_ms: 17,
            provider: Some("ollama".into()),
            model: Some("qwen2.5:14b".into()),
            prompt_tokens: Some(40),
            completion_tokens: Some(20),
        };
        let json = serde_json::to_value(&original).expect("serialises");
        let back: Episode = serde_json::from_value(json).expect("deserialises");
        match back {
            Episode::LlmCall {
                provider, model, ..
            } => {
                assert_eq!(provider.as_deref(), Some("ollama"));
                assert_eq!(model.as_deref(), Some("qwen2.5:14b"));
            }
            other => panic!("expected LlmCall, got {other:?}"),
        }
    }

    /// [`Episode::MemoryRecall`] round-trips through serde, and a legacy
    /// record missing `k` / `returned_fact_ids` (recorded before either
    /// field existed) decodes via `#[serde(default)]` rather than failing
    /// replay.
    #[test]
    fn memory_recall_round_trips_and_defaults_legacy() {
        let ep = Episode::MemoryRecall {
            query: "q".into(),
            k: 5,
            returned_fact_ids: vec![FactId::new("fact_1")],
        };
        let json = serde_json::to_value(&ep).expect("serialises");
        let back: Episode = serde_json::from_value(json).expect("deserialises");
        match back {
            Episode::MemoryRecall {
                query,
                k,
                returned_fact_ids,
            } => {
                assert_eq!(query, "q");
                assert_eq!(k, 5);
                assert_eq!(returned_fact_ids, vec![FactId::new("fact_1")]);
            }
            other => panic!("expected MemoryRecall, got {other:?}"),
        }

        let legacy = serde_json::json!({ "MemoryRecall": { "query": "q" } });
        let ep: Episode = serde_json::from_value(legacy).expect("legacy MemoryRecall decodes");
        match ep {
            Episode::MemoryRecall {
                k,
                returned_fact_ids,
                ..
            } => {
                assert_eq!(k, 0);
                assert!(returned_fact_ids.is_empty());
            }
            other => panic!("expected MemoryRecall, got {other:?}"),
        }
    }
    /// `InMemoryLongTerm` is a substring matcher, and a downstream pipeline
    /// shipped it as its production long-term memory for a week because the
    /// only place that said so was prose. `recall_semantics` makes the same
    /// fact assertable at startup.
    #[test]
    fn in_memory_long_term_declares_substring_recall() {
        use crate::test_utils::InMemoryLongTerm;
        let m = InMemoryLongTerm::default();
        assert_eq!(m.recall_semantics(), RecallSemantics::Substring);
        assert!(!m.recall_semantics().is_semantic());
    }

    /// An implementation that has not declared its rule must not read as
    /// semantic — a caller that cannot establish semantic recall does not
    /// have it.
    #[test]
    fn unspecified_is_not_semantic() {
        assert!(!RecallSemantics::Unspecified.is_semantic());
        assert!(!RecallSemantics::NonRanking.is_semantic());
        assert!(RecallSemantics::Vector.is_semantic());
        assert!(RecallSemantics::Hybrid.is_semantic());
    }
}