car_server_core/assistant/

governance.rs

//! Durable governance primitives for repository-scoped supervised sessions.
//!
//! This module is deliberately pure. The chat loop and daemon sync adapter own
//! I/O; these types define the values that are persisted and the transitions
//! that are legal, so restart and adversarial tests do not need a live model.

use car_inference::tasks::generate::Message;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::path::{Path, PathBuf};

pub const CHECKPOINT_REGISTRY_KIND: &str = "assistant-checkpoint";
pub const ACTION_REGISTRY_KIND: &str = "assistant-action";

/// Canonical, existing repository root accepted by governed-host execution.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RepositoryScope {
    root: PathBuf,
}

impl RepositoryScope {
    /// Validate an explicitly supplied repository root. The filesystem's
    /// canonical path is the authority, which closes `..` and symlink aliases.
    pub fn explicit(path: Option<&Path>) -> Result<Self, String> {
        let path = path.ok_or("governed host execution requires an explicit --dir")?;
        if !path.is_dir() {
            return Err(format!(
                "repository root '{}' is not a directory",
                path.display()
            ));
        }
        let root = path
            .canonicalize()
            .map_err(|e| format!("cannot resolve repository root '{}': {e}", path.display()))?;
        if root.parent().is_none() {
            return Err("repository root cannot be the filesystem root".to_string());
        }
        if let Some(home) = home_dir().and_then(|p| p.canonicalize().ok()) {
            if root == home {
                return Err("repository root cannot be the user's home directory".to_string());
            }
        }
        // A repository-scoped agent must actually point at a repository. A
        // worktree's `.git` may be either a directory or a gitdir file.
        if !root.join(".git").exists() {
            return Err(format!("'{}' is not a Git repository root", root.display()));
        }
        Ok(Self { root })
    }

    pub fn root(&self) -> &Path {
        &self.root
    }

    /// Resolve an existing path and prove it remains beneath this scope.
    pub fn existing_path(&self, path: &Path) -> Result<PathBuf, String> {
        let candidate = if path.is_absolute() {
            path.to_path_buf()
        } else {
            self.root.join(path)
        };
        let resolved = candidate
            .canonicalize()
            .map_err(|e| format!("cannot resolve '{}': {e}", candidate.display()))?;
        if !resolved.starts_with(&self.root) {
            return Err(format!(
                "path '{}' escapes repository scope",
                path.display()
            ));
        }
        Ok(resolved)
    }

    /// Resolve a prospective write. Its nearest existing ancestor is
    /// canonicalized, preventing a symlinked parent from escaping the root.
    pub fn write_path(&self, path: &Path) -> Result<PathBuf, String> {
        let candidate = if path.is_absolute() {
            path.to_path_buf()
        } else {
            self.root.join(path)
        };
        let mut ancestor = candidate.as_path();
        while !ancestor.exists() {
            ancestor = ancestor
                .parent()
                .ok_or_else(|| format!("path '{}' has no existing ancestor", path.display()))?;
        }
        let resolved_ancestor = ancestor
            .canonicalize()
            .map_err(|e| format!("cannot resolve '{}': {e}", ancestor.display()))?;
        if !resolved_ancestor.starts_with(&self.root) {
            return Err(format!(
                "path '{}' escapes repository scope",
                path.display()
            ));
        }
        let suffix = candidate
            .strip_prefix(ancestor)
            .map_err(|_| format!("cannot scope '{}'", candidate.display()))?;
        Ok(resolved_ancestor.join(suffix))
    }
}

fn home_dir() -> Option<PathBuf> {
    std::env::var_os("HOME")
        .or_else(|| std::env::var_os("USERPROFILE"))
        .map(PathBuf::from)
}

/// Names a host credential capability without containing credential material.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub struct CredentialCapability(pub String);

/// Exact scope shown to the operator and covered by the grant digest.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ActionScope {
    pub tool: String,
    pub parameters: Value,
    pub repository_root: PathBuf,
    pub target: String,
    pub environment: String,
    #[serde(default)]
    pub credential_capabilities: Vec<CredentialCapability>,
}

impl ActionScope {
    pub fn canonicalize(mut self) -> Self {
        self.credential_capabilities.sort();
        self.credential_capabilities.dedup();
        self
    }

    /// Content identity for approval and at-most-once dispatch. Secret values
    /// are absent by construction; only capability names participate.
    pub fn action_id(&self, session_id: &str, call_id: &str) -> String {
        let scope = self.clone().canonicalize();
        let value = serde_json::to_value(&scope).expect("ActionScope serializes");
        let mut h = Sha256::new();
        h.update(b"car-supervised-action-v1\x1f");
        h.update(session_id.as_bytes());
        h.update(b"\x1f");
        h.update(call_id.as_bytes());
        h.update(b"\x1f");
        h.update(car_sync::canonical_json(&value).as_bytes());
        format!("action-{:x}", h.finalize())
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ActionState {
    Proposed,
    Approved,
    Denied,
    Dispatched,
    Completed,
    Failed,
    Indeterminate,
}

impl ActionState {
    pub fn is_terminal(self) -> bool {
        matches!(
            self,
            Self::Denied | Self::Completed | Self::Failed | Self::Indeterminate
        )
    }
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct SupervisedActionRecord {
    pub id: String,
    pub session_id: String,
    pub call_id: String,
    pub scope: ActionScope,
    pub state: ActionState,
    /// Runtime-generated receipt or diagnostic metadata. Never model-authored.
    #[serde(default)]
    pub receipt: Option<Value>,
}

impl SupervisedActionRecord {
    pub fn propose(session_id: &str, call_id: &str, scope: ActionScope) -> Self {
        let scope = scope.canonicalize();
        Self {
            id: scope.action_id(session_id, call_id),
            session_id: session_id.to_string(),
            call_id: call_id.to_string(),
            scope,
            state: ActionState::Proposed,
            receipt: None,
        }
    }

    pub fn transition(&mut self, next: ActionState, receipt: Option<Value>) -> Result<(), String> {
        let valid = matches!(
            (self.state, next),
            (
                ActionState::Proposed,
                ActionState::Approved | ActionState::Denied
            ) | (ActionState::Approved, ActionState::Dispatched)
                | (
                    ActionState::Dispatched,
                    ActionState::Completed | ActionState::Failed | ActionState::Indeterminate
                )
        );
        if !valid {
            return Err(format!(
                "invalid supervised action transition {:?} -> {:?}",
                self.state, next
            ));
        }
        self.state = next;
        self.receipt = receipt;
        Ok(())
    }

    /// Resume never automatically replays an action whose effect may have
    /// crossed the process boundary.
    pub fn resume_directive(&self) -> ResumeDirective {
        match self.state {
            ActionState::Proposed => ResumeDirective::AwaitApproval,
            ActionState::Approved => ResumeDirective::Dispatch,
            ActionState::Dispatched => ResumeDirective::MarkIndeterminate,
            ActionState::Denied
            | ActionState::Completed
            | ActionState::Failed
            | ActionState::Indeterminate => ResumeDirective::DoNotDispatch,
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ResumeDirective {
    AwaitApproval,
    Dispatch,
    MarkIndeterminate,
    DoNotDispatch,
}

/// Durable grant. `action_id` is sufficient to bind all scope fields because
/// it is their canonical digest; the redundant scope makes receipts legible.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ActionGrant {
    pub action_id: String,
    pub scope: ActionScope,
    pub approved: bool,
}

impl ActionGrant {
    pub fn authorizes(&self, action: &SupervisedActionRecord) -> bool {
        self.approved
            && self.action_id == action.id
            && self.scope.clone().canonicalize() == action.scope
            && action.state == ActionState::Proposed
    }
}

#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct CompletionMatrix {
    pub local_verification: Option<String>,
    pub remote_main: Option<String>,
    pub ci_cd: Option<String>,
    pub deployment: Option<String>,
    pub health: Option<String>,
    pub production_browser_proof: Option<String>,
}

#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct AssistantCheckpoint {
    pub id: String,
    pub session_id: String,
    pub revision: u64,
    pub repository_root: PathBuf,
    pub messages: Vec<Message>,
    #[serde(default)]
    pub goal: Option<Value>,
    #[serde(default)]
    pub compaction: Option<Value>,
    #[serde(default)]
    pub completion: CompletionMatrix,
}

/// Project a conservative completion matrix from runtime tool receipts in the
/// exact transcript. A field is populated only for an answered, successful
/// call; shell success additionally requires `exit_code == 0`.
pub fn completion_matrix_from_messages(messages: &[Message]) -> CompletionMatrix {
    let mut calls: std::collections::HashMap<String, (String, Value)> =
        std::collections::HashMap::new();
    let mut matrix = CompletionMatrix::default();
    for message in messages {
        match message {
            Message::Assistant { tool_calls, .. } => {
                for call in tool_calls {
                    if let Some(id) = &call.id {
                        calls.insert(
                            id.clone(),
                            (
                                call.name.clone(),
                                serde_json::to_value(&call.arguments).unwrap_or(Value::Null),
                            ),
                        );
                    }
                }
            }
            Message::ToolResult {
                tool_use_id,
                content,
                ..
            } => {
                let Some((tool, params)) = calls.get(tool_use_id) else {
                    continue;
                };
                let parsed = serde_json::from_str::<Value>(content).ok();
                let failed = content.starts_with("[FAILED]")
                    || content.starts_with("[REJECTED]")
                    || parsed
                        .as_ref()
                        .and_then(|value| value.get("error"))
                        .is_some();
                let shell_ok = tool != "shell"
                    || parsed
                        .as_ref()
                        .and_then(|value| value.get("exit_code"))
                        .and_then(Value::as_i64)
                        == Some(0)
                    || parsed
                        .as_ref()
                        .and_then(|value| value.get("ok"))
                        .and_then(Value::as_bool)
                        == Some(true);
                if failed || !shell_ok {
                    continue;
                }
                let command = params
                    .get("command")
                    .and_then(Value::as_str)
                    .unwrap_or_default()
                    .to_ascii_lowercase();
                let command_tokens: Vec<&str> = command
                    .split_whitespace()
                    .map(|token| {
                        token.trim_matches(|ch: char| {
                            !ch.is_ascii_alphanumeric() && ch != '-' && ch != '/' && ch != '.'
                        })
                    })
                    .filter(|token| !token.is_empty())
                    .collect();
                let evidence = format!("{tool} receipt {tool_use_id}: {content}");
                if tool.starts_with("browser_") {
                    matrix.production_browser_proof = Some(evidence.clone());
                }
                if command.contains("git push") {
                    matrix.remote_main = Some(evidence.clone());
                }
                let is_ci = command_tokens
                    .windows(2)
                    .any(|pair| matches!(pair, ["az", "pipelines"] | ["gh", "run"]))
                    || command_tokens.contains(&"pipeline");
                if is_ci {
                    matrix.ci_cd = Some(evidence.clone());
                }
                let is_deployment = command_tokens
                    .iter()
                    .any(|token| matches!(*token, "deploy" | "deployment"))
                    || command.contains("/deploy.")
                    || command.contains("/deploy/");
                if is_deployment {
                    matrix.deployment = Some(evidence.clone());
                }
                if command.contains("health") || command.contains("ready") {
                    matrix.health = Some(evidence.clone());
                }
                if command.contains("test")
                    || command.contains("cargo check")
                    || command.contains("dotnet build")
                    || command.contains("dotnet run")
                    || command.contains("node --test")
                    || command.contains("npm test")
                    || command.contains("pnpm test")
                    || command.contains("yarn test")
                {
                    matrix.local_verification = Some(evidence);
                }
            }
            _ => {}
        }
    }
    matrix
}

/// Persistence seam used by the supervised loop. Production implements this
/// through daemon sync RPCs; tests can use an in-memory oplog-backed adapter.
#[async_trait::async_trait]
pub trait AssistantDurability: Send + Sync {
    async fn load_checkpoint(
        &self,
        session_id: &str,
    ) -> Result<Option<AssistantCheckpoint>, String>;

    /// Append a new exact checkpoint. Implementations assign a strictly
    /// increasing revision and retain `reason` as compaction/transition audit
    /// metadata; callers never maintain a second conversation store.
    async fn checkpoint(
        &self,
        session_id: &str,
        messages: &[Message],
        reason: &str,
        goal: Option<Value>,
    ) -> Result<(), String>;

    async fn load_action(&self, action_id: &str) -> Result<Option<SupervisedActionRecord>, String>;

    async fn record_action(&self, record: &SupervisedActionRecord) -> Result<(), String>;
}

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

    fn fixture_repo() -> tempfile::TempDir {
        let dir = tempfile::tempdir().unwrap();
        fs::create_dir(dir.path().join(".git")).unwrap();
        dir
    }

    fn scope(root: &Path) -> ActionScope {
        ActionScope {
            tool: "shell".into(),
            parameters: json!({"command": "git push origin HEAD:main"}),
            repository_root: root.to_path_buf(),
            target: "origin/main".into(),
            environment: "disposable".into(),
            credential_capabilities: vec![CredentialCapability("git:origin".into())],
        }
    }

    #[test]
    fn explicit_scope_rejects_missing_root_and_home() {
        assert!(RepositoryScope::explicit(None).is_err());
        assert!(RepositoryScope::explicit(Some(Path::new("/"))).is_err());
        if let Some(home) = home_dir() {
            assert!(RepositoryScope::explicit(Some(&home)).is_err());
        }
    }

    #[cfg(unix)]
    #[test]
    fn scope_rejects_symlink_escape_for_reads_and_writes() {
        use std::os::unix::fs::symlink;
        let repo = fixture_repo();
        let outside = tempfile::tempdir().unwrap();
        fs::write(outside.path().join("secret"), "nope").unwrap();
        symlink(outside.path(), repo.path().join("escape")).unwrap();
        let scope = RepositoryScope::explicit(Some(repo.path())).unwrap();
        assert!(scope.existing_path(Path::new("escape/secret")).is_err());
        assert!(scope.write_path(Path::new("escape/new")).is_err());
    }

    #[test]
    fn grant_is_exact_and_parameter_bound() {
        let repo = fixture_repo();
        let mut action = SupervisedActionRecord::propose("s", "c", scope(repo.path()));
        let grant = ActionGrant {
            action_id: action.id.clone(),
            scope: action.scope.clone(),
            approved: true,
        };
        assert!(grant.authorizes(&action));
        action.scope.target = "other/main".into();
        assert!(!grant.authorizes(&action));
    }

    #[test]
    fn dispatched_resume_is_indeterminate_not_replayable() {
        let repo = fixture_repo();
        let mut action = SupervisedActionRecord::propose("s", "c", scope(repo.path()));
        action.transition(ActionState::Approved, None).unwrap();
        action.transition(ActionState::Dispatched, None).unwrap();
        assert_eq!(
            action.resume_directive(),
            ResumeDirective::MarkIndeterminate
        );
        action
            .transition(
                ActionState::Indeterminate,
                Some(json!({"reason": "process restart"})),
            )
            .unwrap();
        assert_eq!(action.resume_directive(), ResumeDirective::DoNotDispatch);
        assert!(action.transition(ActionState::Completed, None).is_err());
    }

    #[test]
    fn telemetry_query_is_not_mislabeled_as_deployment_or_local_verification() {
        let call_id = "telemetry-1";
        let messages = vec![
            Message::Assistant {
                content: String::new(),
                tool_calls: vec![serde_json::from_value(json!({
                    "id": call_id,
                    "name": "shell",
                    "arguments": {
                        "command": "az monitor app-insights query --app ai-fms --analytics-query \"traces | project customDimensions_DeploymentId\""
                    }
                }))
                .unwrap()],
                thinking: vec![],
                            model_id: None,
                local_last_resort: false,
},
            Message::ToolResult {
                tool_use_id: call_id.into(),
                content: json!({"exit_code": 0, "output": "{\"tables\":[]}"}).to_string(),
                provenance: Default::default(),
            },
        ];

        let matrix = completion_matrix_from_messages(&messages);
        assert!(matrix.deployment.is_none());
        assert!(matrix.ci_cd.is_none());
        assert!(matrix.local_verification.is_none());
    }
}