//! Verifies that [`SupervisedAgent::run`] auto-emits Supervisor / Governor /
//! Gate events into the agent's [`EpisodicMemory`] without user code calling
//! [`emit_ops_event`] manually.

use async_trait::async_trait;
use klieo_bus_memory::MemoryBus;
use klieo_core::memory::{Episode, EpisodicMemory};
use klieo_core::test_utils::{fake_context, FakeToolInvoker, InMemoryEpisodic};
use klieo_core::{Agent, AgentContext, RunId, ToolCtx, ToolDef, ToolInvoker};
use klieo_ops::gates::AllowlistGate;
use klieo_ops::governor::TokenBucketGovernor;
use klieo_ops::ops_event::OpsEvent;
use klieo_ops::{OpsRuntime, ProviderId, TenantId, TenantResolver};
use std::sync::Arc;

// ── Fixtures ──────────────────────────────────────────────────────────────────

struct EchoAgent;

#[async_trait]
impl Agent for EchoAgent {
    type Input = String;
    type Output = String;
    type Error = std::io::Error;

    fn name(&self) -> &str {
        "echo"
    }
    fn system_prompt(&self) -> &str {
        ""
    }
    fn tools(&self) -> &[ToolDef] {
        &[]
    }
    async fn run(&self, _ctx: AgentContext, input: String) -> Result<String, std::io::Error> {
        Ok(input)
    }
}

/// Agent that invokes `echo_tool` through ctx.tools so gates get exercised.
struct ToolCallingAgent;

#[async_trait]
impl Agent for ToolCallingAgent {
    type Input = String;
    type Output = String;
    type Error = std::io::Error;

    fn name(&self) -> &str {
        "tool-caller"
    }
    fn system_prompt(&self) -> &str {
        ""
    }
    fn tools(&self) -> &[ToolDef] {
        &[]
    }
    async fn run(&self, ctx: AgentContext, input: String) -> Result<String, std::io::Error> {
        let tool_ctx = ToolCtx::new(ctx.pubsub.clone(), ctx.kv.clone(), ctx.jobs.clone());
        // Deliberately ignore the result — the gate denial we're testing causes
        // an error, but the agent output is still the echoed input.
        let _ = ctx
            .tools
            .invoke("echo_tool", serde_json::json!({}), tool_ctx)
            .await;
        Ok(input)
    }
}

struct FailingAgent;

#[async_trait]
impl Agent for FailingAgent {
    type Input = String;
    type Output = String;
    type Error = std::io::Error;

    fn name(&self) -> &str {
        "failing"
    }
    fn system_prompt(&self) -> &str {
        ""
    }
    fn tools(&self) -> &[ToolDef] {
        &[]
    }
    async fn run(&self, _ctx: AgentContext, _input: String) -> Result<String, std::io::Error> {
        Err(std::io::Error::other("intentional failure"))
    }
}

struct FixedTenant(TenantId);

#[async_trait]
impl TenantResolver for FixedTenant {
    async fn resolve(&self) -> Option<TenantId> {
        Some(self.0.clone())
    }
}

fn make_ctx(episodic: Arc<InMemoryEpisodic>, run_id: RunId) -> AgentContext {
    make_ctx_with_tools(episodic, run_id, Arc::new(FakeToolInvoker::new()))
}

fn make_ctx_with_tools(
    episodic: Arc<InMemoryEpisodic>,
    run_id: RunId,
    tools: Arc<dyn ToolInvoker>,
) -> AgentContext {
    let mut ctx = fake_context("test");
    ctx.episodic = episodic;
    ctx.run_id = run_id;
    ctx.tools = tools;
    ctx
}

fn collect_state_changes(events: &[Episode]) -> Vec<String> {
    events
        .iter()
        .filter_map(|e| {
            let Episode::Ops(payload) = e else {
                return None;
            };
            match OpsEvent::from_episode_payload(payload)? {
                OpsEvent::SupervisorStateChange { state, .. } => Some(state),
                _ => None,
            }
        })
        .collect()
}

// ── Tests ─────────────────────────────────────────────────────────────────────

#[tokio::test]
async fn supervised_agent_emits_running_then_completed() {
    let episodic = Arc::new(InMemoryEpisodic::default());
    let run_id = RunId::new();

    let supervised = OpsRuntime::builder()
        .tenant_resolver(Arc::new(FixedTenant(TenantId("BL_auto".into()))))
        .spawn(EchoAgent)
        .expect("build ok");

    let ctx = make_ctx(episodic.clone(), run_id);
    let out = supervised.run(ctx, "hi".into()).await.expect("run ok");
    assert_eq!(out, "hi");

    let events = episodic.replay(run_id).await.expect("replay ok");
    let states = collect_state_changes(&events);

    assert!(
        states.iter().any(|s| s == "running"),
        "must emit running; got {states:?}"
    );
    assert!(
        states.iter().any(|s| s == "completed"),
        "must emit completed; got {states:?}"
    );
}

#[tokio::test]
async fn supervised_agent_emits_failed_on_error() {
    let episodic = Arc::new(InMemoryEpisodic::default());
    let run_id = RunId::new();

    let supervised = OpsRuntime::builder()
        .tenant_resolver(Arc::new(FixedTenant(TenantId("BL_auto".into()))))
        .spawn(FailingAgent)
        .expect("build ok");

    let ctx = make_ctx(episodic.clone(), run_id);
    let _ = supervised.run(ctx, "ignored".into()).await;

    let events = episodic.replay(run_id).await.expect("replay ok");
    let states = collect_state_changes(&events);

    assert!(
        states.iter().any(|s| s == "failed"),
        "must emit failed on agent error; got {states:?}"
    );
}

#[tokio::test]
async fn governor_wires_into_supervised_agent() {
    let episodic = Arc::new(InMemoryEpisodic::default());
    let run_id = RunId::new();

    let bus = MemoryBus::new();
    let governor = Arc::new(
        TokenBucketGovernor::builder(bus.kv.clone())
            .llm_limit(ProviderId("default".into()), 10)
            .build(),
    );

    let supervised = OpsRuntime::builder()
        .tenant_resolver(Arc::new(FixedTenant(TenantId("BL_auto".into()))))
        .governor(governor)
        .governor_provider(ProviderId("default".into()))
        .spawn(EchoAgent)
        .expect("build ok");

    // EchoAgent never calls ctx.llm, so no permit is consumed. The test
    // verifies that adding a governor does not break the normal run path
    // and that state-change events are still emitted correctly.
    let ctx = make_ctx(episodic.clone(), run_id);
    let out = supervised.run(ctx, "ping".into()).await.expect("run ok");
    assert_eq!(out, "ping");

    let events = episodic.replay(run_id).await.expect("replay ok");
    let states = collect_state_changes(&events);
    assert!(
        states.iter().any(|s| s == "completed"),
        "must emit completed with governor wired; got {states:?}"
    );
}

#[tokio::test]
async fn gate_denial_emits_gate_decision_event() {
    let episodic = Arc::new(InMemoryEpisodic::default());
    let run_id = RunId::new();

    // AllowlistGate that only allows "safe_tool" — "echo_tool" will be denied.
    let gate: Arc<dyn klieo_ops::gates::Gate> = Arc::new(AllowlistGate::new(&["safe_tool"]));

    let inner_tools: Arc<dyn ToolInvoker> =
        Arc::new(FakeToolInvoker::new().with_tool("echo_tool", "echoes back", Ok));

    let supervised = OpsRuntime::builder()
        .tenant_resolver(Arc::new(FixedTenant(TenantId("BL_auto".into()))))
        .gate(gate)
        .spawn(ToolCallingAgent)
        .expect("build ok");

    let ctx = make_ctx_with_tools(episodic.clone(), run_id, inner_tools);
    let _out = supervised.run(ctx, "trigger".into()).await.expect("run ok");

    let events = episodic.replay(run_id).await.expect("replay ok");
    let gate_decisions: Vec<String> = events
        .iter()
        .filter_map(|e| {
            let Episode::Ops(p) = e else {
                return None;
            };
            match OpsEvent::from_episode_payload(p)? {
                OpsEvent::GateDecision { decision, .. } => Some(decision),
                _ => None,
            }
        })
        .collect();

    assert!(
        gate_decisions.iter().any(|d| d == "deny"),
        "must emit GateDecision deny when gate blocks tool; got {gate_decisions:?}"
    );
}

#[tokio::test]
async fn gate_allow_emits_gate_decision_event() {
    let episodic = Arc::new(InMemoryEpisodic::default());
    let run_id = RunId::new();

    // AllowlistGate that allows "echo_tool".
    let gate: Arc<dyn klieo_ops::gates::Gate> = Arc::new(AllowlistGate::new(&["echo_tool"]));

    let inner_tools: Arc<dyn ToolInvoker> =
        Arc::new(FakeToolInvoker::new().with_tool("echo_tool", "echoes back", Ok));

    let supervised = OpsRuntime::builder()
        .tenant_resolver(Arc::new(FixedTenant(TenantId("BL_auto".into()))))
        .gate(gate)
        .spawn(ToolCallingAgent)
        .expect("build ok");

    let ctx = make_ctx_with_tools(episodic.clone(), run_id, inner_tools);
    let out = supervised.run(ctx, "ok".into()).await.expect("run ok");
    assert_eq!(out, "ok");

    let events = episodic.replay(run_id).await.expect("replay ok");
    let gate_decisions: Vec<String> = events
        .iter()
        .filter_map(|e| {
            let Episode::Ops(p) = e else {
                return None;
            };
            match OpsEvent::from_episode_payload(p)? {
                OpsEvent::GateDecision { decision, .. } => Some(decision),
                _ => None,
            }
        })
        .collect();

    assert!(
        gate_decisions.iter().any(|d| d == "allow"),
        "must emit GateDecision allow when gate permits tool; got {gate_decisions:?}"
    );
}
