klieo-ops 0.3.0

//! [`ToolInvoker`] wrapper that runs the registered [`Gate`] stack before
//! every tool invocation and emits [`OpsEvent::GateDecision`] on every
//! decision.

use crate::emit::emit_ops_event;
use crate::gates::{ApprovalError, ApprovalOutcome, Gate, GateDecision, GateRequest};
use crate::ops_event::OpsEvent;
use crate::tenant::current_tenant;
use async_trait::async_trait;
use klieo_core::error::ToolError;
use klieo_core::ids::RunId;
use klieo_core::llm::ToolDef;
use klieo_core::memory::EpisodicMemory;
use klieo_core::tool::{ToolCtx, ToolInvoker};
use std::sync::Arc;
use std::time::Duration;

/// Default wait for human approval before the gate times out the tool call.
/// 10 minutes per spec § 2.3.
const DEFAULT_APPROVAL_TIMEOUT: Duration = Duration::from_secs(600);

/// Wraps an inner [`ToolInvoker`]. Consults every registered [`Gate`] before
/// delegating; first non-`Allow` outcome short-circuits with
/// [`ToolError::Permanent`]. Every decision is emitted into the provided
/// [`EpisodicMemory`] as [`OpsEvent::GateDecision`].
pub struct GatedToolInvoker {
    inner: Arc<dyn ToolInvoker>,
    gates: Vec<Arc<dyn Gate>>,
    episodic: Arc<dyn EpisodicMemory>,
    run_id: RunId,
    approval_timeout: Duration,
}

impl GatedToolInvoker {
    /// Construct a gated wrapper around `inner` with the default approval
    /// timeout (10 minutes).
    #[must_use]
    pub fn new(
        inner: Arc<dyn ToolInvoker>,
        gates: Vec<Arc<dyn Gate>>,
        episodic: Arc<dyn EpisodicMemory>,
        run_id: RunId,
    ) -> Self {
        Self {
            inner,
            gates,
            episodic,
            run_id,
            approval_timeout: DEFAULT_APPROVAL_TIMEOUT,
        }
    }

    /// Override the approval timeout (used by tests and by the builder
    /// `approval_timeout` setter).
    #[must_use]
    pub fn with_approval_timeout(mut self, timeout: Duration) -> Self {
        self.approval_timeout = timeout;
        self
    }

    async fn audit(&self, tool: &str, gate: &str, decision: &str, reason: Option<String>) {
        if let Err(err) = emit_ops_event(
            &*self.episodic,
            self.run_id,
            OpsEvent::GateDecision {
                tenant: current_tenant(),
                tool: tool.into(),
                decision: decision.into(),
                gate: gate.into(),
                policy_ref: None,
                reason,
            },
        )
        .await
        {
            tracing::warn!(
                target: "klieo.ops.audit",
                error = %err,
                "audit emit failed; episode not recorded"
            );
        }
    }
}

#[async_trait]
impl ToolInvoker for GatedToolInvoker {
    async fn invoke(
        &self,
        name: &str,
        args: serde_json::Value,
        ctx: ToolCtx,
    ) -> Result<serde_json::Value, ToolError> {
        let req = GateRequest::new(name, args.clone());

        for gate in &self.gates {
            match gate.evaluate(req.clone()).await {
                GateDecision::Allow => {
                    self.audit(name, gate.name(), "allow", None).await;
                }
                GateDecision::Deny { code, reason } => {
                    self.audit(name, gate.name(), "deny", Some(reason.clone()))
                        .await;
                    return Err(ToolError::Permanent(format!(
                        "gate `{}` denied: code={code} reason={reason}",
                        gate.name()
                    )));
                }
                GateDecision::RequireApproval { ticket, quorum } => {
                    self.audit(
                        name,
                        gate.name(),
                        "require_approval",
                        Some(format!("ticket={ticket} quorum={quorum}")),
                    )
                    .await;
                    match gate
                        .wait_for_approval(ticket.clone(), self.approval_timeout)
                        .await
                    {
                        Ok(ApprovalOutcome::Allow) => {
                            self.audit(
                                name,
                                gate.name(),
                                "approved",
                                Some(format!("ticket={ticket}")),
                            )
                            .await;
                            // Approval granted — continue evaluating remaining gates.
                        }
                        Ok(ApprovalOutcome::Deny { code, reason }) => {
                            self.audit(name, gate.name(), "approval_denied", Some(reason.clone()))
                                .await;
                            return Err(ToolError::Permanent(format!(
                                "gate `{}` approval denied: code={code} reason={reason}",
                                gate.name()
                            )));
                        }
                        Err(ApprovalError::TimedOut { millis }) => {
                            self.audit(
                                name,
                                gate.name(),
                                "approval_timed_out",
                                Some(format!("after {millis}ms")),
                            )
                            .await;
                            return Err(ToolError::Permanent(format!(
                                "gate `{}` approval timed out after {millis}ms",
                                gate.name()
                            )));
                        }
                        Err(ApprovalError::Denied(reason)) => {
                            self.audit(name, gate.name(), "approval_denied", Some(reason.clone()))
                                .await;
                            return Err(ToolError::Permanent(format!(
                                "gate `{}` approval denied: {reason}",
                                gate.name()
                            )));
                        }
                        Err(ApprovalError::Halted) => {
                            self.audit(name, gate.name(), "approval_halted", None).await;
                            return Err(ToolError::Permanent(format!(
                                "gate `{}` approval halted (kill-switch)",
                                gate.name()
                            )));
                        }
                        Err(ApprovalError::NotSupported) => {
                            self.audit(
                                name,
                                gate.name(),
                                "approval_not_supported",
                                Some(
                                    "gate returned RequireApproval but does not implement \
                                     wait_for_approval"
                                        .into(),
                                ),
                            )
                            .await;
                            return Err(ToolError::Permanent(format!(
                                "gate `{}` requires approval but does not support waiting; \
                                 register a FourEyesGate or remove the dual_control classifier",
                                gate.name()
                            )));
                        }
                        Err(ApprovalError::VerificationFailed(msg)) => {
                            self.audit(
                                name,
                                gate.name(),
                                "approval_verification_failed",
                                Some(msg.clone()),
                            )
                            .await;
                            return Err(ToolError::Permanent(format!(
                                "gate `{}` approver verification failed: {msg}",
                                gate.name()
                            )));
                        }
                    }
                }
            }
        }

        self.inner.invoke(name, args, ctx).await
    }

    fn catalogue(&self) -> Vec<ToolDef> {
        self.inner.catalogue()
    }
}