use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;

use cm_db::repo::{messages, run_events, runs, sessions, steps};
use cm_db::DbError;
use cm_domain::{
    shard_of, AgentId, MessageId, MessageRole, MessageWithSteps, SessionId, SessionKey, Step,
    StepStatus, WorkspaceId,
};
use cm_llm::{ChatMessage, ChatRequest, ChatRole, ContentPart, LlmEvent, LlmProvider, StopReason};
use cm_safety::{approvals, checkpoint, grants, NewApproval, ResumeReady, SafetyError};
use cm_tools::{GateDecision, GatePolicy, TaintSet};
use futures::StreamExt;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use sqlx::PgPool;
use tokio::sync::{broadcast, Mutex};
use uuid::Uuid;

use cm_files::{BlobStore, LocalBlobStore};

use crate::events::{RunEventBody, RunEventEnvelope};
use crate::tools::{ToolContext, ToolRegistry};

/// How long a pending approval stays decidable before it expires.
const APPROVAL_TTL: time::Duration = time::Duration::hours(24);

/// Model used for LLM-as-judge roles (the door governor and the topology
/// comparison scorer). Judges use the strongest model — default
/// `claude-opus-4-8` — while everything else runs on the configured default
/// model (`claude-sonnet-4-6`). Override with `CLAWMATES_JUDGE_MODEL`.
pub fn judge_model() -> String {
    std::env::var("CLAWMATES_JUDGE_MODEL").unwrap_or_else(|_| "claude-opus-4-8".to_string())
}

/// Extra named LLM providers (GLM, Kimi, …) selectable as `"<name>:<model>"`
/// alongside the default provider. `Arc<dyn LlmProvider>` isn't `Debug`, so this
/// newtype carries a manual `Debug` that lists only the provider names.
#[derive(Clone, Default)]
pub struct ProviderRegistry(pub HashMap<String, Arc<dyn LlmProvider>>);

impl std::fmt::Debug for ProviderRegistry {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ProviderRegistry")
            .field("names", &self.0.keys().collect::<Vec<_>>())
            .finish()
    }
}

#[derive(Debug, Clone)]
pub struct RuntimeConfig {
    pub model: String,
    pub max_tokens: u32,
    /// Secret broker socket; broker-executed tools require it.
    pub broker_socket: Option<std::path::PathBuf>,
    /// Slack API base (e2e points it at a local sink).
    pub slack_base_url: String,
    /// Sandbox runtime for shell.exec; None disables environment tools.
    pub sandboxes: Option<std::sync::Arc<crate::SandboxManager>>,
    /// Egress-enabled browser containers for browser.goto.
    pub browser: Option<std::sync::Arc<crate::SandboxManager>>,
    /// Themed interactive terminal containers for the Terminal computer app.
    pub terminals: Option<std::sync::Arc<crate::TerminalManager>>,
    /// Extra named providers (GLM/Kimi/…) for judges and topology nodes.
    pub providers: ProviderRegistry,
}

impl RuntimeConfig {
    pub fn with_browser(mut self, browser: std::sync::Arc<crate::SandboxManager>) -> RuntimeConfig {
        self.browser = Some(browser);
        self
    }

    pub fn with_sandboxes(
        mut self,
        sandboxes: std::sync::Arc<crate::SandboxManager>,
    ) -> RuntimeConfig {
        self.sandboxes = Some(sandboxes);
        self
    }

    /// Test/dev defaults: no broker, real Slack base.
    pub fn basic(model: &str, max_tokens: u32) -> RuntimeConfig {
        RuntimeConfig {
            model: model.into(),
            max_tokens,
            broker_socket: None,
            slack_base_url: "https://slack.com/api".into(),
            sandboxes: None,
            browser: None,
            terminals: None,
            providers: ProviderRegistry::default(),
        }
    }
}

#[derive(Debug, thiserror::Error)]
pub enum RuntimeError {
    #[error(transparent)]
    Db(#[from] DbError),
    #[error("safety error: {0}")]
    Safety(#[from] SafetyError),
    #[error("llm error: {0}")]
    Llm(String),
    #[error("checkpoint corrupt: {0}")]
    Checkpoint(String),
}

/// Handle returned to the caller: the run id plus a live event receiver
/// subscribed before the loop starts (no events can be missed).
pub struct StartedRun {
    pub run_id: Uuid,
    pub events: broadcast::Receiver<RunEventEnvelope>,
}

/// A tool call the model requested that has not executed yet.
#[derive(Debug, Clone, Serialize, Deserialize)]
struct PendingTool {
    id: String,
    name: String,
    input: Value,
}

/// The complete, serializable state of an in-flight run. This is what
/// `agent_runs.checkpoint` stores while a run awaits approval; resume
/// deserializes it and continues exactly where the loop stopped.
#[derive(Debug, Serialize, Deserialize)]
struct LoopState {
    session_id: SessionId,
    workspace_id: WorkspaceId,
    agent_id: AgentId,
    reply_message_id: MessageId,
    request: ChatRequest,
    full_text: String,
    step_seq: i32,
    event_seq: i64,
    assistant_parts: Vec<ContentPart>,
    result_parts: Vec<ContentPart>,
    pending_tools: Vec<PendingTool>,
    /// Untrusted sources whose content has entered this run (§15). Once
    /// tainted, every later gated decision carries these sources.
    #[serde(default)]
    taint: Vec<String>,
    /// Token accounting across every provider leg of this run.
    #[serde(default)]
    input_tokens: u64,
    #[serde(default)]
    output_tokens: u64,
}

enum Outcome {
    Completed,
    Suspended,
}

/// Owns active run channels; one instance lives in the server state.
/// Cheap to clone (shared inner state).
#[derive(Clone)]
pub struct Runtime {
    inner: Arc<RuntimeInner>,
}

struct RuntimeInner {
    pool: PgPool,
    provider: Arc<dyn LlmProvider>,
    tools: Arc<ToolRegistry>,
    config: RuntimeConfig,
    blob: Arc<dyn BlobStore>,
    channels: Mutex<HashMap<Uuid, broadcast::Sender<RunEventEnvelope>>>,
}

impl Runtime {
    /// Default blob storage under the system temp dir — fine for dev and
    /// tests; deployments pass their data directory via `with_blob_store`.
    pub fn new(pool: PgPool, provider: Arc<dyn LlmProvider>, config: RuntimeConfig) -> Runtime {
        let blob = Arc::new(LocalBlobStore::new(
            std::env::temp_dir().join("clawmates-blobs"),
        ));
        Runtime::with_blob_store(pool, provider, config, blob)
    }

    pub fn with_blob_store(
        pool: PgPool,
        provider: Arc<dyn LlmProvider>,
        config: RuntimeConfig,
        blob: Arc<dyn BlobStore>,
    ) -> Runtime {
        Runtime {
            inner: Arc::new(RuntimeInner {
                pool,
                provider,
                tools: Arc::new(ToolRegistry::default()),
                config,
                blob,
                channels: Mutex::new(HashMap::new()),
            }),
        }
    }

    pub fn blob(&self) -> Arc<dyn BlobStore> {
        self.inner.blob.clone()
    }

    pub fn pool(&self) -> &PgPool {
        &self.inner.pool
    }

    /// The shared LLM provider (used by the topology comparison endpoint).
    pub fn provider(&self) -> Arc<dyn LlmProvider> {
        self.inner.provider.clone()
    }

    /// Resolve a model spec to `(provider, model)`. A spec like `"glm:glm-4.6"`
    /// or `"kimi:kimi-k2"` selects a named registry provider; anything else
    /// (e.g. `"claude-opus-4-8"`) uses the default provider. This is how judges
    /// and topology nodes pick a different LLM.
    pub fn resolve_provider(&self, spec: &str) -> (Arc<dyn LlmProvider>, String) {
        if let Some((name, model)) = spec.split_once(':') {
            if let Some(provider) = self.inner.config.providers.0.get(name) {
                return (provider.clone(), model.to_string());
            }
        }
        (self.inner.provider.clone(), spec.to_string())
    }

    /// The configured default model id.
    pub fn model(&self) -> &str {
        &self.inner.config.model
    }

    /// Tear down an agent's sandbox + browser containers (on deletion). Returns
    /// whether any container existed and was destroyed.
    pub async fn reap_sandbox(&self, agent_id: cm_domain::AgentId) -> bool {
        let mut any = false;
        if let Some(sb) = &self.inner.config.sandboxes {
            any |= sb.release_agent(agent_id).await;
        }
        if let Some(br) = &self.inner.config.browser {
            any |= br.release_agent(agent_id).await;
        }
        if let Some(tm) = &self.inner.config.terminals {
            any |= tm.release_agent(agent_id).await;
        }
        any
    }

    /// The themed-terminal manager, if a sandbox engine is configured.
    pub fn terminals(&self) -> Option<std::sync::Arc<crate::TerminalManager>> {
        self.inner.config.terminals.clone()
    }

    /// The configured per-call max output tokens.
    pub fn max_tokens(&self) -> u32 {
        self.inner.config.max_tokens
    }

    /// MCP-shaped tool definition (`{name, description, inputSchema}`) for one
    /// registered tool, or `None` if unknown. Used by the MCP door's
    /// `tools/list`.
    pub fn tool_descriptor_json(&self, name: &str) -> Option<Value> {
        self.inner
            .tools
            .descriptors()
            .into_iter()
            .find(|d| d.name == name)
            .map(|d| {
                json!({
                    "name": d.name,
                    "description": d.description,
                    "inputSchema": d.input_schema,
                })
            })
    }

    /// Whether the named tool is broker-executed (the secret broker consumes the
    /// execution grant and touches the credential — the agent never holds it).
    /// The MCP door must mint an approval+grant for these.
    pub fn tool_broker_executed(&self, name: &str) -> bool {
        self.inner.tools.broker_executed(name)
    }

    /// The §10 approval-card preview for a tool input (used by the door when it
    /// records a gated action).
    pub fn tool_preview(&self, name: &str, input: &Value) -> Value {
        self.inner.tools.preview_of(name, input)
    }

    /// Classify a tool's declared effects into a §15 gated category (assuming
    /// no upstream taint). `None` means the action needs no approval.
    pub fn tool_gate_category(&self, name: &str) -> Option<cm_domain::GatedCategory> {
        let effects = self.inner.tools.effects_of(name);
        let taint = TaintSet::from_strings(&[]);
        match GatePolicy.classify(effects, &taint) {
            GateDecision::RequireApproval(category) => Some(category),
            GateDecision::Allow => None,
        }
    }

    /// Governor agent: ask the **judge model** to judge an action. Returns
    /// `(allow, reason)`. The verdict is the first token (`ALLOW`/`DENY`) of the
    /// model's reply. Best-effort and **fail-open** — if the model is
    /// unreachable it returns `(true, …)` so a governor outage doesn't halt
    /// autonomous agents (the governor is an extra soft check atop deterministic
    /// policy, not the only gate). Judges use the strongest model
    /// ([`judge_model`], default `claude-opus-4-8`); everything else runs on the
    /// configured default model.
    pub async fn judge(&self, system: &str, user: &str) -> (bool, String) {
        let (provider, model) = self.resolve_provider(&judge_model());
        let request = ChatRequest {
            system: system.to_string(),
            model,
            messages: vec![ChatMessage {
                role: ChatRole::User,
                parts: vec![ContentPart::text(user)],
            }],
            tools: vec![],
            max_tokens: 256,
            web_search: false,
        };
        let mut text = String::new();
        match provider.stream(request).await {
            Ok(mut stream) => {
                while let Some(event) = stream.next().await {
                    if let Ok(LlmEvent::TextDelta(t)) = event {
                        text.push_str(&t);
                    }
                }
            }
            Err(e) => return (true, format!("governor unreachable (fail-open): {e}")),
        }
        let allow = !text.to_uppercase().contains("DENY");
        (allow, text.trim().to_string())
    }

    /// One-shot completion: send `system`+`user` to `model`, collect the full
    /// assistant text. Used for non-chat LLM work (e.g. brain enhancement on
    /// `claude-opus-4-8`).
    pub async fn complete(
        &self,
        system: &str,
        user: &str,
        model: &str,
        max_tokens: u32,
        web_search: bool,
    ) -> Result<String, String> {
        let (provider, resolved) = self.resolve_provider(model);
        let request = ChatRequest {
            system: system.to_string(),
            model: resolved,
            messages: vec![ChatMessage {
                role: ChatRole::User,
                parts: vec![ContentPart::text(user)],
            }],
            tools: vec![],
            max_tokens,
            web_search,
        };
        let mut text = String::new();
        let mut stream = provider.stream(request).await.map_err(|e| e.to_string())?;
        while let Some(event) = stream.next().await {
            match event {
                Ok(LlmEvent::TextDelta(t)) => text.push_str(&t),
                Ok(_) => {}
                Err(e) => return Err(e.to_string()),
            }
        }
        Ok(text)
    }

    /// Execute a tool on behalf of the MCP door — capability is already decided
    /// by door policy (the human approver is replaced by an automated policy /
    /// governor). Builds the tool context from runtime config; when an
    /// `approval_id` is supplied and the tool is runtime-executed, the
    /// single-use grant is consumed before the action runs (broker-executed
    /// tools consume it inside the broker).
    pub async fn execute_door_tool(
        &self,
        workspace_id: WorkspaceId,
        agent_id: AgentId,
        name: &str,
        input: Value,
        approval_id: Option<Uuid>,
    ) -> Result<Value, String> {
        let ctx = ToolContext {
            pool: self.inner.pool.clone(),
            workspace_id,
            agent_id,
            blob: self.inner.blob.clone(),
            approval_id,
            broker_socket: self.inner.config.broker_socket.clone(),
            slack_base_url: self.inner.config.slack_base_url.clone(),
            sandboxes: self.inner.config.sandboxes.clone(),
            browser: self.inner.config.browser.clone(),
        };
        if let Some(aid) = approval_id {
            if !self.inner.tools.broker_executed(name) {
                grants::consume(&self.inner.pool, aid)
                    .await
                    .map_err(|e| e.to_string())?;
            }
        }
        self.inner.tools.execute(&ctx, name, input).await
    }

    /// Live-attach to a run that is still streaming.
    pub async fn subscribe(&self, run_id: Uuid) -> Option<broadcast::Receiver<RunEventEnvelope>> {
        self.inner
            .channels
            .lock()
            .await
            .get(&run_id)
            .map(|s| s.subscribe())
    }

    /// Starts a run for a user message and spawns the loop.
    pub async fn send_message(
        &self,
        session_id: SessionId,
        user_text: &str,
    ) -> Result<StartedRun, RuntimeError> {
        let inner = &self.inner;
        let session = sessions::get(&inner.pool, session_id).await?;
        let agent = cm_db::repo::agents::get(&inner.pool, session.agent_id).await?;
        let history = messages::history(&inner.pool, session_id).await?;

        // Brain-augmented system prompt: inject the claw's installed skills +
        // recall relevant memory from its .brain, and record the user turn.
        // Best-effort — falls back to the plain system prompt on any error.
        let skills: Vec<(String, String, String)> =
            cm_db::repo::skills::installed(&inner.pool, agent.id)
                .await
                .unwrap_or_default()
                .into_iter()
                .map(|s| (s.title, s.description, s.body))
                .collect();
        let system_prompt = crate::brain::compose_system(
            &agent.id.to_string(),
            &agent.system_prompt,
            &skills,
            user_text,
            &session_id.to_string(),
        );

        let run_id = runs::create(&inner.pool, session_id).await?;
        let receiver = self.open_channel(run_id).await;

        messages::append(
            &inner.pool,
            session_id,
            MessageRole::User,
            json!({"text": user_text}),
        )
        .await?;
        sessions::touch(&inner.pool, session_id).await?;
        // The reply row exists up front so steps can attach while streaming.
        let reply = messages::append(
            &inner.pool,
            session_id,
            MessageRole::Agent,
            json!({"text": ""}),
        )
        .await?;

        let state = LoopState {
            session_id,
            workspace_id: agent.workspace_id,
            agent_id: agent.id,
            reply_message_id: reply.id,
            request: ChatRequest {
                system: system_prompt,
                messages: chat_messages(&history, user_text),
                tools: inner.tools.descriptors(),
                model: inner.config.model.clone(),
                max_tokens: inner.config.max_tokens,
                web_search: false,
            },
            full_text: String::new(),
            step_seq: 0,
            event_seq: 0,
            assistant_parts: Vec::new(),
            result_parts: Vec::new(),
            pending_tools: Vec::new(),
            taint: Vec::new(),
            input_tokens: 0,
            output_tokens: 0,
        };

        self.spawn_drive(run_id, state, true);
        Ok(StartedRun {
            run_id,
            events: receiver,
        })
    }

    /// Resumes a suspended run after its approval was decided. Exactly one
    /// caller wins the claim; everyone else returns quietly.
    pub async fn resume_run(&self, ready: ResumeReady) -> Result<(), RuntimeError> {
        if !checkpoint::claim_resume(&self.inner.pool, ready.run_id).await? {
            return Ok(());
        }
        let raw = checkpoint::load(&self.inner.pool, ready.run_id).await?;
        let mut state: LoopState =
            serde_json::from_value(raw).map_err(|e| RuntimeError::Checkpoint(e.to_string()))?;
        self.open_channel(ready.run_id).await;

        let runtime = self.clone();
        tokio::spawn(async move {
            let result = runtime.resolve_gated_tool(ready, &mut state).await;
            match result {
                Ok(()) => runtime.spawn_drive_inline(ready.run_id, state).await,
                Err(error) => runtime.fail_run(ready.run_id, &error.to_string()).await,
            }
        });
        Ok(())
    }

    /// The durable resume path: periodically expires overdue approvals and
    /// resumes runs whose approvals were decided (survives crashes between
    /// decision and resumption).
    pub fn spawn_resume_sweeper(&self, interval: Duration) {
        let runtime = self.clone();
        tokio::spawn(async move {
            let mut tick = tokio::time::interval(interval);
            loop {
                tick.tick().await;
                let _ = approvals::sweep_expired(&runtime.inner.pool).await;
                if let Ok(ready) = approvals::decided_unresumed(&runtime.inner.pool).await {
                    for item in ready {
                        let _ = runtime.resume_run(item).await;
                    }
                }
            }
        });
    }

    async fn open_channel(&self, run_id: Uuid) -> broadcast::Receiver<RunEventEnvelope> {
        let (sender, receiver) = broadcast::channel(1024);
        self.inner.channels.lock().await.insert(run_id, sender);
        receiver
    }

    fn spawn_drive(&self, run_id: Uuid, state: LoopState, emit_started: bool) {
        let runtime = self.clone();
        tokio::spawn(async move {
            if emit_started {
                let mut seq = state.event_seq;
                if runtime
                    .emit(run_id, &mut seq, RunEventBody::RunStarted { run_id })
                    .await
                    .is_err()
                {
                    runtime
                        .fail_run(run_id, "could not journal run start")
                        .await;
                    return;
                }
                let mut state = state;
                state.event_seq = seq;
                runtime.spawn_drive_inline(run_id, state).await;
            } else {
                runtime.spawn_drive_inline(run_id, state).await;
            }
        });
    }

    async fn spawn_drive_inline(&self, run_id: Uuid, state: LoopState) {
        match self.drive(run_id, state).await {
            Ok(Outcome::Completed) | Ok(Outcome::Suspended) => {}
            Err(error) => self.fail_run(run_id, &error.to_string()).await,
        }
        self.inner.channels.lock().await.remove(&run_id);
    }

    async fn fail_run(&self, run_id: Uuid, message: &str) {
        let _ = runs::set_state(
            &self.inner.pool,
            run_id,
            cm_domain::RunState::Failed,
            Some(message),
        )
        .await;
        let mut seq = runs::get(&self.inner.pool, run_id)
            .await
            .map(|r| r.last_event_id)
            .unwrap_or(0);
        let _ = self
            .emit(
                run_id,
                &mut seq,
                RunEventBody::Error {
                    message: message.to_owned(),
                },
            )
            .await;
        self.inner.channels.lock().await.remove(&run_id);
    }

    /// Journals an event, then broadcasts it. Persist-before-emit is the
    /// invariant that makes replay equal to what live observers saw.
    async fn emit(
        &self,
        run_id: Uuid,
        seq: &mut i64,
        event: RunEventBody,
    ) -> Result<(), RuntimeError> {
        *seq += 1;
        let payload = serde_json::to_value(&event).expect("event serializes");
        run_events::append(&self.inner.pool, run_id, *seq, event.type_name(), payload).await?;
        runs::set_last_event(&self.inner.pool, run_id, *seq).await?;
        if let Some(sender) = self.inner.channels.lock().await.get(&run_id) {
            let _ = sender.send(RunEventEnvelope { seq: *seq, event });
        }
        Ok(())
    }

    /// Applies the human decision to the gated tool at the head of the
    /// pending queue: approval consumes the single-use grant and executes;
    /// rejection feeds a structured refusal back to the model.
    async fn resolve_gated_tool(
        &self,
        ready: ResumeReady,
        state: &mut LoopState,
    ) -> Result<(), RuntimeError> {
        let Some(tool) = state.pending_tools.first().cloned() else {
            return Err(RuntimeError::Checkpoint(
                "resumed run has no pending tool".into(),
            ));
        };
        state.pending_tools.remove(0);
        let ctx = ToolContext {
            pool: self.inner.pool.clone(),
            workspace_id: state.workspace_id,
            agent_id: state.agent_id,
            blob: self.inner.blob.clone(),
            approval_id: Some(ready.approval_id),
            broker_socket: self.inner.config.broker_socket.clone(),
            slack_base_url: self.inner.config.slack_base_url.clone(),
            sandboxes: self.inner.config.sandboxes.clone(),
            browser: self.inner.config.browser.clone(),
        };

        state.step_seq += 1;
        self.emit_step_started(ready.run_id, state, &tool).await?;

        let (status, output) = if !ready.approved {
            (
                StepStatus::Error,
                json!({"rejected": "The human reviewer rejected this action."}),
            )
        } else if self.inner.tools.broker_executed(&tool.name) {
            // The broker verifies and consumes the grant itself — the
            // strongest path: the runtime never touches the credential.
            match self
                .inner
                .tools
                .execute(&ctx, &tool.name, tool.input.clone())
                .await
            {
                Ok(value) => (StepStatus::Ok, value),
                Err(message) => (StepStatus::Error, json!({"error": message})),
            }
        } else {
            match grants::consume(&self.inner.pool, ready.approval_id).await {
                // The grant is consumed BEFORE the action runs: even a
                // racing resumer cannot execute a gated call twice.
                Ok(()) => match self
                    .inner
                    .tools
                    .execute(&ctx, &tool.name, tool.input.clone())
                    .await
                {
                    Ok(value) => (StepStatus::Ok, value),
                    Err(message) => (StepStatus::Error, json!({"error": message})),
                },
                Err(_) => (
                    StepStatus::Error,
                    json!({"error": "execution grant unavailable (already used or never issued)"}),
                ),
            }
        };

        self.record_step(state, &tool, status, &output).await?;
        self.emit_step_finished(ready.run_id, state, status, &output)
            .await?;
        self.emit_agent_message(ready.run_id, state, &tool, status, &output)
            .await?;
        state.assistant_parts.push(ContentPart::ToolUse {
            id: tool.id.clone(),
            name: tool.name,
            input: tool.input,
        });
        state.result_parts.push(ContentPart::ToolResult {
            tool_use_id: tool.id,
            content: output,
        });
        Ok(())
    }

    /// The run loop. Processes pending tool calls (suspending at the first
    /// gated one), then streams the provider; repeats until the model ends
    /// its turn.
    async fn drive(&self, run_id: Uuid, mut state: LoopState) -> Result<Outcome, RuntimeError> {
        let policy = GatePolicy;
        let ctx = ToolContext {
            pool: self.inner.pool.clone(),
            workspace_id: state.workspace_id,
            agent_id: state.agent_id,
            blob: self.inner.blob.clone(),
            approval_id: None,
            broker_socket: self.inner.config.broker_socket.clone(),
            slack_base_url: self.inner.config.slack_base_url.clone(),
            sandboxes: self.inner.config.sandboxes.clone(),
            browser: self.inner.config.browser.clone(),
        };

        loop {
            while let Some(tool) = state.pending_tools.first().cloned() {
                let effects = self.inner.tools.effects_of(&tool.name);
                let taint = TaintSet::from_strings(&state.taint);
                if let GateDecision::RequireApproval(category) = policy.classify(effects, &taint) {
                    let session = sessions::get(&self.inner.pool, state.session_id).await?;
                    let session_key = SessionKey {
                        agent_id: state.agent_id,
                        shard: shard_of(state.agent_id),
                        session_id: state.session_id,
                        message_id: state.reply_message_id,
                    };
                    let approval = approvals::create(
                        &self.inner.pool,
                        NewApproval {
                            workspace_id: session.workspace_id,
                            run_id,
                            session_key: session_key.to_string(),
                            action_type: tool.name.clone(),
                            category,
                            payload: tool.input.clone(),
                            preview: self.inner.tools.preview_of(&tool.name, &tool.input),
                            requested_by_agent: state.agent_id,
                            taint_sources: taint.as_strings(),
                            expires_at: Some(time::OffsetDateTime::now_utc() + APPROVAL_TTL),
                        },
                    )
                    .await?;

                    let mut seq = state.event_seq;
                    self.emit(
                        run_id,
                        &mut seq,
                        RunEventBody::ApprovalRequired {
                            approval_id: approval.id,
                            category: category.as_str().to_owned(),
                            action_type: approval.action_type.clone(),
                            preview: approval.preview.clone(),
                        },
                    )
                    .await?;
                    self.emit(
                        run_id,
                        &mut seq,
                        RunEventBody::RunSuspended {
                            approval_id: approval.id,
                        },
                    )
                    .await?;
                    state.event_seq = seq;
                    // Checkpoint AFTER journaling so resume continues from
                    // the right sequence number.
                    let snapshot = serde_json::to_value(&state)
                        .map_err(|e| RuntimeError::Checkpoint(e.to_string()))?;
                    checkpoint::suspend(&self.inner.pool, run_id, &snapshot).await?;
                    return Ok(Outcome::Suspended);
                }

                // Ungated: execute now.
                state.pending_tools.remove(0);
                state.step_seq += 1;
                self.emit_step_started(run_id, &mut state, &tool).await?;
                let (status, output) = match self
                    .inner
                    .tools
                    .execute(&ctx, &tool.name, tool.input.clone())
                    .await
                {
                    Ok(value) => (StepStatus::Ok, value),
                    Err(message) => (StepStatus::Error, json!({"error": message})),
                };
                // Taint lands BEFORE the step record: the step that
                // produced untrusted output carries its own taint (§15).
                if status == StepStatus::Ok {
                    if let Some(source) = self.inner.tools.output_taint_of(&tool.name) {
                        let tag = source.as_str().to_owned();
                        if !state.taint.contains(&tag) {
                            state.taint.push(tag);
                        }
                    }
                }
                self.record_step(&state, &tool, status, &output).await?;
                self.emit_step_finished(run_id, &mut state, status, &output)
                    .await?;
                self.emit_agent_message(run_id, &mut state, &tool, status, &output)
                    .await?;
                state.assistant_parts.push(ContentPart::ToolUse {
                    id: tool.id.clone(),
                    name: tool.name.clone(),
                    input: tool.input.clone(),
                });
                state.result_parts.push(ContentPart::ToolResult {
                    tool_use_id: tool.id.clone(),
                    content: output,
                });
            }

            // Feed any finished tool batch back to the model.
            if !state.assistant_parts.is_empty() {
                state.request.messages.push(ChatMessage {
                    role: ChatRole::Assistant,
                    parts: std::mem::take(&mut state.assistant_parts),
                });
                state.request.messages.push(ChatMessage {
                    role: ChatRole::User,
                    parts: std::mem::take(&mut state.result_parts),
                });
            }

            let mut stream = self
                .inner
                .provider
                .stream(state.request.clone())
                .await
                .map_err(|e| RuntimeError::Llm(e.to_string()))?;
            let mut stop = StopReason::EndTurn;
            while let Some(event) = stream.next().await {
                match event.map_err(|e| RuntimeError::Llm(e.to_string()))? {
                    LlmEvent::TextDelta(delta) => {
                        state.full_text.push_str(&delta);
                        let mut seq = state.event_seq;
                        self.emit(run_id, &mut seq, RunEventBody::TextDelta { delta })
                            .await?;
                        state.event_seq = seq;
                    }
                    LlmEvent::ToolUse { id, name, input } => {
                        state.pending_tools.push(PendingTool { id, name, input });
                    }
                    LlmEvent::Usage {
                        input_tokens,
                        output_tokens,
                    } => {
                        state.input_tokens += u64::from(input_tokens);
                        state.output_tokens += u64::from(output_tokens);
                    }
                    LlmEvent::Stop(reason) => stop = reason,
                }
            }

            if state.pending_tools.is_empty() || stop != StopReason::ToolUse {
                break;
            }
        }

        messages::set_content(
            &self.inner.pool,
            state.reply_message_id,
            json!({"text": state.full_text}),
        )
        .await?;
        // Record the assistant's side of the turn in the brain. Only the user's
        // turn was ever written, so recall returned half-conversations: the
        // question without the answer, which is the less useful half.
        // Best-effort, exactly like the user-turn write.
        crate::brain::remember_reply(
            &state.agent_id.to_string(),
            &state.full_text,
            &state.session_id.to_string(),
        );
        // Meter the run (§8.4). Billing failures never fail the run — the
        // usage ledger is the recovery path.
        if let Err(error) = cm_billing::charge(
            &self.inner.pool,
            state.workspace_id,
            state.agent_id,
            Some(run_id),
            state.input_tokens,
            state.output_tokens,
        )
        .await
        {
            eprintln!("billing charge failed for run {run_id}: {error}");
        }
        runs::set_state(
            &self.inner.pool,
            run_id,
            cm_domain::RunState::Completed,
            None,
        )
        .await?;
        let mut seq = state.event_seq;
        self.emit(
            run_id,
            &mut seq,
            RunEventBody::RunCompleted {
                message_id: state.reply_message_id.to_string(),
            },
        )
        .await?;
        Ok(Outcome::Completed)
    }

    async fn emit_step_started(
        &self,
        run_id: Uuid,
        state: &mut LoopState,
        tool: &PendingTool,
    ) -> Result<(), RuntimeError> {
        let mut seq = state.event_seq;
        self.emit(
            run_id,
            &mut seq,
            RunEventBody::StepStarted {
                step_seq: state.step_seq,
                tool: tool.name.clone(),
                input: tool.input.clone(),
            },
        )
        .await?;
        state.event_seq = seq;
        Ok(())
    }

    async fn emit_step_finished(
        &self,
        run_id: Uuid,
        state: &mut LoopState,
        status: StepStatus,
        output: &Value,
    ) -> Result<(), RuntimeError> {
        let mut seq = state.event_seq;
        self.emit(
            run_id,
            &mut seq,
            RunEventBody::StepFinished {
                step_seq: state.step_seq,
                status: status.as_str().to_owned(),
                output: output.clone(),
            },
        )
        .await?;
        state.event_seq = seq;
        Ok(())
    }

    /// After a successful `chat.send`, journal the inter-agent message so the
    /// world feed surfaces agent-to-agent comms live (the A2A observer). No-op
    /// for any other tool or a failed send.
    async fn emit_agent_message(
        &self,
        run_id: Uuid,
        state: &mut LoopState,
        tool: &PendingTool,
        status: StepStatus,
        output: &Value,
    ) -> Result<(), RuntimeError> {
        if tool.name != "chat.send" || status != StepStatus::Ok {
            return Ok(());
        }
        let str_field = |v: &Value, k: &str| {
            v.get(k)
                .and_then(|x| x.as_str())
                .unwrap_or_default()
                .to_owned()
        };
        let mut seq = state.event_seq;
        // Group-room posts fan out to every other active member; 1:1 sends carry
        // a single recipient id.
        if output.get("room").and_then(|v| v.as_bool()) == Some(true) {
            let participant_ids = output
                .get("participants")
                .and_then(|v| v.as_array())
                .map(|a| {
                    a.iter()
                        .filter_map(|x| x.as_str().map(|s| s.to_owned()))
                        .collect()
                })
                .unwrap_or_default();
            self.emit(
                run_id,
                &mut seq,
                RunEventBody::RoomMessage {
                    thread_id: str_field(output, "thread_id"),
                    subject: str_field(output, "subject"),
                    text: str_field(&tool.input, "message"),
                    participant_ids,
                },
            )
            .await?;
            state.event_seq = seq;
            return Ok(());
        }
        let to_agent_id = str_field(output, "to_id");
        if to_agent_id.is_empty() {
            return Ok(());
        }
        self.emit(
            run_id,
            &mut seq,
            RunEventBody::AgentMessage {
                to_agent_id,
                to_name: str_field(output, "to"),
                text: str_field(&tool.input, "message"),
                thread_id: str_field(output, "thread_id"),
            },
        )
        .await?;
        state.event_seq = seq;
        Ok(())
    }

    async fn record_step(
        &self,
        state: &LoopState,
        tool: &PendingTool,
        status: StepStatus,
        output: &Value,
    ) -> Result<(), RuntimeError> {
        steps::append(
            &self.inner.pool,
            &Step {
                id: Uuid::now_v7(),
                message_id: state.reply_message_id,
                seq: state.step_seq,
                kind: "tool_call".into(),
                tool_name: Some(tool.name.clone()),
                input: Some(tool.input.clone()),
                output: Some(output.clone()),
                taint: state.taint.clone(),
                status,
            },
        )
        .await?;
        Ok(())
    }
}

/// Maps persisted history plus the new user message into provider-neutral
/// chat messages. Past step traces are display data; the model sees text.
fn chat_messages(history: &[MessageWithSteps], user_text: &str) -> Vec<ChatMessage> {
    let mut out: Vec<ChatMessage> = history
        .iter()
        .filter_map(|entry| {
            let text = entry.message.content["text"].as_str().unwrap_or_default();
            if text.is_empty() {
                return None;
            }
            let role = match entry.message.role {
                MessageRole::User | MessageRole::System => ChatRole::User,
                MessageRole::Agent => ChatRole::Assistant,
            };
            Some(ChatMessage {
                role,
                parts: vec![ContentPart::text(text)],
            })
        })
        .collect();
    out.push(ChatMessage {
        role: ChatRole::User,
        parts: vec![ContentPart::text(user_text)],
    });
    out
}