klieo-runlog 3.3.1

//! Deterministic-only replay engine.
//!
//! Walks a stored [`RunLog`]'s `Step`s in order. For each step:
//! - `StepKind::LlmCall` — calls the supplied [`LlmClient`]. The recorded
//!   `output` must equal the live response, otherwise [`RunLogError::ReplayMismatch`].
//! - `StepKind::ToolCall` — calls the supplied [`ToolInvoker`]. Same equality check.
//!
//! Returns the assistant's final text — the `output` of the last `LlmCall` step,
//! flattened to a [`String`] via `serde_json::Value::as_str().unwrap_or(json-string)`.
//!
//! ## Determinism caveat
//!
//! Replay only matches values that flow through [`LlmClient`] and
//! [`ToolInvoker`]. Side effects outside these traits (timestamps, randomness,
//! network calls that are not recorded as tool calls) cannot be replayed and
//! will diverge between record and replay runs. Use this engine for regression
//! testing of agent logic against canned LLM/tool outputs, not as a general
//! determinism guarantee.
//!
//! ## Plan deviation: `ToolCtx` parameter
//!
//! The plan's signature was `replay(run_log, llm, tools) -> Result<String, _>`.
//! Live API: [`ToolInvoker::invoke`] requires a [`ToolCtx`] (its third
//! argument), so this implementation takes a fourth `ctx: ToolCtx` parameter.
//! Callers that don't need a real bus can pass a no-op `ToolCtx` built from
//! `klieo_core::test_utils::noop_bus` under the `test-utils` feature.
//! Editing `klieo-core` to relax the trait signature is out of scope for
//! Plan #22 (Plan #11 trait freeze).

use crate::error::RunLogError;
use crate::types::{RunLog, Step, StepKind};
use async_trait::async_trait;
use klieo_core::error::{LlmError, ToolError};
use klieo_core::llm::{
    Capabilities, ChatRequest, ChatResponse, ChunkStream, FinishReason, LlmClient, Message, Role,
    ToolDef, Usage,
};
use klieo_core::tool::{ToolCtx, ToolInvoker};
use std::sync::{Arc, Mutex};

/// Flatten a recorded or replayed JSON value to comparable text: a JSON string
/// yields its inner contents; any other value yields its compact JSON encoding.
/// Both replay walks compare this normalized form so a string output and its
/// quoted JSON encoding are never treated as different.
pub(crate) fn flatten_output(value: &serde_json::Value) -> String {
    value
        .as_str()
        .map(String::from)
        .unwrap_or_else(|| value.to_string())
}

/// The recorded (`expected`) and freshly replayed (`actual`) text for one
/// comparable step, both already normalized through [`flatten_output`] so a
/// caller compares like for like.
pub(crate) struct StepOutcome {
    pub(crate) expected: String,
    pub(crate) actual: String,
}

/// Drive one step against the doubles. An `Err` means the double itself failed
/// (errored or exhausted) — an infrastructure failure the caller must surface,
/// never conflated with a value mismatch (which is a comparison of the returned
/// pair). `Ok(None)` covers steps that re-issue no I/O.
pub(crate) async fn step_outcome(
    step: &Step,
    llm: &Arc<dyn LlmClient>,
    tools: &Arc<dyn ToolInvoker>,
    ctx: &ToolCtx,
) -> Result<Option<StepOutcome>, RunLogError> {
    match step.kind {
        StepKind::LlmCall => {
            let prompt = step
                .input
                .get("prompt")
                .and_then(|v| v.as_str())
                .map(String::from)
                .unwrap_or_else(|| step.input.to_string());
            let req = ChatRequest::new(vec![Message {
                role: Role::User,
                content: prompt,
                tool_calls: vec![],
                tool_call_id: None,
            }]);
            let resp = llm
                .complete(req)
                .await
                .map_err(|e| RunLogError::ReplayStep {
                    step: step.idx,
                    source: Box::new(e),
                })?;
            Ok(Some(StepOutcome {
                expected: flatten_output(&step.output),
                actual: resp.message.content,
            }))
        }
        StepKind::ToolCall => {
            let name = step.name.as_deref().unwrap_or("");
            let live = tools
                .invoke(name, step.input.clone(), ctx.clone())
                .await
                .map_err(|e| RunLogError::ReplayStep {
                    step: step.idx,
                    source: Box::new(e),
                })?;
            Ok(Some(StepOutcome {
                expected: flatten_output(&step.output),
                actual: flatten_output(&live),
            }))
        }
        StepKind::SummaryCheckpoint | StepKind::OpsEvent | StepKind::MemoryRecall => Ok(None),
    }
}

/// Replay a recorded [`RunLog`] against caller-supplied [`LlmClient`] +
/// [`ToolInvoker`] test doubles. Returns the recorded final assistant text.
/// Fails fast on the first value mismatch; the soft-compare counterpart that
/// collects every divergence is [`crate::divergence::replay_with_divergence`].
///
/// `ctx` is forwarded verbatim to every `ToolInvoker::invoke` call. For test
/// scenarios with no bus, build one via
/// `klieo_core::test_utils::noop_bus()` (requires the `test-utils` feature on
/// `klieo-core`).
pub async fn replay(
    run_log: &RunLog,
    llm: Arc<dyn LlmClient>,
    tools: Arc<dyn ToolInvoker>,
    ctx: ToolCtx,
) -> Result<String, RunLogError> {
    let mut last_llm_output: Option<String> = None;
    for step in &run_log.steps {
        let Some(outcome) = step_outcome(step, &llm, &tools, &ctx).await? else {
            continue;
        };
        if outcome.actual != outcome.expected {
            return Err(RunLogError::ReplayMismatch {
                step: step.idx,
                kind: step.kind,
                expected: outcome.expected,
                actual: outcome.actual,
            });
        }
        if step.kind == StepKind::LlmCall {
            last_llm_output = Some(outcome.actual);
        }
    }
    last_llm_output.ok_or(RunLogError::NoLlmCall)
}

/// Production [`LlmClient`] that replays a pre-recorded script of
/// responses in order, returning a typed error when the script
/// exhausts. Equivalent in shape to
/// `klieo_core::test_utils::FakeLlmClient` but lives in the
/// production module so binaries that drive [`replay`] (e.g.
/// `cargo-klieo` and other audit tooling) can depend on it without
/// activating the `test-utils` feature.
///
/// Use [`scripted_llm_from_runlog`] to construct one directly from
/// a stored `RunLog`'s recorded `LlmCall.output` values.
pub struct ScriptedLlmClient {
    name: String,
    script: Mutex<std::collections::VecDeque<String>>,
    capabilities: Capabilities,
}

impl ScriptedLlmClient {
    /// Build a client that emits `script` in order, one entry per
    /// [`LlmClient::complete`] call. After the script exhausts every
    /// subsequent call returns
    /// [`LlmError::BadRequest`] (matches the
    /// `FakeLlmClient` contract used by tests).
    pub fn new(name: impl Into<String>, script: Vec<String>) -> Self {
        Self {
            name: name.into(),
            script: Mutex::new(script.into_iter().collect()),
            capabilities: Capabilities::default(),
        }
    }
}

#[async_trait]
impl LlmClient for ScriptedLlmClient {
    fn name(&self) -> &str {
        &self.name
    }
    fn capabilities(&self) -> &Capabilities {
        &self.capabilities
    }
    async fn complete(&self, _req: ChatRequest) -> Result<ChatResponse, LlmError> {
        let text = {
            let mut q = self
                .script
                .lock()
                .expect("ScriptedLlmClient mutex poisoned");
            q.pop_front()
                .ok_or_else(|| LlmError::BadRequest("ScriptedLlmClient: script exhausted".into()))?
        };
        Ok(ChatResponse::new(
            Message {
                role: Role::Assistant,
                content: text,
                tool_calls: vec![],
                tool_call_id: None,
            },
            Usage::default(),
            FinishReason::Stop,
            self.name.clone(),
        ))
    }
    async fn stream(&self, _req: ChatRequest) -> Result<ChunkStream, LlmError> {
        Err(LlmError::Unsupported(
            "ScriptedLlmClient does not support streaming".into(),
        ))
    }
    async fn embed(&self, _texts: &[String]) -> Result<Vec<klieo_core::llm::Embedding>, LlmError> {
        Err(LlmError::Unsupported(
            "ScriptedLlmClient does not support embedding".into(),
        ))
    }
}

/// Production counterpart to
/// `klieo_runlog::test_utils::fake_llm_from_runlog`. Walks `log`'s
/// `LlmCall` steps in order; each `step.output` (as a JSON string,
/// or its JSON-serialised text for non-string outputs) becomes one
/// entry in the script. Returns a [`ScriptedLlmClient`] suitable for
/// driving [`replay`] from a non-test binary.
pub fn scripted_llm_from_runlog(name: impl Into<String>, log: &RunLog) -> ScriptedLlmClient {
    let script: Vec<String> = log
        .steps
        .iter()
        .filter(|s| s.kind == StepKind::LlmCall)
        .map(|s| {
            s.output
                .as_str()
                .map(String::from)
                .unwrap_or_else(|| s.output.to_string())
        })
        .collect();
    ScriptedLlmClient::new(name, script)
}

/// Zero-tool [`ToolInvoker`] for production replay paths. Catalogue
/// is always empty; any `invoke` call returns
/// [`ToolError::UnknownTool`]. Suitable for [`replay`] runs whose
/// RunLog records no tool calls.
#[derive(Default)]
pub struct NoopToolInvoker;

#[async_trait]
impl ToolInvoker for NoopToolInvoker {
    async fn invoke(
        &self,
        name: &str,
        _args: serde_json::Value,
        _ctx: ToolCtx,
    ) -> Result<serde_json::Value, ToolError> {
        Err(ToolError::UnknownTool(name.into()))
    }
    fn catalogue(&self) -> Vec<ToolDef> {
        Vec::new()
    }
}

/// Production [`ToolInvoker`] that replays a run's recorded `ToolCall` outputs
/// in step order, one per `invoke` call. Tool replay never re-executes a real
/// tool (ADR-046 §5: replaying side effects is unsafe); this returns the
/// captured output instead. After the recording is exhausted every further call
/// returns [`ToolError::UnknownTool`]. Build one with [`scripted_tools_from_runlog`].
pub struct ScriptedToolInvoker {
    outputs: Mutex<std::collections::VecDeque<serde_json::Value>>,
}

impl ScriptedToolInvoker {
    /// Wrap a caller-supplied output queue; prefer
    /// [`scripted_tools_from_runlog`] to build one from a recorded run.
    pub fn new(outputs: Vec<serde_json::Value>) -> Self {
        Self {
            outputs: Mutex::new(outputs.into_iter().collect()),
        }
    }
}

#[async_trait]
impl ToolInvoker for ScriptedToolInvoker {
    async fn invoke(
        &self,
        name: &str,
        _args: serde_json::Value,
        _ctx: ToolCtx,
    ) -> Result<serde_json::Value, ToolError> {
        self.outputs
            .lock()
            .expect("ScriptedToolInvoker mutex poisoned")
            .pop_front()
            .ok_or_else(|| ToolError::UnknownTool(name.into()))
    }
    fn catalogue(&self) -> Vec<ToolDef> {
        Vec::new()
    }
}

/// Build a [`ScriptedToolInvoker`] that replays `log`'s `ToolCall` step outputs
/// in order — the tool-side counterpart to [`scripted_llm_from_runlog`].
pub fn scripted_tools_from_runlog(log: &RunLog) -> ScriptedToolInvoker {
    let outputs: Vec<serde_json::Value> = log
        .steps
        .iter()
        .filter(|s| s.kind == StepKind::ToolCall)
        .map(|s| s.output.clone())
        .collect();
    ScriptedToolInvoker::new(outputs)
}

#[cfg(test)]
mod prod_helper_tests {
    use super::*;
    use crate::types::{RunLog, RunStatus, Step, Usage};
    use chrono::Utc;
    use klieo_core::RunId;

    fn step(idx: u32, kind: StepKind, output: serde_json::Value) -> Step {
        Step {
            idx,
            kind,
            name: None,
            prompt_tokens: None,
            completion_tokens: None,
            cost_usd: None,
            input: serde_json::Value::Null,
            output,
            error: None,
            latency: std::time::Duration::ZERO,
            span_id: None,
        }
    }

    fn run_log_with_mixed_steps() -> RunLog {
        let now = Utc::now();
        RunLog {
            run_id: RunId::new(),
            agent: "t".into(),
            started_at: now,
            finished_at: Some(now),
            status: RunStatus::Completed,
            steps: vec![
                step(0, StepKind::LlmCall, serde_json::json!("hello")),
                step(1, StepKind::ToolCall, serde_json::json!({"hit": true})),
                step(2, StepKind::LlmCall, serde_json::json!("world")),
            ],
            tokens: Usage::default(),
            cost_estimate: None,
        }
    }

    #[tokio::test]
    async fn scripted_llm_from_runlog_filters_to_llm_call_steps_only() {
        let log = run_log_with_mixed_steps();
        let llm = scripted_llm_from_runlog("prod-replay", &log);
        assert_eq!(LlmClient::name(&llm), "prod-replay");

        let r1 = llm.complete(ChatRequest::new(vec![])).await.unwrap();
        assert_eq!(r1.message.content, "hello");
        let r2 = llm.complete(ChatRequest::new(vec![])).await.unwrap();
        assert_eq!(
            r2.message.content, "world",
            "ToolCall step must not consume a slot"
        );
        let err = llm.complete(ChatRequest::new(vec![])).await;
        assert!(err.is_err(), "third call must exhaust");
    }

    #[tokio::test]
    async fn noop_tool_invoker_always_returns_unknown_tool() {
        let invoker = NoopToolInvoker;
        assert!(invoker.catalogue().is_empty());
        let bus = klieo_core::test_utils::noop_bus();
        let ctx = ToolCtx::new(bus.0, bus.2, bus.3);
        let err = invoker
            .invoke("any", serde_json::json!({}), ctx)
            .await
            .unwrap_err();
        assert!(matches!(err, ToolError::UnknownTool(_)));
    }

    #[test]
    fn flatten_output_unwraps_json_string_but_compacts_other_values() {
        assert_eq!(flatten_output(&serde_json::json!("hello")), "hello");
        assert_eq!(flatten_output(&serde_json::json!(42)), "42");
        assert_eq!(flatten_output(&serde_json::json!(true)), "true");
        assert_eq!(flatten_output(&serde_json::json!({"k": 1})), "{\"k\":1}");
        assert_eq!(flatten_output(&serde_json::Value::Null), "null");
        assert_eq!(flatten_output(&serde_json::json!([1, 2])), "[1,2]");
    }

    #[tokio::test]
    async fn scripted_tools_from_runlog_replays_toolcall_outputs_then_exhausts() {
        let log = run_log_with_mixed_steps();
        let tools = scripted_tools_from_runlog(&log);
        let bus = klieo_core::test_utils::noop_bus();
        let ctx = ToolCtx::new(bus.0, bus.2, bus.3);
        let out = tools
            .invoke("calc", serde_json::json!({}), ctx.clone())
            .await
            .unwrap();
        assert_eq!(
            out,
            serde_json::json!({"hit": true}),
            "only the ToolCall step's output is replayed, not the LlmCall steps"
        );
        let err = tools
            .invoke("calc", serde_json::json!({}), ctx)
            .await
            .unwrap_err();
        assert!(matches!(err, ToolError::UnknownTool(_)));
    }

    #[tokio::test]
    async fn scripted_tools_from_runlog_with_no_toolcall_steps_exhausts_immediately() {
        let now = Utc::now();
        let log = RunLog {
            run_id: RunId::new(),
            agent: "t".into(),
            started_at: now,
            finished_at: Some(now),
            status: RunStatus::Completed,
            steps: vec![step(0, StepKind::LlmCall, serde_json::json!("only-llm"))],
            tokens: Usage::default(),
            cost_estimate: None,
        };
        let tools = scripted_tools_from_runlog(&log);
        let bus = klieo_core::test_utils::noop_bus();
        let ctx = ToolCtx::new(bus.0, bus.2, bus.3);
        let err = tools
            .invoke("anything", serde_json::json!({}), ctx)
            .await
            .unwrap_err();
        assert!(matches!(err, ToolError::UnknownTool(_)));
    }
}