klieo-ops 3.3.0

//! Four-eyes gate: emits `RequireApproval` for any tool whose classifier
//! returns `true` and supports `wait_for_approval` for the
//! `GatedToolInvoker` (M8B) suspend-then-resume path.
//!
//! ## Multi-process correctness
//!
//! Pending approval state lives in a shared `KvStore` under
//! `ops.four_eyes.pending/<ticket>`.  Cross-process callers converge on
//! that bucket as the single source of truth; in-process callers also
//! receive an instant wake via a `Notify`.

use super::trait_::{ApprovalError, ApprovalOutcome, Gate, GateDecision, GateRequest};
use crate::approver_registry::{ApproverId, ApproverRegistry};
use crate::escalation::Severity;
use async_trait::async_trait;
use base64::engine::general_purpose::STANDARD as B64;
use base64::Engine as _;
use bytes::Bytes;
use dashmap::DashMap;
use ed25519_dalek::{Signature, Verifier};
use klieo_core::error::BusError;
use klieo_core::KvStore;
use serde::{Deserialize, Serialize};
use std::collections::HashSet;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Notify;
use ulid::Ulid;

const KV_BUCKET: &str = "ops.four_eyes.pending";
const CAS_MAX_RETRIES: u8 = 5;
const POLL_INTERVAL_MS: u64 = 250;
const POLL_JITTER_MS: u64 = 50;
/// Timeout applied to each `run_sync` call so a hung KV store does not
/// deadlock the calling thread indefinitely.
const SYNC_BRIDGE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);

/// Tool name + args → dual-control membership. Closure-shaped so callers
/// can wire arbitrary classification logic.
pub type DualControlClassifier = Arc<dyn Fn(&str, &serde_json::Value) -> bool + Send + Sync>;

// ── KV state types ────────────────────────────────────────────────────────────

/// Serializable representation of one approver submission.
#[derive(Clone, Debug, Serialize, Deserialize)]
struct Submission {
    approver: String,
    signature_hex: String,
    payload_b64: String,
}

/// Terminal outcome stored in KV.
#[derive(Clone, Debug, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
enum KvOutcome {
    Allow,
    Deny { code: String, reason: String },
}

/// Full state stored under `ops.four_eyes.pending/<ticket>`.
#[derive(Clone, Debug, Serialize, Deserialize)]
struct PendingApprovalState {
    severity: Severity,
    submissions: Vec<Submission>,
    outcome: Option<KvOutcome>,
}

impl PendingApprovalState {
    fn new(severity: Severity) -> Self {
        Self {
            severity,
            submissions: Vec::new(),
            outcome: None,
        }
    }
}

// ── In-process notify registry ────────────────────────────────────────────────

/// Per-ticket `Notify` for same-process fast-path wakeup. Cross-process
/// callers fall back to KV polling; both paths converge on the KV outcome.
struct LocalNotify {
    notify: Notify,
}

impl LocalNotify {
    fn new() -> Arc<Self> {
        Arc::new(Self {
            notify: Notify::new(),
        })
    }
}

// ── Gate ──────────────────────────────────────────────────────────────────────

/// Phase B FourEyesGate. Emits `RequireApproval` for tools whose
/// classifier returns `true` (typically a `dual_control` tag check against
/// the agent's `ToolDef`). Supports `wait_for_approval` so
/// `GatedToolInvoker` can suspend the tool call until quorum is met or
/// timeout fires.
///
/// Pending state is persisted to a shared `KvStore`; multiple gate
/// instances (different processes) sharing the same KV bucket converge on
/// the same approval outcome.
pub struct FourEyesGate {
    kv: Arc<dyn KvStore>,
    registry: Arc<dyn ApproverRegistry>,
    quorum: u8,
    classifier: DualControlClassifier,
    local_notifies: Arc<DashMap<String, Arc<LocalNotify>>>,
    /// Severity assigned to new tickets; default `High`.
    severity: Severity,
    /// Timeout applied to each `run_sync` call. Configurable for testing
    /// via `with_sync_bridge_timeout`; defaults to `SYNC_BRIDGE_TIMEOUT`.
    sync_bridge_timeout: Duration,
}

impl FourEyesGate {
    /// Construct a `FourEyesGate` with default severity `High`.
    #[must_use]
    pub fn new(
        kv: Arc<dyn KvStore>,
        registry: Arc<dyn ApproverRegistry>,
        quorum: u8,
        classifier: DualControlClassifier,
    ) -> Self {
        Self {
            kv,
            registry,
            quorum,
            classifier,
            local_notifies: Arc::new(DashMap::new()),
            severity: Severity::High,
            sync_bridge_timeout: SYNC_BRIDGE_TIMEOUT,
        }
    }

    /// Override the sync-bridge timeout used by `submit_approval` and
    /// `deny_approval`. Intended for test code only — allows tests to inject
    /// a short timeout so the sync-bridge timeout path can be exercised
    /// deterministically without waiting 30 seconds.
    #[cfg(any(test, feature = "test-utils"))]
    #[must_use]
    pub fn with_sync_bridge_timeout(mut self, d: Duration) -> Self {
        self.sync_bridge_timeout = d;
        self
    }

    /// Override the initial severity assigned to new approval tickets.
    /// Severity is bumped one level on each requeue by `GatedToolInvoker`.
    #[must_use]
    pub fn with_severity(mut self, severity: Severity) -> Self {
        self.severity = severity;
        self
    }

    /// Return the configured initial severity for new tickets.
    pub fn severity(&self) -> Severity {
        self.severity
    }

    /// Explicitly remove a pending ticket without resolving it.
    ///
    /// The requeue path in `GatedToolInvoker` leaves pending entries alive
    /// across `wait_for_approval` timeouts so that the same ticket can be
    /// waited on again with a fresh timer. Call `cancel_pending` for
    /// explicit cleanup when the invoker decides not to retry.
    pub fn cancel_pending(&self, ticket: &str) {
        self.local_notifies.remove(ticket);
        let kv = self.kv.clone();
        let ticket = ticket.to_string();
        tokio::spawn(async move {
            if let Err(err) = kv.delete(KV_BUCKET, &ticket).await {
                tracing::warn!(
                    target: "klieo.ops.four_eyes",
                    ticket = %ticket,
                    error = %err,
                    "cancel_pending KV delete failed; ticket may leak into stale pending state"
                );
            }
        });
    }

    /// Convenience: returns a classifier that fires for any tool whose
    /// name appears in `tools`.
    #[must_use]
    pub fn dual_control_tools(tools: &[&str]) -> DualControlClassifier {
        let set: HashSet<String> = tools.iter().map(|s| (*s).to_string()).collect();
        Arc::new(move |name: &str, _args: &serde_json::Value| set.contains(name))
    }

    /// Submit an approver signature for a pending ticket. When quorum is
    /// reached, the `wait_for_approval` future is woken with the final
    /// outcome. Called by the operator harness / approval UI.
    pub fn submit_approval(
        &self,
        ticket: &str,
        approver: ApproverId,
        signature: Signature,
        signed_payload: Vec<u8>,
    ) -> Result<(), ApprovalError> {
        let kv = self.kv.clone();
        let registry = self.registry.clone();
        let quorum = self.quorum;
        let ticket_str = ticket.to_string();
        let local_notifies = self.local_notifies.clone();

        run_sync(
            submit_approval_async(SubmitParams {
                kv,
                registry,
                quorum,
                ticket: ticket_str,
                approver,
                signature,
                signed_payload,
                local_notifies,
            }),
            self.sync_bridge_timeout,
        )
    }

    /// Explicitly deny an outstanding ticket (operator action).
    pub fn deny_approval(&self, ticket: &str, reason: impl Into<String>) {
        let kv = self.kv.clone();
        let ticket_str = ticket.to_string();
        let reason = reason.into();
        let local_notifies = self.local_notifies.clone();

        let timeout = self.sync_bridge_timeout;
        let outcome_result = run_sync(
            async move {
                if let Err(err) = cas_write_outcome(
                    &kv,
                    &ticket_str,
                    KvOutcome::Deny {
                        code: "operator.deny".into(),
                        reason,
                    },
                )
                .await
                {
                    tracing::warn!(
                        target: "klieo.ops.four_eyes",
                        error = %err,
                        "deny_approval cas_write_outcome failed; KV state may diverge from local notify"
                    );
                }
                if let Some(n) = local_notifies.get(&ticket_str) {
                    n.notify.notify_waiters();
                }
                Ok::<(), ApprovalError>(())
            },
            timeout,
        );
        if let Err(err) = outcome_result {
            tracing::warn!(
                target: "klieo.ops.four_eyes",
                error = %err,
                "deny_approval run_sync bridge failed"
            );
        }
    }
}

// ── Sync bridge ───────────────────────────────────────────────────────────────

/// Run an async future to completion from a sync context, bounded by
/// `timeout`. A hung KV store will not deadlock the caller indefinitely —
/// after the timeout the wrapped future returns
/// `ApprovalError::VerificationFailed("kv timeout …")`.
///
/// If a Tokio multi-thread runtime is active, uses `block_in_place` to avoid
/// blocking the executor thread. Otherwise (single-thread runtime, non-async
/// callers) spawns a fresh OS thread with its own minimal runtime.
fn run_sync<F>(fut: F, timeout: Duration) -> Result<(), ApprovalError>
where
    F: std::future::Future<Output = Result<(), ApprovalError>> + Send + 'static,
{
    let timed = async move {
        tokio::time::timeout(timeout, fut)
            .await
            .unwrap_or_else(|_| {
                tracing::warn!(
                    target: "klieo.ops.four_eyes.sync_bridge_timeout",
                    timeout_ms = timeout.as_millis(),
                    "sync bridge to async KV exceeded timeout; KV may be wedged"
                );
                Err(ApprovalError::VerificationFailed(format!(
                    "kv timeout: sync bridge exceeded {}ms",
                    timeout.as_millis()
                )))
            })
    };

    match tokio::runtime::Handle::try_current() {
        Ok(handle) => {
            // If we are inside a multi-thread runtime we can block_in_place;
            // for current_thread runtimes we must offload to a separate thread
            // to avoid deadlocking the executor.
            if handle.runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread {
                tokio::task::block_in_place(|| handle.block_on(timed))
            } else {
                std::thread::scope(|s| {
                    s.spawn(|| {
                        tokio::runtime::Builder::new_current_thread()
                            .enable_all()
                            .build()
                            .expect("mini runtime")
                            .block_on(timed)
                    })
                    .join()
                    .expect("sync bridge thread")
                })
            }
        }
        Err(_) => {
            // No runtime at all — build a minimal one.
            tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .expect("mini runtime")
                .block_on(timed)
        }
    }
}

// ── Async helpers ─────────────────────────────────────────────────────────────

/// Read the current state from KV.
async fn read_kv_state(
    kv: &Arc<dyn KvStore>,
    ticket: &str,
) -> Result<Option<(PendingApprovalState, u64)>, ApprovalError> {
    match kv.get(KV_BUCKET, ticket).await {
        Ok(Some(entry)) => {
            let state: PendingApprovalState = serde_json::from_slice(&entry.value)
                .map_err(|e| ApprovalError::VerificationFailed(format!("kv state corrupt: {e}")))?;
            Ok(Some((state, entry.revision)))
        }
        Ok(None) => Ok(None),
        Err(e) => Err(ApprovalError::VerificationFailed(format!(
            "kv read error: {e}"
        ))),
    }
}

/// CAS-write a full state value; retries on conflict up to `CAS_MAX_RETRIES`.
async fn cas_write_state(
    kv: &Arc<dyn KvStore>,
    ticket: &str,
    state: PendingApprovalState,
    expected_revision: u64,
) -> Result<(), BusError> {
    let bytes = Bytes::from(serde_json::to_vec(&state).expect("serializable"));
    kv.cas(KV_BUCKET, ticket, bytes, Some(expected_revision))
        .await
        .map(|_| ())
}

/// Read-modify-CAS to write an outcome into the pending state.
async fn cas_write_outcome(
    kv: &Arc<dyn KvStore>,
    ticket: &str,
    outcome: KvOutcome,
) -> Result<(), ApprovalError> {
    for _ in 0..CAS_MAX_RETRIES {
        let Some((mut state, rev)) = read_kv_state(kv, ticket).await? else {
            return Ok(());
        };
        if state.outcome.is_some() {
            return Ok(());
        }
        state.outcome = Some(outcome.clone());
        match cas_write_state(kv, ticket, state, rev).await {
            Ok(()) => return Ok(()),
            Err(BusError::CasConflict { .. }) => continue,
            Err(e) => {
                return Err(ApprovalError::VerificationFailed(format!(
                    "kv write error: {e}"
                )))
            }
        }
    }
    Ok(())
}

/// Parameters for the async submit flow; grouped to stay under the
/// `too_many_arguments` lint limit.
struct SubmitParams {
    kv: Arc<dyn KvStore>,
    registry: Arc<dyn ApproverRegistry>,
    quorum: u8,
    ticket: String,
    approver: ApproverId,
    signature: Signature,
    signed_payload: Vec<u8>,
    local_notifies: Arc<DashMap<String, Arc<LocalNotify>>>,
}

/// Full submit flow: CAS-append submission, run quorum check, write outcome.
async fn submit_approval_async(p: SubmitParams) -> Result<(), ApprovalError> {
    let SubmitParams {
        kv,
        registry,
        quorum,
        ticket,
        approver,
        signature,
        signed_payload,
        local_notifies,
    } = p;
    let new_submission = Submission {
        approver: approver.0.clone(),
        signature_hex: hex::encode(signature.to_bytes()),
        payload_b64: B64.encode(&signed_payload),
    };

    for _ in 0..CAS_MAX_RETRIES {
        let Some((mut state, rev)) = read_kv_state(&kv, &ticket).await? else {
            return Err(ApprovalError::VerificationFailed("unknown ticket".into()));
        };

        if state.outcome.is_some() {
            // Already resolved — idempotent success.
            return Ok(());
        }

        state.submissions.push(new_submission.clone());

        let unique_count = state
            .submissions
            .iter()
            .map(|s| s.approver.as_str())
            .collect::<HashSet<_>>()
            .len();

        if unique_count < quorum as usize {
            // Quorum not yet met — just persist the new submission.
            match cas_write_state(&kv, &ticket, state, rev).await {
                Ok(()) => return Ok(()),
                Err(BusError::CasConflict { .. }) => continue,
                Err(e) => {
                    return Err(ApprovalError::VerificationFailed(format!(
                        "kv write error: {e}"
                    )))
                }
            }
        }

        // Quorum of unique identities reached — verify all submissions.
        let outcome = verify_quorum(&state.submissions, &registry, quorum)?;
        state.outcome = Some(outcome);

        match cas_write_state(&kv, &ticket, state, rev).await {
            Ok(()) => {
                if let Some(n) = local_notifies.get(&ticket) {
                    n.notify.notify_waiters();
                }
                return Ok(());
            }
            Err(BusError::CasConflict { .. }) => continue,
            Err(e) => {
                return Err(ApprovalError::VerificationFailed(format!(
                    "kv write error: {e}"
                )))
            }
        }
    }

    Ok(())
}

/// Verify all submissions, return the final `KvOutcome`.
fn verify_quorum(
    submissions: &[Submission],
    registry: &Arc<dyn ApproverRegistry>,
    quorum: u8,
) -> Result<KvOutcome, ApprovalError> {
    let mut verified_ids: HashSet<String> = HashSet::new();

    for sub in submissions {
        let id = ApproverId(sub.approver.clone());
        let Some(vk) = registry.lookup(&id) else {
            let reason = format!("approver `{}` not in registry", sub.approver);
            return Err(ApprovalError::VerificationFailed(reason));
        };

        let sig_bytes = hex::decode(&sub.signature_hex)
            .map_err(|e| ApprovalError::VerificationFailed(format!("bad signature hex: {e}")))?;
        let sig = Signature::from_slice(&sig_bytes)
            .map_err(|e| ApprovalError::VerificationFailed(format!("malformed signature: {e}")))?;
        let payload = B64
            .decode(&sub.payload_b64)
            .map_err(|e| ApprovalError::VerificationFailed(format!("bad payload b64: {e}")))?;

        if vk.verify(&payload, &sig).is_err() {
            let reason = format!("approver `{}` signature math-invalid", sub.approver);
            return Err(ApprovalError::VerificationFailed(reason));
        }

        verified_ids.insert(sub.approver.clone());
    }

    if verified_ids.len() < quorum as usize {
        return Err(ApprovalError::VerificationFailed(format!(
            "quorum {} not met after unique-identity dedup; got {}",
            quorum,
            verified_ids.len()
        )));
    }

    Ok(KvOutcome::Allow)
}

/// Poll KV for an outcome, honouring a local `Notify` fast-path.
async fn poll_for_outcome(
    kv: Arc<dyn KvStore>,
    ticket: String,
    notify: Arc<LocalNotify>,
) -> Result<ApprovalOutcome, ApprovalError> {
    loop {
        match read_kv_state(&kv, &ticket).await? {
            Some((state, _)) => {
                if let Some(outcome) = state.outcome {
                    return kv_outcome_to_result(outcome);
                }
            }
            None => return Err(ApprovalError::VerificationFailed("unknown ticket".into())),
        }

        // Wait either for local Notify or poll interval (with jitter).
        let jitter = rand_jitter_ms();
        let sleep = tokio::time::sleep(Duration::from_millis(POLL_INTERVAL_MS + jitter));

        tokio::select! {
            _ = notify.notify.notified() => {}
            _ = sleep => {}
        }
    }
}

/// Convert a `KvOutcome` to the public `ApprovalOutcome` / `ApprovalError`.
fn kv_outcome_to_result(outcome: KvOutcome) -> Result<ApprovalOutcome, ApprovalError> {
    match outcome {
        KvOutcome::Allow => Ok(ApprovalOutcome::Allow),
        KvOutcome::Deny { reason, .. } => Err(ApprovalError::Denied(reason)),
    }
}

/// ±50ms jitter to avoid thundering-herd polling across waiters.
fn rand_jitter_ms() -> u64 {
    use std::collections::hash_map::DefaultHasher;
    use std::hash::{Hash, Hasher};
    use std::time::SystemTime;
    let mut h = DefaultHasher::new();
    SystemTime::now()
        .duration_since(SystemTime::UNIX_EPOCH)
        .unwrap_or_default()
        .subsec_nanos()
        .hash(&mut h);
    h.finish() % (POLL_JITTER_MS * 2)
}

// ── Gate trait impl ───────────────────────────────────────────────────────────

#[async_trait]
impl Gate for FourEyesGate {
    async fn evaluate(&self, req: GateRequest) -> GateDecision {
        if !(self.classifier)(&req.tool_name, &req.args) {
            return GateDecision::Allow;
        }
        let ticket = format!("esc_{}", Ulid::new());

        let state = PendingApprovalState::new(self.severity);
        let bytes = Bytes::from(serde_json::to_vec(&state).expect("serializable"));
        if let Err(e) = self.kv.cas(KV_BUCKET, &ticket, bytes, None).await {
            tracing::error!(
                target: "klieo.ops.four_eyes",
                ticket = %ticket,
                error = %e,
                "failed to create pending state in KV; failing CLOSED"
            );
            return GateDecision::Deny {
                code: "four_eyes.kv_unavailable".into(),
                reason: format!("KV write failed, cannot create approval ticket: {e}"),
            };
        }

        self.local_notifies
            .insert(ticket.clone(), LocalNotify::new());

        GateDecision::RequireApproval {
            ticket,
            quorum: self.quorum,
        }
    }

    fn name(&self) -> &'static str {
        "FourEyesGate"
    }

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

    async fn wait_for_approval(
        &self,
        ticket: String,
        timeout: Duration,
    ) -> Result<ApprovalOutcome, ApprovalError> {
        let notify = self
            .local_notifies
            .get(&ticket)
            .map(|e| e.clone())
            .unwrap_or_else(LocalNotify::new);

        let kv = self.kv.clone();
        let ticket_clone = ticket.clone();

        let poll_result =
            tokio::time::timeout(timeout, poll_for_outcome(kv, ticket_clone, notify)).await;

        match poll_result {
            Ok(Ok(outcome)) => {
                // Resolved — clean up local Notify (KV entry stays for other waiters).
                self.local_notifies.remove(&ticket);
                Ok(outcome)
            }
            Ok(Err(e)) => {
                self.local_notifies.remove(&ticket);
                Err(e)
            }
            Err(_elapsed) => {
                // Leave KV entry and local Notify alive so caller can retry
                // with the same ticket (RequeueAndEscalate path).
                Err(ApprovalError::TimedOut {
                    millis: timeout.as_millis() as u64,
                })
            }
        }
    }
}