use anyhow::{Context, Result};
use chrono::Utc;
use serde_json::json;
use std::path::Path;
use std::time::Duration;
use uuid::Uuid;

// Re-export the shared types under their legacy `codex::bridge::` path so
// existing integration tests and any downstream callers that imported them
// from here keep compiling. The canonical definitions live in `agent::`.
pub use crate::agent::{BridgeEvent, RunPayload};
use crate::cloud::protocol::{ApprovalDecision, ApprovalKind, FailureReason};
use crate::codex::app_server::AppServer;
use crate::codex::jsonrpc::{self, Incoming};
use crate::codex::schema::{
    ApprovalResponseParams, ClientInfo, InitializeParams, NotificationKind, ThreadStartParams,
    TurnInputItem, TurnStartParams,
};

pub struct Bridge {
    pub server: AppServer,
    pub model_default: Option<String>,
    /// Reasoning-effort tier sent on every `turn/start` (`low`/`medium`/
    /// `high`/`xhigh`). `None` omits the field so codex uses the model's
    /// default effort. Sourced from `[runner.codex].effort_default`.
    pub effort_default: Option<String>,
    initialized: bool,
    thread_id: Option<String>,
    /// Notifications that arrived while we were waiting for an RPC response
    /// (e.g. an early `account/reauthRequired` during `initialize`). Drained
    /// by [`Bridge::next_frame`] before reading from the live stream so the
    /// supervisor's translator sees them.
    pending: std::collections::VecDeque<Incoming>,
}

impl Bridge {
    pub async fn spawn(
        binary: &str,
        cwd: &Path,
        model_default: Option<String>,
        effort_default: Option<String>,
    ) -> Result<Self> {
        let server = AppServer::spawn(binary, cwd).await?;
        Ok(Self {
            server,
            model_default,
            effort_default,
            initialized: false,
            thread_id: None,
            pending: std::collections::VecDeque::new(),
        })
    }

    /// Bridge that speaks to an already-constructed AppServer. Used in
    /// integration tests to substitute a fake process.
    pub fn from_server(server: AppServer, model_default: Option<String>) -> Self {
        Self {
            server,
            model_default,
            effort_default: None,
            initialized: false,
            thread_id: None,
            pending: std::collections::VecDeque::new(),
        }
    }

    /// Returns the next frame from Codex, consulting the pending buffer first
    /// so notifications stashed during `await_response` are not lost.
    pub async fn next_frame(&mut self) -> Option<Incoming> {
        if let Some(f) = self.pending.pop_front() {
            return Some(f);
        }
        self.server.inbound.recv().await
    }

    /// Prepare the app-server for a chat turn without sending user input.
    ///
    /// A warm chat bridge keeps the same Codex app-server process alive for
    /// multiple turns. Codex accepts `initialize` once per process, and the
    /// conversation is anchored by the thread created after initialization.
    pub async fn warm(&mut self, cwd: &Path) -> Result<String> {
        self.ensure_initialized().await?;
        if let Some(thread_id) = self.thread_id.as_ref() {
            return Ok(thread_id.clone());
        }
        let thread_id = self.start_thread(cwd).await?;
        self.thread_id = Some(thread_id.clone());
        Ok(thread_id)
    }

    pub async fn run(&mut self, payload: &RunPayload, cwd: &Path) -> Result<BridgeCursor> {
        let thread_id = self.warm(cwd).await?;
        self.start_turn(&thread_id, payload).await?;
        Ok(BridgeCursor {
            run_id: payload.run_id,
            thread_id,
            model: payload.model.clone().or_else(|| self.model_default.clone()),
            seq: 0,
            pending_command: None,
            pending_command_approval_id: None,
            unpaired_waiting_on_approval: false,
        })
    }

    async fn ensure_initialized(&mut self) -> Result<()> {
        if self.initialized {
            return Ok(());
        }
        let id = self.server.alloc_id();
        let line = jsonrpc::request(
            id,
            "initialize",
            &InitializeParams {
                client_info: ClientInfo {
                    name: "pidash".to_string(),
                    version: crate::RUNNER_VERSION.to_string(),
                },
            },
        )?;
        self.server.send_raw(&line).await?;
        // Wait for init response.
        let _ = self.await_response(id, Duration::from_secs(15)).await?;
        let line = jsonrpc::notification("initialized", &serde_json::Value::Null)?;
        self.server.send_raw(&line).await?;
        self.initialized = true;
        Ok(())
    }

    async fn start_thread(&mut self, cwd: &Path) -> Result<String> {
        let id = self.server.alloc_id();
        let line = jsonrpc::request(
            id,
            "thread/start",
            &ThreadStartParams {
                cwd: cwd.to_string_lossy().to_string(),
                model: self.model_default.clone(),
                // Match Claude Code's MVP `bypassPermissions` posture.
                sandbox: "danger-full-access".into(),
                approval_policy: "never".into(),
            },
        )?;
        self.server.send_raw(&line).await?;
        let resp = self.await_response(id, Duration::from_secs(30)).await?;
        resp.get("threadId")
            .or_else(|| resp.get("thread").and_then(|t| t.get("id")))
            .or_else(|| resp.get("thread_id"))
            .and_then(|v| v.as_str())
            .map(|s| s.to_string())
            .context("thread/start missing threadId")
    }

    async fn start_turn(&mut self, thread_id: &str, payload: &RunPayload) -> Result<()> {
        let id = self.server.alloc_id();
        let line = jsonrpc::request(
            id,
            "turn/start",
            &TurnStartParams {
                thread_id: thread_id.to_string(),
                input: vec![TurnInputItem {
                    item_type: "text".into(),
                    text: payload.prompt.clone(),
                }],
                model: payload.model.clone().or_else(|| self.model_default.clone()),
                effort: self.effort_default.clone(),
            },
        )?;
        self.server.send_raw(&line).await
    }

    pub async fn send_approval(
        &mut self,
        approval_id: &str,
        decision: ApprovalDecision,
    ) -> Result<()> {
        let line = jsonrpc::notification(
            "approval/response",
            &ApprovalResponseParams {
                approval_id: approval_id.to_string(),
                decision: match decision {
                    ApprovalDecision::Accept => "accept".into(),
                    ApprovalDecision::Decline => "decline".into(),
                    ApprovalDecision::AcceptForSession => "accept_for_session".into(),
                },
            },
        )?;
        self.server.send_raw(&line).await
    }

    pub async fn interrupt(&mut self) -> Result<()> {
        let id = self.server.alloc_id();
        let line = jsonrpc::request(id, "turn/interrupt", &json!({}))?;
        self.server.send_raw(&line).await
    }

    async fn await_response(&mut self, id: u64, timeout: Duration) -> Result<serde_json::Value> {
        let deadline = tokio::time::Instant::now() + timeout;
        loop {
            let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
            let frame = tokio::time::timeout(remaining, self.server.inbound.recv())
                .await
                .context("codex response timed out")?;
            match frame {
                Some(Incoming::Response {
                    id: got,
                    result,
                    error,
                    ..
                }) if got == id => {
                    if let Some(err) = error {
                        anyhow::bail!("codex rpc error {}: {}", err.code, err.message);
                    }
                    return Ok(result.unwrap_or(serde_json::Value::Null));
                }
                Some(other @ Incoming::Notification { .. }) => {
                    // Buffer notifications received while awaiting an RPC
                    // response so the translator sees them later.
                    self.pending.push_back(other);
                }
                Some(_) => continue,
                None => anyhow::bail!("codex stdout closed while awaiting response"),
            }
        }
    }
}

pub struct BridgeCursor {
    pub run_id: Uuid,
    pub thread_id: String,
    pub model: Option<String>,
    pub seq: u64,
    /// Most recent in-progress `commandExecution` item observed on
    /// `item/started`. Captured so we can synthesize an `ApprovalRequest`
    /// when codex sets `thread/status/changed → activeFlags:
    /// ["waitingOnApproval"]` (codex 0.125.0 stopped firing the legacy
    /// `item/commandExecution/requestApproval` notification, leaving the
    /// in-progress item + the flag as the only signal we get).
    pending_command: Option<PendingCommand>,
    /// Approval id last opened for `pending_command`. Stored so a stray
    /// re-fire of the `waitingOnApproval` flag (e.g. status notifications
    /// sent more than once for the same item) doesn't open duplicate
    /// approval rows.
    pending_command_approval_id: Option<String>,
    /// A `waitingOnApproval` status frame that arrived before its command.
    ///
    /// Codex can emit `waitingOnApproval` before the matching
    /// `item/started` command frame. Keep this unpaired signal so the command
    /// frame can still synthesize the approval request instead of deadlocking.
    unpaired_waiting_on_approval: bool,
}

#[derive(Debug, Clone)]
struct PendingCommand {
    item_id: String,
    command: String,
    cwd: Option<String>,
    raw: serde_json::Value,
}

impl BridgeCursor {
    fn synthesize_pending_command_approval(&mut self) -> Option<BridgeEvent> {
        let pending = self.pending_command.clone()?;
        if self.pending_command_approval_id.is_some() {
            return None;
        }

        let approval_id = Uuid::new_v4().to_string();
        self.pending_command_approval_id = Some(approval_id.clone());
        let reason = Some(format!(
            "codex requesting approval to run: {}",
            pending.command
        ));
        let payload = serde_json::json!({
            "command": pending.command,
            "cwd": pending.cwd,
            "item_id": pending.item_id,
            "synthesized": true,
            "raw": pending.raw,
        });
        Some(BridgeEvent::ApprovalRequest {
            run_id: self.run_id,
            approval_id,
            kind: ApprovalKind::CommandExecution,
            payload,
            reason,
        })
    }

    /// Translate one inbound Codex frame into daemon-level events.
    pub fn translate(&mut self, frame: Incoming) -> Vec<BridgeEvent> {
        self.seq = self.seq.saturating_add(1);
        match frame {
            Incoming::Notification { method, params, .. } => {
                let kind = NotificationKind::from_method(&method);
                if kind.is_approval_request() {
                    let approval_id = params
                        .get("approval_id")
                        .or_else(|| params.get("approvalId"))
                        .and_then(|v| v.as_str())
                        .map(|s| s.to_string())
                        .unwrap_or_else(|| Uuid::new_v4().to_string());
                    let reason = params
                        .get("reason")
                        .and_then(|v| v.as_str())
                        .map(|s| s.to_string());
                    let ak = match kind {
                        NotificationKind::CommandExecutionRequestApproval => {
                            ApprovalKind::CommandExecution
                        }
                        NotificationKind::FileChangeRequestApproval => ApprovalKind::FileChange,
                        _ => ApprovalKind::Other,
                    };
                    vec![BridgeEvent::ApprovalRequest {
                        run_id: self.run_id,
                        approval_id,
                        kind: ak,
                        payload: params,
                        reason,
                    }]
                } else if matches!(kind, NotificationKind::AccountReauthRequired) {
                    vec![BridgeEvent::AwaitingReauth {
                        run_id: self.run_id,
                        detail: params
                            .get("detail")
                            .and_then(|v| v.as_str())
                            .map(|s| s.to_string()),
                    }]
                } else if matches!(kind, NotificationKind::TurnCompleted) {
                    // Treat success only when codex explicitly says so. Any
                    // other value (including a missing field) means the turn
                    // ended without a clean signal — the prior default of
                    // "success" silently masked codex errors that landed
                    // a `turn/completed` with no conclusion right after a
                    // systemError, so the runner reported "completed" on
                    // 400s from OpenAI.
                    let conclusion = params.get("conclusion").and_then(|v| v.as_str());
                    let turn_status = params
                        .get("turn")
                        .and_then(|v| v.get("status"))
                        .and_then(|v| v.as_str());
                    let turn_error = params
                        .get("turn")
                        .and_then(|v| v.get("error"))
                        .filter(|v| !v.is_null());
                    if conclusion == Some("success")
                        || (conclusion.is_none()
                            && turn_error.is_none()
                            && matches!(turn_status, Some("completed" | "succeeded")))
                    {
                        vec![BridgeEvent::Completed {
                            run_id: self.run_id,
                            done_payload: params.get("done").cloned().unwrap_or_else(|| {
                                serde_json::json!({
                                    "conclusion": conclusion.unwrap_or("success"),
                                    "turn": params.get("turn"),
                                    "ended_at": Utc::now().to_rfc3339(),
                                })
                            }),
                        }]
                    } else {
                        vec![BridgeEvent::Failed {
                            run_id: self.run_id,
                            reason: FailureReason::Internal,
                            detail: params
                                .get("error")
                                .and_then(|v| v.as_str())
                                .map(|s| s.to_string())
                                .or_else(|| {
                                    conclusion.map(|c| format!("turn ended with conclusion={c:?}"))
                                })
                                .or_else(|| {
                                    turn_error.map(|e| format!("turn completed with error: {e}"))
                                })
                                .or_else(|| Some("turn/completed without conclusion".to_string())),
                        }]
                    }
                } else if method == "error" {
                    // Codex emits this for transport / API failures (e.g. a
                    // 400 from OpenAI saying the model isn't allowed). When
                    // `willRetry` is false it's terminal, so fail the run
                    // instead of silently logging it as a Raw event and
                    // waiting for a `turn/completed` that may or may not
                    // arrive.
                    let will_retry = params
                        .get("willRetry")
                        .and_then(|v| v.as_bool())
                        .unwrap_or(false);
                    if will_retry {
                        vec![BridgeEvent::Raw {
                            run_id: self.run_id,
                            method,
                            params,
                        }]
                    } else {
                        let detail = params
                            .get("error")
                            .and_then(|e| e.get("message"))
                            .and_then(|v| v.as_str())
                            .map(|s| s.to_string());
                        vec![BridgeEvent::Failed {
                            run_id: self.run_id,
                            reason: FailureReason::Internal,
                            detail,
                        }]
                    }
                } else if method == "thread/status/changed" {
                    // Codex 0.125.0+ signals "I'm parked waiting for the
                    // user to approve a command" by setting this flag,
                    // *without* firing the legacy
                    // `item/commandExecution/requestApproval` notification.
                    // If we don't translate it into an ApprovalRequest, the
                    // run deadlocks: codex blocks on stdin reading the
                    // approval response, the runner blocks on stdout
                    // reading frames that will never arrive.
                    let waiting = params
                        .get("status")
                        .and_then(|s| s.get("activeFlags"))
                        .and_then(|f| f.as_array())
                        .map(|arr| arr.iter().any(|v| v.as_str() == Some("waitingOnApproval")))
                        .unwrap_or(false);
                    if !waiting {
                        self.unpaired_waiting_on_approval = false;
                    } else if self.pending_command.is_some() {
                        // We've already emitted an ApprovalRequest for this
                        // pending command. Re-emitting here would re-send a
                        // ClientMsg::ApprovalRequest to the cloud and append
                        // a fresh "pending" history row — even though the
                        // ApprovalRouter would self-heal via its HashMap.
                        // Skip the duplicate; the previous approval is still
                        // outstanding (or already decided).
                        if self.pending_command_approval_id.is_some() {
                            return vec![BridgeEvent::Raw {
                                run_id: self.run_id,
                                method,
                                params,
                            }];
                        }
                        if let Some(ev) = self.synthesize_pending_command_approval() {
                            return vec![ev];
                        }
                    } else {
                        self.unpaired_waiting_on_approval = true;
                    }
                    vec![BridgeEvent::Raw {
                        run_id: self.run_id,
                        method,
                        params,
                    }]
                } else if method == "item/started" {
                    // Cache the most recent in-progress commandExecution so
                    // a later `waitingOnApproval` flag can refer to it.
                    let mut synthesize_now = false;
                    if let Some(item) = params.get("item")
                        && item.get("type").and_then(|v| v.as_str()) == Some("commandExecution")
                        && item.get("status").and_then(|v| v.as_str()) == Some("inProgress")
                    {
                        let item_id = item
                            .get("id")
                            .and_then(|v| v.as_str())
                            .unwrap_or_default()
                            .to_string();
                        let command = item
                            .get("command")
                            .and_then(|v| v.as_str())
                            .unwrap_or_default()
                            .to_string();
                        let cwd = item
                            .get("cwd")
                            .and_then(|v| v.as_str())
                            .map(|s| s.to_string());
                        self.pending_command = Some(PendingCommand {
                            item_id,
                            command,
                            cwd,
                            raw: item.clone(),
                        });
                        // New item-started → previous waitingOnApproval (if any)
                        // is no longer relevant; allow a fresh approval id next
                        // time codex reports waiting.
                        self.pending_command_approval_id = None;
                        synthesize_now = self.unpaired_waiting_on_approval;
                        self.unpaired_waiting_on_approval = false;
                    }
                    let mut events = vec![BridgeEvent::Raw {
                        run_id: self.run_id,
                        method,
                        params,
                    }];
                    if synthesize_now && let Some(ev) = self.synthesize_pending_command_approval() {
                        events.push(ev);
                    }
                    events
                } else if method == "item/completed" {
                    // The cached command finished; drop the tracking so a
                    // late waitingOnApproval doesn't re-open it.
                    if let Some(item) = params.get("item")
                        && item.get("type").and_then(|v| v.as_str()) == Some("commandExecution")
                    {
                        let same = self
                            .pending_command
                            .as_ref()
                            .and_then(|pc| {
                                item.get("id")
                                    .and_then(|v| v.as_str())
                                    .map(|id| id == pc.item_id)
                            })
                            .unwrap_or(false);
                        if same {
                            self.pending_command = None;
                            self.pending_command_approval_id = None;
                            self.unpaired_waiting_on_approval = false;
                        }
                    }
                    vec![BridgeEvent::Raw {
                        run_id: self.run_id,
                        method,
                        params,
                    }]
                } else {
                    vec![BridgeEvent::Raw {
                        run_id: self.run_id,
                        method,
                        params,
                    }]
                }
            }
            Incoming::Response { .. } => Vec::new(),
        }
    }
}

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

    fn cursor() -> BridgeCursor {
        BridgeCursor {
            run_id: Uuid::new_v4(),
            thread_id: "th_test".into(),
            model: None,
            seq: 0,
            pending_command: None,
            pending_command_approval_id: None,
            unpaired_waiting_on_approval: false,
        }
    }

    fn notif(method: &str, params: serde_json::Value) -> Incoming {
        Incoming::Notification {
            jsonrpc: Some("2.0".into()),
            method: method.into(),
            params,
        }
    }

    #[test]
    fn turn_completed_without_conclusion_fails_run() {
        // Codex sometimes emits a bare `turn/completed` after a systemError.
        // The previous default of "success" silently masked those failures.
        let mut c = cursor();
        let evs = c.translate(notif("turn/completed", json!({})));
        assert_eq!(evs.len(), 1);
        match &evs[0] {
            BridgeEvent::Failed { detail, .. } => {
                assert_eq!(detail.as_deref(), Some("turn/completed without conclusion"));
            }
            other => panic!("expected Failed, got {other:?}"),
        }
    }

    #[test]
    fn turn_completed_with_explicit_success_completes_run() {
        let mut c = cursor();
        let evs = c.translate(notif("turn/completed", json!({"conclusion": "success"})));
        assert!(matches!(evs.as_slice(), [BridgeEvent::Completed { .. }]));
    }

    #[test]
    fn turn_completed_with_completed_turn_status_completes_run() {
        let mut c = cursor();
        let evs = c.translate(notif(
            "turn/completed",
            json!({"turn": {"status": "completed", "error": null}}),
        ));
        assert!(matches!(evs.as_slice(), [BridgeEvent::Completed { .. }]));
    }

    #[test]
    fn turn_completed_with_turn_error_fails_run() {
        let mut c = cursor();
        let evs = c.translate(notif(
            "turn/completed",
            json!({"turn": {"status": "completed", "error": {"message": "boom"}}}),
        ));
        match evs.as_slice() {
            [BridgeEvent::Failed { detail, .. }] => {
                assert!(detail.as_deref().unwrap_or("").contains("boom"));
            }
            other => panic!("expected Failed, got {other:?}"),
        }
    }

    #[test]
    fn turn_completed_with_non_success_conclusion_fails_run() {
        let mut c = cursor();
        let evs = c.translate(notif("turn/completed", json!({"conclusion": "aborted"})));
        match evs.as_slice() {
            [BridgeEvent::Failed { detail, .. }] => {
                assert_eq!(
                    detail.as_deref(),
                    Some("turn ended with conclusion=\"aborted\"")
                );
            }
            other => panic!("expected Failed, got {other:?}"),
        }
    }

    #[test]
    fn error_notification_with_will_retry_false_fails_run() {
        let mut c = cursor();
        let evs = c.translate(notif(
            "error",
            json!({
                "willRetry": false,
                "error": {"message": "model not allowed"}
            }),
        ));
        match evs.as_slice() {
            [BridgeEvent::Failed { detail, .. }] => {
                assert_eq!(detail.as_deref(), Some("model not allowed"));
            }
            other => panic!("expected Failed, got {other:?}"),
        }
    }

    #[test]
    fn error_notification_with_will_retry_true_passes_through_as_raw() {
        let mut c = cursor();
        let evs = c.translate(notif(
            "error",
            json!({
                "willRetry": true,
                "error": {"message": "transient"}
            }),
        ));
        assert!(matches!(evs.as_slice(), [BridgeEvent::Raw { .. }]));
    }

    #[test]
    fn waiting_on_approval_synthesizes_request_from_pending_command() {
        let mut c = cursor();
        // First, codex starts a commandExecution item.
        let _ = c.translate(notif(
            "item/started",
            json!({
                "item": {
                    "id": "it_1",
                    "type": "commandExecution",
                    "status": "inProgress",
                    "command": "rm -rf /tmp/x",
                    "cwd": "/work",
                }
            }),
        ));
        // Then it parks on `waitingOnApproval` without firing the legacy
        // requestApproval notification.
        let evs = c.translate(notif(
            "thread/status/changed",
            json!({"status": {"activeFlags": ["waitingOnApproval"]}}),
        ));
        match evs.as_slice() {
            [
                BridgeEvent::ApprovalRequest {
                    kind,
                    payload,
                    reason,
                    ..
                },
            ] => {
                assert!(matches!(kind, ApprovalKind::CommandExecution));
                assert_eq!(
                    payload.get("command").and_then(|v| v.as_str()),
                    Some("rm -rf /tmp/x")
                );
                assert_eq!(
                    payload.get("synthesized").and_then(|v| v.as_bool()),
                    Some(true)
                );
                assert!(reason.as_deref().unwrap_or("").contains("rm -rf /tmp/x"));
            }
            other => panic!("expected ApprovalRequest, got {other:?}"),
        }
        assert!(c.pending_command_approval_id.is_some());
    }

    #[test]
    fn duplicate_waiting_on_approval_does_not_re_emit() {
        // Regression: a second `waitingOnApproval` flag for the same pending
        // command would re-send ClientMsg::ApprovalRequest to the cloud and
        // append another "pending" history row. Stay quiet (Raw) instead.
        let mut c = cursor();
        let _ = c.translate(notif(
            "item/started",
            json!({
                "item": {
                    "id": "it_1",
                    "type": "commandExecution",
                    "status": "inProgress",
                    "command": "ls",
                }
            }),
        ));
        let first = c.translate(notif(
            "thread/status/changed",
            json!({"status": {"activeFlags": ["waitingOnApproval"]}}),
        ));
        assert!(matches!(
            first.as_slice(),
            [BridgeEvent::ApprovalRequest { .. }]
        ));

        let second = c.translate(notif(
            "thread/status/changed",
            json!({"status": {"activeFlags": ["waitingOnApproval"]}}),
        ));
        assert!(
            matches!(second.as_slice(), [BridgeEvent::Raw { .. }]),
            "second waitingOnApproval must not re-emit ApprovalRequest, got {second:?}"
        );
    }

    #[test]
    fn waiting_on_approval_without_pending_command_is_raw() {
        let mut c = cursor();
        let evs = c.translate(notif(
            "thread/status/changed",
            json!({"status": {"activeFlags": ["waitingOnApproval"]}}),
        ));
        assert!(matches!(evs.as_slice(), [BridgeEvent::Raw { .. }]));
        assert!(c.unpaired_waiting_on_approval);
    }

    #[test]
    fn waiting_on_approval_before_item_started_synthesizes_when_command_arrives() {
        let mut c = cursor();
        let first = c.translate(notif(
            "thread/status/changed",
            json!({"status": {"activeFlags": ["waitingOnApproval"]}}),
        ));
        assert!(matches!(first.as_slice(), [BridgeEvent::Raw { .. }]));

        let evs = c.translate(notif(
            "item/started",
            json!({
                "item": {
                    "id": "it_1",
                    "type": "commandExecution",
                    "status": "inProgress",
                    "command": "pidash workspace me",
                    "cwd": "/work",
                    "processId": null,
                    "source": "agent"
                }
            }),
        ));
        match evs.as_slice() {
            [
                BridgeEvent::Raw { .. },
                BridgeEvent::ApprovalRequest {
                    kind,
                    payload,
                    reason,
                    ..
                },
            ] => {
                assert!(matches!(kind, ApprovalKind::CommandExecution));
                assert_eq!(
                    payload.get("command").and_then(|v| v.as_str()),
                    Some("pidash workspace me")
                );
                assert_eq!(
                    payload.get("synthesized").and_then(|v| v.as_bool()),
                    Some(true)
                );
                assert!(
                    reason
                        .as_deref()
                        .unwrap_or("")
                        .contains("pidash workspace me")
                );
            }
            other => panic!("expected Raw + ApprovalRequest, got {other:?}"),
        }
        assert!(c.pending_command_approval_id.is_some());

        let duplicate = c.translate(notif(
            "thread/status/changed",
            json!({"status": {"activeFlags": ["waitingOnApproval"]}}),
        ));
        assert!(
            matches!(duplicate.as_slice(), [BridgeEvent::Raw { .. }]),
            "duplicate waitingOnApproval must not re-emit ApprovalRequest, got {duplicate:?}"
        );
    }

    #[test]
    fn thread_status_without_waiting_flag_is_raw() {
        let mut c = cursor();
        let evs = c.translate(notif(
            "thread/status/changed",
            json!({"status": {"activeFlags": ["running"]}}),
        ));
        assert!(matches!(evs.as_slice(), [BridgeEvent::Raw { .. }]));
    }

    #[test]
    fn item_completed_clears_cached_pending_command() {
        let mut c = cursor();
        let _ = c.translate(notif(
            "item/started",
            json!({
                "item": {
                    "id": "it_1",
                    "type": "commandExecution",
                    "status": "inProgress",
                    "command": "ls",
                }
            }),
        ));
        let _ = c.translate(notif(
            "thread/status/changed",
            json!({"status": {"activeFlags": ["waitingOnApproval"]}}),
        ));
        assert!(c.pending_command.is_some());
        assert!(c.pending_command_approval_id.is_some());

        let _ = c.translate(notif(
            "item/completed",
            json!({"item": {"id": "it_1", "type": "commandExecution"}}),
        ));
        assert!(c.pending_command.is_none());
        assert!(c.pending_command_approval_id.is_none());

        // A late waitingOnApproval flag must not re-open the cleared command.
        let evs = c.translate(notif(
            "thread/status/changed",
            json!({"status": {"activeFlags": ["waitingOnApproval"]}}),
        ));
        assert!(matches!(evs.as_slice(), [BridgeEvent::Raw { .. }]));
    }

    #[test]
    fn new_item_started_resets_approval_id_so_next_waiting_mints_fresh_id() {
        let mut c = cursor();
        let _ = c.translate(notif(
            "item/started",
            json!({
                "item": {
                    "id": "it_1",
                    "type": "commandExecution",
                    "status": "inProgress",
                    "command": "ls",
                }
            }),
        ));
        let _ = c.translate(notif(
            "thread/status/changed",
            json!({"status": {"activeFlags": ["waitingOnApproval"]}}),
        ));
        let first_id = c.pending_command_approval_id.clone().unwrap();

        // A new commandExecution starts before the previous item completes.
        let _ = c.translate(notif(
            "item/started",
            json!({
                "item": {
                    "id": "it_2",
                    "type": "commandExecution",
                    "status": "inProgress",
                    "command": "pwd",
                }
            }),
        ));
        assert!(c.pending_command_approval_id.is_none());

        let evs = c.translate(notif(
            "thread/status/changed",
            json!({"status": {"activeFlags": ["waitingOnApproval"]}}),
        ));
        match evs.as_slice() {
            [BridgeEvent::ApprovalRequest { approval_id, .. }] => {
                assert_ne!(approval_id, &first_id);
            }
            other => panic!("expected fresh ApprovalRequest, got {other:?}"),
        }
    }
}
