klieo-core 3.16.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.
/// Chosen so `seed_messages(5)` fits inside the window compaction actually
/// loads — `small_compaction_opts().trigger_token_budget * 2`, see
/// `rewrite_short_term_with_summary` — while still exceeding the trigger
/// budget itself.
///
/// At the previous 76 the five-message fixture ran two approx-tokens over that
/// window, so only four messages ever loaded: the "5 loaded, fold 3" tests
/// below were asserting against four. That was invisible while
/// `InMemoryShortTerm::load` ignored `max_tokens`. Keep the fixture inside the
/// window when changing either number.
const PADDING_CHARS: usize = 72;

fn padded(tag: &str) -> String {
    format!("{tag}-{}", "x".repeat(PADDING_CHARS))
}

/// `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()
    }
}

/// `RunOptions` whose history budget is coherent with `compaction`.
///
/// `maybe_compact` rejects a pair where `max_history_tokens` exceeds the window
/// compaction can see, because such a thread would carry history that is never
/// summarised and then discarded. These tests use a pathologically small
/// trigger to make compaction fire on tiny fixtures, so they need an equally
/// small history budget to stay coherent. Derived rather than hardcoded so the
/// two cannot drift apart.
fn run_opts_for(compaction: SummarizeOptions) -> RunOptions {
    let history_budget = visible_window_tokens(&compaction);
    RunOptions::default()
        .with_compaction(compaction)
        .with_max_history_tokens(history_budget)
}

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 = run_opts_for(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(),
        5,
        "short-term must shrink to [brief, summary, turn3, turn4, reply]: {remaining:?}"
    );
    assert!(
        remaining[0].content.contains("turn0"),
        "the brief is kept verbatim at the head, never summarised"
    );
    assert!(remaining[1]
        .content
        .contains("summary of the earlier turns"));
    assert!(remaining[2].content.contains("turn3"));
    assert!(remaining[3].content.contains("turn4"));
    assert_eq!(remaining[4].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 = run_opts_for(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"
    );
}

/// A brief bigger than the whole trigger budget must not make compaction fire
/// on every step for the rest of the run.
///
/// `KEEP_LEADING_MESSAGES` protects the brief unconditionally, so if the
/// trigger is charged for it the total can never fall back under the trigger:
/// each later step re-enters compaction and pays a summariser LLM call to
/// shrink something it is forbidden to touch. The `FakeLlmClient` here offers
/// exactly one reply and no summariser step, so a single unwanted summariser
/// call exhausts the script and fails the run.
#[tokio::test]
async fn an_oversized_brief_alone_does_not_trigger_compaction() {
    let mut ctx = fake_context("compaction-oversized-brief");
    ctx.llm =
        Arc::new(FakeLlmClient::new("fake").with_steps(vec![FakeLlmStep::Text("final".into())]));
    let thread = ThreadId::new("t-oversized-brief");
    ctx.short_term
        .append(
            thread.clone(),
            Message {
                role: Role::User,
                content: "B".repeat(10_000),
                tool_calls: vec![],
                tool_call_id: None,
            },
        )
        .await
        .unwrap();
    seed(&ctx, &thread, tiny_messages(3)).await;

    let opts = run_opts_for(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 { .. })),
        "the brief is the only thing over budget and compaction may not touch it, so \
         summarising cannot help and must not be attempted"
    );
    let remaining = ctx.short_term.load(thread, 100_000).await.unwrap();
    assert!(
        remaining.iter().any(|m| m.content.len() == 10_000),
        "and the brief itself is still there, verbatim"
    );
}

/// A gap marker is a read-time annotation, not history. Compaction persists
/// what it LOADS (`clear` + `append_batch`), so a marker left in the rewritten
/// set would become a stored message: indistinguishable from a real turn, and
/// joined by another one on every later compaction cycle.
#[tokio::test]
async fn compaction_does_not_persist_a_read_time_gap_marker() {
    let mut ctx = fake_context("compaction-marker");
    ctx.llm = Arc::new(FakeLlmClient::new("fake").with_steps(vec![
        FakeLlmStep::Text(padded("reply1")),
        FakeLlmStep::Text("condensed summary".into()),
        FakeLlmStep::Text(padded("reply2")),
    ]));
    let thread = ThreadId::new("t-marker");
    // A brief far past the load budget forces the walk to elide the middle and
    // insert its marker on every load.
    ctx.short_term
        .append(
            thread.clone(),
            Message {
                role: Role::User,
                content: "B".repeat(4_000),
                tool_calls: vec![],
                tool_call_id: None,
            },
        )
        .await
        .unwrap();
    seed(&ctx, &thread, tiny_messages(6)).await;

    let opts = run_opts_for(SummarizeOptions {
        trigger_token_budget: 30,
        keep_recent_messages: 1,
        ..SummarizeOptions::default()
    })
    .with_max_history_tokens(20);
    let _ = run_steps(&ctx, "sys", thread.clone(), opts).await;

    let stored = ctx.short_term.load(thread, 1_000_000).await.unwrap();
    let persisted_markers = stored
        .iter()
        .filter(|m| {
            m.content
                .starts_with(crate::memory::ELIDED_MESSAGES_MARKER_PREFIX)
        })
        .count();
    assert_eq!(
        persisted_markers, 0,
        "a read-time marker must never be written back into the thread: {stored:?}"
    );
}

// ---- 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());
    // `keep_recent_messages: 1` rather than 2: compaction needs at least one
    // message that is neither the brief nor the recent tail before it has
    // anything to compress.
    //
    // The trigger is 30, not 40, because it is charged only for the messages
    // compaction may actually touch — the brief is excluded, since counting a
    // message no summariser may remove would keep the total over the trigger
    // forever and re-summarise the same tail on every later step. The two
    // padded replies here are ~39 approx-tokens between them.
    let opts = run_opts_for(SummarizeOptions {
        trigger_token_budget: 30,
        keep_recent_messages: 1,
        ..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 middle
    // turns are now a summary -- but the seed turn is the BRIEF and is kept
    // verbatim at the head. Before 3.15.0 it was summarised away, which is
    // the defect this assertion now guards.
    let remaining = ctx.short_term.load(thread.clone(), 100_000).await.unwrap();
    assert!(
        remaining.iter().any(|m| m.content == SEED_TEXT),
        "the brief must survive compaction verbatim: {remaining:?}"
    );
    assert!(
        remaining
            .iter()
            .any(|m| m.content.contains("compacted summary")),
        "the middle turns must still have been compacted: {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 = run_opts_for(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, 2,
        "2 messages were folded in: 5 loaded, minus keep_recent 2, minus the brief at index 0 \
         which is never summarised"
    );
    assert_eq!(summary_chars, "condensed summary".chars().count() as u32);
}

// ---- 4b. The brief survives compaction ----

/// THE MEASURED DEFECT. 2026-08-17, in a downstream pipeline: an agent's
/// opening prompt was ~9,600 tokens against the default 6,000-token trigger,
/// so compaction fired on the FIRST step and replaced the task, the repository
/// facts and an explicit language instruction with a 200-word summary. The
/// agent then wrote Python into a Java repository, repeatedly, and the symptom
/// was indistinguishable from the model simply being incapable.
#[tokio::test]
async fn compaction_keeps_the_opening_brief_verbatim() {
    let mut ctx = fake_context("compaction-keeps-brief");
    ctx.llm = Arc::new(FakeLlmClient::new("fake").with_steps(vec![
        FakeLlmStep::Text("condensed summary".into()),
        FakeLlmStep::Text("final".into()),
    ]));
    let thread = ThreadId::new("t-keeps-brief");
    // Five, matching the sibling tests: with `small_compaction_opts`'s tiny
    // budget, a longer thread pushes the true first message outside the
    // visible window entirely, and nothing this fix does can protect a
    // message the summarizer never loads (see `SummarizeOptions`' coverage
    // invariant).
    seed(&ctx, &thread, seed_messages(5)).await;

    let opts = run_opts_for(small_compaction_opts());
    run_steps(&ctx, "sys", thread.clone(), opts).await.unwrap();

    let history = ctx
        .short_term
        .load(thread.clone(), 1_000_000)
        .await
        .unwrap();

    assert!(
        history.first().is_some_and(|m| m.content.contains("turn0")),
        "the brief must still be the first message, verbatim; got {:?}",
        history
            .first()
            .map(|m| m.content.chars().take(40).collect::<String>())
    );
    assert!(
        history
            .iter()
            .any(|m| m.content.contains("compacted summary")),
        "the middle turns must still have been compacted"
    );
}

// ---- 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. Incoherent budget pair is a typed config error ----

/// A `max_history_tokens` above the window compaction can see means a thread
/// the request and resume paths can restore carries history `summarize_history`
/// never loads — which compaction would then discard unsummarised. Rejecting
/// the pair makes that unreachable by configuration rather than merely
/// documented.
#[tokio::test]
async fn history_budget_above_the_compaction_window_is_a_typed_config_error() {
    let ctx = fake_context("compaction-incoherent-budgets");
    let thread = ThreadId::new("t-incoherent-budgets");
    let compaction = small_compaction_opts();
    let window = visible_window_tokens(&compaction);

    let opts = RunOptions::default()
        .with_compaction(compaction)
        .with_max_history_tokens(window + 1);

    let err = run_steps(&ctx, "sys", thread, opts).await.unwrap_err();
    match err {
        Error::Config(ConfigError::InvalidValue { key, reason }) => {
            assert_eq!(key.as_str(), "compaction.trigger_token_budget");
            assert!(
                reason.contains("max_history_tokens"),
                "the reason must name the other half of the pair so the caller \
                 knows both knobs; got: {reason}"
            );
        }
        other => panic!("expected Error::Config(InvalidValue), got {other:?}"),
    }
}

/// Guards the boundary: exactly at the window is coherent, so compaction must
/// run rather than reject. Without this, the check could be off by one in the
/// strict direction and nothing would notice.
#[tokio::test]
async fn history_budget_exactly_at_the_compaction_window_is_accepted() {
    let mut ctx = fake_context("compaction-boundary-budgets");
    ctx.llm = Arc::new(FakeLlmClient::new("fake").with_steps(vec![
        FakeLlmStep::Text("summary".into()),
        FakeLlmStep::Text("final".into()),
    ]));
    let thread = ThreadId::new("t-boundary-budgets");
    seed(&ctx, &thread, seed_messages(5)).await;

    let out = run_steps(&ctx, "sys", thread, run_opts_for(small_compaction_opts()))
        .await
        .unwrap();
    assert_eq!(out, "final");
}

// ---- 7. 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:?}"),
    }
}