car-server-core 0.35.0

//! The `agent.chat` harness — makes the assistant a real conversational agent.
//!
//! [`AssistantService`] holds one runtime + a per-`session_id` conversation
//! thread and turns each `agent.chat` request into a loop run, streaming
//! `agent.chat.event` payloads as they're produced. It is transport-agnostic:
//! the caller supplies an async `emit` sink (the `--serve` path forwards it to
//! `DaemonClient::notify`), so the same service is unit-testable without a live
//! daemon. This is the reusable piece that closes the harness gap — until now
//! only bespoke agents (Milo) implemented the agent side of `agent.chat`.
//!
//! Streaming discipline (per `docs/host-protocol.md`): the loop's synchronous
//! `emit` writes into a bounded channel drained by a dedicated task that awaits
//! `emit`, so a slow downstream never blocks the loop and the ack is never
//! delayed behind token production.

use std::collections::HashMap;
use std::future::Future;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex as StdMutex};
use std::time::Duration;

use car_engine::Runtime;
use car_inference::tasks::generate::{ContentBlock, Message};
use serde_json::{json, Value};
use tokio::sync::{mpsc, oneshot, Mutex as AsyncMutex};

use super::agent_loop::{
    run_assistant_goal_loop, run_assistant_loop_cancellable, ApprovalDecision, ApprovalGate,
    AssistantEvent,
};
use super::AssistantConfig;
use crate::coder::native_loop::TurnGenerator;

/// How long a chat turn waits for a host approval before treating it as declined.
const APPROVAL_TIMEOUT: Duration = Duration::from_secs(300);

/// Deterministic completion check for a goal-driven chat turn.
#[derive(Clone, Debug)]
pub struct ChatGoal {
    pub check: String,
    pub max_iterations: u32,
}

/// A single-action `shell` proposal used to run a chat goal's completion check
/// on the bound substrate, audited through the runtime like any other tool.
fn shell_check_proposal(command: &str) -> car_ir::ActionProposal {
    serde_json::from_value(json!({
        "source": "chat-goal-check",
        "actions": [{
            "id": "goal_check",
            "type": "tool_call",
            "tool": "shell",
            "parameters": { "command": command },
        }],
    }))
    .expect("static shell-check proposal shape")
}

async fn run_shell_check_with_approval(
    runtime: &Runtime,
    cfg: &AssistantConfig,
    approval: Option<&dyn ApprovalGate>,
    command: &str,
) -> i32 {
    if cfg.gated_tools.iter().any(|tool| tool == "shell") {
        let params = json!({ "command": command, "purpose": "goal_check" });
        match approval {
            Some(gate) => match gate.request("shell", &params).await {
                ApprovalDecision::Approved => {}
                ApprovalDecision::Denied(_) => return 1,
            },
            None => return 1,
        }
    }

    let exec = runtime.execute(&shell_check_proposal(command)).await;
    exec.results
        .first()
        .and_then(|r| r.output.as_ref())
        .and_then(|o| o.get("exit_code"))
        .and_then(|v| v.as_i64())
        .unwrap_or(1) as i32
}

/// A conversational assistant over one runtime, multiplexed by `session_id`.
pub struct AssistantService {
    generator: Arc<dyn TurnGenerator>,
    runtime: Arc<Runtime>,
    cfg: AssistantConfig,
    /// System prompt seeded as the first message of every new thread.
    system: String,
    /// Per-session conversation threads (multi-turn continuity).
    threads: AsyncMutex<HashMap<String, Vec<Message>>>,
    /// Per-session cancellation flags, set by [`Self::cancel`].
    cancels: StdMutex<HashMap<String, Arc<AtomicBool>>>,
    /// Pending approvals awaiting a host decision, keyed by approval id. Resolved
    /// by [`Self::resolve_approval`] (driven by the `agent.chat.approve` call).
    approvals: Arc<StdMutex<HashMap<String, oneshot::Sender<bool>>>>,
}

impl AssistantService {
    pub fn new(
        generator: Arc<dyn TurnGenerator>,
        runtime: Arc<Runtime>,
        cfg: AssistantConfig,
        system: String,
    ) -> Self {
        Self {
            generator,
            runtime,
            cfg,
            system,
            threads: AsyncMutex::new(HashMap::new()),
            cancels: StdMutex::new(HashMap::new()),
            approvals: Arc::new(StdMutex::new(HashMap::new())),
        }
    }

    /// Signal the session's in-flight turn to stop before its next model call.
    pub fn cancel(&self, session_id: &str) {
        if let Ok(g) = self.cancels.lock() {
            if let Some(flag) = g.get(session_id) {
                flag.store(true, Ordering::Relaxed);
            }
        }
    }

    /// Resolve a pending approval (from an `agent.chat.approve` reverse-call).
    /// Returns true if an approval by that id was waiting.
    pub fn resolve_approval(&self, approval_id: &str, approved: bool) -> bool {
        let tx = self
            .approvals
            .lock()
            .ok()
            .and_then(|mut g| g.remove(approval_id));
        match tx {
            Some(tx) => tx.send(approved).is_ok(),
            None => false,
        }
    }

    /// Run one chat turn for `session_id`, streaming `agent.chat.event` payloads
    /// (each already stamped with `session_id`) through `emit`. Returns when the
    /// turn reaches a terminal state. `attachments` are image `ContentBlock`s
    /// (`image_base64`/`image_url`) forwarded to a vision model on the first
    /// model call. The caller should have already acked the `agent.chat` request
    /// and spawned this on its own task.
    pub async fn handle_turn<E, Fut>(
        &self,
        session_id: &str,
        prompt: &str,
        attachments: Option<Vec<Value>>,
        emit: E,
    ) where
        E: Fn(Value) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = ()> + Send + 'static,
    {
        // Image attachments → ContentBlocks for the vision path. The daemon
        // already validated the shape; keep only image blocks.
        let images: Vec<ContentBlock> = attachments
            .unwrap_or_default()
            .into_iter()
            .filter_map(|a| serde_json::from_value::<ContentBlock>(a).ok())
            .filter(|c| {
                matches!(
                    c,
                    ContentBlock::ImageBase64 { .. } | ContentBlock::ImageUrl { .. }
                )
            })
            .collect();

        // Fresh cancel flag per turn.
        let cancel = Arc::new(AtomicBool::new(false));
        if let Ok(mut g) = self.cancels.lock() {
            g.insert(session_id.to_string(), cancel.clone());
        }

        // Load (or seed) the thread; append the user turn. Clone out so the loop
        // doesn't hold the threads lock across its awaits.
        let mut messages = {
            let mut g = self.threads.lock().await;
            g.entry(session_id.to_string())
                .or_insert_with(|| {
                    vec![Message::System {
                        content: self.system.clone(),
                    }]
                })
                .clone()
        };
        messages.push(Message::User {
            content: prompt.to_string(),
        });

        // Stream events through a channel drained by a dedicated task, so the
        // loop's synchronous emit never blocks on the async sink.
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
        let drain = tokio::spawn(async move {
            while let Some(v) = rx.recv().await {
                emit(v).await;
            }
        });

        let sid = session_id.to_string();
        // `tx_term` streams the synthesized terminal event after the loop; the
        // loop's emit closure and the gate each hold their own clone. Dropping
        // all three closes the channel so the drain task ends.
        let tx_term = tx.clone();
        let gate = ChatApprovalGate {
            session_id: sid.clone(),
            tx: tx.clone(),
            approvals: self.approvals.clone(),
            counter: Arc::new(AtomicU64::new(0)),
        };
        let outcome = run_assistant_loop_cancellable(
            &*self.generator,
            &self.runtime,
            &self.cfg,
            &mut messages,
            &cancel,
            Some(&gate),
            if images.is_empty() {
                None
            } else {
                Some(images.as_slice())
            },
            {
                let sid = sid.clone();
                move |ev| {
                    if let Some(mut payload) = event_to_wire(ev) {
                        payload["session_id"] = json!(sid);
                        let _ = tx.send(payload);
                    }
                }
            },
        )
        .await;
        drop(gate); // release the gate's channel clone

        // The loop emits `done` on a clean finish and `error` on an inference
        // failure; for a turn cap or a cancel it emits no terminal event, so
        // synthesize one here — a host must always see a terminal.
        match outcome.status {
            "max_turns" => {
                let _ = tx_term
                    .send(json!({ "kind": "done", "text": outcome.summary, "session_id": sid }));
            }
            "cancelled" => {
                let _ = tx_term
                    .send(json!({ "kind": "error", "error": "cancelled", "session_id": sid }));
            }
            _ => {}
        }

        drop(tx_term);
        let _ = drain.await;

        // Persist the thread (unless cancelled mid-turn, where the partial
        // assistant/tool messages would leave a dangling exchange).
        if outcome.status != "cancelled" {
            let mut g = self.threads.lock().await;
            g.insert(session_id.to_string(), messages);
        }
        if let Ok(mut g) = self.cancels.lock() {
            g.remove(session_id);
        }
    }

    /// Run one chat turn as a deterministic goal loop. The user prompt is the
    /// pinned objective; completion is decided by `goal.check` exiting 0 through
    /// the runtime, and every verifier pass streams `goal_evaluated`.
    pub async fn handle_goal_turn<E, Fut>(
        &self,
        session_id: &str,
        prompt: &str,
        _attachments: Option<Vec<Value>>,
        goal: ChatGoal,
        emit: E,
    ) where
        E: Fn(Value) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = ()> + Send + 'static,
    {
        use car_verify::goal::{GoalCondition, GoalGovernor, GoalSpec, GoalStatus};

        let cancel = Arc::new(AtomicBool::new(false));
        if let Ok(mut g) = self.cancels.lock() {
            g.insert(session_id.to_string(), cancel.clone());
        }

        // Goal-loop turns are anchored to the objective every iteration. Seed
        // only the system prompt so prior chat turns cannot repoint the goal.
        let mut messages = vec![Message::System {
            content: format!(
                "{}\n\nYou are working toward a goal. Completion is verified \
                 deterministically by running this shell command:\n  {}\nIt is \
                 done only when that command exits 0. Keep working until it does.",
                self.system, goal.check
            ),
        }];

        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<Value>();
        let drain = tokio::spawn(async move {
            while let Some(v) = rx.recv().await {
                emit(v).await;
            }
        });

        let sid = session_id.to_string();
        let tx_term = tx.clone();
        let gate = ChatApprovalGate {
            session_id: sid.clone(),
            tx: tx.clone(),
            approvals: self.approvals.clone(),
            counter: Arc::new(AtomicU64::new(0)),
        };
        let check_gate = gate.clone();
        let spec = GoalSpec {
            goal: prompt.to_string(),
            condition: GoalCondition::Command {
                id: "goal_check".into(),
                expect_exit: 0,
            },
            governor: GoalGovernor {
                max_turns: Some(goal.max_iterations.max(1)),
                ..Default::default()
            },
        };
        let check = goal.check.clone();
        let result = run_assistant_goal_loop(
            &*self.generator,
            &self.runtime,
            &self.cfg,
            &mut messages,
            &cancel,
            Some(&gate),
            &spec,
            move |_outcome| {
                let cmd = check.clone();
                let check_gate = check_gate.clone();
                async move {
                    let exit = run_shell_check_with_approval(
                        &self.runtime,
                        &self.cfg,
                        Some(&check_gate),
                        &cmd,
                    )
                    .await;
                    let mut g = car_engine::GoalGather::default();
                    g.command_exits.insert("goal_check".into(), exit);
                    g
                }
            },
            {
                let sid = sid.clone();
                move |ev| {
                    // Inner loop `done`/`error` events are iteration-local; the
                    // goal loop sends one terminal event below after the verifier
                    // achieves or halts.
                    if matches!(ev, AssistantEvent::Done { .. } | AssistantEvent::Error(_)) {
                        return;
                    }
                    if let Some(mut payload) = event_to_wire(ev) {
                        payload["session_id"] = json!(sid);
                        let _ = tx.send(payload);
                    }
                }
            },
        )
        .await;
        drop(gate);

        let terminal = match result.run.status {
            GoalStatus::Achieved => {
                json!({ "kind": "done", "text": result.outcome.summary, "session_id": sid })
            }
            GoalStatus::Halted { halt } => json!({
                "kind": "error",
                "error": format!(
                    "goal not reached: {} after {} iteration(s); last check: {}",
                    halt.as_str(),
                    result.run.iterations,
                    result.run.last_reason
                ),
                "session_id": sid,
            }),
        };
        let _ = tx_term.send(terminal);

        drop(tx_term);
        let _ = drain.await;

        if !matches!(result.run.status, GoalStatus::Halted { .. }) {
            let mut g = self.threads.lock().await;
            g.insert(session_id.to_string(), messages);
        }
        if let Ok(mut g) = self.cancels.lock() {
            g.remove(session_id);
        }
    }
}

/// The chat-surface approval gate: emit an `approval_pending` event and park on
/// a oneshot resolved by [`AssistantService::resolve_approval`] (driven by the
/// host's `agent.chat.approve` reverse-call). Times out to "declined".
#[derive(Clone)]
struct ChatApprovalGate {
    session_id: String,
    tx: mpsc::UnboundedSender<Value>,
    approvals: Arc<StdMutex<HashMap<String, oneshot::Sender<bool>>>>,
    counter: Arc<AtomicU64>,
}

#[async_trait::async_trait]
impl ApprovalGate for ChatApprovalGate {
    async fn request(&self, tool: &str, params: &Value) -> ApprovalDecision {
        let n = self.counter.fetch_add(1, Ordering::Relaxed);
        let approval_id = format!("{}-appr-{n}", self.session_id);
        let (otx, orx) = oneshot::channel();
        if let Ok(mut g) = self.approvals.lock() {
            g.insert(approval_id.clone(), otx);
        }
        let _ = self.tx.send(json!({
            "kind": "approval_pending",
            "approval_id": approval_id,
            "tool": tool,
            "params": params,
            "session_id": self.session_id,
        }));
        match tokio::time::timeout(APPROVAL_TIMEOUT, orx).await {
            Ok(Ok(true)) => ApprovalDecision::Approved,
            Ok(Ok(false)) => ApprovalDecision::Denied("declined by user".into()),
            _ => {
                // Timed out or the sender dropped — clean up and treat as denied.
                if let Ok(mut g) = self.approvals.lock() {
                    g.remove(&approval_id);
                }
                ApprovalDecision::Denied("approval timed out".into())
            }
        }
    }
}

/// Translate a loop event into an `agent.chat.event` wire payload. Tool results
/// are internal to the loop and not surfaced as their own kind (the model's
/// subsequent text conveys them).
fn event_to_wire(ev: AssistantEvent) -> Option<Value> {
    match ev {
        AssistantEvent::Text(t) => Some(json!({ "kind": "token", "delta": t })),
        AssistantEvent::ToolCall { name, params } => {
            Some(json!({ "kind": "tool_call", "tool": name, "params": params }))
        }
        AssistantEvent::ToolResult { .. } => None,
        AssistantEvent::Done { text } => Some(json!({ "kind": "done", "text": text })),
        AssistantEvent::Error(e) => Some(json!({ "kind": "error", "error": e })),
        AssistantEvent::GoalEvaluated {
            iteration,
            met,
            grounded,
            reason,
        } => Some(json!({
            "kind": "goal_evaluated",
            "iteration": iteration,
            "met": met,
            "grounded": grounded,
            "reason": reason,
        })),
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::assistant::executor::GeneralExecutor;
    use async_trait::async_trait;
    use car_engine::{LocalSubstrate, Substrate, ToolExecutor};
    use car_inference::{GenerateRequest, InferenceEngine, InferenceResult};
    use std::sync::atomic::AtomicUsize;

    fn turn(text: &str, tool_calls: Value) -> InferenceResult {
        serde_json::from_value(json!({
            "text": text, "tool_calls": tool_calls,
            "trace_id": "t", "model_used": "scripted", "latency_ms": 0,
        }))
        .unwrap()
    }

    struct Script {
        turns: Vec<InferenceResult>,
        cursor: AtomicUsize,
    }
    #[async_trait]
    impl TurnGenerator for Script {
        async fn generate(&self, _r: GenerateRequest) -> Result<InferenceResult, String> {
            let i = self.cursor.fetch_add(1, Ordering::SeqCst);
            self.turns.get(i).cloned().ok_or("exhausted".into())
        }
    }

    async fn runtime(dir: &std::path::Path) -> Arc<Runtime> {
        let substrate: Arc<dyn Substrate> = Arc::new(LocalSubstrate::new());
        let exec: Arc<dyn ToolExecutor> =
            Arc::new(GeneralExecutor::new(substrate.clone(), dir, true));
        let rt = Runtime::new()
            .with_inference(Arc::new(InferenceEngine::new(Default::default())))
            .with_executor(exec)
            .with_substrate(substrate);
        rt.register_agent_basics().await;
        rt.register_tool_entry(
            car_engine::ToolEntry::new(car_ir::builtins::shell()).with_side_effects(true),
        )
        .await;
        Arc::new(rt)
    }

    struct FixedApproval(bool);

    #[async_trait]
    impl ApprovalGate for FixedApproval {
        async fn request(&self, _tool: &str, _params: &Value) -> ApprovalDecision {
            if self.0 {
                ApprovalDecision::Approved
            } else {
                ApprovalDecision::Denied("denied".into())
            }
        }
    }

    fn test_cfg_with_gated_shell() -> AssistantConfig {
        AssistantConfig {
            model: Some("scripted".into()),
            max_turns: 4,
            tools: GeneralExecutor::tool_defs(),
            gated_tools: vec!["shell".into()],
            approval_policy: None,
        }
    }

    #[tokio::test]
    async fn goal_shell_check_does_not_run_without_required_approval() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime(dir.path()).await;
        let cfg = test_cfg_with_gated_shell();
        let target = dir.path().join("should-not-exist");

        let exit = run_shell_check_with_approval(&rt, &cfg, None, "touch should-not-exist").await;

        assert_eq!(exit, 1);
        assert!(
            !target.exists(),
            "gated goal verifier command must not run without approval"
        );
    }

    #[tokio::test]
    async fn goal_shell_check_runs_after_required_approval() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime(dir.path()).await;
        let cfg = test_cfg_with_gated_shell();
        let target = dir.path().join("approved-check");

        let exit = run_shell_check_with_approval(
            &rt,
            &cfg,
            Some(&FixedApproval(true)),
            "touch approved-check",
        )
        .await;

        assert_eq!(exit, 0);
        assert!(target.exists(), "approved verifier command should run");
    }

    #[tokio::test]
    async fn chat_turn_streams_tokens_and_done() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime(dir.path()).await;
        let generator: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![
                turn(
                    "let me compute",
                    json!([{ "id": "c1", "name": "calculate", "arguments": { "expression": "2+2" } }]),
                ),
                turn("It's 4.", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        });
        let cfg = AssistantConfig {
            model: Some("scripted".into()),
            max_turns: 4,
            tools: GeneralExecutor::tool_defs(),
            gated_tools: Vec::new(),
            approval_policy: None,
        };
        let svc = AssistantService::new(generator, rt, cfg, "sys".into());

        let events = Arc::new(StdMutex::new(Vec::<Value>::new()));
        let ev2 = events.clone();
        svc.handle_turn("s1", "what is 2+2?", None, move |v| {
            let ev = ev2.clone();
            async move {
                ev.lock().unwrap().push(v);
            }
        })
        .await;

        let got = events.lock().unwrap().clone();
        // Every event carries the session id.
        assert!(got.iter().all(|e| e["session_id"] == "s1"));
        // A tool_call for calculate was streamed.
        assert!(got
            .iter()
            .any(|e| e["kind"] == "tool_call" && e["tool"] == "calculate"));
        // The last event is the terminal done with the final text.
        let last = got.last().unwrap();
        assert_eq!(last["kind"], "done");
        assert_eq!(last["text"], "It's 4.");

        // Second turn on the same session continues the thread (3 messages
        // seeded: system+user+assistant... at least the thread persisted).
        let thread_len = svc.threads.lock().await.get("s1").map(|m| m.len()).unwrap();
        assert!(thread_len >= 3, "thread should persist across the turn");
    }

    /// The daemon's `try_forward_agent_chat_event` forwards a chat event iff it
    /// is a notification carrying `params.session_id`; the host then dispatches
    /// on `kind`. This asserts every event we emit across a full turn (text,
    /// tool_call, terminal) satisfies that contract — the wire compatibility the
    /// live `agents.chat` → `agent.chat.event` path depends on.
    #[tokio::test]
    async fn every_chat_event_is_forwardable_by_the_daemon() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime(dir.path()).await;
        let generator: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![
                turn(
                    "let me compute",
                    json!([{ "id": "c1", "name": "calculate", "arguments": { "expression": "1+1" } }]),
                ),
                turn("It's 2.", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        });
        let cfg = AssistantConfig {
            model: Some("scripted".into()),
            max_turns: 4,
            tools: GeneralExecutor::tool_defs(),
            gated_tools: Vec::new(),
            approval_policy: None,
        };
        let svc = AssistantService::new(generator, rt, cfg, "sys".into());
        let events = Arc::new(StdMutex::new(Vec::<Value>::new()));
        let ev2 = events.clone();
        svc.handle_turn("sess-42", "1+1?", None, move |v| {
            let ev = ev2.clone();
            async move {
                ev.lock().unwrap().push(v);
            }
        })
        .await;

        const KNOWN_KINDS: [&str; 6] = [
            "token",
            "tool_call",
            "approval_pending",
            "goal_evaluated",
            "done",
            "error",
        ];
        let got = events.lock().unwrap().clone();
        assert!(!got.is_empty());
        for e in &got {
            // Forwarding precondition: session_id present and correct.
            assert_eq!(
                e.get("session_id").and_then(Value::as_str),
                Some("sess-42"),
                "every event must carry its session_id: {e}"
            );
            // Host-dispatchable: a known kind.
            let kind = e.get("kind").and_then(Value::as_str).unwrap_or("");
            assert!(KNOWN_KINDS.contains(&kind), "unknown event kind: {e}");
        }
        // The stream ends in a terminal `done`.
        assert_eq!(got.last().unwrap()["kind"], "done");
    }

    #[test]
    fn goal_evaluated_event_has_host_wire_shape() {
        let wire = event_to_wire(AssistantEvent::GoalEvaluated {
            iteration: 2,
            met: false,
            grounded: true,
            reason: "command goal_check exited 1".into(),
        })
        .expect("goal verifier events should be surfaced to hosts");

        assert_eq!(wire["kind"], "goal_evaluated");
        assert_eq!(wire["iteration"], 2);
        assert_eq!(wire["met"], false);
        assert_eq!(wire["grounded"], true);
        assert_eq!(wire["reason"], "command goal_check exited 1");
    }

    #[tokio::test]
    async fn goal_turn_streams_verifier_events_and_one_terminal() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime(dir.path()).await;
        let generator: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![
                turn("starting", json!([])),
                turn(
                    "creating sentinel",
                    json!([{ "id": "s1", "name": "shell", "arguments": { "command": "touch goal.done" } }]),
                ),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        });
        let cfg = AssistantConfig {
            model: Some("scripted".into()),
            max_turns: 4,
            tools: GeneralExecutor::tool_defs(),
            gated_tools: Vec::new(),
            approval_policy: None,
        };
        let svc = AssistantService::new(generator, rt, cfg, "sys".into());

        let events = Arc::new(StdMutex::new(Vec::<Value>::new()));
        let ev2 = events.clone();
        svc.handle_goal_turn(
            "goal-s1",
            "create goal.done",
            None,
            ChatGoal {
                check: "test -f goal.done".into(),
                max_iterations: 4,
            },
            move |v| {
                let ev = ev2.clone();
                async move {
                    ev.lock().unwrap().push(v);
                }
            },
        )
        .await;

        let got = events.lock().unwrap().clone();
        let verifier: Vec<_> = got
            .iter()
            .filter(|e| e["kind"] == "goal_evaluated")
            .collect();
        assert_eq!(verifier.len(), 2, "one verifier event per goal iteration");
        assert_eq!(verifier[0]["met"], false);
        assert_eq!(verifier[1]["met"], true);
        assert_eq!(verifier[1]["grounded"], true);
        assert_eq!(
            got.iter().filter(|e| e["kind"] == "done").count(),
            1,
            "iteration-local done events must not leak as terminal chat events"
        );
        assert_eq!(got.last().unwrap()["kind"], "done");
        assert_eq!(
            std::fs::read_to_string(dir.path().join("goal.done")).unwrap_or_default(),
            ""
        );
    }

    #[tokio::test]
    async fn chat_gated_write_emits_approval_and_resumes_on_approve() {
        let dir = tempfile::tempdir().unwrap();
        let rt = runtime(dir.path()).await;
        let generator: Arc<dyn TurnGenerator> = Arc::new(Script {
            turns: vec![
                turn(
                    "",
                    json!([{ "id": "w1", "name": "write_file", "arguments": { "path": "z.txt", "content": "zephyr" } }]),
                ),
                turn("done", json!([])),
            ],
            cursor: AtomicUsize::new(0),
        });
        let cfg = AssistantConfig {
            model: Some("scripted".into()),
            max_turns: 4,
            tools: GeneralExecutor::tool_defs(),
            gated_tools: vec!["write_file".into()],
            approval_policy: None,
        };
        let svc = Arc::new(AssistantService::new(generator, rt, cfg, "sys".into()));

        let events = Arc::new(StdMutex::new(Vec::<Value>::new()));
        let ev2 = events.clone();

        // Drive the turn and, concurrently, approve the first pending request.
        let svc_run = svc.clone();
        let turn_task = tokio::spawn(async move {
            svc_run
                .handle_turn("s1", "write z.txt", None, move |v| {
                    let ev = ev2.clone();
                    async move {
                        ev.lock().unwrap().push(v);
                    }
                })
                .await;
        });

        // Poll for the approval_pending event, then approve it.
        let approved = {
            let mut ok = false;
            for _ in 0..200 {
                let id = events
                    .lock()
                    .unwrap()
                    .iter()
                    .find(|e| e["kind"] == "approval_pending")
                    .and_then(|e| e["approval_id"].as_str().map(String::from));
                if let Some(id) = id {
                    assert!(svc.resolve_approval(&id, true));
                    ok = true;
                    break;
                }
                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
            }
            ok
        };
        assert!(
            approved,
            "an approval_pending event should have been emitted"
        );
        turn_task.await.unwrap();

        // The write ran after approval.
        assert_eq!(
            std::fs::read_to_string(dir.path().join("z.txt")).unwrap(),
            "zephyr"
        );
    }
}