klieo-core 3.5.0

//! Integration tests for [`super::maybe_compact`] driven end-to-end
//! through [`crate::runtime::run_steps`], using `klieo_core::test_utils`
//! doubles (fake LLM, in-memory short-term + episodic memory).

use super::*;
use crate::error::{ConfigError, Error};
use crate::llm::{ChatRequest, ChatResponse};
use crate::memory::Episode;
use crate::runtime::{run_steps, CaptureSink, RunOptions};
use crate::test_utils::{fake_context, FakeLlmClient, FakeLlmStep};
use std::sync::{Arc, Mutex};

/// Padded content whose char-count crosses the `chars / 4` approximate-
/// token heuristic predictably: `tag` (short) + a 76-`x` filler, so
/// each message costs ~20 approx-tokens.
fn padded(tag: &str) -> String {
    format!("{tag}-{}", "x".repeat(76))
}

/// `n` alternating User/Assistant messages, each ~20 approx-tokens,
/// tagged `turn0`, `turn1`, ... so tests can assert on exactly which
/// ones a split kept or dropped.
fn seed_messages(n: usize) -> Vec<Message> {
    (0..n)
        .map(|i| Message {
            role: if i % 2 == 0 {
                Role::User
            } else {
                Role::Assistant
            },
            content: padded(&format!("turn{i}")),
            tool_calls: vec![],
            tool_call_id: None,
        })
        .collect()
}

/// `n` tiny messages (well under any budget used in these tests)
/// tagged `m0`, `m1`, ... — isolates the budget check from the
/// "not enough older content" short-circuit in `summarize_history`.
fn tiny_messages(n: usize) -> Vec<Message> {
    (0..n)
        .map(|i| Message {
            role: if i % 2 == 0 {
                Role::User
            } else {
                Role::Assistant
            },
            content: format!("m{i}"),
            tool_calls: vec![],
            tool_call_id: None,
        })
        .collect()
}

fn small_compaction_opts() -> SummarizeOptions {
    SummarizeOptions {
        trigger_token_budget: 50,
        keep_recent_messages: 2,
        ..SummarizeOptions::default()
    }
}

async fn seed(ctx: &crate::agent::AgentContext, thread: &ThreadId, messages: Vec<Message>) {
    for msg in messages {
        ctx.short_term.append(thread.clone(), msg).await.unwrap();
    }
}

// ---- 1. Crossing the budget triggers exactly one summarization ----

#[tokio::test]
async fn compacts_history_over_budget_keeping_last_n_verbatim() {
    let mut ctx = fake_context("compaction-over-budget");
    ctx.llm = Arc::new(FakeLlmClient::new("fake").with_steps(vec![
        FakeLlmStep::Text("summary of the earlier turns".into()),
        FakeLlmStep::Text("final answer".into()),
    ]));
    let thread = ThreadId::new("t-over-budget");
    seed(&ctx, &thread, seed_messages(5)).await;

    let opts = RunOptions::default().with_compaction(small_compaction_opts());
    let out = run_steps(&ctx, "sys", thread.clone(), opts).await.unwrap();
    assert_eq!(out, "final answer");

    let remaining = ctx.short_term.load(thread.clone(), 100_000).await.unwrap();
    assert_eq!(
        remaining.len(),
        4,
        "short-term must shrink to [summary, turn3, turn4, reply]: {remaining:?}"
    );
    assert!(remaining[0]
        .content
        .contains("summary of the earlier turns"));
    assert!(remaining[1].content.contains("turn3"));
    assert!(remaining[2].content.contains("turn4"));
    assert_eq!(remaining[3].content, "final answer");

    let episodes = ctx.episodic.replay(ctx.run_id).await.unwrap();
    let checkpoints = episodes
        .iter()
        .filter(|e| matches!(e, Episode::SummaryCheckpoint { .. }))
        .count();
    assert_eq!(checkpoints, 1, "exactly one summarization must have run");
}

// ---- 2. Under budget never summarizes (negative fixture) ----

#[tokio::test]
async fn stays_under_budget_never_summarizes() {
    let mut ctx = fake_context("compaction-under-budget");
    ctx.llm =
        Arc::new(FakeLlmClient::new("fake").with_steps(vec![FakeLlmStep::Text("final".into())]));
    let thread = ThreadId::new("t-under-budget");
    // 4 messages (more than keep_recent_messages) but each near-zero
    // tokens, isolating the budget check from the "too few older
    // messages" short-circuit.
    seed(&ctx, &thread, tiny_messages(4)).await;

    let opts = RunOptions::default().with_compaction(small_compaction_opts());
    let out = run_steps(&ctx, "sys", thread.clone(), opts).await.unwrap();
    assert_eq!(out, "final");

    let episodes = ctx.episodic.replay(ctx.run_id).await.unwrap();
    assert!(
        !episodes
            .iter()
            .any(|e| matches!(e, Episode::SummaryCheckpoint { .. })),
        "a history under budget must never trigger summarization"
    );

    let remaining = ctx.short_term.load(thread.clone(), 100_000).await.unwrap();
    assert_eq!(
        remaining.len(),
        5,
        "no compaction: original 4 turns plus the new reply, untouched"
    );
}

// ---- 3. Moat test: original history stays recoverable via provenance ----

#[tokio::test]
async fn original_history_survives_in_provenance_after_short_term_compaction() {
    #[derive(Default)]
    struct RecordingSink {
        requests: Mutex<Vec<ChatRequest>>,
    }
    impl CaptureSink for RecordingSink {
        fn record_llm_call(&self, request: &ChatRequest, _response: &ChatResponse) {
            self.requests.lock().unwrap().push(request.clone());
        }
    }

    const SEED_TEXT: &str = "SEED: the original ask";

    let mut ctx = fake_context("compaction-moat");
    // call 1: real reply. call 2: real reply. call 3: history now
    // crosses budget -> summarizer call, then call 3's real reply.
    ctx.llm = Arc::new(FakeLlmClient::new("fake").with_steps(vec![
        FakeLlmStep::Text(padded("reply1")),
        FakeLlmStep::Text(padded("reply2")),
        FakeLlmStep::Text("condensed summary".into()),
        FakeLlmStep::Text(padded("reply3")),
    ]));
    let thread = ThreadId::new("t-moat");
    ctx.short_term
        .append(
            thread.clone(),
            Message {
                role: Role::User,
                content: SEED_TEXT.into(),
                tool_calls: vec![],
                tool_call_id: None,
            },
        )
        .await
        .unwrap();

    let sink = Arc::new(RecordingSink::default());
    let opts = RunOptions::default()
        .with_compaction(SummarizeOptions {
            trigger_token_budget: 40,
            keep_recent_messages: 2,
            ..SummarizeOptions::default()
        })
        .with_capture_sink(sink.clone());

    for _ in 0..3 {
        run_steps(&ctx, "sys", thread.clone(), opts.clone())
            .await
            .unwrap();
    }

    // The working short-term context has been compacted: the seed
    // message is gone from it.
    let remaining = ctx.short_term.load(thread.clone(), 100_000).await.unwrap();
    assert!(
        !remaining.iter().any(|m| m.content == SEED_TEXT),
        "short-term compaction is expected to drop the seed turn: {remaining:?}"
    );

    // But the audit/capture layer recorded it durably, at the moment it
    // was live, before compaction ever ran — recoverable regardless of
    // what later happens to short-term memory.
    let seed_recovered = sink
        .requests
        .lock()
        .unwrap()
        .iter()
        .any(|req| req.messages.iter().any(|m| m.content == SEED_TEXT));
    assert!(
        seed_recovered,
        "original seed content must still be recoverable from captured requests"
    );

    // The episodic ledger accumulated one LlmCall per real step (3),
    // undiminished by the one compaction event.
    let episodes = ctx.episodic.replay(ctx.run_id).await.unwrap();
    let llm_calls = episodes
        .iter()
        .filter(|e| matches!(e, Episode::LlmCall { .. }))
        .count();
    let checkpoints = episodes
        .iter()
        .filter(|e| matches!(e, Episode::SummaryCheckpoint { .. }))
        .count();
    assert_eq!(llm_calls, 3, "episodic ledger must retain all 3 real steps");
    assert_eq!(
        checkpoints, 1,
        "exactly one compaction fired across the 3 calls"
    );
}

// ---- 4. Summarizer LLM call is itself audited ----

#[tokio::test]
async fn summarizer_llm_call_is_recorded_to_the_episodic_trail() {
    let mut ctx = fake_context("compaction-audited");
    ctx.llm = Arc::new(FakeLlmClient::new("fake").with_steps(vec![
        FakeLlmStep::Text("condensed summary".into()),
        FakeLlmStep::Text("final".into()),
    ]));
    let thread = ThreadId::new("t-audited");
    seed(&ctx, &thread, seed_messages(5)).await;

    let opts = RunOptions::default().with_compaction(small_compaction_opts());
    run_steps(&ctx, "sys", thread.clone(), opts).await.unwrap();

    let episodes = ctx.episodic.replay(ctx.run_id).await.unwrap();
    let checkpoint = episodes.iter().find_map(|e| match e {
        Episode::SummaryCheckpoint {
            input_message_count,
            summary_chars,
            ..
        } => Some((*input_message_count, *summary_chars)),
        _ => None,
    });
    let (input_message_count, summary_chars) =
        checkpoint.expect("the summarizer LLM call must be recorded as Episode::SummaryCheckpoint");
    assert_eq!(
        input_message_count, 3,
        "3 older messages (5 loaded - keep_recent 2) were folded into the summary call"
    );
    assert_eq!(summary_chars, "condensed summary".chars().count() as u32);
}

// ---- 5. Off-switch / compliance carve-out ----

#[tokio::test]
async fn without_compaction_disables_auto_summarization_even_over_budget() {
    let mut ctx = fake_context("compaction-disabled");
    ctx.llm =
        Arc::new(FakeLlmClient::new("fake").with_steps(vec![FakeLlmStep::Text("final".into())]));
    let thread = ThreadId::new("t-disabled");
    seed(&ctx, &thread, seed_messages(5)).await;

    // Same budget that triggers compaction when enabled (test 1), but
    // explicitly disabled here — the compliance carve-out.
    let opts = RunOptions::default()
        .with_compaction(small_compaction_opts())
        .without_compaction();
    let out = run_steps(&ctx, "sys", thread.clone(), opts).await.unwrap();
    assert_eq!(out, "final");

    let episodes = ctx.episodic.replay(ctx.run_id).await.unwrap();
    assert!(
        !episodes
            .iter()
            .any(|e| matches!(e, Episode::SummaryCheckpoint { .. })),
        "compaction must never fire once disabled, regardless of history size"
    );

    let remaining = ctx.short_term.load(thread.clone(), 100_000).await.unwrap();
    assert_eq!(
        remaining.len(),
        6,
        "full original 5 turns plus the new reply, untouched"
    );
}

// ---- 6. Zero budget is a typed config error, not a panic ----

#[tokio::test]
async fn zero_trigger_budget_is_a_typed_config_error_not_a_panic() {
    let ctx = fake_context("compaction-zero-budget");
    let thread = ThreadId::new("t-zero-budget");
    let bad = SummarizeOptions {
        trigger_token_budget: 0,
        ..SummarizeOptions::default()
    };
    let opts = RunOptions::default().with_compaction(bad);

    let err = run_steps(&ctx, "sys", thread, opts).await.unwrap_err();
    match err {
        Error::Config(ConfigError::InvalidValue { key, .. }) => {
            assert_eq!(key.as_str(), "compaction.trigger_token_budget");
        }
        other => panic!("expected Error::Config(InvalidValue), got {other:?}"),
    }
}