klieo-ops 0.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.

use super::trait_::{ApprovalError, ApprovalOutcome, Gate, GateDecision, GateRequest};
use crate::approver_registry::{ApproverId, ApproverRegistry};
use async_trait::async_trait;
use dashmap::DashMap;
use ed25519_dalek::{Signature, Verifier};
use std::collections::HashSet;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Notify;
use ulid::Ulid;

/// 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>;

/// 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.
pub struct FourEyesGate {
    registry: Arc<dyn ApproverRegistry>,
    quorum: u8,
    classifier: DualControlClassifier,
    pending: Arc<DashMap<String, Arc<PendingApproval>>>,
}

struct PendingApproval {
    notify: Notify,
    state: parking_lot::Mutex<ApprovalState>,
}

#[derive(Default)]
struct ApprovalState {
    /// Collected `(approver, signature, signed_payload)` triples kept
    /// until quorum is reached, then verified in one pass.
    submissions: Vec<(ApproverId, Signature, Vec<u8>)>,
    /// Terminal outcome once resolved; `None` while still pending.
    outcome: Option<ApprovalOutcome>,
}

impl FourEyesGate {
    /// Construct a `FourEyesGate`.
    #[must_use]
    pub fn new(
        registry: Arc<dyn ApproverRegistry>,
        quorum: u8,
        classifier: DualControlClassifier,
    ) -> Self {
        Self {
            registry,
            quorum,
            classifier,
            pending: Arc::new(DashMap::new()),
        }
    }

    /// 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 entry = self
            .pending
            .get(ticket)
            .ok_or_else(|| ApprovalError::VerificationFailed("unknown ticket".into()))?
            .clone();

        let mut state = entry.state.lock();
        state
            .submissions
            .push((approver, signature, signed_payload));

        let unique_count = state
            .submissions
            .iter()
            .map(|(a, _, _)| a)
            .collect::<HashSet<_>>()
            .len();

        if unique_count < self.quorum as usize {
            return Ok(());
        }

        // Quorum of unique identities reached — verify all submissions.
        let mut verified_ids: HashSet<ApproverId> = HashSet::new();
        for (approver, sig, payload) in &state.submissions {
            let Some(vk) = self.registry.lookup(approver) else {
                let reason = format!("approver `{approver}` not in registry");
                state.outcome = Some(ApprovalOutcome::Deny {
                    code: "approver.unknown".into(),
                    reason: reason.clone(),
                });
                entry.notify.notify_waiters();
                return Err(ApprovalError::VerificationFailed(reason));
            };
            if vk.verify(payload, sig).is_err() {
                let reason = format!("approver `{approver}` signature math-invalid");
                state.outcome = Some(ApprovalOutcome::Deny {
                    code: "approver.bad_signature".into(),
                    reason: reason.clone(),
                });
                entry.notify.notify_waiters();
                return Err(ApprovalError::VerificationFailed(reason));
            }
            verified_ids.insert(approver.clone());
        }

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

        state.outcome = Some(ApprovalOutcome::Allow);
        entry.notify.notify_waiters();
        Ok(())
    }

    /// Explicitly deny an outstanding ticket (operator action).
    pub fn deny_approval(&self, ticket: &str, reason: impl Into<String>) {
        if let Some(entry) = self.pending.get(ticket) {
            let mut state = entry.state.lock();
            state.outcome = Some(ApprovalOutcome::Deny {
                code: "operator.deny".into(),
                reason: reason.into(),
            });
            entry.notify.notify_waiters();
        }
    }
}

#[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());
        self.pending.insert(
            ticket.clone(),
            Arc::new(PendingApproval {
                notify: Notify::new(),
                state: parking_lot::Mutex::new(ApprovalState::default()),
            }),
        );
        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 entry = match self.pending.get(&ticket) {
            Some(e) => e.clone(),
            None => return Err(ApprovalError::VerificationFailed("unknown ticket".into())),
        };

        let outcome = tokio::time::timeout(timeout, async {
            loop {
                if let Some(out) = entry.state.lock().outcome.clone() {
                    return out;
                }
                entry.notify.notified().await;
            }
        })
        .await
        .map_err(|_| ApprovalError::TimedOut {
            millis: timeout.as_millis() as u64,
        })?;

        self.pending.remove(&ticket);

        match &outcome {
            ApprovalOutcome::Allow => Ok(outcome),
            ApprovalOutcome::Deny { reason, .. } => Err(ApprovalError::Denied(reason.clone())),
        }
    }
}