//! The session's active context view: an installed LCM frontier.
//!
//! A frontier is what the session has *decided* to carry forward. Raw events below its
//! `drained_through_seq` are represented by `summary` plus the source children they were built
//! from, not by replaying them into every provider request. Everything above it is still raw.
//!
//! Two properties are structural rather than asserted:
//!
//! * Installation is ONE durable append. A crash before it leaves the previous view active; a
//!   crash after it leaves the new one. There is no window in which the session has neither.
//! * `plan` is pure and takes the previous frontier explicitly, so the same event page always
//!   yields the same decision. Nothing about wall-clock or a mutable counter can change a view
//!   that was already installed.

use anyhow::Result;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};

use crate::store::Event;

/// The event kind an installed frontier is recorded as.
pub const FRONTIER_KIND: &str = "lcm_frontier";

/// Scratchpad record types that carry conversation. The daemon assembles a request from `input`
/// events and from scratchpad records of these types, and from nothing else, so these are what a
/// frontier's size is measured in. Kernel placement, cell output and transient scratchpad noise
/// are covered by the range without deciding where it ends.
const CONVERSATION_SCRATCHPAD_TYPES: &[&str] = &["model_response", "tool_result"];

/// Whether an event carries conversation, and therefore counts toward the kept window.
fn carries_conversation(event: &Event) -> bool {
    match event.kind.as_str() {
        "input" => true,
        "scratchpad" => event
            .payload
            .get("type")
            .and_then(Value::as_str)
            .is_some_and(|kind| CONVERSATION_SCRATCHPAD_TYPES.contains(&kind)),
        "tool.result" => true,
        _ => false,
    }
}

/// Bounded size of a derived recollection, in characters.
const RECOLLECTION_BUDGET: usize = 4096;

/// Conversation-carrying events that stay raw below a newly installed frontier.
///
/// The daemon passes the operator's configured session window here, so there is one window in the
/// system rather than a second constant that disagrees the moment config changes.
pub const DEFAULT_KEEP_EVENTS: usize = 20;

/// One archived group of covered events, as the frontier remembers them.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct SourceChild {
    pub kind: String,
    pub first_seq: i64,
    pub last_seq: i64,
    pub count: usize,
}

/// Where exact actor testimony came from. Testimony without its source is a claim, not testimony.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ActorTestimonySource {
    pub actor_id: String,
    pub session_id: String,
    pub source_ref: String,
}

/// An installed frontier: the active view of a session's context.
///
/// Deliberately NOT `Serialize`/`Deserialize`. The canonical wire form is `to_payload` /
/// `from_payload`, which names the fields the record actually stores (`created_at_ms`, and
/// testimony as one nested object). A derived shape would be a SECOND serialization of the same
/// type that disagrees with the stored one - `to_value` would look reasonable and produce a
/// payload `from_payload` rejects - so the type has exactly one.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Frontier {
    /// Monotone per session. Every install increments it, so a stale install is detectable.
    pub epoch: u64,
    /// The covered prefix, by event sequence. Everything at or below is represented by the
    /// summary; everything above is still raw.
    pub drained_through_seq: i64,
    /// The derived summary. Never presented as original evidence.
    pub summary: String,
    /// How the summary was derived, so a reader is never misled about its provenance.
    pub derived_by: String,
    pub created_ms: i64,
    pub source_children: Vec<SourceChild>,
    /// Exact actor testimony, carried as itself when one is installed. Absent stays absent.
    pub actor_testimony: Option<String>,
    /// The identity the testimony came from, carried beside it so an exact quote is attributable.
    pub actor_testimony_source: Option<ActorTestimonySource>,
}

impl Frontier {
    /// A view with nothing in it, for tests that care about one field.
    #[cfg(test)]
    fn for_test() -> Self {
        Self {
            epoch: 1,
            drained_through_seq: 1,
            summary: String::new(),
            derived_by: "test".to_string(),
            created_ms: 0,
            source_children: Vec::new(),
            actor_testimony: None,
            actor_testimony_source: None,
        }
    }

    /// The payload shape recorded in the event log.
    pub fn to_payload(&self) -> Value {
        json!({
            "type": "lcm_frontier",
            "epoch": self.epoch,
            "drained_through_seq": self.drained_through_seq,
            "summary": self.summary,
            "derived_by": self.derived_by,
            "created_at_ms": self.created_ms,
            "source_children": self.source_children,
            "actor_testimony": testimony_payload(
                self.actor_testimony.as_deref(),
                self.actor_testimony_source.as_ref(),
            ),
        })
    }

    /// Reads a frontier back out of a recorded payload. A malformed record is not a frontier:
    /// it is refused rather than half-read, because a half-read view silently loses context.
    pub fn from_payload(payload: &Value) -> Result<Self> {
        let is_frontier = payload.get("type").and_then(Value::as_str) == Some("lcm_frontier")
            || payload.get("kind").and_then(Value::as_str) == Some("lcm_frontier");
        if !is_frontier {
            anyhow::bail!("not an lcm_frontier payload");
        }
        let epoch = payload
            .get("epoch")
            .and_then(Value::as_u64)
            .ok_or_else(|| anyhow::anyhow!("frontier has no epoch"))?;
        let drained_through_seq = payload
            .get("drained_through_seq")
            .and_then(Value::as_i64)
            .ok_or_else(|| anyhow::anyhow!("frontier has no drained_through_seq"))?;
        let summary = payload
            .get("summary")
            .and_then(Value::as_str)
            .ok_or_else(|| anyhow::anyhow!("frontier has no summary"))?
            .to_string();
        let derived_by = payload
            .get("derived_by")
            .and_then(Value::as_str)
            .unwrap_or("unknown")
            .to_string();
        let created_ms = payload
            .get("created_at_ms")
            .and_then(Value::as_i64)
            .unwrap_or(0);
        let source_children: Vec<SourceChild> = payload
            .get("source_children")
            .map(|value| serde_json::from_value(value.clone()))
            .transpose()?
            .unwrap_or_default();
        let testimony = payload.get("actor_testimony");
        let actor_testimony = testimony
            .and_then(|t| t.get("exact_text"))
            .and_then(Value::as_str)
            .map(str::to_string);
        let actor_testimony_source = testimony
            .and_then(|t| t.get("source"))
            .map(|source| serde_json::from_value(source.clone()))
            .transpose()?;
        Ok(Self {
            epoch,
            drained_through_seq,
            summary,
            derived_by,
            created_ms,
            source_children,
            actor_testimony,
            actor_testimony_source,
        })
    }
}

/// The `actor_testimony` object: absent when there is no testimony, and otherwise the exact text
/// and the identity it came from, so a reader never has to guess either.
fn testimony_payload(exact: Option<&str>, source: Option<&ActorTestimonySource>) -> Value {
    if exact.is_none() && source.is_none() {
        return Value::Null;
    }
    let mut object = serde_json::Map::new();
    if let Some(text) = exact {
        object.insert("exact_text".into(), Value::String(text.to_string()));
    }
    if let Some(source) = source {
        object.insert(
            "source".into(),
            serde_json::to_value(source).unwrap_or(Value::Null),
        );
    }
    Value::Object(object)
}

/// Plans the frontier a session's event page implies, or `None` when nothing should be installed.
///
/// `keep` is the number of most recent conversation-carrying events that stay raw. A frontier is
/// installed only when at least that many would be drained: finite thresholds, so a session does
/// not reinstall a view on every turn (the hysteresis the LCM keeps for the same reason).
pub fn plan_frontier(
    events: &[Event],
    previous: Option<&Frontier>,
    keep: usize,
) -> Option<Frontier> {
    let covered = previous
        .map(|frontier| frontier.drained_through_seq)
        .unwrap_or(0);
    let uncovered: Vec<&Event> = events.iter().filter(|event| event.id > covered).collect();

    let conversation: Vec<&Event> = uncovered
        .iter()
        .copied()
        .filter(|event| carries_conversation(event))
        .collect();
    if conversation.len() <= keep {
        return None;
    }
    // Finite threshold with hysteresis: an install must drain at least as much conversation as it
    // keeps. Draining one event per turn would reinstall a view on practically every turn, which
    // is the continual tiny compaction the LCM thresholds exist to prevent.
    let drained_conversation = conversation.len() - keep;
    if drained_conversation < keep.max(1) {
        return None;
    }
    // The drain boundary is the last event before the kept window. Coverage is a prefix by
    // sequence, so it also covers the non-conversation records in between.
    let last_drained = conversation[conversation.len() - keep - 1].id;
    let drained: Vec<&Event> = uncovered
        .iter()
        .copied()
        .filter(|event| event.id <= last_drained)
        .collect();
    if drained.is_empty() {
        return None;
    }

    // A frontier's summary is CUMULATIVE: it must describe everything the session carries, not
    // just what this install added. Summarizing only the delta would silently drop the previous
    // view's content the moment a newer view replaced it, which is a context loss disguised as an
    // install. The oldest lines are what fall off when the budget is reached.
    let summary = carry_forward(previous, &drained);
    if summary.trim().is_empty() {
        // An empty recollection is not a view. Refusing here is what keeps the previous view
        // active instead of installing one that says nothing.
        return None;
    }

    let mut source_children: Vec<SourceChild> = Vec::new();
    for event in &drained {
        match source_children.last_mut() {
            Some(child) if child.kind == event.kind => {
                child.last_seq = event.id;
                child.count += 1;
            }
            _ => source_children.push(SourceChild {
                kind: event.kind.clone(),
                first_seq: event.id,
                last_seq: event.id,
                count: 1,
            }),
        }
    }

    let created_ms = drained.last().map(|event| event.created_ms).unwrap_or(0);
    Some(Frontier {
        epoch: previous.map(|frontier| frontier.epoch).unwrap_or(0) + 1,
        drained_through_seq: last_drained,
        summary,
        derived_by: "deterministic_prune".to_string(),
        created_ms,
        source_children,
        // A prune does not author testimony, and inventing one would be exactly the misrepresentation
        // this field exists to prevent. Installed testimony is therefore carried forward UNCHANGED
        // for as long as it stands: it is exact text the session is keeping, so a summary of it is
        // not a replacement for it, and dropping it because a drain happened to cover its source
        // would silently delete the one thing that is not allowed to be paraphrased.
        // Supersession is explicit: an installer that means to replace it passes its own testimony,
        // which overrides these fields exactly as an explicit orientation change overrides a carrier.
        actor_testimony: previous.and_then(|frontier| frontier.actor_testimony.clone()),
        actor_testimony_source: previous
            .and_then(|frontier| frontier.actor_testimony_source.clone()),
    })
}

/// A bounded, deterministic recollection of a drained range: this install's delta.
///
/// It is a summary, so it is never presented as original evidence: each line keeps the event id it
/// came from, and the exact text remains reachable from the store.
fn recollection(drained: &[&Event]) -> String {
    let mut out = String::new();
    for event in drained {
        if !carries_conversation(event) {
            continue;
        }
        let text = event
            .payload
            .get("content")
            .or_else(|| event.payload.get("text"))
            .or_else(|| event.payload.get("output"))
            .and_then(Value::as_str)
            .unwrap_or("")
            .to_string();
        let text = text.trim();
        if text.is_empty() {
            continue;
        }
        let clipped: String = text.chars().take(240).collect();
        let label = event
            .payload
            .get("type")
            .and_then(Value::as_str)
            .unwrap_or(event.kind.as_str());
        out.push_str(&format!("- [{label}#{}] {clipped}\n", event.id));
    }
    out
}

/// The previous view's summary followed by this install's delta, trimmed to the budget from the
/// front: the newest content is what must survive, and the covered records stay in the store.
fn carry_forward(previous: Option<&Frontier>, drained: &[&Event]) -> String {
    let delta = recollection(drained);
    let mut combined = match previous {
        Some(frontier) if !frontier.summary.trim().is_empty() => {
            format!("{}\n{}", frontier.summary.trim_end(), delta)
        }
        _ => delta,
    };
    if combined.len() > RECOLLECTION_BUDGET {
        let overflow = combined.len() - RECOLLECTION_BUDGET;
        // A byte offset is not a character boundary. Slicing at one panics the runner task, and the
        // text that triggers it is the ordinary case - any non-ASCII history, an emoticon in a
        // message - so the cut walks forward to the next boundary before it is used.
        let mut start = overflow;
        while start < combined.len() && !combined.is_char_boundary(start) {
            start += 1;
        }
        let cut = combined[start..]
            .find('\n')
            .map(|offset| start + offset + 1)
            .unwrap_or(start);
        combined = format!(
            "- (older recollection trimmed; those records remain in the store)\n{}",
            &combined[cut..]
        );
    }
    combined
}

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

    fn conversation(id: i64, kind: &str, text: &str) -> Event {
        Event {
            id,
            kind: kind.to_string(),
            payload: json!({"type": "model_response", "content": text}),
            created_ms: id,
        }
    }

    /// Non-ASCII history is the ordinary case, and a budget that trims it must not panic.
    ///
    /// The prefix is multi-byte on purpose: the trim crosses it at a byte offset that is almost
    /// never a character boundary, which is exactly what used to panic the runner task.
    #[test]
    fn a_multi_byte_recollection_is_trimmed_to_a_character_boundary() {
        let previous = Frontier {
            summary: format!("{}\n- [input#1] earlier", "😀ことば".repeat(400)),
            ..Frontier::for_test()
        };
        let events = [
            conversation(9, "input", "the newest turn"),
            conversation(10, "input", "the newest turn continued"),
        ];
        let drained: Vec<&Event> = events.iter().collect();
        let carried = carry_forward(Some(&previous), &drained);
        assert!(
            carried.starts_with("- (older recollection trimmed"),
            "an over-budget recollection must say it was trimmed"
        );
        assert!(
            carried.len() <= RECOLLECTION_BUDGET + 128,
            "the trimmed recollection must be bounded, got {}",
            carried.len()
        );
        // The cut lands on a line boundary, and the newest content is what survives it.
        assert!(
            carried.trim_end().ends_with("the newest turn continued"),
            "the newest recollection must survive the trim, got {:?}",
            &carried[carried.len().saturating_sub(80)..]
        );
    }

    /// Testimony is carried until an explicit supersession, and a plan does not invent one.
    #[test]
    fn testimony_is_carried_forward_until_superseded() {
        let previous = Frontier {
            summary: "- [input#1] earlier".to_string(),
            actor_testimony: Some("I said the exact words".to_string()),
            actor_testimony_source: Some(ActorTestimonySource {
                actor_id: "dawn".to_string(),
                session_id: "s".to_string(),
                source_ref: "ref-1".to_string(),
            }),
            ..Frontier::for_test()
        };
        let events: Vec<Event> = (1..8)
            .map(|id| {
                conversation(
                    id,
                    "input",
                    &format!("turn {id} of ordinary history that will be drained"),
                )
            })
            .collect();
        let planned = plan_frontier(&events, Some(&previous), 2).expect("a view is earned");
        assert_eq!(
            planned.actor_testimony.as_deref(),
            Some("I said the exact words"),
            "exact testimony must survive an install that did not author one"
        );
        assert_eq!(
            planned
                .actor_testimony_source
                .as_ref()
                .map(|s| s.actor_id.as_str()),
            Some("dawn"),
            "testimony stays attributable"
        );
        // A different testimony replaces it, which is the only way it changes.
        let superseded = Frontier {
            actor_testimony: Some("the newer exact words".to_string()),
            ..planned.clone()
        };
        assert_eq!(
            superseded.actor_testimony.as_deref(),
            Some("the newer exact words")
        );
    }

    /// The threshold, stated exactly: an install must drain at least as much as it keeps.
    #[test]
    fn a_tiny_drain_does_not_earn_a_new_view() {
        let one_past = |count: i64| -> Vec<Event> {
            (1..=count)
                .map(|id| conversation(id, "input", "one turn"))
                .collect()
        };
        // One event past a keep of two drains a single event: not a view.
        assert!(
            plan_frontier(&one_past(3), None, 2).is_none(),
            "draining one event per turn is the continual tiny compaction the threshold exists to prevent"
        );
        // Two past it drains as much as it keeps: that is a view.
        assert!(
            plan_frontier(&one_past(4), None, 2).is_some(),
            "a drain as large as the keep is what earns a view"
        );
    }
}