//! `tools/list` + `tools/call` dispatcher.
//!
//! Two sources contribute to the published tool list:
//!
//! 1. The NATS request/reply tools from `schemas.json` (catalog passthrough).
//!    Each one is dispatched through `KlodiClient::request(<subject>, params)`;
//!    the marketplace's reply lands in `CallToolResult::structured`.
//! 2. A set of local tools (no 1:1 NATS subject) that run inside this process:
//!    - `klodi_setup_status` — reads `${KLODI_HOME}/{nats.creds,config.json,policies/}`
//!    - `klodi_setup_reseed_policies` — non-destructive seed of `policies/{negotiation_style,security}.md`
//!    - `klodi_health` — round-trips through `users.whoami`
//!    - `klodi_channel_message` — direct JetStream publish
//!    - `klodi_watch` — composite: `searches.create` (or one-shot `listings.search`) + `${KLODI_HOME}/buy/<slug>.md`
//!    - `klodi_unwatch` — composite: `searches.delete` + delete `${KLODI_HOME}/buy/<slug>.md`
//!
//! Per spec §§ 2/5, registration and repair (`klodi_register`,
//! `klodi_setup_repair`) are NOT MCP tools on the Rust path — they live in
//! the per-host CLI binary (`klodi-<host>-register`). `klodi_setup_status`'s
//! `next_action` field surfaces the binary name to the agent.

use super::handler::KlodiMcpHandler;
use super::schemas::catalog;
use crate::_rust_host::buy_sell_files::{self, ActionOnMatch, BuyFile, slugify};
use crate::_rust_host::policy_seed;
use crate::_rust_host::setup_status::klodi_setup_status_with_register_cli;
use crate::_natsclient::KlodiError;
use crate::_natsclient::catalog::ToolName;
use rmcp::ErrorData as McpError;
use rmcp::model::{
    CallToolResult, Content, JsonObject, ListToolsResult, Tool,
};
use serde_json::{Map, Value, json};
use std::sync::Arc;

const LOCAL_TOOL_HEALTH: &str = "klodi_health";
const LOCAL_TOOL_SETUP_STATUS: &str = "klodi_setup_status";
const LOCAL_TOOL_SETUP_RESEED_POLICIES: &str = "klodi_setup_reseed_policies";
const LOCAL_TOOL_CHANNEL_MESSAGE: &str = "klodi_channel_message";
const LOCAL_TOOL_WATCH: &str = "klodi_watch";
const LOCAL_TOOL_UNWATCH: &str = "klodi_unwatch";

/// I-4: post a structured note into the operator's ZeroClaw session.
/// Only registered + dispatched when the `zeroclaw_session` feature is
/// on AND `McpConfig::operator_channel` is `Some`. Daemon-only adapters
/// don't expose this tool.
const LOCAL_TOOL_REPORT_TO_OPERATOR: &str = "klodi_report_to_operator";

/// Approval-gate retry parameter the agent passes back. Reserved field
/// the plugin recognises on EVERY tool call — gated tools use it to
/// match the retry against the persisted pending entry. Listed here so
/// the agent and the plugin agree on the exact field name.
const APPROVAL_REQUEST_ID_FIELD: &str = "_klodi_approval_request_id";

/// Approval-gate retry parameter carrying the operator's verbatim chat
/// reply. The plugin's affirmation regex runs against this value.
const APPROVAL_TEXT_FIELD: &str = "_klodi_approval_operator_text";

/// Maximum age before [`zeroclaw_approval::reap_expired`] drops a
/// pending entry. 24h gives the operator a full day to come back and
/// approve a long-running negotiation thread; older entries are almost
/// always abandoned and would otherwise grow unbounded under
/// `${KLODI_HOME}/approvals/`.
const APPROVAL_MAX_AGE_SECONDS: i64 = 24 * 60 * 60;

const BUY_FILE_HINT: &str =
    "Append your standing-search strategy (target price, walk-away rules, dialogue digest) \
     to the body below the frontmatter. The agent reads this file before responding to \
     search.match wakes.";

/// Build the tool list. Each NATS-passthrough entry pairs the catalog's
/// description + JSON Schema with the catalog-declared subject. Local
/// tools carry hand-written schemas because they have no NATS analogue.
pub(super) fn list_all_tools() -> ListToolsResult {
    ListToolsResult::with_all_items(build_tool_list())
}

fn build_tool_list() -> Vec<Tool> {
    let mut out: Vec<Tool> = Vec::with_capacity(32);
    let cat = catalog();
    for (name, entry) in cat.tools.iter() {
        // Cross-check the public name appears in the strongly-typed
        // ToolName enum. If catalog and enum disagree, codegen is stale.
        if ToolName::from_name(name).is_none() {
            tracing::warn!(
                tool = name.as_str(),
                "schemas.json tool not in ToolName enum — codegen drift, skipping"
            );
            continue;
        }
        out.push(make_tool(name, &entry.description, &entry.params));
    }

    out.push(make_tool(
        LOCAL_TOOL_SETUP_STATUS,
        "Inspect klodi setup state. Reports phase (unconfigured | registering | needs_policy | ready), \
         file-presence flags, policy-fill state, and a structured `next_action` describing the single \
         next step (run a CLI binary, call another klodi tool, edit a file, or tighten file perms). \
         Call this at the start of every session and any time a connection or policy issue surfaces.",
        &json!({
            "type": "object",
            "properties": {},
            "additionalProperties": false
        }),
    ));
    out.push(make_tool(
        LOCAL_TOOL_SETUP_RESEED_POLICIES,
        "Re-seed ${KLODI_HOME}/policies/{negotiation_style,security}.md from the embedded skill \
         bundle, non-destructively. Existing files are preserved verbatim — use this to restore a \
         deleted policy file without touching the user's edits to the others. Returns per-file \
         seed flags: `{ negotiation_style_seeded, security_policy_seeded }`.",
        &json!({
            "type": "object",
            "properties": {},
            "additionalProperties": false
        }),
    ));
    out.push(make_tool(
        LOCAL_TOOL_HEALTH,
        "Probe klodi connectivity. Round-trips through users.whoami and reports the \
         persistent NATS connection status.",
        &json!({
            "type": "object",
            "properties": {},
            "additionalProperties": false
        }),
    ));
    out.push(make_tool(
        LOCAL_TOOL_CHANNEL_MESSAGE,
        "Publish a message on an open negotiation channel. Direct JetStream publish; \
         the recipient wakes when the message lands in their channels consumer.",
        &json!({
            "type": "object",
            "properties": {
                "channel_id": {
                    "description": "UUID of the open channel",
                    "type": "string",
                    "pattern": "^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$"
                },
                "content": {
                    "description": "Message body (1..2000 chars)",
                    "type": "string",
                    "minLength": 1,
                    "maxLength": 2000
                }
            },
            "required": ["channel_id", "content"],
            "additionalProperties": false
        }),
    ));
    out.push(make_tool(
        LOCAL_TOOL_WATCH,
        "Create a standing (persistent) search OR run a one-shot search. \
         persist=true registers the search server-side AND writes a buy file at \
         ${KLODI_HOME}/buy/<slug>.md (the agent reads this file when search.match wakes \
         arrive). persist=false (default) is a one-shot equivalent of klodi_search. \
         `delivery` is a discriminated union — pickup, ship, digital, or any (default).",
        &json!({
            "type": "object",
            "properties": {
                "query": { "type": "string", "description": "Text search" },
                "category": { "type": "string", "description": "Listing category" },
                "max_price": {
                    "type": "integer",
                    "description": "Maximum acceptable asking price, in cents",
                    "minimum": 0
                },
                "delivery": {
                    "type": "object",
                    "description": "Delivery filter — pickup, ship, digital, or any (default)",
                    "additionalProperties": true
                },
                "limit": {
                    "type": "integer",
                    "description": "Result cap for one-shot mode (default 20)",
                    "minimum": 1
                },
                "persist": {
                    "type": "boolean",
                    "description": "true = register server-side standing search; false = one-shot"
                },
                "target_price": {
                    "type": "integer",
                    "description": "Target price in cents — recorded in buy file frontmatter",
                    "minimum": 0
                },
                "action_on_match": {
                    "type": "string",
                    "enum": ["notify", "negotiate"],
                    "description": "What to do on match (default notify)"
                }
            },
            "additionalProperties": false
        }),
    ));
    out.push(make_tool(
        LOCAL_TOOL_UNWATCH,
        "Close out a standing search: deletes the server-side registration AND the \
         ${KLODI_HOME}/buy/<slug>.md file. Irreversible — call klodi_watch persist=true to recreate.",
        &json!({
            "type": "object",
            "properties": {
                "buy_slug": {
                    "type": "string",
                    "description": "Slug of the buy file to remove (e.g. gaming-laptop-abc123)"
                }
            },
            "required": ["buy_slug"],
            "additionalProperties": false
        }),
    ));
    out.push(make_tool(
        LOCAL_TOOL_REPORT_TO_OPERATOR,
        "Post a structured note directly to the operator's ZeroClaw chat session. \
         Use whenever the operator should know something the agent has done, learned, \
         or is about to ask. Examples: 'I accepted offer #abc for €600, awaiting your \
         confirmation', 'search.match found 3 listings — here are the top picks', \
         'no path to settle this negotiation, dropping with seller's permission'. \
         The note appears in the same chat the operator opened to talk to klodi; \
         it does NOT trigger a fresh agent context. Severity is informational only — \
         the gateway renders all severities the same way today.",
        &json!({
            "type": "object",
            "properties": {
                "summary": {
                    "type": "string",
                    "description": "One-line headline the operator sees first (1..200 chars)",
                    "minLength": 1,
                    "maxLength": 200
                },
                "details": {
                    "type": "string",
                    "description": "Optional multi-line body. Markdown OK.",
                    "maxLength": 4000
                },
                "severity": {
                    "type": "string",
                    "enum": ["info", "warn", "error"],
                    "description": "Visual cue for the headline. Default 'info'."
                },
                "structured": {
                    "type": "object",
                    "description": "Optional structured payload, embedded as a fenced JSON block",
                    "additionalProperties": true
                }
            },
            "required": ["summary"],
            "additionalProperties": false
        }),
    ));
    out
}

fn make_tool(name: &str, description: &str, schema: &Value) -> Tool {
    let object = match schema {
        Value::Object(map) => map.clone(),
        _ => {
            tracing::warn!(tool = name, "tool schema is not an object — using empty");
            JsonObject::new()
        }
    };
    Tool::new(
        name.to_owned(),
        description.to_owned(),
        Arc::new(object),
    )
}

/// Dispatch a `tools/call` request. Branches on the tool name into
/// either a NATS passthrough or one of the local handlers.
///
/// I-5: any call (passthrough or local) goes through the approval gate
/// first when the tool is in the gated list AND the MCP server has an
/// operator channel configured. Gated tools without operator-channel
/// support fall through (the host's own approval mechanism is
/// authoritative in that mode).
pub(super) async fn dispatch(
    handler: &KlodiMcpHandler,
    name: &str,
    arguments: Option<JsonObject>,
) -> Result<CallToolResult, McpError> {
    // `args` needs to be `mut` so the zeroclaw_session approval-gate
    // path can strip the reserved fields out of the agent's call. The
    // moltis/ironclaw build doesn't compile in that path — they get an
    // unused-mut warning suppressed by `allow` rather than a parallel
    // immutable binding. Two reasons we can't cfg-split the binding:
    // (a) zeroclaw's vendor.py strips `#[cfg(feature = "zeroclaw_session")]`
    // lines so both halves of a cfg-split would go live in the staged
    // crate, and (b) cfg(not(feature = ...)) survives the strip so the
    // immutable arm would still fire when the feature is "always-on".
    #[allow(unused_mut)]
    let mut args = arguments.unwrap_or_default();

    // Local-only zeroclaw_session tool — handle before the gate, since
    // it doesn't need approval and isn't in the catalog.
    if name == LOCAL_TOOL_REPORT_TO_OPERATOR {
        return dispatch_report_to_operator(handler, args).await;
    }

    // Approval-gate evaluation — strips the reserved fields out of
    // `args` before passing them on so downstream handlers never see
    // them. Returns Some(result) when the gate is closed (first call,
    // or denied retry) and None when it's open (no gating, or approved
    // retry).
    if let Some(gated_response) = approval_gate(handler, name, &mut args).await? {
        return Ok(gated_response);
    }

    // Locally-handled tools take priority over passthrough lookup so a
    // catalog rename can't accidentally shadow a local handler.
    match name {
        LOCAL_TOOL_SETUP_STATUS => return dispatch_setup_status(handler).await,
        LOCAL_TOOL_SETUP_RESEED_POLICIES => return dispatch_setup_reseed_policies(handler).await,
        LOCAL_TOOL_HEALTH => return dispatch_health(handler).await,
        LOCAL_TOOL_CHANNEL_MESSAGE => return dispatch_channel_message(handler, args).await,
        LOCAL_TOOL_WATCH => return dispatch_watch(handler, args).await,
        LOCAL_TOOL_UNWATCH => return dispatch_unwatch(handler, args).await,
        _ => {}
    }

    if let Some(tool) = ToolName::from_name(name) {
        return dispatch_passthrough(handler, tool, args).await;
    }

    Err(McpError::invalid_params(
        format!("unknown klodi tool: {name}"),
        Some(json!({ "tool": name })),
    ))
}

/// Approval-gate evaluation. Returns:
///
/// - `Ok(None)` — tool is not gated, OR operator channel is not configured,
///   OR the retry was approved (and the reserved fields are stripped from
///   `args` so downstream handlers don't see them).
/// - `Ok(Some(result))` — first call (prompt posted, "approval_required"
///   response surfaced to the agent), denied retry, or "still pending"
///   ambiguous retry.
/// - `Err(...)` — internal failure (couldn't post the prompt, couldn't
///   persist state, etc.).
async fn approval_gate(
    handler: &KlodiMcpHandler,
    name: &str,
    args: &mut JsonObject,
) -> Result<Option<CallToolResult>, McpError> {
    // Operator channel is the only thing that makes the gate enforceable
    // — without it, we can't post the prompt to the operator's session
    // and there's nothing to retry against. Bail out cleanly so
    // daemon-only adapters keep working.
    let channel = match handler.operator_channel() {
        Some(c) => c.clone(),
        None => return Ok(None),
    };

    // Reap stale pending entries. Doing this on every gated call adds
    // O(n) directory work to gated tools only — well below the cost of
    // a NATS round-trip. Failures here log but don't block the actual
    // dispatch.
    if let Err(err) = crate::_rust_host::zeroclaw_approval::reap_expired(
        handler.klodi_home(),
        APPROVAL_MAX_AGE_SECONDS,
    ) {
        tracing::warn!(error = %err, "klodi_approval_reap_failed");
    }

    let request_id_value = args.remove(APPROVAL_REQUEST_ID_FIELD);
    let approval_text_value = args.remove(APPROVAL_TEXT_FIELD);
    let args_for_decision = Value::Object(args.clone());

    if !crate::_rust_host::zeroclaw_approval::should_gate(name, &args_for_decision) {
        // Tool isn't gated; the reserved fields (if the agent passed
        // them anyway) have been stripped, so downstream handlers see a
        // clean args object.
        return Ok(None);
    }

    let request_id = request_id_value.and_then(|v| v.as_str().map(str::to_string));
    let approval_text = approval_text_value.and_then(|v| v.as_str().map(str::to_string));

    match (request_id, approval_text) {
        (None, _) => {
            // First call — post prompt, persist pending state, surface
            // the request_id back to the agent so it can retry.
            let summary = render_summary(name, &args_for_decision);
            let pending = crate::_rust_host::zeroclaw_approval::new_pending(
                name,
                &args_for_decision,
                summary.clone(),
            );
            let prompt = crate::_rust_host::zeroclaw_approval::format_prompt(
                name,
                &pending.request_id,
                &summary,
            );
            // Persist BEFORE posting so a crash mid-post doesn't leave
            // an unresolvable approval id on the operator's chat.
            crate::_rust_host::zeroclaw_approval::persist(handler.klodi_home(), &pending)
                .map_err(|err| McpError::internal_error(
                    format!("persisting approval state: {err}"),
                    None,
                ))?;
            crate::_rust_host::zeroclaw_ws::send_session_message(
                &channel.ws_config,
                &channel.session_id,
                &prompt,
            )
            .await
            .map_err(|err| {
                // Drop the pending entry so the agent can retry the gate
                // from scratch instead of hitting a "request_id missing"
                // dead-end. Best-effort.
                let _ = crate::_rust_host::zeroclaw_approval::clear(
                    handler.klodi_home(),
                    &pending.request_id,
                );
                McpError::internal_error(
                    format!("posting approval prompt to operator session: {err:#}"),
                    None,
                )
            })?;
            Ok(Some(structured_with_text(json!({
                "approval_required": true,
                "request_id": pending.request_id,
                "tool": name,
                "summary": summary,
                "instructions": format!(
                    "Wait for the operator's reply in their ZeroClaw chat session. \
                     When they answer, retry this same tool call with the same args \
                     plus '{APPROVAL_REQUEST_ID_FIELD}' set to '{}' and \
                     '{APPROVAL_TEXT_FIELD}' set to the operator's verbatim reply text.",
                    pending.request_id,
                ),
            }))))
        }
        (Some(rid), Some(text)) => {
            let pending = match crate::_rust_host::zeroclaw_approval::load(
                handler.klodi_home(),
                &rid,
            ) {
                Ok(Some(p)) => p,
                Ok(None) => {
                    return Ok(Some(structured_with_text(json!({
                        "approval_required": true,
                        "request_id": rid,
                        "denied": true,
                        "reason": format!(
                            "no pending approval matches request_id={rid}. \
                             It may have expired (>24h) or never been issued. \
                             Drop the reserved fields and retry to start a fresh approval flow."
                        ),
                    }))));
                }
                Err(err) => {
                    return Err(McpError::internal_error(
                        format!("loading approval state: {err}"),
                        None,
                    ));
                }
            };
            if pending.tool != name {
                return Ok(Some(structured_with_text(json!({
                    "approval_required": true,
                    "request_id": rid,
                    "denied": true,
                    "reason": format!(
                        "request_id {rid} was issued for tool {} but the retry called {name}. \
                         Drop the reserved fields and retry to start a fresh approval flow.",
                        pending.tool,
                    ),
                }))));
            }
            match crate::_rust_host::zeroclaw_approval::evaluate_retry(&pending, &text, &args_for_decision) {
                crate::_rust_host::zeroclaw_approval::ApprovalDecision::Approved => {
                    let _ = crate::_rust_host::zeroclaw_approval::clear(
                        handler.klodi_home(),
                        &rid,
                    );
                    // Open the gate — let the call fall through to the
                    // normal dispatcher.
                    Ok(None)
                }
                crate::_rust_host::zeroclaw_approval::ApprovalDecision::Denied => {
                    let _ = crate::_rust_host::zeroclaw_approval::clear(
                        handler.klodi_home(),
                        &rid,
                    );
                    Ok(Some(structured_with_text(json!({
                        "approval_required": true,
                        "request_id": rid,
                        "denied": true,
                        "reason": "operator declined the request",
                    }))))
                }
                crate::_rust_host::zeroclaw_approval::ApprovalDecision::NotGranted { reason } => {
                    Ok(Some(structured_with_text(json!({
                        "approval_required": true,
                        "request_id": rid,
                        "still_pending": true,
                        "reason": reason,
                    }))))
                }
            }
        }
        (Some(rid), None) => Ok(Some(structured_with_text(json!({
            "approval_required": true,
            "request_id": rid,
            "still_pending": true,
            "reason": format!(
                "request_id {rid} present but '{APPROVAL_TEXT_FIELD}' missing — \
                 retry the call with the operator's verbatim chat reply",
            ),
        })))),
    }
}

/// Compact, operator-readable summary of the about-to-be-gated call.
/// Plain JSON-of-args by default; we'd want richer per-tool formatting
/// later but the JSON shape matches what the agent already sees.
fn render_summary(tool: &str, args: &Value) -> String {
    let pretty = serde_json::to_string_pretty(args)
        .unwrap_or_else(|_| args.to_string());
    format!("Tool: `{tool}`\n\nArgs:\n```json\n{pretty}\n```")
}
async fn dispatch_report_to_operator(
    handler: &KlodiMcpHandler,
    args: JsonObject,
) -> Result<CallToolResult, McpError> {
    let channel = handler.operator_channel().ok_or_else(|| {
        McpError::invalid_request(
            "klodi_report_to_operator: this MCP server has no ZeroClaw operator session bound. \
             Run klodi-zeroclaw-daemon first so ${KLODI_HOME}/zeroclaw.session is populated."
                .to_string(),
            None,
        )
    })?;
    let summary = args
        .get("summary")
        .and_then(Value::as_str)
        .ok_or_else(|| {
            McpError::invalid_params(
                "klodi_report_to_operator: 'summary' (string) is required".to_string(),
                None,
            )
        })?;
    let details = args.get("details").and_then(Value::as_str);
    let severity = args
        .get("severity")
        .and_then(Value::as_str)
        .unwrap_or("info");
    let structured = args.get("structured");

    let icon = match severity {
        "warn" => "⚠️",
        "error" => "🛑",
        _ => "ℹ️",
    };
    let mut content = format!("{icon} **{summary}**");
    if let Some(d) = details {
        content.push_str("\n\n");
        content.push_str(d);
    }
    if let Some(payload) = structured {
        let pretty = serde_json::to_string_pretty(payload)
            .unwrap_or_else(|_| payload.to_string());
        content.push_str("\n\n```json\n");
        content.push_str(&pretty);
        content.push_str("\n```");
    }

    crate::_rust_host::zeroclaw_ws::send_session_message(
        &channel.ws_config,
        &channel.session_id,
        &content,
    )
    .await
    .map_err(|err| {
        McpError::internal_error(
            format!("posting note to operator session: {err:#}"),
            None,
        )
    })?;

    Ok(structured_with_text(json!({
        "posted": true,
        "session_id": channel.session_id,
        "severity": severity,
        "characters": content.chars().count(),
    })))
}

async fn dispatch_passthrough(
    handler: &KlodiMcpHandler,
    tool: ToolName,
    args: JsonObject,
) -> Result<CallToolResult, McpError> {
    let client = handler.klodi_client().await?;
    let payload = Value::Object(args);
    let result: Value = client
        .request(tool.subject(), &payload, None)
        .await
        .map_err(map_klodi_err)?;
    Ok(structured_with_text(result))
}

async fn dispatch_setup_status(
    handler: &KlodiMcpHandler,
) -> Result<CallToolResult, McpError> {
    let status =
        klodi_setup_status_with_register_cli(handler.klodi_home(), handler.register_cli());
    let body = serde_json::to_value(&status).map_err(|err| {
        McpError::internal_error(
            format!("encoding setup_status: {err}"),
            None,
        )
    })?;
    Ok(structured_with_text(body))
}

async fn dispatch_setup_reseed_policies(
    handler: &KlodiMcpHandler,
) -> Result<CallToolResult, McpError> {
    let report = policy_seed::seed_policies_if_absent(handler.klodi_home())
        .map_err(|err| McpError::internal_error(format!("reseed_policies: {err}"), None))?;
    let body = serde_json::to_value(&report).map_err(|err| {
        McpError::internal_error(
            format!("encoding seed report: {err}"),
            None,
        )
    })?;
    Ok(structured_with_text(body))
}

async fn dispatch_health(handler: &KlodiMcpHandler) -> Result<CallToolResult, McpError> {
    let started = std::time::Instant::now();
    let probe = match handler.klodi_client().await {
        Ok(client) => match client
            .request::<Value, _>(
                ToolName::KlodiWhoami.subject(),
                &json!({}),
                None,
            )
            .await
        {
            Ok(reply) => json!({
                "ok": true,
                "connected": client.is_connected().await,
                "user_id": reply.get("user_id"),
                "handle": reply.get("handle"),
                "latency_ms": started.elapsed().as_millis() as u64,
            }),
            Err(err) => json!({
                "ok": false,
                "connected": client.is_connected().await,
                "issue": format!("{err}"),
            }),
        },
        Err(err) => json!({
            "ok": false,
            "connected": false,
            "issue": format!("{err:?}"),
        }),
    };
    Ok(structured_with_text(probe))
}

async fn dispatch_channel_message(
    handler: &KlodiMcpHandler,
    args: JsonObject,
) -> Result<CallToolResult, McpError> {
    let channel_id = args
        .get("channel_id")
        .and_then(Value::as_str)
        .ok_or_else(|| {
            McpError::invalid_params(
                "klodi_channel_message: channel_id (string) is required".to_owned(),
                None,
            )
        })?;
    let content = args
        .get("content")
        .and_then(Value::as_str)
        .ok_or_else(|| {
            McpError::invalid_params(
                "klodi_channel_message: content (string) is required".to_owned(),
                None,
            )
        })?;

    let client = handler.klodi_client().await?;
    let ack = client
        .publish_channel_message(channel_id, content)
        .await
        .map_err(map_klodi_err)?;
    let body = json!({
        "sequence": ack.sequence,
        "event_id": ack.event_id,
        "message_id": ack.message_id,
        "created_at": ack.created_at,
    });
    Ok(structured_with_text(body))
}

async fn dispatch_watch(
    handler: &KlodiMcpHandler,
    args: JsonObject,
) -> Result<CallToolResult, McpError> {
    let persist = args
        .get("persist")
        .and_then(Value::as_bool)
        .unwrap_or(false);
    if persist {
        dispatch_watch_persist(handler, args).await
    } else {
        dispatch_watch_one_shot(handler, args).await
    }
}

async fn dispatch_watch_one_shot(
    handler: &KlodiMcpHandler,
    args: JsonObject,
) -> Result<CallToolResult, McpError> {
    let payload = Value::Object(compact_search_payload(&args));
    let client = handler.klodi_client().await?;
    let result: Value = client
        .request(ToolName::KlodiSearch.subject(), &payload, None)
        .await
        .map_err(map_klodi_err)?;
    Ok(structured_with_text(result))
}

async fn dispatch_watch_persist(
    handler: &KlodiMcpHandler,
    args: JsonObject,
) -> Result<CallToolResult, McpError> {
    let query_raw = args
        .get("query")
        .and_then(Value::as_str)
        .unwrap_or("")
        .to_string();
    let id_suffix: String = uuid::Uuid::new_v4().to_string().chars().take(6).collect();
    let title = if query_raw.is_empty() { "watch" } else { &query_raw };
    let slug = slugify(title, &id_suffix);

    // Catalog payload = strict subset of args plus the chosen slug.
    let mut payload = Map::new();
    payload.insert("slug".to_string(), Value::String(slug.clone()));
    if !query_raw.is_empty() {
        payload.insert("query".to_string(), Value::String(query_raw.clone()));
    }
    if let Some(category) = args.get("category").cloned() {
        payload.insert("category".to_string(), category);
    }
    if let Some(max_price) = args.get("max_price").cloned() {
        payload.insert("max_price".to_string(), max_price);
    }
    let delivery = args
        .get("delivery")
        .cloned()
        .unwrap_or_else(|| json!({ "method": "any" }));
    payload.insert("delivery".to_string(), delivery.clone());

    let client = handler.klodi_client().await?;
    let mut result: Value = client
        .request(
            ToolName::KlodiSearchesCreate.subject(),
            &Value::Object(payload),
            None,
        )
        .await
        .map_err(map_klodi_err)?;

    // Side effect: persist the buy file. Failure here is surfaced as an
    // error so the caller can decide whether to retry — the server-side
    // search is already created at this point, so a failed buy-file
    // write leaves a half-state the agent should reconcile.
    let action_on_match = match args.get("action_on_match").and_then(Value::as_str) {
        Some("negotiate") => ActionOnMatch::Negotiate,
        _ => ActionOnMatch::Notify,
    };
    let buy_dir = handler.klodi_home().join("buy");
    let buy_file_path = buy_dir.join(format!("{slug}.md"));
    let buy_file = BuyFile {
        query: query_raw,
        max_price: args.get("max_price").and_then(Value::as_i64),
        target_price: args.get("target_price").and_then(Value::as_i64),
        delivery,
        action_on_match,
        slug: slug.clone(),
        body: String::new(),
    };
    buy_sell_files::write_buy_file_at(&buy_dir, &buy_file)
        .map_err(|err| McpError::internal_error(format!("writing buy file: {err}"), None))?;

    if let Value::Object(ref mut map) = result {
        map.insert(
            "buy_file".to_string(),
            json!({
                "slug": slug,
                "path": buy_file_path.to_string_lossy(),
                "hint": BUY_FILE_HINT,
            }),
        );
    }
    Ok(structured_with_text(result))
}

async fn dispatch_unwatch(
    handler: &KlodiMcpHandler,
    args: JsonObject,
) -> Result<CallToolResult, McpError> {
    let slug = args
        .get("buy_slug")
        .and_then(Value::as_str)
        .ok_or_else(|| {
            McpError::invalid_params(
                "klodi_unwatch: buy_slug (string) is required".to_owned(),
                None,
            )
        })?
        .to_string();

    let client = handler.klodi_client().await?;
    let _: Value = client
        .request(
            ToolName::KlodiSearchesDelete.subject(),
            &json!({ "slug": slug }),
            None,
        )
        .await
        .map_err(map_klodi_err)?;

    let buy_file_path = handler.klodi_home().join("buy").join(format!("{slug}.md"));
    let buy_file_removed = buy_sell_files::delete_buy_file_at(&buy_file_path)
        .map_err(|err| McpError::internal_error(format!("removing buy file: {err}"), None))?;

    Ok(structured_with_text(json!({
        "removed": true,
        "slug": slug,
        "buy_file_removed": buy_file_removed,
    })))
}

/// Strip adapter-internal fields before forwarding a search payload to
/// the marketplace's `listings.search` subject, and drop empty values.
fn compact_search_payload(args: &JsonObject) -> Map<String, Value> {
    let mut out = Map::new();
    for (key, value) in args {
        if matches!(
            key.as_str(),
            "persist" | "action_on_match" | "target_price",
        ) {
            continue;
        }
        if value.is_null() {
            continue;
        }
        if let Value::String(s) = value {
            if s.is_empty() {
                continue;
            }
        }
        out.insert(key.clone(), value.clone());
    }
    out
}

fn structured_with_text(value: Value) -> CallToolResult {
    // MCP clients vary in which content channel they prefer.
    // Setting both a structured body and a text fallback keeps every
    // client useful — agents that look at `structured_content` get the
    // typed tree, agents that only render text get a JSON-formatted
    // line. The two are derived from the same value, so they cannot
    // disagree.
    let text = serde_json::to_string(&value).unwrap_or_else(|_| value.to_string());
    let mut result = CallToolResult::structured(value);
    result.content = vec![Content::text(text)];
    result
}

fn map_klodi_err(err: KlodiError) -> McpError {
    match &err {
        KlodiError::Marketplace { code, message, details } => McpError::invalid_request(
            format!("{code}: {}", if message.is_empty() { code } else { message }),
            Some(json!({
                "error": code,
                "message": message,
                "details": details,
            })),
        ),
        _ => McpError::internal_error(format!("{err}"), None),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn build_tool_list_includes_passthrough_and_local() {
        let tools = build_tool_list();
        let names: Vec<&str> = tools.iter().map(|t| t.name.as_ref()).collect();
        // Passthrough sample
        assert!(names.contains(&"klodi_list_create"));
        assert!(names.contains(&"klodi_offer_create"));
        // Local tools
        assert!(names.contains(&"klodi_setup_status"));
        assert!(names.contains(&"klodi_setup_reseed_policies"));
        assert!(names.contains(&"klodi_health"));
        assert!(names.contains(&"klodi_channel_message"));
        assert!(names.contains(&"klodi_watch"));
        assert!(names.contains(&"klodi_unwatch"));
        // Tools that intentionally do NOT appear on the Rust MCP surface:
        // klodi_register / klodi_register_poll / klodi_setup_repair /
        // klodi_setup_reseed_skill all live in the per-host CLI binary
        // (klodi-<host>-register) or in the embedded skill bundle, not as
        // agent tools.
        assert!(!names.contains(&"klodi_register"));
        assert!(!names.contains(&"klodi_setup_repair"));
        assert!(
            tools.len() >= 26 + 6,
            "expected at least 26 passthrough + 6 local tools — got {}",
            tools.len(),
        );
    }
    #[test]
    fn report_to_operator_tool_listed_when_zeroclaw_session_feature_enabled() {
        let tools = build_tool_list();
        let names: Vec<&str> = tools.iter().map(|t| t.name.as_ref()).collect();
        assert!(
            names.contains(&"klodi_report_to_operator"),
            "klodi_report_to_operator must appear when zeroclaw_session is on; got {names:?}",
        );
        let report_tool = tools
            .iter()
            .find(|t| t.name.as_ref() == "klodi_report_to_operator")
            .expect("found above");
        // Schema shape — `summary` required, `severity` enum, `structured`
        // optional object.
        let schema = &report_tool.input_schema;
        let required = schema.get("required").and_then(Value::as_array).unwrap();
        assert!(required.iter().any(|v| v.as_str() == Some("summary")));
        let props = schema.get("properties").and_then(Value::as_object).unwrap();
        assert!(props.contains_key("summary"));
        assert!(props.contains_key("severity"));
        assert!(props.contains_key("details"));
    }

    #[test]
    fn every_passthrough_tool_carries_an_object_schema() {
        for tool in build_tool_list() {
            let schema_type = tool.input_schema.get("type").and_then(Value::as_str);
            assert_eq!(
                schema_type,
                Some("object"),
                "tool {} must have an object input schema",
                tool.name,
            );
        }
    }

    #[test]
    fn watch_compact_payload_drops_adapter_fields() {
        let mut args = JsonObject::new();
        args.insert("query".to_string(), json!("kindle"));
        args.insert("max_price".to_string(), json!(8000));
        args.insert("persist".to_string(), json!(true));
        args.insert("action_on_match".to_string(), json!("notify"));
        args.insert("target_price".to_string(), json!(6000));
        args.insert("empty_string".to_string(), json!(""));
        let compacted = compact_search_payload(&args);
        assert!(compacted.contains_key("query"));
        assert!(compacted.contains_key("max_price"));
        assert!(!compacted.contains_key("persist"));
        assert!(!compacted.contains_key("action_on_match"));
        assert!(!compacted.contains_key("target_price"));
        assert!(!compacted.contains_key("empty_string"));
    }
}
