procyon 0.3.0

//! The gate a tool passes through before it is allowed to change anything.
//!
//! It lives here, on the registry, rather than in the turn loop in `main.rs`, because the turn
//! loop is not the only thing that runs tools: `agent::subagent` calls `registry.execute` too, so
//! a gate in the loop would have left `spawn_agent`, `run_skill` and `party_mode` as a way around
//! it. Every caller goes through `execute`, so the gate goes there.
//!
//! What may run is not decided here. This module asks and remembers; `crate::risk` decides. The
//! split matters because the decision has three outcomes and only one of them is a question: a
//! `Deny` is refused without a prompt, since putting it to the user would imply the answer could
//! be yes.

use std::collections::HashSet;

use serde_json::Value;
use tokio::sync::mpsc;

use crate::channels::{AgentUpdate, ApprovalDecision, ApprovalRequest, CancelFlag};
use crate::risk::{self, Capability, Decision};

/// The approver this session is using, once one exists.
///
/// A specialist runs in its own registry, built inside `talk_to` where the UI's channels are out of
/// reach — so without this it would either get a fresh approver that can ask nobody (refusing every
/// write a `DeploymentEngineer` exists to make) or no approver at all (which is worse). Sharing the
/// session's means a nested agent's write reaches the same prompt, and a grant the user already gave
/// still counts. There is one session per process, so there is one of these.
static SESSION: std::sync::OnceLock<std::sync::Arc<Approver>> = std::sync::OnceLock::new();

/// Records the approver for nested registries to find. Ignored if one is already installed: the
/// first belongs to the session, and a later caller replacing it would be quietly re-deciding who
/// gets asked.
pub fn install(approver: std::sync::Arc<Approver>) {
    let _ = SESSION.set(approver);
}

/// The session's approver, if one has been installed.
pub fn session() -> Option<std::sync::Arc<Approver>> {
    SESSION.get().cloned()
}

/// Asks the user, and remembers what they said for the rest of the session.
pub struct Approver {
    updates: mpsc::UnboundedSender<AgentUpdate>,
    // A `Mutex` because `execute` takes `&self`: the registry is shared, and a receiver has to be
    // taken mutably. It also serialises the asking, which is what we want — two prompts on screen
    // at once would have no way to say which answer belonged to which.
    decisions: tokio::sync::Mutex<mpsc::UnboundedReceiver<ApprovalDecision>>,
    /// Keyed by `risk::scope`, not by tool name: "always allow writing src/lib.rs" is not the same
    /// grant as "always allow writing".
    allowed: std::sync::Mutex<HashSet<String>>,
    cancel: CancelFlag,
    /// Set when there is no user to ask. See [`Approver::unattended`].
    unattended: Option<Unattended>,
}

/// What to do with a `Confirm` when nobody is there to answer it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Unattended {
    /// Refuse. The default, and the only safe reading of silence.
    Refuse,
    /// Proceed, because the operator said so when they started the run.
    Proceed,
}

impl Approver {
    pub fn new(
        updates: mpsc::UnboundedSender<AgentUpdate>,
        decisions: mpsc::UnboundedReceiver<ApprovalDecision>,
        cancel: CancelFlag,
    ) -> Self {
        Self {
            updates,
            decisions: tokio::sync::Mutex::new(decisions),
            allowed: std::sync::Mutex::new(HashSet::new()),
            cancel,
            unattended: None,
        }
    }

    /// An approver for a run with no interface: `procyon --exec`, CI, a benchmark harness.
    ///
    /// `allow_changes` is stated once, when the run starts, by whoever started it. That is the only
    /// place such a decision can honestly be made — mid-run there is nobody to ask, and a gate that
    /// silently proceeds because the screen is missing is not a gate. Refusing is the default, and
    /// the refusal says how to grant it so a script does not look broken.
    pub fn unattended(allow_changes: bool) -> Self {
        // The channels exist only to satisfy the type: nothing is sent, and nothing would be read.
        let (updates, _updates_rx) = mpsc::unbounded_channel();
        let (_decisions_tx, decisions) = mpsc::unbounded_channel();
        Self {
            updates,
            decisions: tokio::sync::Mutex::new(decisions),
            allowed: std::sync::Mutex::new(HashSet::new()),
            cancel: CancelFlag::default(),
            unattended: Some(if allow_changes {
                Unattended::Proceed
            } else {
                Unattended::Refuse
            }),
        }
    }

    /// `Ok(())` to proceed. `Err` carries what the model is told instead — it is a `tool_result`,
    /// so it has to read as an outcome the model can act on, not as an internal state.
    pub async fn approve(
        &self,
        tool: &str,
        capability: Capability,
        input: &Value,
    ) -> Result<(), String> {
        let assessment = risk::assess(tool, capability, input);

        match assessment.decision {
            Decision::Allow => return Ok(()),
            // Not a question. The user cannot lift this from inside the session — that is the
            // whole property — so a prompt would be theatre.
            Decision::Deny(reason) => return Err(reason),
            Decision::Confirm => {}
        }

        // Answered before anything is put on a channel: an unattended run has no screen, so a
        // prompt would be a turn that waits forever.
        match self.unattended {
            Some(Unattended::Proceed) => return Ok(()),
            Some(Unattended::Refuse) => {
                return Err(format!(
                    "{} needs approval and this run has no interface to ask. Nothing ran. Start \
                     the run with --allow-changes if it is meant to change things.",
                    assessment.detail
                ))
            }
            None => {}
        }

        if self
            .allowed
            .lock()
            .map(|set| set.contains(&assessment.scope))
            .unwrap_or(false)
        {
            return Ok(());
        }
        // Asking about a turn the user has already stopped would put a prompt on screen for work
        // that is being abandoned anyway.
        if self.cancel.is_raised() {
            return Err("The user interrupted the turn, so this tool did not run.".to_string());
        }

        let mut decisions = self.decisions.lock().await;

        if self
            .updates
            .send(AgentUpdate::Approval(ApprovalRequest {
                tool: tool.to_string(),
                detail: assessment.detail,
                scope: assessment.scope.clone(),
            }))
            .is_err()
        {
            // Nothing is listening, so nobody can approve. Refusing is the only safe reading.
            return Err(
                "Could not ask the user for approval, so this tool did not run.".to_string(),
            );
        }

        let decision = tokio::select! {
            answer = decisions.recv() => answer,
            // Esc has to reach a turn that is parked on a question, or the prompt becomes a way to
            // wedge the app: the agent would wait forever for an answer the user is done giving.
            _ = self.cancel.wait() => None,
        };

        match decision {
            Some(ApprovalDecision::Once) => Ok(()),
            Some(ApprovalDecision::Always) => {
                if let Ok(mut set) = self.allowed.lock() {
                    set.insert(assessment.scope);
                }
                Ok(())
            }
            Some(ApprovalDecision::Deny) => Err(format!(
                "The user declined to let {} run. Do not retry it; ask what they want instead.",
                tool
            )),
            // The channel closed, or the turn was cancelled while the question was on screen.
            None => Err("The tool was not approved, so it did not run.".to_string()),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    fn approver() -> (
        Approver,
        mpsc::UnboundedReceiver<AgentUpdate>,
        mpsc::UnboundedSender<ApprovalDecision>,
        CancelFlag,
    ) {
        let (updates, updates_rx) = mpsc::unbounded_channel();
        let (decisions_tx, decisions) = mpsc::unbounded_channel();
        let cancel = CancelFlag::default();
        (
            Approver::new(updates, decisions, cancel.clone()),
            updates_rx,
            decisions_tx,
            cancel,
        )
    }

    #[tokio::test]
    async fn a_read_is_never_put_to_the_user() {
        let (approver, mut updates, _tx, _cancel) = approver();
        assert!(approver
            .approve("read_file", Capability::ReadOnly, &json!({}))
            .await
            .is_ok());
        assert!(
            updates.try_recv().is_err(),
            "a read must not raise a prompt"
        );
    }

    #[tokio::test]
    async fn approving_once_does_not_approve_the_next_call() {
        let (approver, mut updates, tx, _cancel) = approver();
        let call = json!({"path": "a.rs"});

        tx.send(ApprovalDecision::Once).unwrap();
        assert!(approver
            .approve("write_file", Capability::Write, &call)
            .await
            .is_ok());
        assert!(updates.try_recv().is_ok(), "the first call must ask");

        tx.send(ApprovalDecision::Once).unwrap();
        assert!(approver
            .approve("write_file", Capability::Write, &call)
            .await
            .is_ok());
        assert!(
            updates.try_recv().is_ok(),
            "'once' means once; the second call must ask again"
        );
    }

    #[tokio::test]
    async fn always_stops_asking_for_that_scope_only() {
        let (approver, mut updates, tx, _cancel) = approver();
        let lib = json!({"path": "src/lib.rs"});

        tx.send(ApprovalDecision::Always).unwrap();
        assert!(approver
            .approve("write_file", Capability::Write, &lib)
            .await
            .is_ok());
        let _ = updates.try_recv();

        // No decision is sent this time: if it asked, this would hang rather than return.
        assert!(approver
            .approve("write_file", Capability::Write, &lib)
            .await
            .is_ok());
        assert!(
            updates.try_recv().is_err(),
            "'always' must stop the prompt for the scope it was granted for"
        );

        // ...but the grant was for a file, which is what the question named. A different file is a
        // different question.
        tx.send(ApprovalDecision::Deny).unwrap();
        assert!(approver
            .approve(
                "write_file",
                Capability::Write,
                &json!({"path": "src/main.rs"})
            )
            .await
            .is_err());

        // ...and it says nothing at all about a tool that signs.
        tx.send(ApprovalDecision::Deny).unwrap();
        assert!(approver
            .approve(
                "caatinga_deploy",
                Capability::Signing,
                &json!({"network": "testnet"})
            )
            .await
            .is_err());
    }

    #[tokio::test]
    async fn a_denial_tells_the_model_not_to_retry() {
        let (approver, _updates, tx, _cancel) = approver();

        tx.send(ApprovalDecision::Deny).unwrap();
        let err = approver
            .approve("write_file", Capability::Write, &json!({"path": "a.rs"}))
            .await
            .unwrap_err();

        // The model reads this as a tool_result; without the instruction it treats a denial as a
        // transient failure and calls the same tool again.
        assert!(err.contains("Do not retry"), "got {}", err);
    }

    // A refusal the user cannot lift is not a question, so it must not reach the screen as one.
    #[tokio::test]
    async fn a_refusal_is_not_put_to_the_user_as_a_question() {
        let (approver, mut updates, _tx, _cancel) = approver();

        let err = approver
            .approve(
                "stellar_invoke",
                Capability::Signing,
                &json!({"source": "S".repeat(56)}),
            )
            .await
            .unwrap_err();

        assert!(err.contains("identity alias"), "got {}", err);
        assert!(
            updates.try_recv().is_err(),
            "a denial must not raise a prompt"
        );
    }

    // A run with no interface cannot ask, and a gate that proceeds because the screen is missing
    // is not a gate. It also must not park on a question nobody will ever answer.
    #[tokio::test]
    async fn an_unattended_run_refuses_rather_than_waits() {
        let approver = Approver::unattended(false);

        let err = approver
            .approve("write_file", Capability::Write, &json!({"path": "a.rs"}))
            .await
            .unwrap_err();

        assert!(err.contains("a.rs"), "got {}", err);
        // Without this a script looks broken rather than unauthorised.
        assert!(err.contains("--allow-changes"), "got {}", err);
    }

    #[tokio::test]
    async fn an_unattended_run_told_to_proceed_proceeds() {
        let approver = Approver::unattended(true);
        assert!(approver
            .approve("write_file", Capability::Write, &json!({"path": "a.rs"}))
            .await
            .is_ok());
    }

    // The operator's up-front yes covers approvals. It does not reach a refusal, which is the
    // whole distinction the risk engine exists to keep.
    #[tokio::test]
    async fn an_unattended_yes_does_not_override_a_refusal() {
        let approver = Approver::unattended(true);
        let err = approver
            .approve(
                "stellar_invoke",
                Capability::Signing,
                &json!({"source": "S".repeat(56)}),
            )
            .await
            .unwrap_err();
        assert!(err.contains("identity alias"), "got {}", err);
    }

    // A prompt the user cannot escape is a way to wedge the app.
    #[tokio::test]
    async fn cancelling_releases_a_turn_parked_on_a_question() {
        let (approver, _updates, _tx, cancel) = approver();

        let raise = tokio::spawn(async move {
            tokio::time::sleep(std::time::Duration::from_millis(80)).await;
            cancel.raise();
        });

        let result = approver
            .approve("write_file", Capability::Write, &json!({}))
            .await;
        raise.await.unwrap();

        assert!(result.is_err(), "a cancelled question must not approve");
    }

    // If the answer can never arrive, the safe reading is no.
    #[tokio::test]
    async fn a_closed_channel_refuses_rather_than_proceeds() {
        let (approver, _updates, tx, _cancel) = approver();
        drop(tx);
        assert!(approver
            .approve("write_file", Capability::Write, &json!({}))
            .await
            .is_err());
    }
}