klieo-eval 3.0.0

//! Live agent-loop re-drive eval (ADR-048).
//!
//! [`eval_capture_live`] re-runs the *current* agent loop ([`run_steps`])
//! against a recorded [`Capture`], feeding the recorded structured LLM
//! responses in order via [`ReplayLlmClient`] and the recorded tool outputs via
//! the runlog's scripted tool invoker. It then compares the re-driven final
//! assistant text to the recording. A mismatch — or a divergence (the current
//! code requesting more LLM calls than the recording holds) — is an
//! agent-logic / orchestration regression.
//!
//! The harness builds a throwaway in-process [`AgentContext`]: a real in-memory
//! short-term memory (the loop accumulates the conversation), no-op long-term /
//! episodic ports (the bare loop only records episodes, which we discard), and
//! an in-process [`MemoryBus`]. The system prompt and tool catalogue are not
//! captured, so request fingerprints cannot be reproduced here; matching is by
//! call order (see [`ReplayLlmClient`]).

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

use async_trait::async_trait;
use klieo_bus_memory::MemoryBus;
use klieo_core::agent::AgentContext;
use klieo_core::error::MemoryError;
use klieo_core::llm::{Message, Role};
use klieo_core::memory::{
    Episode, EpisodicMemory, Fact, LongTermMemory, RunFilter, RunSummary, Scope, ShortTermMemory,
};
use klieo_core::runtime::{run_steps, RunOptions};
use klieo_core::{FactId, RunId, ThreadId};
use klieo_runlog::replay::scripted_tools_from_runlog;
use klieo_runlog::{Capture, ReplayLlmClient};
use std::sync::Arc;
use tokio_util::sync::CancellationToken;

const REDRIVE_THREAD: &str = "klieo-eval-live";

/// Outcome of one re-driven fixture. `reproduced` is the gate; when it is
/// false, exactly one of `replayed_final` (the run finished with a different
/// answer) or `divergence` (the run could not finish) explains why.
#[derive(Debug, Clone, PartialEq)]
#[non_exhaustive]
pub struct LiveEvalMetrics {
    /// The re-driven run produced the same final assistant text as the
    /// recording. The headline regression signal.
    pub reproduced: bool,
    /// The recording's final assistant text — the oracle `reproduced` is
    /// checked against.
    pub recorded_final: String,
    /// Final assistant text from the re-driven run; `None` if the run diverged
    /// before completing (see `divergence`).
    pub replayed_final: Option<String>,
    /// Set when the re-driven run could not complete the recorded sequence —
    /// the current code requested more LLM calls than recorded, or otherwise
    /// errored. Carries the reason for an operator; its presence means
    /// `reproduced` is false.
    pub divergence: Option<String>,
}

/// Reports whether the current agent loop, re-driven against `capture`, still
/// produces the recorded final output. Never errors: a run that cannot complete
/// (more LLM calls than recorded, or a runtime failure) is reported in
/// [`LiveEvalMetrics::divergence`], not as an `Err`.
pub async fn eval_capture_live(capture: &Capture) -> LiveEvalMetrics {
    let llm = Arc::new(ReplayLlmClient::from_llm_io(
        "klieo-eval-live",
        &capture.llm_io,
    ));
    // Keep a concrete handle so we can assert the recording was fully drained
    // after the re-drive — fewer calls than recorded is a divergence the
    // final-output oracle cannot see.
    let replay_probe = llm.clone();
    let tools = Arc::new(scripted_tools_from_runlog(&capture.run_log));
    let bus = MemoryBus::new();
    let thread = ThreadId::new(REDRIVE_THREAD);

    let initial = capture
        .llm_io
        .first()
        .map(|io| io.prompt.clone())
        .unwrap_or_default();
    let short_term = Arc::new(LiveShortTerm::seeded(
        thread.clone(),
        Message {
            role: Role::User,
            content: initial,
            tool_calls: vec![],
            tool_call_id: None,
        },
    ));

    let ctx = AgentContext::new(
        llm,
        short_term,
        Arc::new(NoopLongTerm),
        Arc::new(NoopEpisodic),
        bus.pubsub.clone(),
        bus.kv.clone(),
        bus.request_reply.clone(),
        bus.jobs.clone(),
        tools,
        RunId::new(),
        CancellationToken::new(),
        capture.run_log.agent.clone(),
    );

    let recorded_final = capture
        .llm_io
        .last()
        .map(|io| io.completion.clone())
        .unwrap_or_default();

    // The system prompt is not captured; matching is by call order, so an empty
    // prompt here does not affect which recorded response answers each call.
    match run_steps(&ctx, "", thread, RunOptions::default()).await {
        Ok(replayed) => {
            let undrained = replay_probe.remaining();
            if undrained > 0 {
                return LiveEvalMetrics {
                    reproduced: false,
                    recorded_final,
                    replayed_final: Some(replayed),
                    divergence: Some(format!(
                        "re-drive consumed fewer LLM calls than recorded: {undrained} recorded \
                         response(s) left un-replayed"
                    )),
                };
            }
            LiveEvalMetrics {
                reproduced: replayed == recorded_final,
                recorded_final,
                replayed_final: Some(replayed),
                divergence: None,
            }
        }
        Err(err) => LiveEvalMetrics {
            reproduced: false,
            recorded_final,
            replayed_final: None,
            divergence: Some(err.to_string()),
        },
    }
}

/// In-memory short-term memory for the re-drive harness — the only memory port
/// that needs real behaviour (the loop appends each turn and re-loads history).
#[derive(Default)]
struct LiveShortTerm {
    threads: Mutex<HashMap<ThreadId, Vec<Message>>>,
}

impl LiveShortTerm {
    fn seeded(thread: ThreadId, message: Message) -> Self {
        let mut threads = HashMap::new();
        threads.insert(thread, vec![message]);
        Self {
            threads: Mutex::new(threads),
        }
    }

    fn lock(&self) -> std::sync::MutexGuard<'_, HashMap<ThreadId, Vec<Message>>> {
        self.threads.lock().expect("LiveShortTerm mutex poisoned")
    }
}

#[async_trait]
impl ShortTermMemory for LiveShortTerm {
    async fn append(&self, thread: ThreadId, msg: Message) -> Result<(), MemoryError> {
        self.lock().entry(thread).or_default().push(msg);
        Ok(())
    }
    async fn load(
        &self,
        thread: ThreadId,
        _max_tokens: usize,
    ) -> Result<Vec<Message>, MemoryError> {
        Ok(self.lock().get(&thread).cloned().unwrap_or_default())
    }
    async fn clear(&self, thread: ThreadId) -> Result<(), MemoryError> {
        self.lock().remove(&thread);
        Ok(())
    }
}

/// No-op episodic store: the bare loop records `Started`/`LlmCall`/`Completed`,
/// none of which the re-drive consumes.
struct NoopEpisodic;

#[async_trait]
impl EpisodicMemory for NoopEpisodic {
    async fn record(&self, _run: RunId, _event: Episode) -> Result<(), MemoryError> {
        Ok(())
    }
    async fn replay(&self, _run: RunId) -> Result<Vec<Episode>, MemoryError> {
        Ok(Vec::new())
    }
    async fn list_runs(&self, _filter: RunFilter) -> Result<Vec<RunSummary>, MemoryError> {
        Ok(Vec::new())
    }
}

struct NoopLongTerm;

#[async_trait]
impl LongTermMemory for NoopLongTerm {
    async fn remember(&self, _scope: Scope, _fact: Fact) -> Result<FactId, MemoryError> {
        Ok(FactId::new("klieo-eval-live-noop"))
    }
    async fn recall(
        &self,
        _scope: Scope,
        _query: &str,
        _k: usize,
    ) -> Result<Vec<Fact>, MemoryError> {
        Ok(Vec::new())
    }
    async fn forget(&self, _id: FactId) -> Result<(), MemoryError> {
        Ok(())
    }
}