klieo-runlog 0.41.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::Replay`].
//! - `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, 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};

/// Replay a recorded [`RunLog`] against caller-supplied [`LlmClient`] +
/// [`ToolInvoker`] test doubles. Returns the recorded final assistant text.
///
/// `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 {
        match step.kind {
            StepKind::LlmCall => {
                // klieo-core's LlmClient takes a ChatRequest, not a bare prompt
                // string. We reconstruct a single User message from the
                // recorded `input` (best-effort: extract a "prompt" field if
                // present, otherwise serialise the whole input as JSON text).
                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::Replay(format!("step {} llm error: {e}", step.idx))
                })?;
                let live = resp.message.content;
                let recorded = step
                    .output
                    .as_str()
                    .map(String::from)
                    .unwrap_or_else(|| step.output.to_string());
                if live != recorded {
                    return Err(RunLogError::Replay(format!(
                        "LLM mismatch at step {} (recorded={recorded:?} live={live:?})",
                        step.idx
                    )));
                }
                last_llm_output = Some(live);
            }
            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::Replay(format!("step {} tool error: {e}", step.idx))
                    })?;
                if live != step.output {
                    return Err(RunLogError::Replay(format!(
                        "Tool mismatch at step {} (recorded={} live={})",
                        step.idx, step.output, live
                    )));
                }
            }
            StepKind::SummaryCheckpoint => {
                // Summarizer checkpoints are bookkeeping — they do not
                // re-issue an LLM call during replay (the underlying
                // transcript is intentionally not retained on the
                // step). Skip without affecting `last_llm_output`.
            }
            StepKind::OpsEvent => {
                // Operational-layer events (supervisor, gates, governor)
                // are informational audit records. They carry no I/O to
                // re-issue during replay — skip them transparently.
            }
        }
    }
    last_llm_output.ok_or_else(|| RunLogError::Replay("no LlmCall step in RunLog".into()))
}

/// 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 {
            message: Message {
                role: Role::Assistant,
                content: text,
                tool_calls: vec![],
                tool_call_id: None,
            },
            usage: Usage::default(),
            finish_reason: FinishReason::Stop,
        })
    }
    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()
    }
}

#[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,
            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(_)));
    }
}