//! Host built-ins callable from `run_code` (ADR 0005): MCP resources and
//! watches. They are catalog entries whose routing name starts with
//! [`PREFIX`], so they never collide with an MCP tool.

use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;

use serde_json::{json, Value};

use super::watch::WatchManager;
use crate::resources::ResourceSource;
use crate::tools::{ToolCallOutput, ToolDefinition, ToolError};

/// Routing prefix for built-ins; not a valid MCP tool name prefix in practice.
pub const PREFIX: &str = "asyncai::";

#[derive(Default)]
pub struct Builtins {
    pub resources: Option<Arc<dyn ResourceSource>>,
    pub watches: Option<Arc<WatchManager>>,
}

fn string() -> Value {
    json!({"type": "string"})
}

fn object(properties: Value, required: &[&str]) -> Value {
    json!({"type": "object", "properties": properties, "required": required})
}

impl Builtins {
    /// Python-facing definitions; names are Python identifiers.
    pub fn definitions(&self) -> Vec<ToolDefinition> {
        let mut definitions = Vec::new();
        if self.resources.is_none() {
            return definitions;
        }
        let mut list = ToolDefinition::new(
            "list_resources",
            Some("List MCP resources and resource templates. `uri` is the URI, or an RFC 6570 template when `template` is true (fill in the {placeholders}); `example` is ready-to-adapt code; `subscribable` ones can be watched.\nExample: [(r[\"uri\"], r.get(\"example\")) for r in await list_resources() if r[\"subscribable\"]]"),
            object(json!({}), &[]),
        );
        list.output_schema = Some(json!({"type": "array", "items": {
            "type": "object",
            "properties": {
                "uri": string(), "template": {"type": "boolean"}, "name": string(),
                "description": string(), "example": string(), "mime_type": string(),
                "server": string(), "subscribable": {"type": "boolean"},
            },
            "required": ["uri", "template", "name", "server", "subscribable"],
            "additionalProperties": false,
        }}));
        list.read_only = Some(true);
        definitions.push(list);

        let mut read = ToolDefinition::new(
            "read_resource",
            Some("Read one MCP resource; JSON arrives parsed.\nExample: await read_resource(uri=\"memo://notes\")"),
            object(json!({"uri": string()}), &["uri"]),
        );
        read.read_only = Some(true);
        definitions.push(read);

        if self.watches.is_none() {
            return definitions;
        }
        let mut start = ToolDefinition::new(
            "watch_start",
            Some("Watch an MCP resource for changes (pushed, no polling). Asked to monitor or watch something: call this first, then refine. Changes arrive later as <watch-event> messages; bursts are merged and a line repeated for the same object is dropped. Without on_event each event becomes a line like `MODIFIED ns/name (status.phase: Pending→Running)`. `on_event` is optional Python run per event with `event` defined: {type: ADDED|MODIFIED|DELETED, object, old (previous object or None), changed (dotted paths, MODIFIED only)}; its last expression is the line, None drops the event. on_event may only call read-only tools; helpers like dig() work there.\nExample: await watch_start(name=\"vms\", uri=\"kcp-events://watch/root/computevms\", on_event=\"f\\\"{event['type']} {dig(event, 'object.metadata.name')} {dig(event, 'old.status.phase')}->{dig(event, 'object.status.phase')}\\\" if event['type'] != 'MODIFIED' or 'status.phase' in event.get('changed', []) else None\")"),
            object(
                json!({"name": string(), "uri": string(), "on_event": string()}),
                &["name", "uri"],
            ),
        );
        start.output_schema = Some(object(
            json!({"name": string(), "uri": string(), "shape": {"enum": ["state", "event-log"]}}),
            &[],
        ));
        definitions.push(start);

        let mut stop = ToolDefinition::new(
            "watch_stop",
            Some(
                "Stop a watch. Returns whether it existed.\nExample: await watch_stop(name=\"cm\")",
            ),
            object(json!({"name": string()}), &["name"]),
        );
        stop.output_schema = Some(json!({"type": "boolean"}));
        definitions.push(stop);

        let mut list_watches = ToolDefinition::new(
            "watch_list",
            Some("Active watches.\nExample: [w[\"name\"] for w in await watch_list()]"),
            object(json!({}), &[]),
        );
        list_watches.output_schema = Some(json!({"type": "array", "items": object(
            json!({"name": string(), "uri": string(), "shape": string(), "events": {"type": "integer"}, "notices": {"type": "integer"}}),
            &[],
        )}));
        list_watches.read_only = Some(true);
        definitions.push(list_watches);
        definitions
    }

    /// Boxed so the watch task → filter → built-in → `watch_start` →
    /// spawn(watch task) cycle has a nameable `Send` future.
    pub fn call<'a>(
        &'a self,
        name: &'a str,
        arguments: Value,
    ) -> Pin<Box<dyn Future<Output = Result<ToolCallOutput, ToolError>> + Send + 'a>> {
        Box::pin(self.call_inner(name, arguments))
    }

    async fn call_inner(&self, name: &str, arguments: Value) -> Result<ToolCallOutput, ToolError> {
        let text = |key: &str| {
            arguments
                .get(key)
                .and_then(Value::as_str)
                .map(str::to_string)
                .ok_or_else(|| ToolError::new(format!("{name}() requires {key}")))
        };
        let resources = self
            .resources
            .as_ref()
            .ok_or_else(|| ToolError::new("no MCP resources available"))?;
        let watches = || {
            self.watches
                .as_ref()
                .ok_or_else(|| ToolError::new(format!("{name}() is not available here")))
        };
        let json_output = |value: Value| Ok(ToolCallOutput::text(value.to_string()));
        match name {
            "list_resources" => json_output(json!(resources.list_resources().await?)),
            "read_resource" => Ok(ToolCallOutput::text(
                resources.read_resource(&text("uri")?).await?.text,
            )),
            "watch_start" => {
                let on_event = match arguments.get("on_event") {
                    None | Some(Value::Null) => None,
                    Some(Value::String(code)) if !looks_like_object_repr(code) => {
                        Some(code.clone())
                    }
                    Some(other) => {
                        return Ok(ToolCallOutput::error(format!(
                            "on_event must be Python source text (a str), not a function or object (got {}). \
                             Functions do not cross into the watch scope; pass the body as a string, e.g. \
                             on_event=\"f\\\"{{event['type']}} {{dig(event, 'object.metadata.name')}}\\\"\"",
                            crate::resources::truncate(&other.to_string(), 80)
                        )))
                    }
                };
                match watches()?
                    .start(&text("name")?, &text("uri")?, on_event)
                    .await
                {
                    Ok(value) => json_output(value),
                    Err(message) => Ok(ToolCallOutput::error(message)),
                }
            }
            "watch_stop" => json_output(json!(watches()?.stop(&text("name")?))),
            "watch_list" => json_output(watches()?.list()),
            other => Err(ToolError::new(format!("unknown built-in {other}"))),
        }
    }
}

/// A Python object that crossed the sandbox boundary as its `repr()`:
/// `Repr("<function 'fmt' at 0xec>")`, `<function f at ...>`, `<lambda>`.
fn looks_like_object_repr(code: &str) -> bool {
    let code = code.trim();
    let inner = code
        .strip_prefix("Repr(")
        .map(|rest| rest.trim_start_matches(['"', '\'']))
        .unwrap_or(code);
    [
        "<function",
        "<lambda",
        "<bound method",
        "<built-in",
        "<coroutine",
    ]
    .iter()
    .any(|prefix| inner.starts_with(prefix))
}