mahbot 0.7.1

//! Tool implementations for agent-callable capabilities.

use anyhow::Context;
pub(crate) mod active_models;
pub(crate) mod alarms;
pub(crate) mod analyze;
pub mod bun;
pub mod chrome;
pub mod chrome_daemon;
pub mod chrome_release;
// Computer-use tool (GUI observe/act). Only Linux offers it: on macOS the
// capability is withdrawn and the implementation is kept compiled but dormant
// and private there (see `computer/macos.rs`), so nothing in the product can
// reach it. Everywhere else the stub backend is never advertised.
#[cfg(target_os = "macos")]
mod computer;
#[cfg(not(target_os = "macos"))]
pub mod computer;
pub(crate) mod custom;
pub(crate) mod edit;
pub(crate) mod image_gen;
pub(crate) mod implement;
pub(crate) mod listing;
pub(crate) mod mahbot_config;
pub(crate) mod mahbot_debug;
pub(crate) mod manager_chat;
pub mod media_catalog;
pub(crate) mod path;
pub(crate) mod read;
pub(crate) mod read_document;
pub(crate) mod research;
pub(crate) mod search;
pub(crate) mod search_archived_tickets;
pub(crate) mod shell;
pub(crate) mod sleep;
pub(crate) mod ticket;
pub(crate) mod video_edit;
pub(crate) mod video_gen;
pub(crate) mod web_search;
pub(crate) mod workspace_control;

/// Maximum file size for edit and search tool operations and the dashboard editor
/// (10 MB), and the read tool's cap for any read that is not a document container;
/// a container it accepts up to `FILE_MAX_BYTES` (50 MB), like an inbound
/// attachment. Guards against OOM when agents or the GUI attempt to read very large
/// files. ReadTool/EditTool enforce it via `check_size_within()` (the read tool's
/// FIFO arm bounds its own accumulation); SearchTool and the Iced Editor use it
/// directly.
pub(crate) const MAX_FILE_SIZE_BYTES: u64 = 10 * 1024 * 1024;

/// Maximum size for a single reference image in bytes. The 2 MB figure is a
/// client-side sanity budget, not a documented OpenRouter number; base64 adds
/// ~33% overhead so we cap raw image data at 1.5 MB (1_500_000 bytes) to stay
/// well under.
/// Over-cap references are compressed by the loader (see `load_reference_image`).
/// Note: the aggregate body budget (`MAX_REQUEST_BODY_BYTES`) re-encodes even
/// under-cap references when the serialized request exceeds it, so literal
/// pass-through holds up to ~1.499 MB per single reference (base64 of 1.5 MB
/// already reaches the 2 MB budget once the data-URI prefix and JSON envelope
/// are added).
const MAX_REFERENCE_IMAGE_BYTES: u64 = 1_500_000;

/// Aggregate budget for the serialized generation request body. The per-image
/// cap does not bound the total body when multiple references are present, so
/// the final serialized body is checked against this budget and references are
/// compressed further until it fits. The 2 MB figure is a client-side sanity
/// budget mirroring the pre-existing ~2 MB provider body-limit premise (not a
/// documented OpenRouter number).
const MAX_REQUEST_BODY_BYTES: usize = 2_000_000;

/// Conservative cap on reference images per generation request, applied only
/// when the model catalog is unavailable (fail-open). Per-model caps come from
/// the catalog; provider acceptance of `input_references` varies, so this is a
/// memory-bounding sanity limit — the aggregate body budget is the real
/// backstop for what reaches the wire.
pub(crate) const MAX_REFERENCE_IMAGES_PER_REQUEST: usize = 16;

/// Canonical list of argument aliases for file path parameters.
///
/// The aliases `"file"` and `"filename"` are remapped to `"path"` for every
/// tool call by [`normalize_tool_arguments`]. Tools that do not accept a
/// `"path"` parameter are unaffected — unknown keys are discarded downstream.
const PATH_ALIAS_KEYS: &[&str] = &["file", "filename"];

/// Check that a file's size is within `limit` bytes, naming `label` in the
/// error. The shared gate for the file-reading paths, whose limits differ: the
/// read/edit cap above, and [`crate::util::FILE_MAX_BYTES`] for a converted
/// document container.
fn check_size_within(meta: &std::fs::Metadata, limit: u64, label: &str) -> anyhow::Result<()> {
    if meta.len() > limit {
        anyhow::bail!("{label}: {} bytes (limit: {limit} bytes)", meta.len());
    }
    Ok(())
}

// ── Generation request reference images ─────────────────────────────────

/// Build the OpenRouter `input_references` array (image_url entries) for
/// generation requests. Single source of truth for the reference request shape.
fn reference_json(references: &[crate::util::ReferenceImage]) -> serde_json::Value {
    serde_json::json!(
        references
            .iter()
            .map(|r| serde_json::json!({
                "type": "image_url",
                "image_url": { "url": r.data_uri() }
            }))
            .collect::<Vec<_>>()
    )
}

/// Key of the reference-image array in the generation request body.
const INPUT_REFERENCES_KEY: &str = "input_references";

/// Ensure the final serialized request body fits `max_body_bytes`, compressing
/// references (largest first, falling through to the next-largest when a
/// reference's ladder is exhausted) one step at a time until it does. The
/// per-image cap does not bound the total body (N refs × cap), so this is the
/// real guard against oversized generation requests.
///
/// The reference array is always (re)synced from the current reference state
/// when references are present, so a caller cannot silently drop them; the
/// loop rewrites it after each compression step.
pub(crate) fn fit_request_body_budget(
    body: &mut serde_json::Value,
    references: &mut [crate::util::ReferenceImage],
    max_body_bytes: usize,
) -> anyhow::Result<()> {
    if !references.is_empty() {
        body[INPUT_REFERENCES_KEY] = reference_json(references);
    }
    loop {
        if serde_json::to_vec(body)?.len() <= max_body_bytes {
            // The request body is final — drop the retained source bytes so
            // they don't sit in memory through the generation fetch/poll.
            for r in references {
                r.release_source_bytes();
            }
            return Ok(());
        }
        let Some(idx) = references
            .iter()
            .enumerate()
            .filter(|(_, r)| r.has_compression_left())
            .max_by_key(|(_, r)| r.data_uri().len())
            .map(|(i, _)| i)
        else {
            break;
        };
        references[idx].compress_more()?;
        body[INPUT_REFERENCES_KEY] = reference_json(references);
    }
    let body_len = serde_json::to_vec(body)?.len();
    if references.is_empty() {
        anyhow::bail!(
            "Generation request body is too large ({body_len} bytes; limit {max_body_bytes} bytes). \
             Shorten the prompt.",
        );
    }
    let advice = if references.len() > 1 {
        "Reduce the reference image size, the number of references, or the prompt length."
    } else {
        "Reduce the reference image size or shorten the prompt."
    };
    anyhow::bail!(
        "Generation request body is too large ({body_len} bytes after compression; \
         limit {max_body_bytes} bytes). {advice}",
    );
}

// ── Re-exports ─────────────────────────────────────────────────────────

pub(crate) use alarms::{AddAlarmTool, ListAlarmsTool, RemoveAlarmTool};
pub(crate) use analyze::{AnalyzeTool, DispatchMode};
pub(crate) use chrome::ChromeTool;
#[cfg(not(target_os = "macos"))]
pub(crate) use computer::ComputerTool;
pub(crate) use custom::CustomTool;
pub(crate) use edit::EditTool;
pub(crate) use image_gen::ImageGenTool;
pub(crate) use implement::ImplementTool;
pub(crate) use mahbot_config::MahbotConfigTool;
pub(crate) use mahbot_debug::MahbotDebugTool;
pub(crate) use manager_chat::SendMessageToManagerTool;
pub(crate) use read::ReadTool;
pub(crate) use research::ResearchTool;
pub(crate) use search::SearchTool;
pub(crate) use search_archived_tickets::SearchArchivedTicketsTool;
pub(crate) use shell::{ShellMode, ShellTool};
pub(crate) use sleep::SleepTool;
pub(crate) use ticket::{
    AddCommentTool, CreateTicketTool, GetTicketTool, ListTicketsTool, UpdateTicketTool,
};
pub(crate) use video_edit::VideoEditTool;
pub(crate) use video_gen::VideoGenTool;
pub(crate) use web_search::{WebSearchBackend, WebSearchTool};
pub(crate) use workspace_control::WorkspaceControlTool;

use crate::agent::message_router::{AgentJob, MessageKind};
use crate::{Tool, Workspace};
use futures_util::FutureExt;
use std::borrow::Cow;
use std::path::PathBuf;
use std::time::{Duration, Instant};

// ── JSON helpers ─────────────────────────────────────────────────────────

use crate::util::json::{
    get_bool, get_object, get_opt_bool, get_opt_i64, get_opt_str, get_opt_u64, get_str,
    get_str_array, get_usize, wrong_type,
};

/// Prepend a note about argument normalization so the model can see what was
/// silently corrected (e.g. flattened fields, XML wrapping).
pub(crate) fn with_normalization_notes(output: String, notes: &[String]) -> String {
    if notes.is_empty() {
        return output;
    }
    format!("[normalized] {}\n{output}", notes.join("; "))
}

/// Append a bracketed note (`[exit status: 3]`, `[leftover process] …`) under
/// `text` as its own paragraph: the text's trailing whitespace is trimmed so the
/// note is never preceded by a blank gap, and a run that produced nothing is left
/// as just the note.
pub(crate) fn with_note(text: &str, note: &str) -> String {
    let mut out = text.trim_end().to_string();
    if !out.is_empty() {
        out.push_str("\n\n");
    }
    out.push_str(note);
    out
}

/// Build a JSON schema for tool parameters.
///
/// Wraps `properties` in the standard `{"type": "object", "properties": {...}}`
/// envelope and conditionally adds `"required"` only when the slice is non-empty.
///
/// This eliminates repetitive boilerplate across tool implementations.
/// Tools with non-standard top-level keys in their top-level schema
/// (e.g., `oneOf` in WebSearchTool) should not use this directly;
/// they may still use it internally as a building block (e.g.,
/// ChromeTool's [`action_entry_schema`] calls it for each inner entry).
#[must_use]
pub(crate) fn tool_params_schema(
    properties: &serde_json::Value,
    required: &[&str],
) -> serde_json::Value {
    let mut schema = serde_json::json!({
        "type": "object",
        "properties": properties,
    });
    if !required.is_empty() {
        schema["required"] = serde_json::json!(required);
    }
    schema
}

/// Build a single action entry for a `oneOf` action list.
///
/// Wraps a `"{"name": inner}"` object in the standard `{"type":"object",
/// "properties":{name: inner}, "required":[name], "additionalProperties":false,
/// "description":description}` envelope used by the chrome and computer tools'
/// nested action schemas. `inner` is built by [`tool_params_schema`], which adds
/// `"required"` only when non-empty, and then hard-closes the action object.
#[must_use]
pub(crate) fn action_entry_schema(
    name: &str,
    description: &str,
    required: &[&str],
    properties: &serde_json::Value,
) -> serde_json::Value {
    let mut inner = tool_params_schema(properties, required);
    inner["additionalProperties"] = serde_json::json!(false);

    serde_json::json!({
        "type": "object",
        "properties": {
            (name): inner
        },
        "required": [name],
        "additionalProperties": false,
        "description": description
    })
}

// ── scrub ────────────────────────────────────────────────────────────

use crate::util::scrub_credentials;

/// Scrub credentials from tool output; delegates the scrubbing policy to the tool.
///
/// Call flow:
///
/// ```text
/// agent::execute_tool
///   └─ scrub_tool_output(tool, args, output)   (tools/mod.rs)
///        ├─ tool.should_scrub_output(args)      (lib.rs, per-tool override)
///        │    ├─ returns true  → scrub_credentials(output)  (util/mod.rs)
///        │    └─ returns false → output as-is
/// ```
///
/// Integration with shell tool's internal scrubbing:
/// The shell tool's `apply_profile_pipeline` (shell/mod.rs) scrubs stdout and stderr
/// at pipeline entry, so `ShellTool::should_scrub_output` returns `false` to prevent
/// this function from double-scrubbing. See the `ShellTool::should_scrub_output` doc
/// comment for the rationale. Any tool that performs its own credential scrubbing
/// internally must follow the same pattern — return `false` from `should_scrub_output`.
#[must_use]
pub(crate) fn scrub_tool_output(
    tool: &dyn Tool,
    call_arguments: &serde_json::Value,
    output: &str,
) -> String {
    if tool.should_scrub_output(call_arguments) {
        scrub_credentials(output)
    } else {
        output.to_string()
    }
}

/// Prefix of a failed tool result — shared so the research wrap-up's
/// success detection checks the same marker [`format_tool_failure_feedback`]
/// produces (a successful result never starts with it).
pub(crate) const TOOL_FAILURE_MARKER: &str = "Tool call failed.";

#[must_use]
pub(crate) fn format_tool_failure_feedback(
    tool_name: &str,
    tool_args: &serde_json::Value,
    reason: &str,
) -> String {
    // The `reason` parameter is pre-scrubbed by the caller
    // ([`failure_outcome`](crate::agent::Agent::failure_outcome)) and passed
    // through as-is to avoid double-scrubbing. The `tool_args` are scrubbed
    // here since they're formatted for display.
    let args_preview = scrub_credentials(&crate::util::truncate(&tool_args.to_string(), 1000));
    format!(
        "{TOOL_FAILURE_MARKER}\n\
         tool: {tool_name}\n\
         arguments: {args_preview}\n\
         reason:\n{reason}"
    )
}

/// Build the canonical `internal` tool error for a mahbot-side fault the
/// model cannot recover from (lost state, missing internal rows, all-worker
/// failure): no argument change or retry can fix it — a user or engineer
/// must investigate the mahbot service.
#[must_use]
pub(crate) fn internal_fault(cause: &str) -> anyhow::Error {
    anyhow::anyhow!(
        "internal: {cause} (mahbot-side fault) — \
         hint: no model action can recover this; a user or engineer must \
         investigate the mahbot service"
    )
}

/// Internal error-carrier for a tool call whose durable work was drain-cut by
/// shutdown/drain. NEVER persisted — the emission-time frame already recorded
/// the call durably, so the result is simply absent and the universal
/// resume-completion step re-runs/resumes it. Carried as an `anyhow::Error`
/// out of the sync analyze/implement dispatch so the caller's tool-execution
/// layer can mark the outcome `suspended`.
#[derive(Debug)]
pub(crate) struct CallSuspended;

impl std::fmt::Display for CallSuspended {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(
            f,
            "tool call suspended by shutdown/drain — durable work left launched for deterministic resume"
        )
    }
}

impl std::error::Error for CallSuspended {}

// ── Sync durable-core glue ──────────────────────────────────────────────
//
// The analyze/implement durable lifecycle is mirrored by the two tools
// (sync dispatch, async dispatch, boot resume, result envelopes). ALL of
// that glue lives on [`SyncDurableCore`] so there is exactly one
// implementation: the per-tool sync wrappers (`run_sync_analyze` /
// `run_sync_implement`), the async-dispatch tool branches, and the
// boot-resume rounds delegate to it.

/// The job-scoped parameters every durable sync core takes. Beyond
/// `(ws, task)` these are the values the sync wrappers read from the
/// task-locals / job row, so the core never touches `CURRENT_TOOL_*` itself.
pub(crate) struct CoreJobArgs<'a> {
    pub job_id: &'a str,
    pub caller_role: crate::Role,
    pub user_name: &'a str,
    pub channel: &'a str,
    pub resume: bool,
    /// Caller session pin persisted on the jobs row at fresh spawn so the
    /// live resume-completion step can find caller-owned launched jobs.
    /// Always `None` on resume — the row already carries it.
    pub caller_agent_id: Option<&'a str>,
    /// Sync-call discriminator: a checkpoint DB error fails the tool call /
    /// round so it is retried; async & boot-resume paths warn-and-continue
    /// (the outcome is recomputable on the next resume).
    pub fail_on_checkpoint_error: bool,
}

/// The single outcome form of the durable sync cores (analyze/implement):
/// `Terminal` wraps a produced result (Ok or Err — both terminalize with a
/// durable envelope), `DrainCut` means the round was cut by drain/shutdown —
/// the job stays status='launched' for resume (boot or live), nothing is
/// routed or terminalized now.
pub(crate) enum SyncCoreOutcome {
    Terminal(Result<String, anyhow::Error>),
    DrainCut,
}

/// The sync analyze/implement durable cores, normalized behind one interface
/// so the sync dispatch, the resume-completion loop and the per-tool wrappers
/// all reuse a single body.
#[derive(Copy, Clone)]
pub(crate) enum SyncDurableCore {
    Analyze,
    Implement,
}

impl SyncDurableCore {
    /// The log/site label for this core ("Analyze" / "Implement").
    #[must_use]
    pub(crate) fn label(self) -> &'static str {
        match self {
            Self::Analyze => "Analyze",
            Self::Implement => "Implement",
        }
    }

    /// Resolve a tool name to the corresponding sync durable core.
    #[must_use]
    pub(crate) fn from_tool_name(name: &str) -> Option<Self> {
        match name {
            "analyze" => Some(Self::Analyze),
            "implement" => Some(Self::Implement),
            _ => None,
        }
    }

    /// Run the durable core — both per-tool bodies return [`SyncCoreOutcome`]
    /// directly.
    pub(crate) async fn run(
        self,
        ws: &crate::Workspace,
        task: &str,
        args: CoreJobArgs<'_>,
    ) -> anyhow::Result<SyncCoreOutcome> {
        match self {
            Self::Analyze => crate::tools::analyze::run_analyze_with_job(ws, task, args).await,
            Self::Implement => {
                crate::tools::implement::run_implement_with_job(ws, task, args).await
            }
        }
    }

    /// Shared sync dispatch: read the task-locals the way the two wrappers do
    /// today, generate a fresh job id, run the core with `resume: false`, then
    /// terminalize-on-terminal / surface [`CallSuspended`] on drain-cut.
    pub(crate) async fn run_sync_dispatch(
        self,
        ws: &crate::Workspace,
        task: &str,
        caller_role: crate::Role,
    ) -> anyhow::Result<String> {
        let caller_agent_id = crate::agent::CURRENT_TOOL_AGENT_ID
            .try_with(Clone::clone)
            .unwrap_or(None);
        let user_name = crate::agent::tool_user_name();
        let channel = crate::agent::tool_channel();
        let conn = &crate::session::store().conn;

        // This dispatch never binds to prior jobs: if the model re-issues a call
        // after the resume-completion already settled and delivered, no launched
        // job remains and this spawns fresh.

        let job_id = crate::generate_id();
        let args = CoreJobArgs {
            job_id: &job_id,
            caller_role,
            user_name: &user_name,
            channel: &channel,
            resume: false,
            caller_agent_id: caller_agent_id.as_deref(),
            fail_on_checkpoint_error: caller_agent_id.is_some(),
        };
        match self.run(ws, task, args).await? {
            SyncCoreOutcome::Terminal(Ok(text)) => {
                crate::jobs::terminalize_job(conn, &job_id).await?;
                Ok(text)
            }
            SyncCoreOutcome::Terminal(Err(e)) => {
                // The round completed with failure — terminalize so a later
                // resume never re-runs the failed slots/coder forever.
                if let Err(te) = crate::jobs::terminalize_job(conn, &job_id).await {
                    tracing::warn!(
                        job = %job_id,
                        error = %te,
                        "Failed to terminalize failed {} job",
                        self.label(),
                    );
                }
                Err(e)
            }
            SyncCoreOutcome::DrainCut => {
                tracing::info!(
                    job = %job_id,
                    "{} round cut short by drain — job stays launched for deterministic resume",
                    self.label(),
                );
                Err(anyhow::Error::new(crate::tools::CallSuspended))
            }
        }
    }

    /// Shared sync resume: run the [`crate::jobs::resume_job_preamble_discrete`]
    /// preamble, then the core with `resume: true`, mapping the outcome to
    /// [`crate::jobs::SyncResumeOutcome`].
    ///
    /// `sync_resume` selects checkpoint-error propagation: a live sync resume
    /// (the caller's resume-completion step, agent/mod.rs) passes `true` — a
    /// checkpoint DB error fails the round; the async boot-resume rounds pass
    /// `false` — warn-and-continue, the outcomes are recomputable on the next
    /// resume.
    pub(crate) async fn resume_sync_core(
        self,
        ws: &crate::Workspace,
        job_id: &str,
        sync_resume: bool,
    ) -> anyhow::Result<crate::jobs::SyncResumeOutcome> {
        let label = self.label();
        let (caller, caller_role) = match crate::jobs::resume_job_preamble_discrete(
            &crate::session::store().conn,
            job_id,
            &format!("{label} resume"),
            &format!("{label} resume"),
        )
        .await
        {
            crate::jobs::ResumePreamble::Proceed(caller, caller_role) => (caller, caller_role),
            crate::jobs::ResumePreamble::DrainCut => {
                return Ok(crate::jobs::SyncResumeOutcome::DrainCut);
            }
            crate::jobs::ResumePreamble::Gone => {
                return Ok(crate::jobs::SyncResumeOutcome::Gone);
            }
            // An unreadable row is a resume-core infra failure, not an abandon:
            // fail the round so the caller-driven path never surfaces an
            // "explicitly abandoned" note for a live job (the row stays
            // launched and the next cycle retries).
            crate::jobs::ResumePreamble::Unreadable(e) => {
                return Err(e.context(format!("{label} resume: job row unreadable")));
            }
        };
        let args = CoreJobArgs {
            job_id,
            caller_role,
            user_name: &caller.user_name,
            channel: &caller.channel,
            resume: true,
            caller_agent_id: None,
            fail_on_checkpoint_error: sync_resume,
        };
        let run = self.run(ws, &caller.task, args).await;
        match run {
            Ok(SyncCoreOutcome::Terminal(result)) => Ok(crate::jobs::SyncResumeOutcome::Terminal(
                caller_role,
                caller,
                result,
            )),
            Ok(SyncCoreOutcome::DrainCut) => {
                tracing::info!(
                    job = %job_id,
                    "{label} resume drain-cut — outcome checkpointed, job stays for next boot",
                );
                Ok(crate::jobs::SyncResumeOutcome::DrainCut)
            }
            Err(e) => Err(e),
        }
    }

    /// Lowercase tool name ("analyze" / "implement") — the panic-site tag of the
    /// async dispatch scaffold.
    #[must_use]
    pub(crate) fn tool_name(self) -> &'static str {
        match self {
            Self::Analyze => "analyze",
            Self::Implement => "implement",
        }
    }

    /// Message kind of the async result envelope routed to the caller.
    #[must_use]
    pub(crate) fn message_kind(self) -> MessageKind {
        match self {
            Self::Analyze => MessageKind::AnalyzeToolResult,
            Self::Implement => MessageKind::ImplementResult,
        }
    }

    /// Build the async `<tag>` envelope message delivered to the caller's agent
    /// channel for this core's round results (success and failure paths). The
    /// caller-generated `job_id` is embedded in the envelope body so the
    /// delivered result is correlatable with the immediate ack placeholder.
    #[must_use]
    pub(crate) fn build_async_message(
        self,
        job_id: &str,
        result: &anyhow::Result<String>,
    ) -> String {
        let tag = match self {
            Self::Analyze => "analyze-tool-result",
            Self::Implement => "implement-tool-result",
        };
        build_async_result_envelope(job_id, result, tag)
    }

    /// Durable async dispatch (SPAWN → run → CHECKPOINT → COMPLETE): run the
    /// core with `resume: false` and no session pin, then terminalize into the
    /// durable envelope under the caller-generated `job_id` (the id is created
    /// in the tool's `execute()` BEFORE the spawn so the immediate ack and the
    /// delivered envelope share one id). Returns `None` only when the round
    /// was cut by drain/shutdown — the job stays status='launched' for boot
    /// resume and NOTHING is routed now. On error the envelope still routes
    /// (errors wrapped in the core's envelope tag) and the job is terminalized.
    pub(crate) async fn dispatch_durable(
        self,
        ws: &crate::Workspace,
        task: &str,
        caller_role: crate::Role,
        user_name: &str,
        channel: &str,
        job_id: &str,
    ) -> Option<AgentJob> {
        let result = match self
            .run(
                ws,
                task,
                CoreJobArgs {
                    job_id,
                    caller_role,
                    user_name,
                    channel,
                    resume: false,
                    caller_agent_id: None,
                    fail_on_checkpoint_error: false,
                },
            )
            .await
        {
            Ok(SyncCoreOutcome::DrainCut) => {
                // Drain-cut: the round's outcomes are already checkpointed — leave
                // the job status='launched' for boot resume. No terminalization, no
                // error envelope: a spurious envelope here would discard the
                // checkpointed outcomes and contradict "jobs stay status='launched'
                // for boot resume".
                tracing::info!(
                    job = %job_id,
                    "{} round cut short by drain — job stays launched for boot resume",
                    self.label(),
                );
                return None;
            }
            Ok(SyncCoreOutcome::Terminal(result)) => result,
            Err(e) => Err(e),
        };
        Some(
            crate::jobs::complete_durable_job(
                job_id,
                self.build_async_message(job_id, &result),
                self.message_kind(),
                caller_role,
                user_name,
                channel,
                &ws.name,
            )
            .await,
        )
    }

    /// Boot-resume a durable round: run the core with `resume: true` through
    /// [`Self::resume_sync_core`], then terminalize into a durable envelope and
    /// route it to the ORIGINAL caller (role/user/channel persisted on the job
    /// row at spawn), never the Manager.
    ///
    /// Aborts quietly on shutdown/drain: no routing, no terminalization — the
    /// job row stays for the next boot (checkpointed outcomes are reused, so
    /// already-completed LLM work is never lost or duplicated).
    pub(crate) async fn resume_durable_round(self, job_id: &str, ws: &crate::Workspace) {
        let label = self.label();
        let (caller_role, caller, result) = match self.resume_sync_core(ws, job_id, false).await {
            Ok(crate::jobs::SyncResumeOutcome::Terminal(caller_role, caller, result)) => {
                (caller_role, caller, result)
            }
            // Drain-cut mid-resume / job row gone (explicitly abandoned): abort
            // quietly — the core logs the reason; a gone job was already noted.
            Ok(crate::jobs::SyncResumeOutcome::DrainCut | crate::jobs::SyncResumeOutcome::Gone) => {
                return;
            }
            // Resume infra failure (roster load / checkpoint) — deliver an error
            // envelope to the ORIGINAL caller, exactly like a round-level failure.
            // The job row is still present (the core never terminalizes), so the
            // caller identity can be re-loaded for the route.
            Err(e) => {
                let Some((caller, caller_role)) = crate::jobs::resume_job_preamble(
                    &crate::session::store().conn,
                    job_id,
                    &format!("{label} resume"),
                    &format!("{label} resume"),
                )
                .await
                else {
                    return;
                };
                let envelope = crate::jobs::complete_durable_job(
                    job_id,
                    self.build_async_message(job_id, &Err(e)),
                    self.message_kind(),
                    caller_role,
                    &caller.user_name,
                    &caller.channel,
                    &ws.name,
                )
                .await;
                crate::agent::message_router::route(
                    &crate::jobs::envelope_target(&envelope),
                    envelope,
                );
                return;
            }
        };
        // Drain/shutdown fired after the round returned: abort quietly WITHOUT
        // routing and WITHOUT deleting the row — the outcomes are checkpointed
        // (next boot reuses them) and routing a partial result here would race
        // the exit.
        if crate::shutdown::aborting() {
            tracing::info!(
                job = %job_id,
                "{label} resume aborted after the round completed — job stays for next boot",
            );
            return;
        }
        let envelope = crate::jobs::complete_durable_job(
            job_id,
            self.build_async_message(job_id, &result),
            self.message_kind(),
            caller_role,
            &caller.user_name,
            &caller.channel,
            &ws.name,
        )
        .await;
        crate::agent::message_router::route(&crate::jobs::envelope_target(&envelope), envelope);
    }

    /// Async dispatch entry of [`crate::tools::analyze::AnalyzeTool`] and
    /// [`crate::tools::implement::ImplementTool`]: read the caller identity from
    /// the `CURRENT_TOOL_*` task-locals, spawn the core's durable dispatch in the
    /// background, catch its panics, and route the resulting envelope to the
    /// caller's agent channel. The core is the single source of the envelope
    /// kind, tag and builder — no call-site knobs can drift. The `job_id` is
    /// generated by the caller (the tool's `execute()`) BEFORE this spawn so the
    /// immediate ack placeholder and the delivered envelope reference one
    /// identical id.
    ///
    /// Semantics, in order:
    ///
    /// 1. `tokio::spawn` + `catch_unwind` so a panic in the dispatch task can
    ///    never leave the caller waiting forever on a result that can never
    ///    arrive — the panic path rebuilds an error envelope here.
    /// 2. The dispatch builds the success envelope itself (with pending_job_id
    ///    set by the completion tx) — route it as-is so the persisted copy and
    ///    the routed copy can never drift.
    /// 3. `Ok(None)` = drain-cut: the job stays status='launched' for boot
    ///    resume — route NOTHING now (a spurious error envelope during the drain
    ///    would discard the checkpointed outcome; the result envelope is
    ///    delivered after boot resume). This silent return is deliberately
    ///    BEFORE the aborting() guard.
    /// 4. `crate::shutdown::aborting()` guard: drain/shutdown fired between the
    ///    dispatch and the route — the pending row (if any) survives for boot
    ///    replay; skip routing rather than deliver into a consumer that has
    ///    stopped pulling. (If the pending INSERT had failed, this also
    ///    suppresses the insert-failure policy's best-effort route — the
    ///    surviving launched job row is resumed at the next boot, so the result
    ///    is deferred, never lost.)
    /// 5. `crate::agent::message_router::route(&crate::jobs::envelope_target(&envelope), envelope)`.
    ///
    /// Note: src/tools/research.rs deliberately does NOT share this scaffold —
    /// its Ok(None) also covers manual cancel, it logs "ended without delivery",
    /// and it omits the pre-route aborting() guard (at-least-once rationale).
    pub(crate) fn spawn_dispatch(
        self,
        ws: &crate::Workspace,
        task: &str,
        caller_role: crate::Role,
        job_id: String,
    ) {
        let user_name = crate::agent::tool_user_name();
        let channel = crate::agent::tool_channel();
        let (ws, task) = (ws.clone(), task.to_string());
        let ws_name = ws.name.clone();
        tokio::spawn(async move {
            let round = std::panic::AssertUnwindSafe(async {
                self.dispatch_durable(&ws, &task, caller_role, &user_name, &channel, &job_id)
                    .await
            })
            .catch_unwind()
            .await;
            let envelope = match round {
                Ok(Some(envelope)) => envelope,
                Ok(None) => {
                    // Drain-cut (see doc point 3): silently route nothing.
                    return;
                }
                Err(panic) => {
                    let panic = crate::util::panic_message(&*panic);
                    let tag = self.tool_name();
                    tracing::error!(panic = %panic, "{tag} round dispatch panicked");
                    AgentJob {
                        content: self.build_async_message(
                            &job_id,
                            &Err(anyhow::anyhow!("{tag} round dispatch panicked: {panic}")),
                        ),
                        workspace_name: ws_name,
                        user_name,
                        channel,
                        kind: self.message_kind(),
                        role: caller_role,
                        reply_target: None,
                        pending_job_id: Some(job_id),
                        originating_workspace: None,
                    }
                }
            };
            if crate::shutdown::aborting() {
                return;
            }
            crate::agent::message_router::route(&crate::jobs::envelope_target(&envelope), envelope);
        });
    }
}

/// Wrap a sub-agent/tool result in the async `<tag>` envelope delivered to
/// the caller's agent channel. The `job_id` is embedded as the first body line
/// so the delivered result is correlatable with the immediate ack placeholder.
/// Failures carry an explicit marker — findings are never silently dropped.
/// Shared with the deep research tool.
pub(crate) fn build_async_result_envelope(
    job_id: &str,
    result: &anyhow::Result<String>,
    tag: &str,
) -> String {
    match result {
        Ok(text) => format!("<{tag}>\n\nJob: {job_id}\n\n{text}</{tag}>"),
        Err(e) => {
            tracing::debug!(error = %e, %tag, "async tool result failed");
            format!("<{tag}>\n\nJob: {job_id}\n\nAn error occurred: {e}</{tag}>")
        }
    }
}

/// A tool call's textual result together with the images it produced — what
/// [`crate::Tool::execute_with_payloads`] returns.
#[derive(Debug)]
pub(crate) struct ToolOutput {
    /// The tool's textual result.
    pub text: String,
    /// Images produced by this call, injected by the agent loop as native
    /// image parts after the round's tool-result block.
    pub image_payloads: Vec<ImagePayload>,
    /// Whether `text` is real content the model must keep, rather than a claim
    /// ABOUT the images this call produced. The single flag behind both
    /// decisions the agent loop makes from a result: with content, image
    /// annotations are APPENDED and any `[IMAGE:path]` marker inside it is
    /// literal text; without it, the annotations REPLACE `text` and such a
    /// marker is a claim that may be turned into an injected image. `true` only
    /// for the read tool's converted-document answer — extracted text, or the
    /// note standing in for what a document could not yield.
    pub text_is_content: bool,
}

/// Pair a textual result with the image payload of the tool's
/// [`Tool::image_payload`] hook — the tail of the default
/// [`Tool::execute_with_payloads`], shared with the read tool's override so the
/// two can never drift.
///
/// The text is flagged as a claim about the image (not content) whether or not a
/// payload comes back, which is what every tool routed here has always meant by
/// its result: a media tool names the file it wrote, the read tool the file it
/// read. Only the read tool's converted-document answer, text the model must
/// keep, bypasses this helper.
pub(crate) async fn with_image_payload<T: Tool + ?Sized>(
    tool: &T,
    text: String,
    ws: &Workspace,
    args: &serde_json::Value,
) -> ToolOutput {
    ToolOutput {
        text,
        image_payloads: tool.image_payload(ws, args).await.into_iter().collect(),
        text_is_content: false,
    }
}

/// Outcome for a tool execution.
#[expect(clippy::struct_excessive_bools)] // independent per-call flags, not states
#[derive(Debug)]
pub(crate) struct ToolExecutionOutcome {
    pub output: String,
    pub success: bool,
    /// Images produced by this call, injected after the round's tool-result block.
    pub image_payloads: Vec<ImagePayload>,
    /// The tool's own [`ToolOutput::text_is_content`], carried through unchanged.
    pub text_is_content: bool,
    /// `true` when the tool's durable work was drain-cut and its result is
    /// intentionally absent (the call's frame is already durably persisted at
    /// emission time; the result settles on the next resume-completion). Such
    /// outcomes are SKIPPED by the tool-result commit — no row is written and
    /// the call stays dangling.
    pub suspended: bool,
    /// `true` when a successful outcome should end the agent run gracefully
    /// after the round's results commit (sleep tool); failed/unknown calls
    /// never set it.
    pub ends_turn: bool,
}

/// Discriminates the tool that produced an [`ImagePayload`] so the agent loop can
/// label the injected image with the actual opening phrase ("Read image file",
/// "Generated image file", "Chrome screenshot") without knowing which tool
/// produced it — the loop selects the annotation purely by dedup.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ImagePayloadSource {
    /// On-disk raster read by the read tool.
    Read,
    /// Tool-produced image (e.g. image_gen, or a page rasterized or extracted
    /// from a converted document), or one derived by the agent loop from a tool's
    /// `[IMAGE:path]` marker.
    Generated,
    /// Screenshot captured by the chrome tool (injected as a native image for
    /// the agent's own visual analysis).
    Chrome,
    /// Screen capture from the computer tool (injected as a native image for
    /// the agent's own visual analysis). Nothing produces it on macOS, where
    /// computer use is withdrawn.
    #[cfg_attr(
        target_os = "macos",
        expect(dead_code, reason = "computer use is withdrawn on macOS")
    )]
    Computer,
}

impl ImagePayloadSource {
    /// Opening phrase of the injected-image annotation, e.g. "Read image file"
    /// for the read tool's on-disk raster, "Generated image file" for a
    /// tool-produced image, "Chrome screenshot" for the chrome's capture.
    #[must_use]
    pub(crate) fn opener(self) -> &'static str {
        match self {
            Self::Read => "Read image file",
            Self::Generated => "Generated image file",
            Self::Chrome => "Chrome screenshot",
            Self::Computer => "Computer screenshot",
        }
    }
}

/// Per-call image payload produced by a tool that reads an on-disk raster image
/// (the Read tool), or a tool-produced/derived image (e.g. image_gen) the agent
/// loop derives from a real on-disk `[IMAGE:path]` marker. The agent loop injects
/// `[IMAGE:{data_uri}]` as a synthetic persisted User-role message after the
/// round's tool results.
///
/// `width`/`height` are the post-EXIF/post-resize dimensions of the *encoded*
/// JPEG the data-URI carries, and `format` is the uppercase source-format label
/// (`"PNG" | "JPEG" | "WEBP"`). Together they let the agent loop build the
/// definitive attached/already-attached annotation from the payload metadata
/// rather than rewriting a hardcoded sentence in a tool-result string.
#[derive(Debug, Clone)]
pub(crate) struct ImagePayload {
    /// Canonical (resolved) path of the image — surfaced in the tool-result
    /// annotation (`{opener} {path} ...`) and the debug log. The dedup
    /// key used by the agent loop is `data_uri`, not `path`.
    pub path: String,
    /// Bounded JPEG data-URI (`data:image/jpeg;base64,...`).
    pub data_uri: String,
    /// Post-EXIF/post-resize width of the encoded JPEG (px).
    pub width: u32,
    /// Post-EXIF/post-resize height of the encoded JPEG (px).
    pub height: u32,
    /// Uppercase source-format label: `"PNG" | "JPEG" | "WEBP"`.
    pub format: String,
    /// `Some` when the read resolved a typo'd path to a unique filename match;
    /// preserved in the tool-result annotation so the model keeps the typo
    /// context (mirrors the `[Recovered path: ...]` note shown for text reads).
    pub recovery_note: Option<String>,
    /// Tool that produced the image, selecting the annotation opener
    /// ("Read image file", "Generated image file", "Chrome screenshot").
    pub source: ImagePayloadSource,
}

/// Prefix `body` with a read's typo-recovery note, when there is one — the one
/// place the `[Recovered path: ...]` line is put in front of a read's answer, so
/// a recovered text read, a recovered raster's annotation and a converted
/// document's answer word it identically.
#[must_use]
pub(crate) fn with_recovery_note(note: Option<&str>, body: String) -> String {
    match note {
        Some(note) => format!("{note}\n{body}"),
        None => body,
    }
}

impl ImagePayload {
    /// Tool-result annotation for a FRESH attachment (the image is injected as a
    /// new synthetic user message).
    #[must_use]
    pub(crate) fn attached_annotation(&self) -> String {
        let base = format!(
            "{} {} ({}x{}, {}). Image content attached to the conversation as a native image.",
            self.source.opener(),
            self.path,
            self.width,
            self.height,
            self.format
        );
        with_recovery_note(self.recovery_note.as_deref(), base)
    }

    /// Tool-result annotation for a DEDUPED repeat read (image already attached;
    /// the read returns a reference rather than re-adding it).
    #[must_use]
    pub(crate) fn already_attached_annotation(&self) -> String {
        let base = format!(
            "{} {} ({}x{}, {}). Image content is already attached to the conversation as a native image.",
            self.source.opener(),
            self.path,
            self.width,
            self.height,
            self.format
        );
        with_recovery_note(self.recovery_note.as_deref(), base)
    }

    /// Build an [`ImagePayload`] from a post-decode [`crate::util::CompressedImageMeta`]
    /// (the bounded data URI + post-EXIF dims + source-format label) and the
    /// caller's resolved path, recovery note, and producing tool.
    ///
    /// This is the infallible construction tail: the caller performs the
    /// fallible decode separately
    /// (`crate::util::local_image_to_compressed_data_uri_with_meta(...).await.ok()?`)
    /// and passes the decoded meta in to build the payload.
    #[must_use]
    pub(crate) fn from_compressed_meta(
        path: &std::path::Path,
        meta: crate::util::CompressedImageMeta,
        recovery_note: Option<String>,
        source: ImagePayloadSource,
    ) -> Self {
        Self {
            path: path.display().to_string(),
            data_uri: meta.data_uri,
            width: meta.width,
            height: meta.height,
            format: meta.format,
            recovery_note,
            source,
        }
    }
}

/// Normalize a tool call name and arguments, repairing common agent mistakes.
///
/// Returns `(normalized_name, normalized_args)`. Stats and dispatch should use
/// the normalized values so recovered calls are attributed to the real tool.
#[must_use]
pub(crate) fn normalize_tool_call(
    name: &str,
    mut args: serde_json::Value,
) -> (String, serde_json::Value) {
    if name == "glob"
        && let Some(obj) = args.as_object_mut()
        && !obj.contains_key("mode")
    {
        obj.insert("mode".to_string(), serde_json::json!("files"));
    }
    let normalized_name = normalize_tool_name(name).to_string();
    normalize_tool_arguments(&normalized_name, &mut args);
    (normalized_name, args)
}

/// Map known tool-name aliases to their canonical names.
///
/// This is the single source of truth for tool-name normalization, shared by
/// [`normalize_tool_call`] (full call normalization) and [`find_tool`] (direct
/// lookup).  Adding a new alias here immediately affects both paths.
///
/// The `"glob"` alias is included because it resolves to `"search"` regardless
/// of arguments; the parallel `mode:"files"` injection is handled separately
/// in [`normalize_tool_call`] when args are available. `pub(crate)` so the
/// research sanitizer can gate on the canonical name before paying for an
/// argument clone.
pub(crate) fn normalize_tool_name(name: &str) -> &str {
    match name {
        "bash" | "run_terminal_cmd" => "shell",
        "grep" | "rg" | "grep_search" | "glob" => "search",
        "read_file" => "read",
        "str_replace" => "edit",
        // pre-rename persisted tool calls (mahbot <0.6) still rebind across restarts
        "browser" => "chrome",
        _ => name,
    }
}

fn normalize_tool_arguments(name: &str, args: &mut serde_json::Value) {
    let Some(obj) = args.as_object_mut() else {
        return;
    };

    // Apply path-aliases (file/filename) universally for every tool call.
    // Tools that do not accept a "path" parameter silently ignore the extra key.
    for &alias in PATH_ALIAS_KEYS {
        remap_arg_key(obj, alias, "path");
    }

    match name {
        "edit" => {
            // Per-tool argument remaps specific to "edit".
            remap_arg_key(obj, "old_str", "old_string");
            remap_arg_key(obj, "new_str", "new_string");
        }
        "shell" => {
            remap_arg_key(obj, "cmd", "command");
            remap_arg_key(obj, "script", "command");
        }
        "get_ticket" | "update_ticket" | "add_comment" => {
            remap_arg_key(obj, "id", "ticket_id");
            remap_arg_key(obj, "ticket", "ticket_id");
        }
        _ => {}
    }
}

/// Move `from` → `to` only when the canonical key is absent.
///
/// When both the alias (`from`) and canonical (`to`) keys are present the
/// canonical value is used.
fn remap_arg_key(obj: &mut serde_json::Map<String, serde_json::Value>, from: &str, to: &str) {
    if !obj.contains_key(to)
        && let Some(v) = obj.remove(from)
    {
        obj.insert(to.to_string(), v);
    }
}

/// Look up a tool by name in a slice of boxed `dyn Tool` values.
///
/// Tool-name aliases are resolved via `normalize_tool_name` so that all
/// callers benefit from the same alias mapping.  Prefer [`normalize_tool_call`]
/// before dispatch when full argument normalization is also desired.
#[must_use]
pub(crate) fn find_tool<'a>(tools: &'a [Box<dyn Tool>], name: &str) -> Option<&'a dyn Tool> {
    let normalized = normalize_tool_name(name);
    tools
        .iter()
        .find(|t| t.name() == normalized)
        .map(Box::as_ref)
}

/// Canonicalize a local media path and confine it to the workspace's
/// `uploads/` (received attachments) and `generated/` (tool outputs) dirs — the
/// shared rule for video_edit's local inputs and the generation tools'
/// reference images. `kind` names the input in the messages.
///
/// Callers must run it before any other filesystem access on the path: a
/// refused path is never stat'ed, opened, or sniffed.
///
/// Roots are canonicalized when they exist; a missing directory cannot contain
/// a canonicalized (existing) file, so it is skipped — which fails closed when
/// neither exists. A relative input resolves against the process CWD rather
/// than the workspace, matching video_edit's local inputs.
pub(crate) async fn resolve_local_media_path(
    input: &str,
    ws: &Workspace,
    kind: &str,
) -> anyhow::Result<PathBuf> {
    // The plain spelling, not the platform's verbatim form: this value is read by
    // the callers and echoed into the refusal text the agent sees.
    let canonical = crate::util::strip_verbatim_prefix(
        &tokio::fs::canonicalize(input)
            .await
            .with_context(|| format!("Local {kind} not found: {input}"))?,
    );
    let roots: Vec<PathBuf> = ["uploads", "generated"]
        .iter()
        .map(|d| ws.as_path().join(d))
        .filter_map(|r| std::fs::canonicalize(&r).ok())
        .collect();
    if !path::is_path_under_roots(&canonical, &roots) {
        tracing::warn!(
            path = %canonical.display(),
            "Local {kind} rejected: not inside workspace uploads or generated dirs"
        );
        anyhow::bail!(
            "forbidden: local {kind} must be inside the workspace uploads directory \
             (received media attachments) or the generated directory (previously \
             generated media), got: {} — hint: move the file into the workspace \
             uploads directory and retry",
            canonical.display()
        );
    }
    Ok(canonical)
}

/// Save generated media bytes to `workspace/generated/{prefix}_{timestamp}.{ext}`.
///
/// Creates the `generated/` directory if needed, generates a millisecond-precision
/// timestamp, writes the file, and returns the full `PathBuf`. The bytes are
/// content-sniffed and stripped (`util::provenance`) on the way out, so the single
/// written file is what the GUI preview, the Telegram delivery and the model-context
/// payload all consume; non-image outputs pass through untouched.
///
/// # Security note
/// This function deliberately bypasses path security (no `resolve_write_target`
/// check) because `generated/` is an ephemeral tool-owned directory within the
/// workspace. Do not use this function for user-uploaded or arbitrary content.
async fn save_generated_file(
    ws: &Workspace,
    bytes: &[u8],
    prefix: &str,
    ext: &str,
) -> anyhow::Result<PathBuf> {
    let generated_dir = ws.as_path().join("generated");
    tokio::fs::create_dir_all(&generated_dir)
        .await
        .with_context(|| {
            format!(
                "Failed to create generated directory at {}",
                generated_dir.display()
            )
        })?;

    let timestamp = crate::util::unix_millis();
    let output_path = generated_dir.join(format!("{prefix}_{timestamp}.{ext}"));

    let stripped = crate::util::provenance::strip_image_provenance(bytes);
    tokio::fs::write(&output_path, stripped.as_ref())
        .await
        .with_context(|| {
            format!(
                "Failed to write generated file to {}",
                output_path.display()
            )
        })?;

    if matches!(stripped, Cow::Owned(_)) {
        // Debug: one record per generated image — the strip running is the normal
        // case, not a finding the log store or the issues view needs.
        tracing::debug!(
            path = %output_path.display(),
            "Stripped AI-provenance metadata from a generated image artifact"
        );
    }

    Ok(output_path)
}

/// Format a video tool result: the `[VIDEO:path]` marker first (so the reply
/// path and GUI keep working), then a "Video content:" description line when
/// the shared video transcription succeeds. Fail-open: marker-only on any
/// failure. The transcription's live tracking resolves via the tool-execution
/// task-local (the owning agent's activity indicator); `None` is passed for
/// workspace because the in-agent path never needs a non-agent call row.
pub(crate) async fn format_video_result(marker: String, output_path: &std::path::Path) -> String {
    match crate::providers::transcribe_video_file(output_path, None).await {
        Some(text) => format!("{marker}\n\nVideo content: {text}"),
        None => marker,
    }
}

// ── Async video jobs (video_gen / video_edit) ───────────────────────────

/// Wall-clock deadline for an async-video job (1 hour) — shared by the
/// video_gen and video_edit tools.
const VIDEO_JOB_TIMEOUT: Duration = Duration::from_hours(1);

/// Polling interval between video job status checks (30 s).
const VIDEO_POLL_INTERVAL: Duration = Duration::from_secs(30);

/// Kind-dependent labels for an OpenRouter async-video job.
///
/// Fields stay separate so every user-facing string is byte-identical with the
/// pre-refactor tools (e.g. gen's download context is bare "Video download",
/// not "Video generation download").
struct VideoJobLabels {
    /// "Video generation" / "Video edit" — submission/poll HTTP contexts,
    /// tracing, "failed" and timeout bails.
    label: &'static str,
    /// "generation" / "editing" — the 402 credits bail.
    gerund: &'static str,
    /// "Video download" / "Video edit download" — download context + MP4 bail.
    download: &'static str,
}

impl VideoJobLabels {
    const GENERATION: Self = Self {
        label: "Video generation",
        gerund: "generation",
        download: "Video download",
    };
    const EDIT: Self = Self {
        label: "Video edit",
        gerund: "editing",
        download: "Video edit download",
    };
}

/// Submit an OpenRouter async-video job (exactly one POST — no retry; each
/// submission is a billable job and the endpoint has no idempotency key),
/// poll for completion (1-hour wall-clock deadline), download the result, and
/// validate it is a real MP4. Returns the video bytes; callers save the file
/// and format the media marker.
#[expect(clippy::too_many_lines)]
async fn fetch_async_video(
    api_base: &str,
    body: &serde_json::Value,
    labels: VideoJobLabels,
) -> anyhow::Result<Vec<u8>> {
    // ── Step 1: Submit video job (exactly one POST — no retry) ─────────
    // Each submission is a billable job; the endpoint has no idempotency key.
    let submit_url = format!("{api_base}/videos");
    let submit_body: serde_json::Value = match crate::util::http::post_json_to_provider(
        &submit_url,
        body,
        &format!("{} submission", labels.label),
    )
    .await
    {
        Ok(v) => v,
        Err(e) => {
            if e.downcast_ref::<crate::util::error::HttpError>()
                .map(|e| e.status)
                == Some(402)
            {
                anyhow::bail!(
                    "Insufficient OpenRouter credits for video {} (HTTP 402). \
                     Please add credits to your OpenRouter account and try again.",
                    labels.gerund,
                );
            }
            return Err(e);
        }
    };

    // OpenRouter returns: { id, polling_url, status, ... }
    let job_id = match submit_body.get("id").and_then(|v| v.as_str()) {
        Some(id) => id.to_string(),
        None => {
            anyhow::bail!("No job ID in submission response: {submit_body}");
        }
    };

    let polling_url = match submit_body.get("polling_url").and_then(|v| v.as_str()) {
        Some(url) => url.to_string(),
        None => format!("{api_base}/videos/{job_id}"),
    };

    tracing::info!(%job_id, "{} job submitted", labels.label);

    // ── Step 2: Poll for completion (1-hour wall-clock deadline) ────────
    let deadline = Instant::now() + VIDEO_JOB_TIMEOUT;
    let mut result_url: Option<String> = None;
    let mut attempt: u32 = 0;

    while Instant::now() < deadline {
        attempt += 1;

        // Sleep the full interval, but never past the deadline.
        let remaining = deadline.saturating_duration_since(Instant::now());
        tokio::time::sleep(VIDEO_POLL_INTERVAL.min(remaining)).await;
        if Instant::now() >= deadline {
            break;
        }

        // Race the poll against the remaining window so a slow poll request
        // cannot push the total wait past the hour.
        let remaining = deadline.saturating_duration_since(Instant::now());
        let poll_body = match tokio::time::timeout(
            remaining,
            crate::util::http::get_json_from_provider(
                &polling_url,
                &format!("{} poll", labels.label),
            ),
        )
        .await
        {
            Ok(Ok(v)) => v,
            Ok(Err(e)) => {
                tracing::debug!(%job_id, attempt, error = %e, "Poll failed");
                continue;
            }
            // Deadline elapsed while the poll request was in flight.
            Err(_) => break,
        };

        let status = poll_body
            .get("status")
            .and_then(|v| v.as_str())
            .unwrap_or("unknown");

        tracing::debug!(%job_id, %status, attempt, "{} poll", labels.label);

        if status == "completed" {
            // Download URL: OpenRouter provides unsigned_urls array or
            // a content endpoint at /api/v1/videos/{jobId}/content?index=0
            result_url = poll_body
                .get("unsigned_urls")
                .and_then(|v| v.as_array())
                .and_then(|arr| arr.first())
                .and_then(|v| v.as_str())
                .map(String::from)
                .or_else(|| {
                    // Fallback: use content endpoint
                    Some(format!("{api_base}/videos/{job_id}/content?index=0"))
                });
            break;
        }

        if status == "failed" || status == "cancelled" || status == "expired" {
            let err_msg = poll_body
                .get("error")
                .and_then(|v| v.as_str())
                .unwrap_or("Unknown error");
            anyhow::bail!("{} failed: {err_msg}", labels.label);
        }
    }

    let Some(download_url) = result_url else {
        anyhow::bail!(
            "{} did not complete within the 1-hour timeout period",
            labels.label
        );
    };

    // ── Step 3: Download the video ──────────────────────────────────────
    // The result URL requires the bearer key despite the "unsigned" name.
    // The per-request timeout is the remaining job window, so large files on
    // slow connections are not cut short by the shared client's 2-minute cap.
    let remaining = deadline.saturating_duration_since(Instant::now());
    if remaining.is_zero() {
        anyhow::bail!(
            "{} did not complete within the 1-hour timeout period",
            labels.label
        );
    }
    let video_bytes =
        crate::util::http::get_bytes_from_provider(&download_url, labels.download, remaining)
            .await?;

    // Validate the payload is a real MP4 (no Content-Length on the response;
    // an error page would slip through otherwise).
    if video_bytes.len() <= 100_000 || &video_bytes[4..8] != b"ftyp" {
        anyhow::bail!(
            "{} returned an invalid file ({} bytes, no ftyp header)",
            labels.download,
            video_bytes.len(),
        );
    }

    Ok(video_bytes)
}

// ── Tests ──────────────────────────────────────────────────────────────

#[cfg(test)]
mod tests {
    use super::*;
    use crate::Tool;
    use crate::ToolSpec;
    use crate::workspace::test_ws_named;
    use tempfile::TempDir;

    // ── ToolSpec serde ───────────────────────────────────────────

    #[test]
    fn tool_spec_serde_roundtrip() {
        let spec = ToolSpec {
            name: "test".into(),
            description: "A test tool".into(),
            parameters: serde_json::json!({"type": "object"}),
        };
        let parsed: ToolSpec =
            serde_json::from_str(&serde_json::to_string(&spec).unwrap()).unwrap();
        assert_eq!(parsed.name, "test");
    }

    // ── ImagePayload source verb ──────────────────────────────────

    #[test]
    fn image_payload_generated_verb_annotation() {
        let payload = ImagePayload {
            path: "/gen/img.png".into(),
            data_uri: "data:image/jpeg;base64,aaa".into(),
            width: 4,
            height: 4,
            format: "PNG".into(),
            recovery_note: None,
            source: ImagePayloadSource::Generated,
        };
        let fresh = payload.attached_annotation();
        assert!(
            fresh.starts_with("Generated image file /gen/img.png"),
            "fresh: {fresh}"
        );
        assert!(
            !fresh.starts_with("Read image file"),
            "must not be Read: {fresh}"
        );
        let dup = payload.already_attached_annotation();
        assert!(
            dup.starts_with("Generated image file /gen/img.png"),
            "dup: {dup}"
        );
        assert!(dup.contains("already attached"), "dup: {dup}");
    }

    // ── find_tool aliases ──────────────────────────────────────────

    #[test]
    fn find_tool_aliases() {
        let tools: Vec<Box<dyn Tool>> = vec![
            Box::new(SearchTool),
            Box::new(ShellTool::new(ShellMode::Full)),
            Box::new(ReadTool::general()),
            Box::new(EditTool),
        ];

        // Each case: (alias, expected_tool_name or None).
        let cases: &[(&str, Option<&str>)] = &[
            // Canonical names
            ("search", Some("search")),
            ("shell", Some("shell")),
            ("read", Some("read")),
            ("edit", Some("edit")),
            // Shell aliases
            ("bash", Some("shell")),
            ("run_terminal_cmd", Some("shell")),
            // Search aliases
            ("grep", Some("search")),
            ("rg", Some("search")),
            ("grep_search", Some("search")),
            ("glob", Some("search")),
            // Read aliases
            ("read_file", Some("read")),
            // Edit aliases
            ("str_replace", Some("edit")),
            // Unknown tool
            ("unknown", None),
        ];

        for &(input, expected) in cases {
            let found = find_tool(&tools, input);
            assert_eq!(found.map(Tool::name), expected, "find_tool({input:?})");
        }
    }

    #[test]
    fn normalize_tool_call_repairs_names_and_args() {
        let (name, args) = normalize_tool_call("bash", serde_json::json!({"cmd": "echo hi"}));
        assert_eq!(name, "shell");
        assert_eq!(args["command"], "echo hi");

        let (name, args) = normalize_tool_call("glob", serde_json::json!({"query": "main.rs"}));
        assert_eq!(name, "search");
        assert_eq!(args["mode"], "files");

        let (name, args) = normalize_tool_call("get_ticket", serde_json::json!({"id": "mahbot-1"}));
        assert_eq!(name, "get_ticket");
        assert_eq!(args["ticket_id"], "mahbot-1");

        // "read" tool: file/filename → path
        let (name, args) = normalize_tool_call("read", serde_json::json!({"file": "src/main.rs"}));
        assert_eq!(name, "read");
        assert_eq!(args["path"], "src/main.rs");

        let (name, args) =
            normalize_tool_call("read_file", serde_json::json!({"filename": "lib.rs"}));
        assert_eq!(name, "read");
        assert_eq!(args["path"], "lib.rs");

        // "edit" tool: file/filename → path, old_str/new_str → old_string/new_string
        let (name, args) = normalize_tool_call(
            "edit",
            serde_json::json!({"file": "main.rs", "old_str": "foo", "new_str": "bar"}),
        );
        assert_eq!(name, "edit");
        assert_eq!(args["path"], "main.rs");
        assert_eq!(args["old_string"], "foo");
        assert_eq!(args["new_string"], "bar");

        // Canonical "path" key is preserved when both "path" and alias are present.
        // remap_arg_key only moves when canonical key is absent, so "file" remains.
        let (name, args) = normalize_tool_call(
            "read",
            serde_json::json!({"path": "canonical.rs", "file": "alias.rs"}),
        );
        assert_eq!(name, "read");
        assert_eq!(args["path"], "canonical.rs");
        assert!(args.as_object().unwrap().contains_key("file"));
    }

    #[test]
    // ── media_marker coverage ────────────────────────────────────────
    fn all_media_tools_implement_media_marker() {
        // Each media-generation tool must return Some from media_marker()
        let tools: [(&str, Box<dyn Tool>); 3] = [
            ("ImageGenTool", Box::new(ImageGenTool)),
            ("VideoGenTool", Box::new(VideoGenTool)),
            ("VideoEditTool", Box::new(VideoEditTool)),
        ];
        for (name, tool) in &tools {
            let marker = tool.media_marker();
            assert!(
                marker.is_some(),
                "{name} should return Some from media_marker()"
            );
            let marker = marker.unwrap();
            // Validate format: `[KIND:` where KIND is uppercase letters
            assert!(
                marker.starts_with('['),
                "{name} marker {marker:?} should start with '['"
            );
            assert!(
                marker.ends_with(':'),
                "{name} marker {marker:?} should end with ':'"
            );
            let kind = &marker[1..marker.len() - 1]; // strip [ and :
            assert!(
                !kind.is_empty() && kind.chars().all(char::is_uppercase),
                "{name} marker kind {kind:?} should be non-empty uppercase letters"
            );
            // Validate against the canonical MEDIA_MARKER_RE pattern
            let full_marker = format!("{marker}/some/path]");
            assert!(
                crate::util::MEDIA_MARKER_RE.is_match(&full_marker),
                "{name} marker + path should match MEDIA_MARKER_RE, got: {full_marker:?}"
            );
        }
    }

    // ── normalize_tool_call tests ─────────────────────────────────────

    /// Verify that [`normalize_tool_call`] universally remaps every alias in
    /// [`PATH_ALIAS_KEYS`] to `"path"` — regardless of whether the tool
    /// explicitly accepts a `"path"` parameter.
    ///
    /// This test verifies path-aliasing for both a path-accepting tool (`read`)
    /// and a non-path tool (`shell`). The remap happens unconditionally;
    /// tools that do not use `"path"` simply ignore the extra key.
    ///
    /// This test explicitly iterates the constant so the loop-based approach in
    /// [`normalize_tool_arguments`] is verified against all current aliases.
    /// If an alias is added to [`PATH_ALIAS_KEYS`], this test immediately
    /// exercises it — preventing any gap between the lookup path and the
    /// normalization path.
    #[test]
    fn normalize_tool_call_remaps_all_path_aliases() {
        for &alias in PATH_ALIAS_KEYS {
            for (tool_name, extra) in &[
                ("read", serde_json::json!({})),
                ("shell", serde_json::json!({"cmd": "ls"})),
            ] {
                let mut input = serde_json::json!({});
                input[alias] = serde_json::json!("src/main.rs");
                if let Some(obj) = extra.as_object() {
                    for (k, v) in obj {
                        input[k] = v.clone();
                    }
                }
                let (name, args) = normalize_tool_call(tool_name, input);
                assert_eq!(
                    name, *tool_name,
                    "tool name should not change for {tool_name} with alias {alias}"
                );
                // The alias is remapped to "path" unconditionally — even for tools
                // that don't use the "path" parameter (unknown keys are discarded).
                assert_eq!(
                    args["path"], "src/main.rs",
                    "alias {alias} should be remapped to 'path' for tool {tool_name}"
                );
                // The alias key itself should have been removed since "path" was absent.
                assert!(
                    !args.as_object().unwrap().contains_key(alias),
                    "alias key {alias} should be removed after normalization for {tool_name}"
                );
                // Shell-specific remaps must be unaffected.
                if *tool_name == "shell" {
                    assert_eq!(args["command"], "ls");
                }
            }
        }
    }

    // ── resolve_local_media_path tests ─────────────────────────────────────

    /// The rule video_edit's local inputs and the generation tools' reference
    /// images share: only the workspace's `uploads/` and `generated/` dirs are
    /// reachable, and anywhere else is refused with the reason.
    #[tokio::test]
    async fn resolve_local_media_path_confines_to_uploads_and_generated() {
        let tmp = TempDir::new().expect("tempdir");
        let ws = crate::Workspace::from_path(tmp.path());
        let write = |name: &str| {
            let path = tmp.path().join(name);
            std::fs::create_dir_all(path.parent().unwrap()).unwrap();
            std::fs::write(&path, b"media").unwrap();
            path.to_string_lossy().to_string()
        };

        // Both media dirs are accepted, including `generated/` when `uploads/`
        // does not exist yet.
        assert!(
            resolve_local_media_path(&write("generated/image_1.png"), &ws, "image")
                .await
                .is_ok()
        );
        assert!(
            resolve_local_media_path(&write("uploads/clip.mp4"), &ws, "clip")
                .await
                .is_ok()
        );

        // Anywhere else in the workspace is refused, naming both accepted dirs.
        let err = resolve_local_media_path(&write("config.toml"), &ws, "clip")
            .await
            .expect_err("a file outside uploads/generated must be refused");
        assert!(
            err.to_string()
                .contains("must be inside the workspace uploads directory"),
            "unexpected error: {err}"
        );

        // A workspace with neither dir refuses everything — a missing dir
        // cannot contain a canonicalized file, so the rule fails closed.
        let bare = TempDir::new().expect("tempdir");
        let bare_ws = crate::Workspace::from_path(bare.path());
        let root_file = bare.path().join("clip.mp4");
        std::fs::write(&root_file, b"media").unwrap();
        assert!(
            resolve_local_media_path(&root_file.to_string_lossy(), &bare_ws, "clip")
                .await
                .is_err()
        );
    }

    // ── save_generated_file tests ──────────────────────────────────────────

    #[tokio::test]
    async fn save_generated_file_creates_file() {
        let tmp = TempDir::new().expect("tempdir");
        let ws = test_ws_named(&tmp.path().to_string_lossy(), "test");

        let data = b"hello world";
        let path = save_generated_file(&ws, data, "img", "png")
            .await
            .expect("save_generated_file should succeed");

        assert!(path.exists(), "file should exist: {}", path.display());
        let content = tokio::fs::read_to_string(&path).await.unwrap();
        assert_eq!(content, "hello world");

        // Verify filename format: {prefix}_{timestamp}.{ext}
        let file_name = path.file_name().unwrap().to_str().unwrap();
        assert!(
            file_name.starts_with("img_"),
            "filename should start with 'img_': {file_name}",
        );
        assert!(
            std::path::Path::new(file_name)
                .extension()
                .is_some_and(|ext| ext.eq_ignore_ascii_case("png")),
            "filename should end with '.png': {file_name}",
        );

        let _ = tokio::fs::remove_dir_all(tmp.path()).await;
    }

    #[tokio::test]
    async fn save_generated_file_creates_directory_if_missing() {
        let tmp = TempDir::new().expect("tempdir");
        let ws = test_ws_named(&tmp.path().join("nested").to_string_lossy(), "test");

        let data = b"test content";
        let path = save_generated_file(&ws, data, "vid", "mp4")
            .await
            .expect("save_generated_file should create dirs");

        assert!(path.exists(), "file should exist: {}", path.display());
        assert!(
            path.starts_with(tmp.path().join("nested")),
            "file should be inside workspace"
        );

        let _ = tokio::fs::remove_dir_all(tmp.path()).await;
    }

    // ── Reference-image body budget ─────────────────────────────────

    #[tokio::test]
    async fn reference_body_budget_compresses_further() {
        const BUDGET: usize = 150_000;
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join("ref.png");
        std::fs::write(&path, crate::util::test::noisy_png(384, 384)).unwrap();
        let mut refs = vec![
            crate::util::load_reference_image(&path, MAX_REFERENCE_IMAGE_BYTES)
                .await
                .unwrap(),
        ];
        let original = refs[0].data_uri().to_string();
        let mut body = serde_json::json!({
            "model": "test",
            "prompt": "test",
            "input_references": reference_json(&refs),
        });
        // The per-image cap passes a 384×384 ref through unchanged (~590 KB
        // base64 body); a small budget forces the aggregate ladder.
        assert!(serde_json::to_vec(&body).unwrap().len() > BUDGET);
        fit_request_body_budget(&mut body, &mut refs, BUDGET).unwrap();
        assert!(serde_json::to_vec(&body).unwrap().len() <= BUDGET);
        assert_ne!(
            refs[0].data_uri(),
            original,
            "reference should be compressed"
        );
    }

    #[tokio::test]
    async fn body_budget_falls_through_to_smaller_reference() {
        // When the largest reference's ladder is exhausted, the loop must
        // fall through to the next-largest instead of bailing.
        let tmp = TempDir::new().unwrap();
        let big = tmp.path().join("big.png");
        let small = tmp.path().join("small.png");
        std::fs::write(&big, crate::util::test::noisy_png(512, 512)).unwrap();
        std::fs::write(&small, crate::util::test::noisy_png(128, 128)).unwrap();
        let mut refs = vec![
            crate::util::load_reference_image(&big, MAX_REFERENCE_IMAGE_BYTES)
                .await
                .unwrap(),
            crate::util::load_reference_image(&small, MAX_REFERENCE_IMAGE_BYTES)
                .await
                .unwrap(),
        ];
        let small_original = refs[1].data_uri().to_string();
        let mut body = serde_json::json!({
            "model": "test",
            "prompt": "test",
            "input_references": reference_json(&refs),
        });
        // Exhaust the big reference's ladder, then measure the raw body so the
        // budget can be pinned to a window that only the small reference can
        // close — deterministically forcing the fall-through path.
        while refs[0].has_compression_left() {
            refs[0].compress_more().unwrap();
        }
        body[INPUT_REFERENCES_KEY] = reference_json(&refs);
        let raw_body = serde_json::to_vec(&body).unwrap().len();
        fit_request_body_budget(&mut body, &mut refs, raw_body - 1).unwrap();
        assert!(serde_json::to_vec(&body).unwrap().len() < raw_body);
        assert_ne!(
            refs[1].data_uri(),
            small_original,
            "smaller reference should have been compressed"
        );
    }
}