car_engine/

goal.rs

//! Live bridge for the goal loop: project [`GoalInputs`] from a [`Runtime`]'s
//! **ground truth** — its event-log receipts, its shared state, its
//! transactional consistency — so `car-verify`'s pure
//! [`car_verify::goal::evaluate_goal`] decides completion against what actually
//! happened, not a transcript read.
//!
//! See `docs/proposals/goal-loop.md`. This is the deterministic Evaluator half
//! of the goal loop. It is the seam the flagship assistant and the external
//! (Codex / Claude Code) executors plug into: each iteration, the caller runs
//! any command / model-judge checks it owns, then calls
//! [`Runtime::gather_goal_inputs`] to fold in the runtime-owned signals, then
//! runs `evaluate_goal`. The loop driver itself
//! ([`car_verify::goal::run_goal_loop`]) stays injected and crate-free.

use crate::Runtime;
use car_eventlog::tool_receipts::ToolClaim;
use car_eventlog::EventKind;
use car_ir::ActionProposal;
use car_verify::goal::{GoalCondition, GoalInputs};
use car_verify::plan_check::{check_plan, PlanCheckRequest};
use car_verify::transaction::check_transaction;
use std::collections::HashMap;

/// What to project into a [`GoalInputs`] for one evaluation. The
/// runtime-owned signals (`claims`, `plan`, `transaction_proposal`,
/// `state_keys`) are gathered by [`Runtime::gather_goal_inputs`]; the
/// caller-owned checks (`command_exits`, `model_verdicts`) are passed through
/// verbatim, since running a shell check or a model judge belongs to the
/// executor, not the runtime (the same injected split the driver uses).
#[derive(Debug, Clone, Default)]
pub struct GoalGather {
    /// The model's tool claims for this iteration (from its tool_calls / IR),
    /// cross-checked against the event-log receipts. Empty ⇒ nothing claimed,
    /// which is vacuously grounded.
    pub claims: Vec<ToolClaim>,
    /// Scope the receipt check to one proposal's events.
    pub proposal_id: Option<String>,
    /// A STRIPS plan to forward-check (drives `plan_achieved`).
    pub plan: Option<PlanCheckRequest>,
    /// A proposal to check for transactional consistency against current state
    /// (drives `state_consistent` — the state-drift guard).
    pub transaction_proposal: Option<ActionProposal>,
    /// State keys to include in the snapshot for `StatePredicate` conditions;
    /// empty ⇒ the whole state.
    pub state_keys: Vec<String>,
    /// Exit codes of command checks the caller already ran on the substrate.
    pub command_exits: HashMap<String, i32>,
    /// Verdicts of model judges the caller already gathered.
    pub model_verdicts: HashMap<String, bool>,
}

impl Runtime {
    /// Project a [`GoalInputs`] from this runtime's ground truth plus the
    /// caller-supplied checks. Reads receipts from the event log
    /// ([`Runtime::verify_tool_receipts`]), the live state snapshot, and — when
    /// a proposal is supplied — transactional consistency. Never fabricates a
    /// signal: a check whose input wasn't requested is left `None`, so the
    /// pure evaluator fails it closed.
    pub async fn gather_goal_inputs(&self, gather: &GoalGather) -> GoalInputs {
        // Receipts: always computed (empty claims ⇒ grounded=true, vacuously),
        // so a no-claims iteration still satisfies a `ToolReceiptsGrounded`
        // leaf rather than failing it closed for lack of input.
        let receipts_grounded = Some(
            self.verify_tool_receipts(&gather.claims, gather.proposal_id.as_deref())
                .await
                .grounded,
        );

        let plan_achieved = gather.plan.as_ref().map(|p| check_plan(p).valid);

        let full = self.state.snapshot();
        let state = if gather.state_keys.is_empty() {
            full
        } else {
            gather
                .state_keys
                .iter()
                .filter_map(|k| full.get(k).map(|v| (k.clone(), v.clone())))
                .collect()
        };

        let state_consistent = gather.transaction_proposal.as_ref().map(|p| {
            let (values, versions) = self.state.versioned_snapshot();
            check_transaction(p, &versions, Some(&values)).consistent
        });

        GoalInputs {
            receipts_grounded,
            plan_achieved,
            state_consistent,
            state,
            command_exits: gather.command_exits.clone(),
            model_verdicts: gather.model_verdicts.clone(),
        }
    }

    /// Append a durable goal-verifier result to the runtime event log.
    ///
    /// This is the audit counterpart to the live `GoalEvaluated` stream event:
    /// hosts can miss a websocket frame, but the session journal still records
    /// why the verifier continued or accepted the run. `model`/`model_tier`
    /// attribute the verdict to the model that produced the turn, so an
    /// ungrounded completion is queryable as "fails on local, passes on cloud"
    /// — the same provenance `record_turn_completed` stamps on the default path.
    ///
    /// Returns the appended event. The event names no session, so a caller
    /// whose runtime is shared by several conversations keeps it to say which
    /// conversation it belongs to (car#2284).
    pub async fn record_goal_evaluated(
        &self,
        goal: &str,
        condition: &GoalCondition,
        iteration: u32,
        met: bool,
        grounded: bool,
        reason: &str,
        model: &str,
    ) -> car_eventlog::Event {
        let mut data = HashMap::new();
        data.insert("goal".to_string(), serde_json::json!(goal));
        data.insert(
            "condition".to_string(),
            serde_json::to_value(condition).unwrap_or(serde_json::Value::Null),
        );
        data.insert("iteration".to_string(), serde_json::json!(iteration));
        data.insert("met".to_string(), serde_json::json!(met));
        data.insert("grounded".to_string(), serde_json::json!(grounded));
        data.insert("reason".to_string(), serde_json::json!(reason));
        data.insert("model_id".to_string(), serde_json::json!(model));
        data.insert(
            "model_tier".to_string(),
            serde_json::json!(model_tier(model)),
        );
        self.log
            .lock()
            .await
            .append(EventKind::GoalEvaluated, None, None, data)
            .clone()
    }

    /// Append a durable completion-decision record for an assistant/coder turn
    /// loop. The default (goal-less) loop declares success the instant the model
    /// emits no tool calls, with no truncation or outcome check — so a truncated
    /// or turn-capped run is indistinguishable from a real finish at the
    /// terminal. Recording `decision`/`stop_reason`/`was_truncated`/`turns` makes
    /// that decision a first-class, queryable event: the false-success and
    /// never-finished signals live here on the ungrounded default path, which
    /// emits no `GoalEvaluated`.
    ///
    /// Returns the appended event, for the same reason as
    /// [`Self::record_goal_evaluated`].
    pub async fn record_turn_completed(
        &self,
        decision: &str,
        stop_reason: Option<&str>,
        was_truncated: bool,
        turns: u32,
        model: &str,
    ) -> car_eventlog::Event {
        let data = turn_completed_data(decision, stop_reason, was_truncated, turns, model);
        self.log
            .lock()
            .await
            .append(EventKind::TurnCompleted, None, None, data)
            .clone()
    }
}

/// Build the durable `TurnCompleted` event-data map.
///
/// Shared by the assistant loop's [`Runtime::record_turn_completed`] and the
/// coder loop's terminal (`EventSink::record_turn_completed` in
/// `car-server-core`), so the two paths never drift on the keys the harness
/// miners parse (`harness_adapt::diagnose`, `failed_trace_events`): `decision`,
/// `stop_reason`, `was_truncated`, `turns`, and the P0c provenance
/// `model_id` (authoritative) / `model_tier` (derived local|cloud grouping).
///
/// The coder loop stamps one further key on top of these — `models_served`,
/// every model that served a turn in that iteration (car#1333) — which no miner
/// parses and which the assistant path has no answer to, since nothing there
/// consumes authorship. Non-drift is a promise about the keys above, not about
/// the whole record.
pub fn turn_completed_data(
    decision: &str,
    stop_reason: Option<&str>,
    was_truncated: bool,
    turns: u32,
    model: &str,
) -> HashMap<String, serde_json::Value> {
    let mut data = HashMap::new();
    data.insert("decision".to_string(), serde_json::json!(decision));
    data.insert("stop_reason".to_string(), serde_json::json!(stop_reason));
    data.insert(
        "was_truncated".to_string(),
        serde_json::json!(was_truncated),
    );
    data.insert("turns".to_string(), serde_json::json!(turns));
    data.insert("model_id".to_string(), serde_json::json!(model));
    data.insert(
        "model_tier".to_string(),
        serde_json::json!(model_tier(model)),
    );
    data
}

/// Best-effort `local` | `cloud` | `unknown` tier from a model id's provider
/// prefix (the segment before the first `/`).
///
/// `model_id` is the authoritative attribution field; `model_tier` is a
/// convenience grouping so "fails on local vs cloud" — the framing of Matt's
/// field report (works on Codex/Opus, breaks on a local model) — can be
/// queried without re-deriving the prefix. Shared by the `GoalEvaluated` /
/// `TurnCompleted` event emitters and the offline bench corpus (`car-bench`)
/// so the provider allow-lists never drift between call sites.
pub fn model_tier(model: &str) -> &'static str {
    match model.split('/').next().unwrap_or("") {
        "mlx" | "mlx-vlm" | "vllm-mlx" | "local" | "apple" | "qwen" => "local",
        "anthropic" | "openai" | "google" | "gemini" | "parslee" => "cloud",
        _ => "unknown",
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::ToolExecutor;
    use car_eventlog::tool_receipts::{ClaimKind, ToolClaim};
    use car_ir::{Action, ActionProposal, ActionType};
    use car_verify::goal::{evaluate_goal, GoalCondition};
    use serde_json::Value;
    use std::sync::Arc;

    struct EchoExec;
    #[async_trait::async_trait]
    impl ToolExecutor for EchoExec {
        async fn execute(&self, tool: &str, params: &Value) -> Result<Value, String> {
            match tool {
                "echo" => Ok(params.get("message").cloned().unwrap_or(Value::Null)),
                _ => Err(format!("unknown tool: {tool}")),
            }
        }
    }

    fn tool_call(tool: &str, params: HashMap<String, Value>) -> Action {
        let mut a = Action::tool_call(tool);
        a.parameters = params;
        a
    }

    fn state_write(key: &str, value: Value) -> Action {
        let mut a = Action::new(ActionType::StateWrite);
        a.parameters = [
            ("key".to_string(), Value::from(key)),
            ("value".to_string(), value),
        ]
        .into();
        a
    }

    fn proposal(actions: Vec<Action>) -> ActionProposal {
        ActionProposal {
            id: "goal-test".to_string(),
            source: "test".to_string(),
            actions,
            timestamp: chrono::Utc::now(),
            context: HashMap::new(),
        }
    }

    /// End-to-end over REAL ground truth: execute a proposal that calls a tool
    /// and writes state, then gather inputs and evaluate a goal condition —
    /// proving `receipts_grounded` reads the event log and `StatePredicate`
    /// reads live state.
    #[tokio::test]
    async fn gather_reads_real_receipts_and_state() {
        let rt = Runtime::new().with_executor(Arc::new(EchoExec));
        rt.register_tool("echo").await;

        let p = proposal(vec![
            tool_call("echo", [("message".to_string(), Value::from("hi"))].into()),
            state_write("tests_passing", Value::Bool(true)),
        ]);
        let r = rt.execute(&p).await;
        assert!(r.all_succeeded(), "setup proposal must succeed");

        // A truthful claim ("I invoked echo") + the real state write.
        let gather = GoalGather {
            claims: vec![ToolClaim {
                kind: ClaimKind::Invoked,
                tool: "echo".to_string(),
                call_id: None,
                count: None,
                text: None,
            }],
            proposal_id: Some("goal-test".to_string()),
            ..Default::default()
        };
        let inputs = rt.gather_goal_inputs(&gather).await;
        assert_eq!(inputs.receipts_grounded, Some(true));
        assert_eq!(inputs.state.get("tests_passing"), Some(&Value::Bool(true)));

        let cond = GoalCondition::AllOf {
            conditions: vec![
                GoalCondition::ToolReceiptsGrounded,
                GoalCondition::StatePredicate {
                    key: "tests_passing".into(),
                    equals: Value::Bool(true),
                },
            ],
        };
        let verdict = evaluate_goal(&cond, &inputs);
        assert!(verdict.met && verdict.grounded, "{}", verdict.reason);
    }

    /// The `/goal` blindspot, closed: a fabricated tool claim makes the
    /// grounded-receipts condition fail even though the model "said" it ran.
    #[tokio::test]
    async fn fabricated_claim_fails_the_receipts_condition() {
        let rt = Runtime::new().with_executor(Arc::new(EchoExec));
        rt.register_tool("echo").await;
        let r = rt
            .execute(&proposal(vec![tool_call(
                "echo",
                [("message".to_string(), Value::from("hi"))].into(),
            )]))
            .await;
        assert!(r.all_succeeded());

        // The model claims it deployed — but no deploy ever ran.
        let gather = GoalGather {
            claims: vec![ToolClaim {
                kind: ClaimKind::Invoked,
                tool: "deploy".to_string(),
                call_id: None,
                count: None,
                text: Some("I deployed it".to_string()),
            }],
            proposal_id: Some("goal-test".to_string()),
            ..Default::default()
        };
        let inputs = rt.gather_goal_inputs(&gather).await;
        assert_eq!(inputs.receipts_grounded, Some(false));

        let verdict = evaluate_goal(&GoalCondition::ToolReceiptsGrounded, &inputs);
        assert!(
            !verdict.met,
            "a hallucinated tool claim must not satisfy the goal"
        );
        assert!(verdict.reason.contains("hallucinated"));
    }

    #[tokio::test]
    async fn record_goal_evaluated_appends_auditable_event() {
        let rt = Runtime::new();
        let condition = GoalCondition::Command {
            id: "tests".to_string(),
            expect_exit: 0,
        };
        rt.record_goal_evaluated(
            "make tests pass",
            &condition,
            2,
            true,
            true,
            "command `test` exited 0",
            "mlx/qwen3-8b:4bit",
        )
        .await;

        let log = rt.log.lock().await;
        let event = log.events().last().expect("goal event should be recorded");
        assert_eq!(event.kind, car_eventlog::EventKind::GoalEvaluated);
        assert_eq!(
            event.data.get("goal"),
            Some(&serde_json::json!("make tests pass"))
        );
        assert_eq!(
            event.data.get("condition"),
            Some(&serde_json::json!({"kind": "command", "id": "tests", "expect_exit": 0}))
        );
        assert_eq!(event.data.get("iteration"), Some(&serde_json::json!(2)));
        assert_eq!(event.data.get("met"), Some(&serde_json::json!(true)));
        assert_eq!(event.data.get("grounded"), Some(&serde_json::json!(true)));
        assert_eq!(
            event.data.get("reason"),
            Some(&serde_json::json!("command `test` exited 0"))
        );
        // Model provenance stamped on the goal path too (mirrors TurnCompleted),
        // so an ungrounded completion is attributable to local vs cloud.
        assert_eq!(
            event.data.get("model_id"),
            Some(&serde_json::json!("mlx/qwen3-8b:4bit"))
        );
        assert_eq!(
            event.data.get("model_tier"),
            Some(&serde_json::json!("local"))
        );
    }

    #[test]
    fn model_tier_maps_provider_prefixes() {
        // Local inference backends.
        assert_eq!(model_tier("mlx/qwen3-8b:4bit"), "local");
        assert_eq!(model_tier("qwen/qwen3-8b"), "local");
        assert_eq!(model_tier("apple/foundation"), "local");
        // Cloud providers.
        assert_eq!(model_tier("anthropic/claude-opus-4-8"), "cloud");
        assert_eq!(model_tier("openai/gpt-5.5"), "cloud");
        assert_eq!(model_tier("google/gemini-2.5-pro"), "cloud");
        assert_eq!(model_tier("parslee/reasoning"), "cloud");
        // Unrecognized prefix and the empty-model (no generate completed) case
        // both fall through to "unknown" rather than being misattributed.
        assert_eq!(model_tier("acme/whatever"), "unknown");
        assert_eq!(model_tier(""), "unknown");
    }

    #[tokio::test]
    async fn record_turn_completed_appends_auditable_event() {
        let rt = Runtime::new();
        rt.record_turn_completed(
            "empty_tool_calls",
            Some("length"),
            true,
            7,
            "mlx/qwen3-8b:4bit",
        )
        .await;

        let log = rt.log.lock().await;
        let event = log.events().last().expect("turn event should be recorded");
        assert_eq!(event.kind, car_eventlog::EventKind::TurnCompleted);
        assert_eq!(
            event.data.get("decision"),
            Some(&serde_json::json!("empty_tool_calls"))
        );
        assert_eq!(
            event.data.get("stop_reason"),
            Some(&serde_json::json!("length"))
        );
        assert_eq!(
            event.data.get("was_truncated"),
            Some(&serde_json::json!(true))
        );
        assert_eq!(event.data.get("turns"), Some(&serde_json::json!(7)));
        assert_eq!(
            event.data.get("model_id"),
            Some(&serde_json::json!("mlx/qwen3-8b:4bit"))
        );
        // provider prefix "mlx" → local tier (the local-vs-cloud attribution).
        assert_eq!(
            event.data.get("model_tier"),
            Some(&serde_json::json!("local"))
        );
    }
}