//! Regression tests for `RequeueAndEscalate` approval-timeout policy (spec § 2.3).
//!
//! Three scenarios:
//! 1. `timeout_once_then_approve` — first window times out, approval arrives in
//!    the retry window → tool succeeds and events are recorded.
//! 2. `exhaust_max_requeues` — all windows time out → terminal Permanent error +
//!    `EscalationResolved` event recorded.
//! 3. `explicit_deny_mid_requeue` — first window times out (one requeue queued),
//!    then operator explicitly denies the ticket → no further requeues, Permanent.

use async_trait::async_trait;
use ed25519_dalek::{Signer, SigningKey};
use klieo_bus_memory::MemoryBus;
use klieo_core::error::ToolError;
use klieo_core::ids::RunId;
use klieo_core::memory::EpisodicMemory;
use klieo_core::test_utils::{noop_bus, FakeToolInvoker, InMemoryEpisodic};
use klieo_core::tool::{ToolCtx, ToolInvoker};
use klieo_ops::approver_registry::{ApproverId, StaticApproverRegistry};
use klieo_ops::gates::{
    ApprovalError, ApprovalOutcome, ApprovalTimeoutPolicy, FourEyesGate, Gate, GateDecision,
    GateRequest,
};
use klieo_ops::ops_event::OpsEvent;
use klieo_ops::runtime::gated_invoker::GatedToolInvoker;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;

fn fresh_kv() -> Arc<dyn klieo_core::KvStore> {
    MemoryBus::new().kv
}

// ── helpers ──────────────────────────────────────────────────────────────────

fn signing_key(seed: u8) -> SigningKey {
    SigningKey::from_bytes(&[seed; 32])
}

fn registry_with(pairs: &[(&str, u8)]) -> (StaticApproverRegistry, HashMap<String, SigningKey>) {
    let mut vkeys = HashMap::new();
    let mut skeys = HashMap::new();
    for (id, seed) in pairs {
        let sk = signing_key(*seed);
        vkeys.insert(ApproverId((*id).to_string()), sk.verifying_key());
        skeys.insert((*id).to_string(), sk);
    }
    (StaticApproverRegistry::from_map(vkeys), skeys)
}

fn tool_ctx() -> ToolCtx {
    let (pubsub, _rr, kv, jobs) = noop_bus();
    ToolCtx::new(pubsub, kv, jobs)
}

fn payout_invoker() -> Arc<dyn ToolInvoker> {
    Arc::new(
        FakeToolInvoker::new()
            .with_tool("payout", "transfer funds", |_| Ok(serde_json::json!("ok"))),
    )
}

/// Wraps `FourEyesGate` and forwards the first ticket over a channel so tests
/// can interact with the pending entry from a spawned task.
struct TicketCapture {
    inner: Arc<FourEyesGate>,
    sender: Mutex<Option<tokio::sync::oneshot::Sender<String>>>,
}

#[async_trait]
impl Gate for TicketCapture {
    async fn evaluate(&self, req: GateRequest) -> GateDecision {
        let decision = self.inner.evaluate(req).await;
        if let GateDecision::RequireApproval { ref ticket, .. } = decision {
            if let Some(tx) = self.sender.lock().await.take() {
                let _ = tx.send(ticket.clone());
            }
        }
        decision
    }

    fn name(&self) -> &'static str {
        self.inner.name()
    }

    fn may_require_approval(&self) -> bool {
        true
    }

    async fn wait_for_approval(
        &self,
        ticket: String,
        timeout: Duration,
    ) -> Result<ApprovalOutcome, ApprovalError> {
        self.inner.wait_for_approval(ticket, timeout).await
    }
}

// ── tests ─────────────────────────────────────────────────────────────────────

/// First approval window times out; approval arrives in the retry window.
/// Tool must succeed and `EscalationStateChanged` must be recorded.
#[tokio::test]
async fn timeout_once_then_approve() {
    let (reg, signers) = registry_with(&[("alice", 1), ("bob", 2)]);
    let gate = Arc::new(FourEyesGate::new(
        fresh_kv(),
        Arc::new(reg),
        2,
        FourEyesGate::dual_control_tools(&["payout"]),
    ));

    let (tx, rx) = tokio::sync::oneshot::channel::<String>();
    let capturing = Arc::new(TicketCapture {
        inner: gate.clone(),
        sender: Mutex::new(Some(tx)),
    });

    let episodic = Arc::new(InMemoryEpisodic::default());
    let run_id = RunId::new();
    let invoker = GatedToolInvoker::new(
        payout_invoker(),
        vec![capturing as Arc<dyn Gate>],
        episodic.clone(),
        run_id,
    )
    // Very short window so the first wait times out quickly.
    .with_approval_timeout(Duration::from_millis(60))
    .with_approval_timeout_policy(ApprovalTimeoutPolicy::RequeueAndEscalate { max_requeues: 3 });

    let gate_submit = gate.clone();
    let submit_handle = tokio::spawn(async move {
        let ticket = rx.await.expect("ticket channel");
        // Let the first window expire, then submit in the retry window.
        tokio::time::sleep(Duration::from_millis(80)).await;
        let payload = b"approve-payout";
        let sig_a = signers["alice"].sign(payload);
        let sig_b = signers["bob"].sign(payload);
        gate_submit
            .submit_approval(&ticket, ApproverId("alice".into()), sig_a, payload.to_vec())
            .expect("alice submit");
        gate_submit
            .submit_approval(&ticket, ApproverId("bob".into()), sig_b, payload.to_vec())
            .expect("bob submit");
    });

    let result: Result<serde_json::Value, ToolError> = invoker
        .invoke("payout", serde_json::json!({}), tool_ctx())
        .await;

    submit_handle.await.expect("submit task");

    let value = result.expect("tool must succeed after requeue + approval");
    assert_eq!(value, serde_json::json!("ok"));

    // Verify that an EscalationStateChanged event was recorded.
    let episodes = episodic.replay(run_id).await.expect("replay");
    let has_state_changed = episodes.iter().any(|ep| {
        if let klieo_core::memory::Episode::Ops(v) = ep {
            if let Some(OpsEvent::EscalationStateChanged { .. }) = OpsEvent::from_episode_payload(v)
            {
                return true;
            }
        }
        false
    });
    assert!(
        has_state_changed,
        "EscalationStateChanged must be recorded on timeout requeue"
    );
}

/// All requeue windows exhaust → terminal Permanent + EscalationResolved recorded.
#[tokio::test]
async fn exhaust_max_requeues() {
    let (reg, _) = registry_with(&[("alice", 1)]);
    let gate = Arc::new(FourEyesGate::new(
        fresh_kv(),
        Arc::new(reg),
        2,
        FourEyesGate::dual_control_tools(&["payout"]),
    ));

    let episodic = Arc::new(InMemoryEpisodic::default());
    let run_id = RunId::new();
    let invoker = GatedToolInvoker::new(
        payout_invoker(),
        vec![gate as Arc<dyn Gate>],
        episodic.clone(),
        run_id,
    )
    .with_approval_timeout(Duration::from_millis(30))
    .with_approval_timeout_policy(ApprovalTimeoutPolicy::RequeueAndEscalate { max_requeues: 2 });

    let result: Result<serde_json::Value, ToolError> = invoker
        .invoke("payout", serde_json::json!({}), tool_ctx())
        .await;

    let err = result.expect_err("must fail after exhausting requeues");
    let ToolError::Permanent(msg) = err else {
        panic!("expected Permanent, got {err:?}");
    };
    assert!(
        msg.contains("exhausted"),
        "error must mention 'exhausted'; got: {msg}"
    );

    // Verify EscalationResolved event was recorded.
    let episodes = episodic.replay(run_id).await.expect("replay");
    let resolved = episodes.iter().any(|ep| {
        if let klieo_core::memory::Episode::Ops(v) = ep {
            if let Some(OpsEvent::EscalationResolved { outcome, .. }) =
                OpsEvent::from_episode_payload(v)
            {
                return outcome == "timed_out";
            }
        }
        false
    });
    assert!(
        resolved,
        "EscalationResolved {{ outcome: 'timed_out' }} must be recorded"
    );
}

/// First window times out (one requeue pending), then operator explicitly
/// denies the ticket. No further requeues; result is Permanent.
#[tokio::test]
async fn explicit_deny_mid_requeue() {
    let (reg, _) = registry_with(&[("alice", 1)]);
    let gate = Arc::new(FourEyesGate::new(
        fresh_kv(),
        Arc::new(reg),
        2,
        FourEyesGate::dual_control_tools(&["payout"]),
    ));

    let (tx, rx) = tokio::sync::oneshot::channel::<String>();
    let capturing = Arc::new(TicketCapture {
        inner: gate.clone(),
        sender: Mutex::new(Some(tx)),
    });

    let episodic = Arc::new(InMemoryEpisodic::default());
    let invoker = GatedToolInvoker::new(
        payout_invoker(),
        vec![capturing as Arc<dyn Gate>],
        episodic.clone(),
        RunId::new(),
    )
    .with_approval_timeout(Duration::from_millis(60))
    .with_approval_timeout_policy(ApprovalTimeoutPolicy::RequeueAndEscalate { max_requeues: 3 });

    let gate_deny = gate.clone();
    let deny_handle = tokio::spawn(async move {
        let ticket = rx.await.expect("ticket channel");
        // Let the first window expire (requeue triggered), then deny.
        tokio::time::sleep(Duration::from_millis(80)).await;
        gate_deny.deny_approval(&ticket, "compliance officer denied mid-requeue");
    });

    let result: Result<serde_json::Value, ToolError> = invoker
        .invoke("payout", serde_json::json!({}), tool_ctx())
        .await;

    deny_handle.await.expect("deny task");

    let err = result.expect_err("must fail on mid-requeue explicit deny");
    let ToolError::Permanent(msg) = err else {
        panic!("expected Permanent, got {err:?}");
    };
    assert!(
        msg.contains("denied"),
        "error must mention 'denied'; got: {msg}"
    );
}
