//! End-to-end: GatedToolInvoker suspends on RequireApproval, resumes once
//! quorum is supplied, and surfaces timed-out errors when no approvals arrive.
//!
//! Tests drive GatedToolInvoker directly (no SupervisedAgent wrapper) for
//! minimal friction. SupervisedAgent integration is covered by the
//! auto_emission + M8C conformance test suites.

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::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, FourEyesGate, Gate, GateDecision, GateRequest,
};
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"))),
    )
}

/// Gate wrapper that captures the ticket from the first RequireApproval
/// decision and forwards it over the provided channel.
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 ─────────────────────────────────────────────────────────────────────

/// Happy-path: quorum is reached after the invoker parks; the inner tool
/// executes and returns its result.
#[tokio::test]
async fn four_eyes_gate_invoker_runs_tool_after_approval() {
    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 invoker = GatedToolInvoker::new(
        payout_invoker(),
        vec![capturing as Arc<dyn Gate>],
        episodic,
        RunId::new(),
    )
    .with_approval_timeout(Duration::from_secs(5));

    // Submit approvals from a separate task after the ticket is known.
    let gate_submit = gate.clone();
    let submit_handle = tokio::spawn(async move {
        let ticket = rx.await.expect("ticket channel");
        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 quorum");
    assert_eq!(value, serde_json::json!("ok"), "inner tool must have run");
}

/// Timeout path: no approvals submitted → invoker returns Permanent error
/// containing "timed out".
#[tokio::test]
async fn four_eyes_gate_invoker_times_out_with_no_approvers() {
    let (reg, _) = registry_with(&[("alice", 1), ("bob", 2)]);
    let gate = Arc::new(FourEyesGate::new(
        fresh_kv(),
        Arc::new(reg),
        2,
        FourEyesGate::dual_control_tools(&["payout"]),
    ));

    let episodic = Arc::new(InMemoryEpisodic::default());
    let invoker = GatedToolInvoker::new(
        payout_invoker(),
        vec![gate as Arc<dyn Gate>],
        episodic,
        RunId::new(),
    )
    .with_approval_timeout(Duration::from_millis(80));

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

    let err = result.expect_err("must fail with timeout");
    let ToolError::Permanent(msg) = err else {
        panic!("expected Permanent, got {err:?}");
    };
    assert!(
        msg.contains("timed out"),
        "error message must mention 'timed out'; got: {msg}"
    );
}

/// Non-dual-control tools bypass the gate without any approval flow.
#[tokio::test]
async fn non_dual_control_tool_passes_through_gate() {
    let (reg, _) = registry_with(&[]);
    let gate = Arc::new(FourEyesGate::new(
        fresh_kv(),
        Arc::new(reg),
        2,
        FourEyesGate::dual_control_tools(&["payout"]),
    ));

    let inner = Arc::new(
        FakeToolInvoker::new()
            .with_tool("read_balance", "read only", |_| Ok(serde_json::json!(42))),
    );
    let episodic = Arc::new(InMemoryEpisodic::default());
    let invoker = GatedToolInvoker::new(inner, vec![gate as Arc<dyn Gate>], episodic, RunId::new())
        .with_approval_timeout(Duration::from_millis(100));

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

    assert_eq!(
        result.expect("non-dual-control tool allowed"),
        serde_json::json!(42)
    );
}

/// Explicit deny from the gate's deny_approval results in Permanent error
/// containing "denied".
#[tokio::test]
async fn four_eyes_gate_invoker_surfaces_explicit_deny() {
    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,
        RunId::new(),
    )
    .with_approval_timeout(Duration::from_secs(5));

    let gate_deny = gate.clone();
    let deny_handle = tokio::spawn(async move {
        let ticket = rx.await.expect("ticket channel");
        gate_deny.deny_approval(&ticket, "compliance officer denied");
    });

    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 explicit deny");
    let ToolError::Permanent(msg) = err else {
        panic!("expected Permanent, got {err:?}");
    };
    assert!(
        msg.contains("denied"),
        "error message must mention 'denied'; got: {msg}"
    );
}
