// Package mcp implements the Oikos MCP interface (plan R3-10).
// Uses the official MCP Go SDK with Streamable HTTP transport.
package mcp

import (
	"bytes"
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"html"
	"io"
	"log/slog"
	"net"
	"net/http"
	"net/url"
	"os"
	"regexp"
	"strings"
	"sync"
	"time"

	"github.com/dtoro/oikos/internal/db"
	"github.com/dtoro/oikos/internal/db/sqlcgen"
	"github.com/dtoro/oikos/internal/observability"
	"github.com/dtoro/oikos/internal/policy"
	"github.com/google/jsonschema-go/jsonschema"
	"github.com/google/uuid"
	"github.com/modelcontextprotocol/go-sdk/mcp"
	"golang.org/x/crypto/ssh"
)

// prop is one input-schema property (name → type + description).
type prop struct {
	name, typ, desc string
}

// objSchema builds an "object" JSON Schema from a list of properties. The
// MCP SDK requires every tool to declare an object input schema so tools
// are self-describing to the agent; a nil schema panics at registration.
func objSchema(props ...prop) *jsonschema.Schema {
	s := &jsonschema.Schema{Type: "object", Properties: map[string]*jsonschema.Schema{}}
	for _, p := range props {
		s.Properties[p.name] = &jsonschema.Schema{Type: p.typ, Description: p.desc}
	}
	return s
}

// NewHandler creates an http.Handler that serves the Oikos MCP server.
// agentID is the Nomos agent entity UUID; tool calls are logged to agent_activity.
func NewHandler(pool *db.Pool, token string, agentID uuid.UUID) http.Handler {
	s := newServer(pool, agentID)
	handler := mcp.NewStreamableHTTPHandler(func(r *http.Request) *mcp.Server {
		if token != "" {
			if r.Header.Get("Authorization") != "Bearer "+token {
				return nil
			}
		}
		return s
	}, nil)
	return handler
}

// toolHandler is the function signature registered via AddTool.
type toolHandler = mcp.ToolHandler

func newServer(pool *db.Pool, agentID uuid.UUID) *mcp.Server {
	s := mcp.NewServer(&mcp.Implementation{Name: "oikos", Version: "dev"}, &mcp.ServerOptions{
		Logger: slog.Default(),
	})

	register := func(tool *mcp.Tool, handler toolHandler) {
		s.AddTool(tool, withActivityLogging(pool, agentID, tool.Name, handler))
	}

	register(&mcp.Tool{Name: "get_entity", Description: "Get an entity by slug or UUID",
		InputSchema: objSchema(prop{"slug_or_id", "string", "Entity slug (e.g. host:hubris) or UUID"}),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		idOrSlug, _ := args["slug_or_id"].(string)
		return queryEntity(ctx, pool, idOrSlug), nil
	})

	register(&mcp.Tool{Name: "list_entities", Description: "List entities filtered by type, state, or search",
		InputSchema: objSchema(
			prop{"type", "string", "Filter by entity type"},
			prop{"state", "string", "Filter by lifecycle state"},
			prop{"q", "string", "Substring match on slug or name"},
			prop{"limit", "integer", "Max rows (default 50)"}),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		limit := int(getFloat(args, "limit", 50))
		return queryRows(ctx, pool, `
			SELECT e.slug, e.type, e.name, e.state, e.version, e.created_at, e.updated_at
			FROM entities e
			WHERE ($1::text IS NULL OR e.type = $1)
			  AND ($2::text IS NULL OR e.state = $2)
			  AND ($3::text IS NULL OR e.slug ILIKE '%'||$3||'%' OR e.name ILIKE '%'||$3||'%')
			ORDER BY e.slug LIMIT $4`,
			nStr(args["type"]), nStr(args["state"]), nStr(args["q"]), limit), nil
	})

	register(&mcp.Tool{Name: "get_relations", Description: "Get relationships for an entity",
		InputSchema: objSchema(prop{"entity_id", "string", "Entity slug"}),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		slug, _ := args["entity_id"].(string)
		return queryRows(ctx, pool, `
			SELECT r.type, src.slug AS source, tgt.slug AS target
			FROM relationships r
			JOIN entities src ON src.id = r.source_id
			JOIN entities tgt ON tgt.id = r.target_id
			WHERE (src.slug = $1 OR tgt.slug = $1) AND r.valid_to IS NULL
			ORDER BY r.type`, slug), nil
	})

	register(&mcp.Tool{Name: "get_blast_radius", Description: "Find entities affected if this entity goes down",
		InputSchema: objSchema(
			prop{"entity_id", "string", "Entity slug"},
			prop{"depth", "integer", "Traversal depth (default 3)"}),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		slug, _ := args["entity_id"].(string)
		depth := int(getFloat(args, "depth", 3))
		return queryRows(ctx, pool,
			"SELECT e.slug, CAST(b.depth AS int) FROM blast_radius((SELECT id FROM entities WHERE slug = $1), $2) b JOIN entities e ON e.id = b.entity_id",
			slug, depth), nil
	})

	register(&mcp.Tool{Name: "get_health_summary", Description: "Current fleet health summary",
		InputSchema: objSchema(),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		return queryRows(ctx, pool, `
			SELECT e.slug, e.type, st.health, st.last_check_at
			FROM entity_status st JOIN entities e ON e.id = st.entity_id
			WHERE e.type <> 'check'
			ORDER BY e.slug`), nil
	})

	register(&mcp.Tool{Name: "get_audit_trail", Description: "Query the audit log",
		InputSchema: objSchema(prop{"entity_id", "string", "Filter by affected entity UUID"}),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		return queryRows(ctx, pool, `
			SELECT id, ts, actor_type, action, entity_id::text, method, path, correlation_id
			FROM audit_log
			WHERE ($1::text IS NULL OR entity_id::text = $1)
			ORDER BY ts DESC LIMIT 50`, nStr(args["entity_id"])), nil
	})

	register(&mcp.Tool{Name: "search_knowledge", Description: "Full-text search across documentation (PostgreSQL FTS with ts_rank ranking)",
		InputSchema: objSchema(prop{"query", "string", "Search terms"}),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		q := nStr(args["query"])
		return queryRows(ctx, pool, `
			SELECT ke.title, e.slug,
			       ts_rank(ke.search, plainto_tsquery('english', $1)) AS rank,
			       ts_headline('english', ke.content, plainto_tsquery('english', $1),
			         'MaxWords=40, MinWords=15, ShortWord=3, MaxFragments=3,
			          FragmentDelimiter=" ... "') AS snippet,
			       ke.source, ke.tags
			FROM knowledge_entities ke
			JOIN entities e ON e.id = ke.entity_id
			WHERE ke.search @@ plainto_tsquery('english', $1)
			ORDER BY rank DESC
			LIMIT 20`, q), nil
	})

	register(&mcp.Tool{Name: "get_entity_knowledge", Description: "All documents, investigations, and runbooks linked to an entity",
		InputSchema: objSchema(prop{"entity_slug", "string", "Entity slug (e.g. lxc:jellyfin, service:caddy)"}),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		slug, _ := args["entity_slug"].(string)
		return queryRows(ctx, pool, `
			SELECT ke.title, ke.source, e.type AS kind, e.slug,
			       ts_headline('english', ke.content, plainto_tsquery('english', '')) AS headline
			FROM knowledge_entities ke
			JOIN entities e ON e.id = ke.entity_id
			JOIN relationships r ON r.source_id = ke.entity_id
			JOIN entities target ON target.id = r.target_id
			WHERE target.slug = $1
			  AND r.valid_to IS NULL
			  AND r.type IN ('documents', 'about')
			UNION
			SELECT ke.title, ke.source, e.type AS kind, e.slug,
			       ts_headline('english', ke.content, plainto_tsquery('english', '')) AS headline
			FROM knowledge_entities ke
			JOIN entities e ON e.id = ke.entity_id
			JOIN relationships r ON r.source_id = ke.entity_id
			JOIN entity_types target_type ON target_type.name = (SELECT type FROM entities WHERE slug = $1)
			JOIN entities ent ON ent.type = target_type.name AND ent.slug = $1
			WHERE r.valid_to IS NULL
			  AND r.type = 'procedure-for'
			ORDER BY 1`, slug), nil
	})

	register(&mcp.Tool{Name: "upsert_knowledge", Description: "Write back what you learned so future sessions (and future you) benefit — this is how the system gets smarter over time. Use it AFTER solving a non-obvious problem, deploying a service, or discovering a gotcha: record the finding, the fix, and any caveats. Re-calling with the same title updates the existing note instead of duplicating. This is the ONLY way to persist knowledge; a chat message alone is forgotten. search_knowledge/get_entity_knowledge read it back.",
		InputSchema: objSchema(
			prop{"title", "string", "Short, specific, searchable title (e.g. 'Dragonfly memlock rlimit in unprivileged LXCs', not 'notes')."},
			prop{"content", "string", "The knowledge itself, in markdown. Be concrete: symptom, root cause, the exact fix/commands, and any caveats. Written for someone hitting this fresh."},
			prop{"about", "string", "Optional entity slug this knowledge concerns (e.g. lxc:typetype, host:strong) — links the note to that entity so get_entity_knowledge surfaces it."},
			prop{"tags", "string", "Optional comma-separated tags (e.g. 'docker,networking,gotcha')."},
			prop{"kind", "string", "One of: investigation (a finding/incident analysis — default), document (reference), runbook (a repeatable procedure)."},
		),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		return upsertKnowledge(ctx, pool, args)
	})

	register(&mcp.Tool{Name: "query_metrics", Description: "Query time-series metrics",
		InputSchema: objSchema(prop{"hours", "integer", "Look-back window in hours (default 24)"}),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		hours := int(getFloat(args, "hours", 24))
		return queryRows(ctx, pool, `
			SELECT time_bucket('1 hour', ts) AS bucket,
			       entity_id::text, metric,
			       ROUND(avg(value)::numeric, 2) AS avg,
			       ROUND(min(value)::numeric, 2) AS min,
			       ROUND(max(value)::numeric, 2) AS max
			FROM metric_samples
			WHERE ts > now() - make_interval(hours => $1)
			GROUP BY bucket, entity_id, metric
			ORDER BY bucket DESC LIMIT 100`, hours), nil
	})

	// ─── Phase 4: new tools ──────────────────────────────────────────

	register(&mcp.Tool{Name: "get_signal_history", Description: "Query open and recent signals",
		InputSchema: objSchema(
			prop{"entity_slug", "string", "Filter by target entity slug"},
			prop{"state", "string", "Filter by signal state (raised, resolved)"},
			prop{"limit", "integer", "Max rows (default 50)"}),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		limit := int(getFloat(args, "limit", 50))
		return queryRows(ctx, pool, `
			SELECT s.entity_id::text, s.kind, s.severity, s.state,
			       s.occurrence_count, e.slug AS target_slug,
			       s.first_seen_at, s.last_seen_at
			FROM signals s
			LEFT JOIN entities e ON e.id = s.target_entity_id
			WHERE ($1::text IS NULL OR e.slug = $1)
			  AND ($2::text IS NULL OR s.state = $2)
			ORDER BY s.last_seen_at DESC LIMIT $3`,
			nStr(args["entity_slug"]), nStr(args["state"]), limit), nil
	})

	register(&mcp.Tool{Name: "get_patterns", Description: "List learned action patterns",
		InputSchema: objSchema(
			prop{"status", "string", "Filter by status (hypothesized, validated, active)"},
			prop{"entity_type", "string", "Filter by applies_type"},
			prop{"action", "string", "Filter by action"}),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		return queryRows(ctx, pool, `
			SELECT p.entity_id::text, p.applies_type, p.action, p.pattern,
			       p.confidence, p.evidence_count, p.success_count, p.failure_count,
			       p.status, p.quarantined, p.version, p.last_validated_at
			FROM patterns p
			WHERE ($1::text IS NULL OR p.status = $1)
			  AND ($2::text IS NULL OR p.applies_type = $2)
			  AND ($3::text IS NULL OR p.action = $3)
			ORDER BY p.applies_type, p.action`,
			nStr(args["status"]), nStr(args["entity_type"]), nStr(args["action"])), nil
	})

	register(&mcp.Tool{Name: "get_skills", Description: "List available automation skills",
		InputSchema: objSchema(
			prop{"status", "string", "Filter by status (active, inactive, deprecated)"},
		),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		return queryRows(ctx, pool, `
			SELECT s.entity_id::text, s.version, s.name, LEFT(s.procedure::text, 300) AS procedure_preview,
			       s.applies_type, s.action, s.status, s.success_rate,
			       s.changed_by::text, s.change_reason, s.last_used_at
			FROM skills s
			WHERE ($1::text IS NULL OR s.status = $1)
			ORDER BY s.name, s.version DESC`,
			nStr(args["status"])), nil
	})

	register(&mcp.Tool{Name: "request_execution", Description: "Request a gated execution (agent-only mutation path). Supported actions: restart, systemctl, pct_exec, apt_upgrade, pct_create. pct_create provisions a NEW LXC and installs its service in one approved step — you do NOT need follow-up pct_exec calls for package installs.",
		InputSchema: objSchema(
			prop{"target", "string", "Target entity slug. For pct_create this MUST be the Proxmox HOST that will run the container (e.g. host:strong) — NOT the new LXC's name. For restart/systemctl/apt_upgrade/pct_exec use the target service/LXC slug (e.g. lxc:caddy)."},
			prop{"action", "string", "Action: restart, systemctl, pct_exec, apt_upgrade, pct_create"},
			prop{"params", "string", "For systemctl: 'enable|disable|reload'. For pct_exec: the shell command. For apt_upgrade: 'audit|upgrade'. For pct_create: a JSON object string with keys: vmid (int, required, unused id), hostname (string, required), cores (int), memory (MB int), disk_gb (int), ip (CIDR e.g. 192.168.8.50/24, or omit/\"dhcp\" — DHCP is the safe default, see below), gw (gateway ip, static only), bridge (e.g. vmbr0/vmbr1 — WHICH BRIDGE REACHES WHICH SUBNET IS DIFFERENT PER HOST, never assume vmbr0; see below), storage (default local-lvm), template (optional — omit to auto-pick newest debian on the host), privileged (bool), nesting (bool), mounts ([]string of 'src,mp=/dst'). Example: {\"vmid\":150,\"hostname\":\"typetype\",\"cores\":2,\"memory\":2048,\"disk_gb\":16,\"ip\":\"192.168.8.50/24\",\"gw\":\"192.168.8.2\",\"bridge\":\"vmbr1\",\"nesting\":true}. pct_create is ATOMIC — it ONLY creates and starts the container (no services/post_install params anymore). Once it completes you will be automatically re-invoked with the result; install packages and run setup by issuing your OWN `run` calls against the new lxc:<hostname> target, one step at a time — you'll see each step's real output and can fix exactly the one that fails, instead of one opaque multi-minute install that either fully works or fully doesn't. STATIC IP RULE: before setting ip/gw/bridge to anything other than DHCP, use list_entities/get_entity_knowledge to find an EXISTING lxc on the SAME host whose IP is in the same /28 block, and copy its exact gw+bridge — do not invent a gateway. If no such neighbor exists, prefer ip:\"dhcp\" (proven to work, gets a real routable address) over guessing; a wrong bridge/gateway pair fails a fast pre-flight ping check now (seconds, not minutes) but is still a wasted turn — better to not guess at all."},
		),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		targetSlug, _ := args["target"].(string)
		action, _ := args["action"].(string)
		params, _ := args["params"].(string)
		if targetSlug == "" || action == "" {
			return textResult("error: target and action required"), nil
		}

		var targetID uuid.UUID
		if err := pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", targetSlug).Scan(&targetID); err != nil {
			return textResult(fmt.Sprintf("target not found: %s", targetSlug)), nil
		}

		// restart, pct_exec, and systemctl (outside enable/disable) route
		// through the same classify→gate path as `run` instead of executing
		// immediately over SSH with a hardcoded risk_class='reversible_low'
		// that was never actually checked against anything. Found live
		// 2026-07-10: a chat request to "restart caddy" — the fleet's
		// reverse proxy — executed instantly with zero approval, because
		// this action bypassed the classifier entirely. classifyAndGate
		// applies the same read-only/config-mutation/destructive
		// classification and approval flow the `run` tool already uses.
		if action == "restart" || action == "pct_exec" || (action == "systemctl" && params != "enable" && params != "disable") {
			svc := strings.TrimPrefix(targetSlug, "lxc:")
			var cmd, purpose string
			switch action {
			case "restart":
				cmd = fmt.Sprintf("systemctl restart %s; sleep 1; systemctl is-active %s", svc, svc)
				purpose = "restart " + svc
			case "pct_exec":
				cmd = params
				purpose = "pct_exec (legacy) on " + targetSlug
			case "systemctl":
				cmd = fmt.Sprintf("systemctl %s %s; sleep 1; systemctl is-active %s", params, svc, svc)
				purpose = "systemctl " + params + " " + svc
			}
			return classifyAndGate(ctx, pool, agentID, targetID, targetSlug, cmd, purpose, ""), nil
		}

		// Deduplicate: if a pending execution already exists for the same
		// target+action, return the existing one instead of creating a
		// duplicate. Prevents the LLM from re-requesting the same gated
		// action in a tool-calling loop. Only blocks when a pending
		// execution exists; completed/failed ones don't block.
		if action == "systemctl" || action == "apt_upgrade" || action == "pct_create" {
			execNamePrefix := action + " on " + targetSlug
			var existingID string
			err := pool.QueryRow(ctx, `
				SELECT e.id::text FROM entities e
				JOIN executions ex ON ex.entity_id = e.id
				WHERE e.type = 'execution' AND e.name LIKE $1 AND ex.status = 'pending_approval'
				ORDER BY e.created_at DESC LIMIT 1`, execNamePrefix+"%").Scan(&existingID)
			if err == nil && existingID != "" {
				return textResult(fmt.Sprintf("%s on %s is already queued for approval — execution %s. Wait for operator approval. Do not re-request.",
					action, targetSlug, existingID)), nil
			}
		}

		id, _ := uuid.NewV7()
		correlationID := uuid.New().String()

		// Full UUID, not a truncated prefix: UUIDv7's leading bytes encode a
		// millisecond timestamp, so an 8-char prefix collides for real under
		// back-to-back requests (observed live: two `run` calls seconds
		// apart hit entities_slug_key). The full string is guaranteed unique.
		execName := action + " on " + targetSlug + " (" + id.String() + ")"
		execSlug := "exec:" + targetSlug + ":" + id.String()
		_, err := pool.Exec(ctx, `INSERT INTO entities (id, slug, type, name, attributes) VALUES ($1, $2, 'execution', $3, '{}')`,
			id, execSlug, execName)
		if err != nil {
			return textResult(fmt.Sprintf("error: failed to create execution: %v", err)), nil
		}
		pool.Exec(ctx, `INSERT INTO executions (entity_id, target_entity_id, action, risk_class, status, correlation_id, agent_id) VALUES ($1, $2, $3, 'reversible_low', 'running', $4, $5) ON CONFLICT DO NOTHING`,
			id, targetID, action+":"+params, correlationID, agentID)

		// Execute reversible actions immediately. restart/pct_exec/systemctl
		// (outside enable/disable) never reach here — they're routed through
		// classifyAndGate above, before this dedup+insert block.
		switch action {
		case "systemctl":
			// Only enable/disable reach this case now.
			svc := strings.TrimPrefix(targetSlug, "lxc:")
			pool.Exec(ctx, `UPDATE executions SET status='pending_approval', risk_class='config_mutation' WHERE entity_id=$1`, id)
			createApproval(ctx, pool, id, targetID, "systemctl", svc+":"+params, "config_mutation")
			return textResult(fmt.Sprintf("systemctl %s on %s requires approval — execution %s queued", params, svc, id)), nil

		case "apt_upgrade":
			if params == "audit" {
				host, user, err := resolveHost(ctx, pool, targetSlug)
				if err != nil {
					return textResult(fmt.Sprintf("resolve: %v", err)), nil
				}
				out, err := sshExec(ctx, host, user, "apt update -qq 2>&1 >/dev/null; apt list --upgradable 2>/dev/null | tail -n +2 | wc -l; apt list --upgradable 2>/dev/null | tail -n +2 | head -20")
				if err != nil {
					return textResult(fmt.Sprintf("apt audit error: %v", err)), nil
				}
				return textResult("apt audit:\n" + out), nil
			}
			// During an active assent window, auto-approve.
			if assentWindowActive(ctx, pool, agentID) {
				pool.Exec(ctx, `UPDATE executions SET status='pending_approval', risk_class='config_mutation' WHERE entity_id=$1`, id)
				createApproval(ctx, pool, id, targetID, "apt_upgrade", params, "config_mutation")
				// Do NOT pre-flip approvals/executions status here (that was
				// the previous, broken "autoApprove" helper). DecideApproval
				// (invoked below) is the ONE place that transitions
				// pending_approval -> approved and dispatches the real SSH
				// work — it specifically looks for status='pending_approval'
				// to find what to run. Pre-flipping the status past that
				// state meant DecideApproval's own lookup found nothing,
				// silently no-opped, and the execution sat at 'approved'
				// forever with nothing actually running. Found live: every
				// assent-window auto-approved pct_create/apt_upgrade has
				// never actually executed, via this exact bug. Calling
				// executeApprovedViaAPI directly against the untouched
				// pending_approval row makes this identical to the manual
				// Approve-button path, just without a human click.
				//
				// context.Background(), NOT ctx: ctx is scoped to this MCP
				// tool call, cancelled the instant the chat turn's HTTP
				// response completes (every normal turn) — a goroutine
				// meant to outlive the request must not inherit its context.
				go executeApprovedViaAPI(context.Background(), id, targetSlug, "apt_upgrade:"+params)
				slog.Info("mcp: apt_upgrade auto-approved via assent window", "execution_id", id)
				return textResult(fmt.Sprintf("apt_upgrade on %s auto-approved via assent window — execution %s running.", targetSlug, id)), nil
			}
			// upgrade requires approval — queue
			pool.Exec(ctx, `UPDATE executions SET status='pending_approval', risk_class='config_mutation' WHERE entity_id=$1`, id)
			createApproval(ctx, pool, id, targetID, "apt_upgrade", params, "config_mutation")
			return textResult(fmt.Sprintf("apt_upgrade on %s requires approval — execution %s queued", targetSlug, id)), nil

		case "pct_create":
			// During an active assent window, auto-approve and execute
			// instead of queuing — the operator already approved the plan.
			if assentWindowActive(ctx, pool, agentID) {
				pool.Exec(ctx, `UPDATE executions SET status='pending_approval', risk_class='config_mutation' WHERE entity_id=$1`, id)
				createApproval(ctx, pool, id, targetID, "pct_create", params, "config_mutation")
				// See the apt_upgrade case above for why there's no
				// pre-flip-status "autoApprove" step here anymore, and why
				// this uses context.Background().
				go executeApprovedViaAPI(context.Background(), id, targetSlug, "pct_create:"+params)
				slog.Info("mcp: pct_create auto-approved via assent window", "execution_id", id)
				return textResult(fmt.Sprintf("pct_create on %s auto-approved via assent window — execution %s running. The LXC is being provisioned now.", targetSlug, id)), nil
			}
			pool.Exec(ctx, `UPDATE executions SET status='pending_approval', risk_class='config_mutation' WHERE entity_id=$1`, id)
			createApproval(ctx, pool, id, targetID, "pct_create", params, "config_mutation")
			return textResult(fmt.Sprintf("pct_create on %s requires approval — execution %s queued. The LXC will be provisioned once approved.", targetSlug, id)), nil

		default:
			return textResult(fmt.Sprintf("unknown action: %s. Supported: restart, systemctl, pct_exec, apt_upgrade, pct_create", action)), nil
		}
	})

	register(&mcp.Tool{Name: "run", Description: "Run ANY shell command against any host or LXC. This is the general execution primitive — prefer it over asking the operator to run something manually, and don't wait for a matching fixed action to exist. Every command is automatically risk-classified: read-only inspection (cat, systemctl status, docker ps, journalctl, df, git status, ...) runs immediately; anything that changes state requires operator approval (granted by the operator replying \"go ahead\"/\"yes\" in chat, or via the Approve button); commands matching a destructive pattern (rm -rf, dd, mkfs, pct/qm destroy, DROP TABLE, reboot, piping curl into a shell, ...) always require approval regardless of what you declare. You cannot talk your way past the destructive check by declaring a lower risk.",
		InputSchema: objSchema(
			prop{"target", "string", "Target entity slug: host:<slug> (e.g. host:strong) or lxc:<slug> (e.g. lxc:caddy). LXC commands run via pct exec on its Proxmox host automatically."},
			prop{"command", "string", "The shell command to run. Can be a full script (multi-line, &&-chained). Runs as root."},
			prop{"purpose", "string", "One sentence: why you're running this. Shown to the operator alongside the approval — be specific, this is what they're approving."},
			prop{"declared_risk", "string", "Optional self-assessment: read_only, reversible_low, config_mutation, or destructive. This can only ESCALATE the automatic classification, never lower it — declaring a mutating command as read_only has no effect."},
		),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		targetSlug, _ := args["target"].(string)
		command, _ := args["command"].(string)
		purpose, _ := args["purpose"].(string)
		declaredRisk, _ := args["declared_risk"].(string)
		if targetSlug == "" || command == "" {
			return textResult("error: target and command are required"), nil
		}

		var targetID uuid.UUID
		if err := pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", targetSlug).Scan(&targetID); err != nil {
			return textResult(fmt.Sprintf("target not found: %s", targetSlug)), nil
		}

		return classifyAndGate(ctx, pool, agentID, targetID, targetSlug, command, purpose, declaredRisk), nil
	})

	register(&mcp.Tool{Name: "http_get", Description: "Fetch a public web page or raw file (e.g. a GitHub README/raw URL) and return sanitized text. Use this to research how to deploy a service before provisioning. HTTP/HTTPS only; body is truncated to ~16KB.",
		InputSchema: objSchema(
			prop{"url", "string", "Absolute http(s) URL to fetch"},
		),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		rawURL, _ := args["url"].(string)
		return httpGet(ctx, rawURL), nil
	})

	register(&mcp.Tool{Name: "get_execution_status", Description: "Check the status of a requested execution",
		InputSchema: objSchema(
			prop{"execution_id", "string", "Execution UUID (from request_execution output)"},
		),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		execID, _ := args["execution_id"].(string)
		if execID == "" {
			return textResult("execution_id required"), nil
		}
		eid, err := uuid.Parse(execID)
		if err != nil {
			// Try finding by exec slug prefix
			var found uuid.UUID
			err2 := pool.QueryRow(ctx, "SELECT entity_id FROM executions WHERE entity_id::text LIKE $1 LIMIT 1", execID+"%").Scan(&found)
			if err2 != nil {
				return textResult(fmt.Sprintf("execution not found: %s", execID)), nil
			}
			eid = found
		}
		return queryRows(ctx, pool, `
			SELECT e.entity_id::text, e.action, e.risk_class, e.status,
			       e.result::text, e.duration_ms, e.started_at::text,
			       e.completed_at::text, e.correlation_id
			FROM executions e
			WHERE e.entity_id = $1`, eid), nil
	})

	register(&mcp.Tool{Name: "get_trend", Description: "Get metric trends for an entity",
		InputSchema: objSchema(
			prop{"entity_id", "string", "Entity slug"},
			prop{"days", "integer", "Look-back window in days (default 7)"}),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		slug, _ := args["entity_id"].(string)
		days := int(getFloat(args, "days", 7))
		return queryRows(ctx, pool, `
			SELECT metric,
			       ROUND(avg(value)::numeric, 2) AS avg_val,
			       ROUND(stddev(value)::numeric, 2) AS std_val,
			       count(*) AS sample_count,
			       ROUND(regr_slope(value, EXTRACT(EPOCH FROM ts)::numeric)::numeric, 4) AS slope
			FROM metric_samples ms
			JOIN entities e ON e.id = ms.entity_id
			WHERE e.slug = $1 AND ts >= now() - make_interval(days => $2)
			GROUP BY metric
			ORDER BY metric`, slug, days), nil
	})

	register(&mcp.Tool{Name: "get_event_timeline", Description: "Get recent events",
		InputSchema: objSchema(
			prop{"severity", "string", "Filter by severity (info, warn, error)"},
			prop{"entity_slug", "string", "Filter by entity slug"},
			prop{"limit", "integer", "Max rows (default 50)"}),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		limit := int(getFloat(args, "limit", 50))
		return queryRows(ctx, pool, `
			SELECT ev.ts, ev.type, ev.severity, ev.source, e.slug AS entity_slug,
			       ev.data::text AS message, ev.correlation_id
			FROM events ev
			LEFT JOIN entities e ON e.id = ev.entity_id
			WHERE ($1::text IS NULL OR ev.severity = $1)
			  AND ($2::text IS NULL OR e.slug = $2)
			ORDER BY ev.ts DESC LIMIT $3`,
			nStr(args["severity"]), nStr(args["entity_slug"]), limit), nil
	})

	register(&mcp.Tool{Name: "get_agent_activity", Description: "Agent self-inspection: query agent activity log",
		InputSchema: objSchema(
			prop{"limit", "integer", "Max rows (default 50)"}),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		limit := int(getFloat(args, "limit", 50))
		return queryRows(ctx, pool, `
			SELECT id, ts, agent_id::text, session_id, activity_type, tool_name,
			       entity_id::text, left(input_summary, 200) AS input_summary,
			       left(output_summary, 200) AS output_summary,
			       duration_ms, token_count, success, correlation_id
			FROM agent_activity
			WHERE agent_id = $1
			ORDER BY ts DESC LIMIT $2`, agentID, limit), nil
	})

	// ─── Phase 5: operational MCP tools ──────────────────────────────

	register(&mcp.Tool{Name: "list_lxcs", Description: "List all LXC containers with ID, host, IP, and state",
		InputSchema: objSchema(),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		return queryRows(ctx, pool, `
			SELECT e.slug, e.name, e.attributes->>'pve_id' AS pve_id,
			       e.attributes->>'lan_ip' AS lan_ip,
			       st.health, st.last_check_at
			FROM entities e
			LEFT JOIN entity_status st ON st.entity_id = e.id
			WHERE e.type = 'lxc'
			ORDER BY (e.attributes->>'pve_id')::int`), nil
	})

	register(&mcp.Tool{Name: "ping_service", Description: "Check if a service is reachable via HTTP",
		InputSchema: objSchema(prop{"service_slug", "string", "Service entity slug"}),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		slug, _ := args["service_slug"].(string)
		if slug == "" {
			return textResult("service_slug is required"), nil
		}
		rows, err := pool.Query(ctx, `
			SELECT st.health, st.last_check_at, e.attributes->>'url' AS url
			FROM entity_status st
			JOIN entities e ON e.id = st.entity_id
			WHERE e.slug = $1`, slug)
		if err != nil {
			return textResult(fmt.Sprintf("query error: %v", err)), nil
		}
		defer rows.Close()
		if !rows.Next() {
			return textResult(fmt.Sprintf("service not found: %s", slug)), nil
		}
		var health, lastCheck, url string
		rows.Scan(&health, &lastCheck, &url)
		if url == "" {
			url = "(no URL in entity attributes)"
		}
		return textResult(fmt.Sprintf("health=%s last_check=%s url=%s", health, lastCheck, url)), nil
	})

	register(&mcp.Tool{Name: "tail_log", Description: "Get recent log lines from a service via journalctl",
		InputSchema: objSchema(
			prop{"service_slug", "string", "Service entity slug (e.g. lxc:caddy)"},
			prop{"lines", "integer", "Number of lines (default 50)"}),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		slug, _ := args["service_slug"].(string)
		n := int(getFloat(args, "lines", 50))
		if slug == "" {
			return textResult("service_slug is required"), nil
		}
		host, user, err := resolveHost(ctx, pool, slug)
		if err != nil {
			return textResult(fmt.Sprintf("resolve host: %v", err)), nil
		}
		svc := strings.TrimPrefix(slug, "lxc:")
		out, err := sshExec(ctx, host, user, fmt.Sprintf("journalctl -u %s -n %d --no-pager 2>&1 || true", svc, n))
		if err != nil {
			return textResult(fmt.Sprintf("ssh: %v", err)), nil
		}
		return textResult(out), nil
	})

	register(&mcp.Tool{Name: "get_service_status", Description: "Check systemd service status on a host",
		InputSchema: objSchema(
			prop{"service_slug", "string", "Service entity slug (e.g. lxc:caddy)"}),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		slug, _ := args["service_slug"].(string)
		if slug == "" {
			return textResult("service_slug is required"), nil
		}
		host, user, err := resolveHost(ctx, pool, slug)
		if err != nil {
			return textResult(fmt.Sprintf("resolve host: %v", err)), nil
		}
		svc := strings.TrimPrefix(slug, "lxc:")
		out, err := sshExec(ctx, host, user,
			fmt.Sprintf("systemctl is-active %s; systemctl is-enabled %s; systemctl show %s -p ActiveEnterTimestamp -p SubState 2>&1 || true", svc, svc, svc))
		if err != nil {
			return textResult(fmt.Sprintf("ssh: %v", err)), nil
		}
		return textResult(out), nil
	})

	register(&mcp.Tool{Name: "get_lxc_state", Description: "Get LXC container resource state from Proxmox host",
		InputSchema: objSchema(
			prop{"lxc_slug", "string", "LXC entity slug (e.g. lxc:caddy)"}),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		slug, _ := args["lxc_slug"].(string)
		if slug == "" {
			return textResult("lxc_slug is required"), nil
		}
		var pveID string
		err := pool.QueryRow(ctx, "SELECT attributes->>'pve_id' FROM entities WHERE slug = $1", slug).Scan(&pveID)
		if err != nil || pveID == "" {
			return textResult(fmt.Sprintf("LXC not found or missing pve_id: %s", slug)), nil
		}
		// Resolve the Proxmox host — find the host that runs this LXC
		var hostID uuid.UUID
		err = pool.QueryRow(ctx, `
			SELECT t.id FROM entities t
			JOIN relationships r ON r.source_id = t.id
			JOIN entities s ON s.id = r.target_id
			WHERE s.slug = $1 AND r.type = 'hosts' AND r.valid_to IS NULL
			LIMIT 1`, slug).Scan(&hostID)
		if err != nil {
			// Fallback: use the inventory host attribute if no relationship
			var hostSlug string
			err = pool.QueryRow(ctx, "SELECT attributes->>'host' FROM entities WHERE slug = $1", slug).Scan(&hostSlug)
			if err != nil || hostSlug == "" {
				return textResult(fmt.Sprintf("cannot resolve Proxmox host for %s", slug)), nil
			}
			var host, user string
			host, user, err = resolveHost(ctx, pool, "host:"+hostSlug)
			if err != nil {
				return textResult(fmt.Sprintf("resolve: %v", err)), nil
			}
			out, err2 := sshExec(ctx, host, user, fmt.Sprintf("pct status %s --verbose 2>&1 || true", pveID))
			if err2 != nil {
				return textResult(fmt.Sprintf("ssh: %v", err2)), nil
			}
			return textResult(out), nil
		}
		var hostSlug string
		pool.QueryRow(ctx, "SELECT slug FROM entities WHERE id = $1", hostID).Scan(&hostSlug)
		host, user, err := resolveHost(ctx, pool, hostSlug)
		if err != nil {
			return textResult(fmt.Sprintf("resolve host: %v", err)), nil
		}
		out, err := sshExec(ctx, host, user, fmt.Sprintf("pct status %s --verbose 2>&1 || true", pveID))
		if err != nil {
			return textResult(fmt.Sprintf("ssh: %v", err)), nil
		}
		return textResult(out), nil
	})

	// ─── Client introspection tools (plan: client-lifecycle Phase 3) ──

	register(&mcp.Tool{Name: "whoami", Description: "Get the current entity record, peers, and health for a host",
		InputSchema: objSchema(prop{"hostname", "string", "Hostname of the calling machine"}),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		hostname, _ := args["hostname"].(string)
		if hostname == "" {
			return textResult("error: hostname required"), nil
		}
		slug := "ws:" + hostname
		return queryRows(ctx, pool, `
			SELECT e.slug, e.type, e.name, e.state,
			       COALESCE(st.health, 'unknown') AS health,
			       COALESCE(st.last_check_at::text, '') AS last_check,
			       e.attributes->>'mesh_ip' AS mesh_ip,
			       e.attributes->>'age_pubkey' AS age_pubkey,
			       e.enrolled_at
			FROM entities e
			LEFT JOIN entity_status st ON st.entity_id = e.id
			WHERE e.slug = $1
			ORDER BY e.slug`, slug), nil
	})

	register(&mcp.Tool{Name: "explain", Description: "Compact context card for a service: type, state, health, relations, risk",
		InputSchema: objSchema(prop{"service_slug", "string", "Service entity slug (e.g. service:jellyfin, lxc:caddy)"}),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		slug, _ := args["service_slug"].(string)
		if slug == "" {
			return textResult("error: service_slug required"), nil
		}
		return queryRows(ctx, pool, `
			SELECT e.slug, e.type, e.name, e.state,
			       COALESCE(st.health, 'unknown') AS health,
			       COALESCE(st.last_check_at::text, '') AS last_check,
			       e.version, e.updated_at,
			       COALESCE(e.attributes::text, '{}') AS attrs
			FROM entities e
			LEFT JOIN entity_status st ON st.entity_id = e.id
			WHERE e.slug = $1`, slug), nil
	})

	register(&mcp.Tool{Name: "preflight", Description: "Risk classification for an action on a service",
		InputSchema: objSchema(
			prop{"service_slug", "string", "Entity slug"},
			prop{"action", "string", "Planned action (restart, deploy, destroy, etc.)"}),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		slug, _ := args["service_slug"].(string)
		action, _ := args["action"].(string)
		if slug == "" || action == "" {
			return textResult("error: service_slug and action required"), nil
		}
		return queryRows(ctx, pool, `
			SELECT e.slug, e.type, e.state,
			       CASE
			         WHEN $2 IN ('restart', 'logs', 'status') THEN 'reversible_low'
			         WHEN $2 IN ('deploy', 'upgrade', 'configure') THEN 'config_mutation'
			         WHEN $2 IN ('destroy', 'wipe', 'revoke') THEN 'destructive'
			         ELSE 'read_only'
			       END AS risk_class,
			       CASE
			         WHEN $2 IN ('read_only','reversible_low') THEN 'auto-act'
			         WHEN $2 = 'config_mutation' THEN 'operator-approval'
			         ELSE 'operator-approval+confirmation'
			       END AS approval
			FROM entities e WHERE e.slug = $1`, slug, action), nil
	})

	register(&mcp.Tool{Name: "get_change_history", Description: "Last N change-ledger entries for an entity",
		InputSchema: objSchema(
			prop{"entity_slug", "string", "Entity slug"},
			prop{"limit", "integer", "Max entries (default 20)"}),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		slug, _ := args["entity_slug"].(string)
		limit := int(getFloat(args, "limit", 20))
		return queryRows(ctx, pool, `
			SELECT al.ts AS timestamp, al.actor_type, al.actor_id::text AS actor_label,
			       al.action, al.method, al.path,
			       al.detail::text AS details
			FROM audit_log al
			JOIN entities e ON e.id = al.entity_id
			WHERE e.slug = $1
			ORDER BY al.ts DESC
			LIMIT $2`, slug, limit), nil
	})

	register(&mcp.Tool{Name: "get_state_snapshot", Description: "Last scheduler Observe-pass: fleet health, disk, drift count",
		InputSchema: objSchema(),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		return queryRows(ctx, pool, `
			SELECT e.slug, e.type, e.state,
			       COALESCE(st.health, 'unknown') AS health,
			       COALESCE(st.last_check_at::text, '') AS last_check
			FROM entities e
			LEFT JOIN entity_status st ON st.entity_id = e.id
			WHERE e.state IS NOT NULL
			   OR st.health IS NOT NULL
			ORDER BY st.health, e.slug
			LIMIT 200
		`), nil
	})

	register(&mcp.Tool{Name: "list_my_secrets", Description: "List secrets accessible to this client by public key",
		InputSchema: objSchema(prop{"caller_pubkey", "string", "Age public key of the caller (optional)"}),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		pubkey, _ := args["caller_pubkey"].(string)
		// Match entities where age_pubkey attribute contains the caller's key.
		query := `
			SELECT e.slug, e.type, e.name,
			       e.attributes->>'age_pubkey' AS age_pubkey
			FROM entities e
			WHERE e.attributes->>'age_pubkey' IS NOT NULL`
		var dbArgs []any
		if pubkey != "" {
			query += ` AND e.attributes->>'age_pubkey' = $1`
			dbArgs = append(dbArgs, pubkey)
		}
		query += ` ORDER BY e.slug LIMIT 100`
		return queryRows(ctx, pool, query, dbArgs...), nil
	})

	return s
}

// withActivityLogging wraps a tool handler to record agent_activity rows.
func withActivityLogging(pool *db.Pool, agentID uuid.UUID, toolName string, next mcp.ToolHandler) mcp.ToolHandler {
	if agentID == uuid.Nil {
		return next
	}
	return func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		start := time.Now()
		result, err := next(ctx, req)
		duration := int(time.Since(start).Milliseconds())

		// Build input summary (first 500 chars of args)
		inputSummary := ""
		if req != nil && len(req.Params.Arguments) > 0 {
			inputSummary = string(req.Params.Arguments)
		}
		if len(inputSummary) > 500 {
			inputSummary = inputSummary[:500]
		}

		// Build output summary
		outputSummary := ""
		success := err == nil
		if result != nil {
			for _, c := range result.Content {
				if tc, ok := c.(*mcp.TextContent); ok {
					outputSummary = tc.Text
					break
				}
			}
		}
		if err != nil {
			outputSummary = err.Error()
			success = false
		}
		if len(outputSummary) > 500 {
			outputSummary = outputSummary[:500]
		}

		correlationID := uuid.New().String()

		_, logErr := pool.Exec(ctx, `
			INSERT INTO agent_activity
			    (agent_id, activity_type, tool_name, input_summary, output_summary,
			     duration_ms, success, correlation_id)
			VALUES ($1, $2, $3, $4, $5, $6, $7, $8)`,
			agentID, "tool_call", toolName, inputSummary, outputSummary,
			duration, success, correlationID)
		if logErr != nil {
			slog.Warn("mcp: log agent_activity", "error", logErr)
		}

		return result, err
	}
}

// ─── Helpers ──────────────────────────────────────────────────────────

func argsMap(req *mcp.CallToolRequest) map[string]any {
	if req == nil || len(req.Params.Arguments) == 0 {
		return nil
	}
	var m map[string]any
	json.Unmarshal(req.Params.Arguments, &m)
	return m
}

func getFloat(m map[string]any, key string, def float64) float64 {
	if m == nil {
		return def
	}
	switch v := m[key].(type) {
	case float64:
		return v
	case int:
		return float64(v)
	case json.Number:
		f, err := v.Float64()
		if err != nil {
			return def
		}
		return f
	}
	return def
}

func nStr(v any) any {
	if v == nil {
		return nil
	}
	s, _ := v.(string)
	if s == "" {
		return nil
	}
	return s
}

func textResult(s string) *mcp.CallToolResult {
	return &mcp.CallToolResult{
		Content: []mcp.Content{&mcp.TextContent{Text: s}},
	}
}

// jsonOut builds a valid {"output": "..."} JSON payload for an execution's
// result column. Command output contains quotes/backslashes/control chars, so
// it must be JSON-marshaled — a hand-built string fails the ::jsonb cast and
// silently drops the status update, leaving the execution stuck.
func jsonOut(out string) []byte {
	b, _ := json.Marshal(map[string]any{"output": out})
	return b
}

// jsonErr builds a valid {"error": "..."} JSON payload for an execution's
// result column — same rationale as jsonOut, for the failure path.
func jsonErr(format string, args ...any) []byte {
	b, _ := json.Marshal(map[string]any{"error": fmt.Sprintf(format, args...)})
	return b
}

func queryEntity(ctx context.Context, pool *db.Pool, idOrSlug string) *mcp.CallToolResult {
	var id uuid.UUID
	if u, err := uuid.Parse(idOrSlug); err == nil {
		id = u
	} else {
		pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", idOrSlug).Scan(&id)
	}
	if id == uuid.Nil {
		return textResult(fmt.Sprintf("entity not found: %s", idOrSlug))
	}
	return queryRows(ctx, pool, `
		SELECT slug, type, name, state, attributes,
		       maintenance_until::text, version, created_at, updated_at
		FROM entities WHERE id = $1`, id)
}

func queryRows(ctx context.Context, pool *db.Pool, query string, args ...any) *mcp.CallToolResult {
	rows, err := pool.Query(ctx, query, args...)
	if err != nil {
		return textResult(fmt.Sprintf("error: %v", err))
	}
	defer rows.Close()

	cols := rows.FieldDescriptions()
	var items []map[string]any
	items = make([]map[string]any, 0)
	for rows.Next() {
		vals, err := rows.Values()
		if err != nil {
			continue
		}
		m := make(map[string]any)
		for i, col := range cols {
			m[string(col.Name)] = fmt.Sprintf("%v", vals[i])
		}
		items = append(items, m)
	}
	if err := rows.Err(); err != nil {
		return textResult(fmt.Sprintf("error: %v", err))
	}

	data, _ := json.MarshalIndent(items, "", "  ")
	return textResult(string(data))
}

// ─── SSH helpers ─────────────────────────────────────────────────────────

var (
	sshUser   string
	sshKey    []byte
	sshPool   = make(map[string]*ssh.Client)
	sshPoolMu sync.Mutex
)

func initSSH() {
	if sshUser == "" {
		sshUser = os.Getenv("OIKOS_SSH_USER")
		if sshUser == "" {
			sshUser = "root"
		}
	}
	keyPath := os.Getenv("OIKOS_SSH_KEY_PATH")
	if keyPath == "" {
		keyPath = "/etc/oikos/ssh_key"
	}
	if len(sshKey) == 0 {
		var err error
		sshKey, err = os.ReadFile(keyPath)
		if err != nil {
			slog.Warn("mcp ssh: cannot read key", "path", keyPath, "error", err)
		}
	}
}

// sshExecTimeout bounds how long a single remote command may run — see the
// matching constant/comment in httpapi/phase3.go. Without it, a hung remote
// command (piped install script stuck retrying DNS, etc.) blocks this
// goroutine forever with no way for the caller to ever get an answer.
const sshExecTimeout = 10 * time.Minute

func sshExec(ctx context.Context, host, user, command string) (string, error) {
	initSSH()
	if len(sshKey) == 0 {
		return "", fmt.Errorf("no SSH key available")
	}
	if user == "" {
		user = sshUser
	}

	addr := host + ":22"
	signer, err := ssh.ParsePrivateKey(sshKey)
	if err != nil {
		return "", fmt.Errorf("parse key: %w", err)
	}

	cfg := &ssh.ClientConfig{
		User:            user,
		Auth:            []ssh.AuthMethod{ssh.PublicKeys(signer)},
		HostKeyCallback: ssh.InsecureIgnoreHostKey(),
		Timeout:         10 * time.Second,
	}

	client, err := ssh.Dial("tcp", addr, cfg)
	if err != nil {
		return "", fmt.Errorf("dial %s: %w", host, err)
	}
	defer client.Close()

	session, err := client.NewSession()
	if err != nil {
		return "", fmt.Errorf("session: %w", err)
	}
	defer session.Close()

	type result struct {
		out []byte
		err error
	}
	done := make(chan result, 1)
	go func() {
		out, err := session.CombinedOutput(command)
		done <- result{out, err}
	}()

	select {
	case r := <-done:
		text := strings.TrimSpace(string(r.out))
		// A non-zero exit MUST surface as an error — matching the fix
		// applied to httpapi's sshExec (this copy still had the original
		// bug: only erroring when there was no output at all, so a command
		// that failed but printed something was silently reported as
		// success).
		if r.err != nil {
			if text != "" {
				return text, fmt.Errorf("%w: %s", r.err, text)
			}
			return text, fmt.Errorf("exec: %w", r.err)
		}
		return text, nil
	case <-time.After(sshExecTimeout):
		session.Close()
		client.Close()
		return "", fmt.Errorf("timed out after %s waiting for command to finish on %s", sshExecTimeout, host)
	case <-ctx.Done():
		session.Close()
		client.Close()
		return "", ctx.Err()
	}
}

func resolveHost(ctx context.Context, pool *db.Pool, entitySlug string) (hostIP string, sshUser string, err error) {
	var attrs string
	err = pool.QueryRow(ctx, "SELECT attributes::text FROM entities WHERE slug = $1", entitySlug).Scan(&attrs)
	if err != nil {
		return "", "", fmt.Errorf("entity not found: %s", entitySlug)
	}

	var m map[string]interface{}
	if err := json.Unmarshal([]byte(attrs), &m); err != nil {
		return "", "", fmt.Errorf("parse attributes: %w", err)
	}

	if ip, ok := m["lan_ip"].(string); ok && ip != "" {
		return ip, sshUser, nil
	}
	if mesh, ok := m["mesh"].(map[string]interface{}); ok {
		for _, proto := range []string{"netbird", "tailscale"} {
			if p, ok := mesh[proto].(map[string]interface{}); ok {
				if ip, ok := p["ip"].(string); ok && ip != "" {
					return ip, sshUser, nil
				}
			}
		}
	}
	return "", "", fmt.Errorf("no IP found for %s", entitySlug)
}

// htmlTagRe strips HTML tags for the naive text extraction in httpGet.
var htmlTagRe = regexp.MustCompile(`(?s)<(script|style)[^>]*>.*?</(script|style)>|<[^>]+>`)

// httpGet fetches a public URL and returns sanitized, size-capped text so the
// agent can read a service's README/site before provisioning. Guards: scheme
// allow-list, request timeout, 16KB body cap, and blocking of RFC1918/loopback
// hosts to avoid using the tool as an SSRF pivot into the private mesh.
func httpGet(ctx context.Context, rawURL string) *mcp.CallToolResult {
	if rawURL == "" {
		return textResult("error: url required")
	}
	u, err := url.Parse(strings.TrimSpace(rawURL))
	if err != nil || (u.Scheme != "http" && u.Scheme != "https") {
		return textResult("error: url must be an absolute http(s) URL")
	}
	if isPrivateHost(u.Hostname()) {
		return textResult("error: refusing to fetch private/loopback address")
	}

	cctx, cancel := context.WithTimeout(ctx, 15*time.Second)
	defer cancel()
	hreq, err := http.NewRequestWithContext(cctx, http.MethodGet, u.String(), nil)
	if err != nil {
		return textResult(fmt.Sprintf("error: %v", err))
	}
	hreq.Header.Set("User-Agent", "oikos-nomos/1.0 (+homelab agent)")
	hreq.Header.Set("Accept", "text/plain, text/html, application/json;q=0.9, */*;q=0.5")

	client := &http.Client{Timeout: 20 * time.Second}
	resp, err := client.Do(hreq)
	if err != nil {
		return textResult(fmt.Sprintf("error: fetch failed: %v", err))
	}
	defer resp.Body.Close()

	const cap = 256 * 1024 // read a bit extra pre-strip; final output capped below
	body, _ := io.ReadAll(io.LimitReader(resp.Body, cap))
	ct := resp.Header.Get("Content-Type")
	text := sanitizeBody(ct, string(body))
	return textResult(fmt.Sprintf("GET %s → %d %s\n\n%s", u.String(), resp.StatusCode, ct, text))
}

// sanitizeBody strips scripts/styles/tags from HTML, unescapes entities,
// collapses whitespace, and caps the result to ~16KB of readable text.
func sanitizeBody(contentType, raw string) string {
	text := raw
	if strings.Contains(contentType, "html") {
		text = htmlTagRe.ReplaceAllString(text, " ")
		text = html.UnescapeString(text)
		text = strings.Join(strings.Fields(text), " ")
	}
	if len(text) > 16*1024 {
		text = text[:16*1024] + "\n…[truncated]"
	}
	return text
}

// isPrivateHost reports whether host is loopback, link-local, or RFC1918.
func isPrivateHost(host string) bool {
	host = strings.ToLower(host)
	if host == "localhost" || strings.HasSuffix(host, ".local") || strings.HasSuffix(host, ".internal") {
		return true
	}
	ip := net.ParseIP(host)
	if ip == nil {
		return false // hostname; DNS may still resolve private — acceptable for a homelab tool
	}
	return ip.IsLoopback() || ip.IsPrivate() || ip.IsLinkLocalUnicast() || ip.IsUnspecified()
}

// resolveExecTarget resolves any target slug (host: or lxc:) to the SSH
// endpoint that will actually run the command, and a wrap function that turns
// a plain shell command into whatever must actually be sent over that SSH
// connection: identity for a host, `pct exec <pve_id> -- ...` for an LXC.
//
// The lxc.attributes.host value is stored WITHOUT a "host:" prefix (e.g.
// "strong", not "host:strong") — see pct_create's entity registration. The
// pre-existing pct_exec handler queried resolveHost with that bare value
// directly, which can never match a "host:*" slug and always fails; this
// prefixes it correctly.
func resolveExecTarget(ctx context.Context, pool *db.Pool, targetSlug string) (host, user string, wrap func(cmd string) string, err error) {
	if strings.HasPrefix(targetSlug, "host:") {
		host, user, err = resolveHost(ctx, pool, targetSlug)
		return host, user, func(cmd string) string { return cmd }, err
	}
	if strings.HasPrefix(targetSlug, "lxc:") {
		var pveID, hostAttr string
		// COALESCE the host column: many older LXC entities (seeded from
		// inventory, not provisioned by pct_create) have pve_id but no host
		// attribute at all. Scanning a SQL NULL into a plain string errors
		// the whole row, wrongly reporting "missing pve_id" even when it was
		// present — COALESCE avoids the NULL, "" is handled below.
		if qerr := pool.QueryRow(ctx, "SELECT attributes->>'pve_id', COALESCE(attributes->>'host', '') FROM entities WHERE slug = $1", targetSlug).Scan(&pveID, &hostAttr); qerr != nil || pveID == "" {
			return "", "", nil, fmt.Errorf("LXC not found or missing pve_id: %s", targetSlug)
		}
		hostSlug := hostAttr
		if hostSlug == "" {
			hostSlug = "hubris" // documented default Proxmox host when unset
		}
		if !strings.HasPrefix(hostSlug, "host:") {
			hostSlug = "host:" + hostSlug
		}
		host, user, err = resolveHost(ctx, pool, hostSlug)
		id := pveID
		return host, user, func(cmd string) string {
			b64 := base64.StdEncoding.EncodeToString([]byte(cmd))
			return fmt.Sprintf("pct exec %s -- bash -c 'echo %s | base64 -d | bash'", id, b64)
		}, err
	}
	return "", "", nil, fmt.Errorf("unsupported target %q: must be host:<slug> or lxc:<slug>", targetSlug)
}

// classifyAndGate is the shared classify→execute-or-queue path for every
// mutating command, used by both the general `run` tool and
// request_execution's restart/systemctl/pct_exec actions. Those legacy
// actions used to execute immediately over SSH with a hardcoded
// risk_class='reversible_low' that was never actually evaluated against the
// command — found live 2026-07-10 when a chat request to restart caddy (the
// fleet's reverse proxy) executed instantly with no approval at all. Routing
// every mutating path through the same classifier + approval-queue logic
// closes that gap without special-casing each caller.
func classifyAndGate(ctx context.Context, pool *db.Pool, agentID, targetID uuid.UUID, targetSlug, command, purpose, declaredRisk string) *mcp.CallToolResult {
	riskClass := policy.ClassifyCommand(command, declaredRisk)
	runParams, _ := json.Marshal(map[string]string{"command": command, "purpose": purpose})
	actionCol := "run:" + string(runParams)

	// Dedup: an identical pending command (same target, command, and
	// purpose) blocks a re-request — stops a tool-calling loop from queuing
	// the same approval repeatedly.
	var existingID string
	derr := pool.QueryRow(ctx, `
		SELECT e.id::text FROM entities e
		JOIN executions ex ON ex.entity_id = e.id
		WHERE e.type = 'execution' AND ex.target_entity_id = $1
		  AND ex.action = $2 AND ex.status = 'pending_approval'
		ORDER BY e.created_at DESC LIMIT 1`,
		targetID, actionCol).Scan(&existingID)
	if derr == nil && existingID != "" {
		return textResult(fmt.Sprintf("An identical command is already queued for approval on %s — execution %s. Wait for the operator, don't re-request.", targetSlug, existingID))
	}

	id, _ := uuid.NewV7()
	correlationID := uuid.New().String()
	execName := "run on " + targetSlug + " (" + id.String() + ")"
	execSlug := "exec:" + targetSlug + ":" + id.String()
	if _, err := pool.Exec(ctx, `INSERT INTO entities (id, slug, type, name, attributes) VALUES ($1, $2, 'execution', $3, '{}')`,
		id, execSlug, execName); err != nil {
		return textResult(fmt.Sprintf("error: failed to create execution: %v", err))
	}
	pool.Exec(ctx, `INSERT INTO executions (entity_id, target_entity_id, action, risk_class, status, correlation_id, agent_id) VALUES ($1, $2, $3, $4, 'running', $5, $6) ON CONFLICT DO NOTHING`,
		id, targetID, actionCol, riskClass, correlationID, agentID)

	if riskClass == policy.RiskReadOnly {
		host, user, wrap, rerr := resolveExecTarget(ctx, pool, targetSlug)
		if rerr != nil {
			pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, id, jsonErr("%s", rerr.Error()))
			return textResult(fmt.Sprintf("resolve target: %v", rerr))
		}
		out, xerr := sshExec(ctx, host, user, wrap(command))
		if xerr != nil {
			pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, id, jsonErr("%s: %s", xerr.Error(), out))
			return textResult(fmt.Sprintf("run on %s: ERROR %v\n%s", targetSlug, xerr, out))
		}
		pool.Exec(ctx, `UPDATE executions SET status='completed', result=$2::jsonb WHERE entity_id=$1`, id, jsonOut(out))
		return textResult(fmt.Sprintf("run on %s (read_only, auto): %s", targetSlug, out))
	}

	// Assent window: if the operator recently approved a plan in this
	// agent's chat session, config_mutation commands auto-run without
	// re-approval. This is the "approve the plan, carry it out" path — the
	// operator approved the overall direction; individual config steps
	// within the window don't each need a separate yes. Destructive
	// commands never auto-run, regardless of window.
	if riskClass == policy.RiskConfigMutation && assentWindowActive(ctx, pool, agentID) {
		host, user, wrap, rerr := resolveExecTarget(ctx, pool, targetSlug)
		if rerr != nil {
			pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, id, jsonErr("%s", rerr.Error()))
			return textResult(fmt.Sprintf("resolve target: %v", rerr))
		}
		out, xerr := sshExec(ctx, host, user, wrap(command))
		if xerr != nil {
			pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, id, jsonErr("%s: %s", xerr.Error(), out))
			return textResult(fmt.Sprintf("run on %s: ERROR %v\n%s", targetSlug, xerr, out))
		}
		pool.Exec(ctx, `UPDATE executions SET status='completed', result=$2::jsonb WHERE entity_id=$1`, id, jsonOut(out))
		slog.Info("mcp: run auto-executed via assent window", "target", targetSlug, "execution_id", id)
		return textResult(fmt.Sprintf("run on %s (config_mutation, auto via assent window): %s", targetSlug, out))
	}

	// Destructive window: a narrow, TARGET-scoped grant opened only after an
	// operator's explicit typed confirmation ("I confirm") on this same
	// target — never by loose assent. Exists for multi-step destructive
	// recovery (e.g. a failed destroy needing stop, then destroy) so the
	// operator isn't asked to re-type "I confirm" for every single command
	// against the thing they just confirmed.
	if riskClass == policy.RiskDestructive && destructiveWindowActive(ctx, pool, agentID, targetSlug) {
		host, user, wrap, rerr := resolveExecTarget(ctx, pool, targetSlug)
		if rerr != nil {
			pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, id, jsonErr("%s", rerr.Error()))
			return textResult(fmt.Sprintf("resolve target: %v", rerr))
		}
		out, xerr := sshExec(ctx, host, user, wrap(command))
		if xerr != nil {
			pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`, id, jsonErr("%s: %s", xerr.Error(), out))
			return textResult(fmt.Sprintf("run on %s: ERROR %v\n%s", targetSlug, xerr, out))
		}
		pool.Exec(ctx, `UPDATE executions SET status='completed', result=$2::jsonb WHERE entity_id=$1`, id, jsonOut(out))
		slog.Info("mcp: run auto-executed via destructive window", "target", targetSlug, "execution_id", id)
		return textResult(fmt.Sprintf("run on %s (destructive, auto via confirmed-target window): %s", targetSlug, out))
	}

	pool.Exec(ctx, `UPDATE executions SET status='pending_approval', risk_class=$2 WHERE entity_id=$1`, id, riskClass)
	createApproval(ctx, pool, id, targetID, "run", string(runParams), riskClass)
	confirmNote := ""
	if riskClass == policy.RiskDestructive {
		confirmNote = " This is classified DESTRUCTIVE — flag that clearly to the operator; it needs explicit confirmation, not just a casual \"go ahead\"."
	}
	return textResult(fmt.Sprintf("run on %s requires approval (risk: %s) — execution %s queued.%s Present the command and purpose to the operator and wait; do not re-request.",
		targetSlug, riskClass, id, confirmNote))
}

// autoApprove updates the approval + execution status in the DB to approved,
// mirroring what DecideApproval does. Returns true on success. This is used
// by the assent-window path to skip the operator-approval queue when the
// operator already approved the overall plan via chat assent.
// executeApprovedViaAPI calls the HTTP API's approval-decision endpoint to
// trigger the actual execution. The API server (phase3.executeApprovedAction)
// handles the real SSH work (pct create, apt upgrade, etc.) in a goroutine.
// We POST to the decision endpoint to reuse the exact same execution path
// as a manual Approve-button click, ensuring the audit trail is consistent.
func executeApprovedViaAPI(ctx context.Context, execID uuid.UUID, targetSlug, actionStr string) {
	apiBase := os.Getenv("OIKOS_API_BASE")
	if apiBase == "" {
		apiBase = "http://api:8090"
	}
	body, _ := json.Marshal(map[string]string{"decision": "approve"})
	client := &http.Client{Timeout: 10 * time.Second}
	req, err := http.NewRequestWithContext(ctx, http.MethodPost,
		apiBase+"/api/v1/approvals/"+execID.String()+"/decision", bytes.NewReader(body))
	if err != nil {
		slog.Error("mcp: executeApprovedViaAPI request", "error", err)
		return
	}
	req.Header.Set("Content-Type", "application/json")
	resp, err := client.Do(req)
	if err != nil {
		slog.Error("mcp: executeApprovedViaAPI call", "error", err)
		return
	}
	defer resp.Body.Close()
	if resp.StatusCode != http.StatusOK {
		// A non-200 here means the real SSH work was never dispatched — this
		// is the call that actually triggers executeApprovedAction via
		// DecideApproval. (A previous version of this comment claimed a
		// non-200 was fine because a since-removed "autoApprove" step had
		// already triggered execution via a raw DB update — it hadn't; that
		// was the bug where auto-approved pct_create/apt_upgrade never
		// actually ran. There is no other path that dispatches the work.)
		slog.Error("mcp: executeApprovedViaAPI non-200 — execution was NOT dispatched", "status", resp.StatusCode, "execution", execID)
	}
}

// assentWindowActive checks whether the operator has recently approved a plan
// in this agent's chat session. The agent sets an assent_window.agent:<uuid>
// key in autonomy_settings with an expiry timestamp when chat-assent grants
// a pending execution. While active, config_mutation commands auto-run
// without re-approval — the operator approved the overall plan, not each step.
func assentWindowActive(ctx context.Context, pool *db.Pool, agentID uuid.UUID) bool {
	if agentID == uuid.Nil {
		return false
	}
	var expiresStr string
	err := pool.QueryRow(ctx,
		"SELECT value FROM autonomy_settings WHERE key = $1",
		"assent_window.agent:"+agentID.String()).Scan(&expiresStr)
	if err != nil {
		return false
	}
	expires, err := time.Parse(time.RFC3339, expiresStr)
	if err != nil {
		return false
	}
	return time.Now().UTC().Before(expires)
}

// destructiveWindowActive reports whether targetSlug has a live, explicitly-
// confirmed destructive grant for this agent. Key format
// ("destructive_window.agent:<id>.target:<slug>") must match
// cmd/nomos/store.go's openDestructiveWindow — both processes read/write the
// same autonomy_settings row. Scoped to one target so a typed confirmation
// for destroying container A can never be read as authorizing anything
// against container B.
func destructiveWindowActive(ctx context.Context, pool *db.Pool, agentID uuid.UUID, targetSlug string) bool {
	if agentID == uuid.Nil || targetSlug == "" {
		return false
	}
	var expiresStr string
	err := pool.QueryRow(ctx,
		"SELECT value FROM autonomy_settings WHERE key = $1",
		"destructive_window.agent:"+agentID.String()+".target:"+targetSlug).Scan(&expiresStr)
	if err != nil {
		return false
	}
	expires, err := time.Parse(time.RFC3339, expiresStr)
	if err != nil {
		return false
	}
	return time.Now().UTC().Before(expires)
}

// knowledgeSlugRe strips a title down to a slug segment.
var knowledgeSlugRe = regexp.MustCompile(`[^a-z0-9]+`)

func knowledgeSlug(kind, title string) string {
	s := strings.ToLower(strings.TrimSpace(title))
	s = knowledgeSlugRe.ReplaceAllString(s, "-")
	s = strings.Trim(s, "-")
	if s == "" {
		s = "note"
	}
	if len(s) > 80 {
		s = s[:80]
	}
	return kind + ":nomos/" + s
}

// upsertKnowledge is the agent's write-back path — the missing half of the
// knowledge loop (search_knowledge/get_entity_knowledge could only read).
// Without this, everything the agent learned lived only in an ephemeral chat
// message and was lost; the system could never actually "get better." A
// knowledge doc IS an entity (type document/investigation/runbook) with a row
// in knowledge_entities; re-titling the same thing updates in place rather
// than duplicating. Optionally linked to the entity it's about so
// get_entity_knowledge surfaces it there.
func upsertKnowledge(ctx context.Context, pool *db.Pool, args map[string]any) (*mcp.CallToolResult, error) {
	title, _ := args["title"].(string)
	content, _ := args["content"].(string)
	about, _ := args["about"].(string)
	tagsRaw, _ := args["tags"].(string)
	kind, _ := args["kind"].(string)

	title = strings.TrimSpace(title)
	content = strings.TrimSpace(content)
	if title == "" || content == "" {
		return textResult("error: title and content are required"), nil
	}
	switch kind {
	case "document", "investigation", "runbook":
	case "":
		kind = "investigation"
	default:
		return textResult(fmt.Sprintf("error: kind must be document, investigation, or runbook (got %q)", kind)), nil
	}

	var tags []string
	for _, t := range strings.Split(tagsRaw, ",") {
		if t = strings.TrimSpace(t); t != "" {
			tags = append(tags, t)
		}
	}

	slug := knowledgeSlug(kind, title)

	// Upsert the knowledge-doc entity, getting its id whether it already
	// existed or we just created it.
	docID, _ := uuid.NewV7()
	err := pool.QueryRow(ctx, `
		INSERT INTO entities (id, slug, type, name, attributes)
		VALUES ($1, $2, $3, $4, '{}')
		ON CONFLICT (slug) DO UPDATE SET name = EXCLUDED.name, updated_at = now()
		RETURNING id`, docID, slug, kind, title).Scan(&docID)
	if err != nil {
		return textResult(fmt.Sprintf("error creating knowledge entity: %v", err)), nil
	}

	// Upsert the knowledge content (search column is generated, don't set it).
	_, err = pool.Exec(ctx, `
		INSERT INTO knowledge_entities (entity_id, title, content, source, tags, updated_at)
		VALUES ($1, $2, $3, 'nomos-agent', $4, now())
		ON CONFLICT (entity_id) DO UPDATE
		  SET title = EXCLUDED.title, content = EXCLUDED.content,
		      tags = EXCLUDED.tags, updated_at = now()`,
		docID, title, content, tags)
	if err != nil {
		return textResult(fmt.Sprintf("error writing knowledge: %v", err)), nil
	}

	// Link it to the entity it's about, if given and not already linked.
	linked := ""
	if about = strings.TrimSpace(about); about != "" {
		var targetID uuid.UUID
		if qerr := pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", about).Scan(&targetID); qerr == nil {
			pool.Exec(ctx, `
				INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
				SELECT $1, $2, 'documents', '{"by":"nomos"}'::jsonb, now()
				WHERE NOT EXISTS (
					SELECT 1 FROM relationships
					WHERE source_id = $1 AND target_id = $2 AND type = 'documents' AND valid_to IS NULL)`,
				docID, targetID)
			linked = " and linked to " + about
		} else {
			linked = fmt.Sprintf(" (note: entity %q not found, saved unlinked)", about)
		}
	}

	_ = observability.Event(ctx, sqlcgen.New(pool), "knowledge.upserted", &docID, "info", "mcp", "",
		map[string]any{"slug": slug, "title": title, "kind": kind})

	return textResult(fmt.Sprintf("Saved knowledge %q as %s%s. It's now searchable via search_knowledge and will surface in future sessions.", title, slug, linked)), nil
}

func createApproval(ctx context.Context, pool *db.Pool, execID, targetID uuid.UUID, action, params, riskClass string) {
	p := map[string]any{"action": action, "params": params, "execution_id": execID.String()}
	payload, _ := json.Marshal(p)
	// approvals.entity_id is PK + FK to entities(id). Reuse the execution's
	// entity (already inserted by request_execution) so the FK is satisfied —
	// a fresh UUID here had no matching entities row, so the INSERT silently
	// failed, orphaning the execution and never alerting the operator. One
	// execution maps to at most one approval, so the 1:1 identity holds.
	if _, err := pool.Exec(ctx, `
		INSERT INTO approvals (entity_id, subject_entity_id, action, risk_class,
			kind, payload, status, expires_at, created_at)
		VALUES ($1, $2, $3, $4, 'execution', $5::jsonb, 'pending',
			now() + interval '1 hour', now())`,
		execID, targetID, action, riskClass, string(payload)); err != nil {
		slog.Error("createApproval: insert approval", "error", err, "execution", execID)
		return
	}
	if _, err := pool.Exec(ctx, `UPDATE executions SET approval_id = $1 WHERE entity_id = $1`, execID); err != nil {
		slog.Error("createApproval: link approval to execution", "error", err, "execution", execID)
	}

	// Emit for SSE fan-out — the operator-facing moment: an agent-requested
	// gated action is now awaiting a decision.
	_ = observability.Event(ctx, sqlcgen.New(pool), "approval.created", &execID, "warning", "mcp", "",
		map[string]any{"action": action, "params": params, "risk_class": riskClass})
}