klieo-runlog 3.5.0

//! Order-keyed structured LLM double for agent-loop re-drive (ADR-048).
//!
//! Unlike [`crate::replay::ScriptedLlmClient`] (text-only, always `Stop`), this
//! double reconstructs the recorded structured response — `finish_reason` +
//! `tool_calls` — so a re-driven `run_steps` takes the same tool-dispatch /
//! continuation branches the recording took. It answers the agent's calls **in
//! recorded order** (the Nth call gets the Nth recorded response).
//!
//! ## Why order, not fingerprint
//!
//! `request_fingerprint` covers the whole request (system prompt + history +
//! tool defs + sampling params), but a `Capture` stores neither the system
//! prompt nor the tool defs, and the replay tool invoker exposes an empty
//! catalogue — so a re-driven request can never reproduce a recorded
//! fingerprint. Order-keying is therefore the matcher; the hard divergence
//! signal is **script exhaustion** (current code requesting more LLM calls than
//! the recording holds). Recorded fingerprints are kept on the capture for a
//! future drift diagnostic that reconstructs exact requests; this double does
//! not consume them.

use std::collections::VecDeque;
use std::sync::Mutex;

use async_trait::async_trait;
use klieo_core::error::LlmError;
use klieo_core::llm::{
    Capabilities, ChatRequest, ChatResponse, ChunkStream, Embedding, FinishReason, LlmClient,
    Message, Role, ToolCall, Usage,
};

use crate::types::LlmIo;

#[derive(Clone)]
struct Recorded {
    completion: String,
    finish_reason: FinishReason,
    tool_calls: Vec<ToolCall>,
    /// Model from the original capture; `"replay:unknown"` for pre-2.0 captures
    /// that did not record a model name.
    model: String,
}

/// Replays a capture's recorded LLM responses in order, reconstructing the
/// structured `finish_reason` + `tool_calls`. Build with [`Self::from_llm_io`].
pub struct ReplayLlmClient {
    name: String,
    capabilities: Capabilities,
    queue: Mutex<VecDeque<Recorded>>,
}

impl ReplayLlmClient {
    /// Build from a capture's `llm_io` sidecar (call order preserved).
    pub fn from_llm_io(name: impl Into<String>, llm_io: &[LlmIo]) -> Self {
        let queue = llm_io
            .iter()
            .map(|io| Recorded {
                completion: io.completion.clone(),
                // Pre-2.0 captures lack a recorded finish_reason; a text-only
                // record could only have ended in Stop.
                finish_reason: io.finish_reason.unwrap_or(FinishReason::Stop),
                tool_calls: io.tool_calls.clone(),
                // Pre-2.0 captures lack a recorded model; sentinel signals the gap.
                model: io.model.clone().unwrap_or_else(|| "replay:unknown".into()),
            })
            .collect();
        Self {
            name: name.into(),
            capabilities: Capabilities::default(),
            queue: Mutex::new(queue),
        }
    }

    /// Number of recorded responses not yet replayed. After a faithful re-drive
    /// this is `0`; a non-zero value means the current code made *fewer* LLM
    /// calls than the recording — a divergence the final-output oracle alone
    /// cannot see (a shorter path to the same answer).
    pub fn remaining(&self) -> usize {
        self.queue
            .lock()
            .expect("ReplayLlmClient mutex poisoned")
            .len()
    }
}

#[async_trait]
impl LlmClient for ReplayLlmClient {
    fn name(&self) -> &str {
        &self.name
    }
    fn capabilities(&self) -> &Capabilities {
        &self.capabilities
    }
    async fn complete(&self, _req: ChatRequest) -> Result<ChatResponse, LlmError> {
        let recorded = {
            let mut queue = self.queue.lock().expect("ReplayLlmClient mutex poisoned");
            queue.pop_front().ok_or_else(|| {
                // The double stands in for the LLM server; an exhausted script
                // is a server-side "cannot answer", not a malformed client
                // request — hence `Server`, not `BadRequest`.
                LlmError::Server(
                    "re-drive divergence: current code requested more LLM calls than the recording"
                        .into(),
                )
            })?
        };
        Ok(ChatResponse::new(
            Message {
                role: Role::Assistant,
                content: recorded.completion,
                tool_calls: recorded.tool_calls,
                tool_call_id: None,
            },
            Usage::default(),
            recorded.finish_reason,
            recorded.model,
        ))
    }
    async fn stream(&self, _req: ChatRequest) -> Result<ChunkStream, LlmError> {
        Err(LlmError::Unsupported(
            "ReplayLlmClient does not support streaming".into(),
        ))
    }
    async fn embed(&self, _texts: &[String]) -> Result<Vec<Embedding>, LlmError> {
        Err(LlmError::Unsupported(
            "ReplayLlmClient does not support embedding".into(),
        ))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use klieo_core::llm::ResponseFormat;

    fn request(content: &str) -> ChatRequest {
        ChatRequest {
            messages: vec![Message {
                role: Role::User,
                content: content.into(),
                tool_calls: vec![],
                tool_call_id: None,
            }],
            tools: vec![],
            temperature: None,
            max_tokens: None,
            response_format: ResponseFormat::Text,
            stop: vec![],
            timeout: None,
        }
    }

    #[tokio::test]
    async fn replays_structured_responses_in_order() {
        let tool_step = LlmIo::new("", "").with_response_shape(
            FinishReason::ToolCalls,
            vec![ToolCall::new("1", "search", serde_json::json!({}))],
        );
        let final_step = LlmIo::new("", "done");
        let client = ReplayLlmClient::from_llm_io("replay", &[tool_step, final_step]);

        let first = client.complete(request("a")).await.unwrap();
        assert_eq!(first.finish_reason, FinishReason::ToolCalls);
        assert_eq!(first.message.tool_calls.len(), 1);

        let second = client.complete(request("b")).await.unwrap();
        assert_eq!(second.finish_reason, FinishReason::Stop);
        assert_eq!(second.message.content, "done");
    }

    #[tokio::test]
    async fn remaining_tracks_undrained_responses() {
        let client =
            ReplayLlmClient::from_llm_io("replay", &[LlmIo::new("", "one"), LlmIo::new("", "two")]);
        assert_eq!(client.remaining(), 2);
        client.complete(request("a")).await.unwrap();
        assert_eq!(client.remaining(), 1, "one response left after one call");
        client.complete(request("b")).await.unwrap();
        assert_eq!(client.remaining(), 0, "fully drained");
    }

    #[tokio::test]
    async fn exhaustion_is_a_divergence_error() {
        let client = ReplayLlmClient::from_llm_io("replay", &[LlmIo::new("", "only")]);
        client.complete(request("a")).await.unwrap();
        let err = client.complete(request("b")).await.unwrap_err();
        assert!(
            matches!(err, LlmError::Server(m) if m.contains("re-drive divergence")),
            "a call past the recording is the hard divergence signal"
        );
    }

    #[tokio::test]
    async fn stream_and_embed_are_unsupported() {
        let client = ReplayLlmClient::from_llm_io("replay", &[LlmIo::new("a", "b")]);
        assert!(matches!(
            client.stream(request("a")).await,
            Err(LlmError::Unsupported(_))
        ));
        assert!(matches!(
            client.embed(&["x".to_string()]).await,
            Err(LlmError::Unsupported(_))
        ));
    }

    #[tokio::test]
    async fn ignores_request_content_answering_by_order() {
        // The double answers by position, not by request — re-drive cannot
        // reproduce the original request, so request content is irrelevant.
        let client = ReplayLlmClient::from_llm_io("replay", &[LlmIo::new("recorded", "answer")]);
        let resp = client.complete(request("totally different")).await.unwrap();
        assert_eq!(resp.message.content, "answer");
    }
}