oikos/internal/mcp/server.go

// 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 annotateJSONResult(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), "entity_table"), 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). Returns a short snippet per hit, not the full note — call get_knowledge_content with the returned slug to read the whole thing.",
		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 annotateJSONResult(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), "knowledge_results"), nil
	})

	register(&mcp.Tool{Name: "get_entity_knowledge", Description: "All documents, investigations, and runbooks linked to an entity. Returns a headline per note, not the full text — call get_knowledge_content with the returned slug to read the whole thing.",
		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 annotateJSONResult(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), "knowledge_results"), nil
	})

	register(&mcp.Tool{Name: "get_knowledge_content", Description: "Full markdown body of one document/investigation/runbook, by its own entity slug. search_knowledge and get_entity_knowledge only return short snippets/headlines — once you know which note you need (from either of those, or because you already know its slug), call this to read the whole thing before acting on it.",
		InputSchema: objSchema(prop{"slug", "string", "The knowledge entity's own slug (e.g. document:containers/101-jellyfin, runbook:client-enrollment) — not the slug of an entity it's about."}),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		slug, _ := args["slug"].(string)
		return queryRows(ctx, pool, `
			SELECT ke.title, e.slug, e.type AS kind, ke.content, ke.source, ke.tags, ke.updated_at::text
			FROM knowledge_entities ke
			JOIN entities e ON e.id = ke.entity_id
			WHERE e.slug = $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 find it, get_knowledge_content reads the full body 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(s) this knowledge concerns. Pass a single slug (e.g. 'lxc:nfs-export') or a JSON array of slugs (e.g. '[\"lxc:nfs-export\", \"lxc:gitea\"]') to link to multiple entities. get_entity_knowledge surfaces it for each."},
			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: "update_entity_attributes", Description: "Merge new/changed attributes into an entity — the OTHER half of avoiding knowledge-base drift (upsert_knowledge records what you learned; this keeps the entity's own facts current). Use it when you discover something concrete about an entity's actual state that the graph doesn't reflect yet: a new IP, a version number, a config value, a discovered port — anything a FUTURE task would otherwise have to rediscover from scratch. Does NOT require approval (this updates the knowledge graph, not the live infrastructure). Merges shallowly — existing keys not mentioned are kept; keys you pass overwrite.",
		InputSchema: objSchema(
			prop{"slug", "string", "Entity slug to update (e.g. lxc:typetype, host:strong)."},
			prop{"attributes", "string", "JSON object string of attributes to merge in, e.g. {\"lan_ip\":\"192.168.8.50\",\"os\":\"debian-12\"}."},
		),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		slug, _ := args["slug"].(string)
		attrsStr, _ := args["attributes"].(string)
		if slug == "" || attrsStr == "" {
			return textResult("error: slug and attributes are required"), nil
		}
		var attrs map[string]any
		if err := json.Unmarshal([]byte(attrsStr), &attrs); err != nil {
			return textResult(fmt.Sprintf("error: attributes is not valid JSON: %v", err)), nil
		}
		attrsJSON, _ := json.Marshal(attrs)
		ct, err := pool.Exec(ctx, `
			UPDATE entities SET attributes = attributes || $2::jsonb, updated_at = now()
			WHERE slug = $1`, slug, string(attrsJSON))
		if err != nil {
			return textResult(fmt.Sprintf("error updating %s: %v", slug, err)), nil
		}
		if ct.RowsAffected() == 0 {
			return textResult(fmt.Sprintf("error: entity %q not found", slug)), nil
		}
		return textResult(fmt.Sprintf("Updated %s with %d attribute(s).", slug, len(attrs))), nil
	})

	register(&mcp.Tool{Name: "create_relationship", Description: "Record a relationship you discovered between two entities — the graph-structure half of keeping the knowledge base current (alongside update_entity_attributes and upsert_knowledge). Use it when you learn that one entity depends on, hosts, routes to, etc. another, and that edge isn't in the graph yet. type must be an existing relationship type (see get_relations output on similar entities for examples: hosts, provides, depends-on, configured-by, about, documents, ...). Idempotent — re-calling the same source/target/type is a no-op. Does NOT require approval.",
		InputSchema: objSchema(
			prop{"source", "string", "Source entity slug."},
			prop{"target", "string", "Target entity slug."},
			prop{"type", "string", "Relationship type name (must already exist in the ontology)."},
		),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		args := argsMap(req)
		source, _ := args["source"].(string)
		target, _ := args["target"].(string)
		relType, _ := args["type"].(string)
		if source == "" || target == "" || relType == "" {
			return textResult("error: source, target, and type are required"), nil
		}
		var sourceID, targetID uuid.UUID
		if err := pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", source).Scan(&sourceID); err != nil {
			return textResult(fmt.Sprintf("error: source entity %q not found", source)), nil
		}
		if err := pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", target).Scan(&targetID); err != nil {
			return textResult(fmt.Sprintf("error: target entity %q not found", target)), nil
		}
		_, err := pool.Exec(ctx, `
			INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
			SELECT $1, $2, $3, '{"by":"nomos"}'::jsonb, now()
			WHERE NOT EXISTS (
				SELECT 1 FROM relationships
				WHERE source_id = $1 AND target_id = $2 AND type = $3 AND valid_to IS NULL)`,
			sourceID, targetID, relType)
		if err != nil {
			return textResult(fmt.Sprintf("error creating relationship: %v (is %q a valid relationship type?)", err, relType)), nil
		}
		return textResult(fmt.Sprintf("Recorded: %s —%s→ %s", source, relType, target)), nil
	})

	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 annotateJSONResult(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), "metric_chart"), 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
	})

	// ── request_execution (legacy fixed enum) retired 2026-07-14 ──
	// All mutations now route through `run`. The handler functions
	// (runRexecRestart, runRexecSystemctl, etc.) are kept as reference
	// for future runbook extraction — especially pct_create DNS/VMID logic.
	// DO NOT re-register this tool. See plans/2026-07-10-general-gated-execution.md.

	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)
		sessionID, _ := args["_session_id"].(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, sessionID), 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 annotateJSONResult(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), "change_log"), nil
	})

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

	register(&mcp.Tool{Name: "list_lxcs", Description: "List all LXC containers with ID, host, IP, state, and last-audited hint. Pass state=\"active\" to exclude destroyed/deprecated containers. The last_audited_at column shows the most recent knowledge entry (investigation or document tagged audit/update) linked via an 'about' edge — use it to skip re-running `run` against LXCs that were already audited recently.",
		InputSchema: objSchema(
			prop{"state", "string", "Optional: filter by entity state (active, destroyed, …)"},
		),
	}, func(ctx context.Context, req *mcp.CallToolRequest) (*mcp.CallToolResult, error) {
		state, _ := argsMap(req)["state"].(string)
		var statePtr *string
		if state != "" {
			statePtr = &state
		}
		return annotateJSONResult(queryRows(ctx, pool, `
			SELECT e.slug, e.name, e.attributes->>'pve_id' AS pve_id,
			       e.attributes->>'lan_ip' AS lan_ip,
			       e.state,
			       st.health, st.last_check_at,
			       (SELECT MAX(k.created_at)
			        FROM relationships r
			        JOIN knowledge_entities k ON k.entity_id = r.source_id
			        WHERE r.target_id = e.id
			          AND r.type = 'about'
			          AND r.valid_to IS NULL
			          AND (k.tags @> ARRAY['audit']::text[]
			            OR k.tags @> ARRAY['update']::text[]
			            OR k.title ILIKE '%audit%'
			            OR k.title ILIKE '%update%')
			       ) AS last_audited_at
			FROM entities e
			LEFT JOIN entity_status st ON st.entity_id = e.id
			WHERE e.type = 'lxc'
			  AND ($1::text IS NULL OR e.state = $1)
			ORDER BY CASE WHEN e.state = 'active' THEN 0 ELSE 1 END,
			         (e.attributes->>'pve_id')::int`, statePtr), "lxc_list"), 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 annotateJSONResult(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), "entity_card"), 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 annotateJSONResult(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), "entity_card"), 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 annotateJSONResult(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), "change_log"), 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 annotateJSONResult(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
		`), "fleet_snapshot"), 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()

		entityID := resolveArgEntityID(ctx, pool, argsMap(req))
		var entityIDArg any
		if entityID != uuid.Nil {
			entityIDArg = entityID
		}

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

		return result, err
	}
}

// entityArgKeys lists tool-argument keys, in priority order, that commonly
// carry the target entity's slug or UUID. Tool input schemas aren't
// consistent about naming this (target, entity_slug, slug, service_slug,
// lxc_slug, entity_id all appear across server.go's tool registrations), so
// this is a best-effort lookup used to tag agent_activity rows with the
// entity a tool call acted on.
var entityArgKeys = []string{
	"target", "entity_slug", "slug", "slug_or_id",
	"service_slug", "lxc_slug", "entity_id", "about",
}

// resolveArgEntityID best-effort resolves the entity a tool call acted on
// from its arguments, trying entityArgKeys in order. Returns uuid.Nil if no
// key is present or none resolves to a known entity.
func resolveArgEntityID(ctx context.Context, pool *db.Pool, args map[string]any) uuid.UUID {
	for _, key := range entityArgKeys {
		v, _ := args[key].(string)
		if v == "" {
			continue
		}
		if u, err := uuid.Parse(v); err == nil {
			return u
		}
		var id uuid.UUID
		if err := pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", v).Scan(&id); err == nil {
			return id
		}
	}
	return uuid.Nil
}

// ─── 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))
}

func annotateJSONResult(result *mcp.CallToolResult, rendererID string) *mcp.CallToolResult {
	if len(result.Content) == 0 {
		return result
	}
	tc, ok := result.Content[0].(*mcp.TextContent)
	if !ok || tc.Text == "" {
		return result
	}
	var items []map[string]any
	if err := json.Unmarshal([]byte(tc.Text), &items); err != nil {
		return result
	}
	wrapper := map[string]any{
		"__renderer": rendererID,
		"data":       items,
	}
	data, _ := json.MarshalIndent(wrapper, "", "  ")
	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() {
		// Recovers a panic in CombinedOutput (SSH library internals, rare but
		// not impossible) and reports it as a failed command instead of
		// crashing the whole api process — every gated action runs through
		// this function, so an unrecovered panic here would take down every
		// concurrently-running task's execution, not just this one. Without
		// this, a panic would ALSO silently degrade to "wait out the full
		// timeout" (done never receives, the select below falls through to
		// its time.After case) rather than crashing outright — recovering
		// and sending an immediate result is strictly better: the caller
		// finds out now, not after sshExecTimeout.
		defer func() {
			if r := recover(); r != nil {
				done <- result{nil, fmt.Errorf("panic in ssh exec: %v", r)}
			}
		}()
		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, sessionID 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)
	pool.Exec(ctx, `
		INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
		SELECT $1, $2, 'targets', '{"by":"nomos"}'::jsonb, now()
		WHERE NOT EXISTS (
			SELECT 1 FROM relationships
			WHERE source_id = $1 AND target_id = $2 AND type = 'targets' AND valid_to IS NULL)`,
		id, targetID)
	if sessionID != "" {
		pool.Exec(ctx, `
			INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
			SELECT t.id, $1, 'involves', '{"by":"nomos"}'::jsonb, now()
			FROM entities t WHERE t.slug = $2
			AND NOT EXISTS (
				SELECT 1 FROM relationships
				WHERE source_id = t.id AND target_id = $1 AND type = 'involves' AND valid_to IS NULL)`,
			id, "task:"+sessionID)
	}

	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))
	}

	// Plan window: if a plan was proposed (and possibly approved), this
	// command is part of an in-flight plan. The plan IS the approval —
	// config_mutation commands within an active plan auto-run without
	// per-action approval. Created by propose_plan, checked by planWindowActive.
	if riskClass == policy.RiskConfigMutation && sessionID != "" && planWindowActive(ctx, pool, sessionID) {
		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 plan window", "target", targetSlug, "execution_id", id)
		return textResult(fmt.Sprintf("run on %s (config_mutation, auto via plan): %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, sessionID) {
		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, sessionID) {
		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)
	}
}

// planWindowActive reports whether a plan has been proposed (or approved)
// for this session. Created by propose_plan, the window allows config_mutation
// commands to auto-execute without per-action approval — the plan IS the
// approval. Plan-approve-once policy (2026-07-14).
func planWindowActive(ctx context.Context, pool *db.Pool, sessionID string) bool {
	if sessionID == "" {
		return false
	}
	var val string
	err := pool.QueryRow(ctx,
		"SELECT value FROM autonomy_settings WHERE key = $1",
		"nomos:plan:"+sessionID).Scan(&val)
	return err == nil && val == "active"
}

// assentWindowActive checks whether the operator has recently approved a plan
// in THIS TASK's chat session. The agent sets an
// assent_window.agent:<uuid>.session:<id> 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. Scoped by session, not
// just agent: with one agent:nomos entity serving every concurrent task, an
// agent-only key would let approving Task A's plan silently auto-run
// unapproved actions from a concurrently-running Task B. sessionID comes
// from the `_session_id` nomos injects into every tool call's wire args
// (never part of any tool's declared InputSchema, so the model never
// supplies or sees it) — see cmd/nomos/agent.go's tool dispatch loop.
func assentWindowActive(ctx context.Context, pool *db.Pool, agentID uuid.UUID, sessionID string) bool {
	if agentID == uuid.Nil || sessionID == "" {
		return false // fail closed: no session to scope to means no window
	}
	var expiresStr string
	err := pool.QueryRow(ctx,
		"SELECT value FROM autonomy_settings WHERE key = $1",
		"assent_window.agent:"+agentID.String()+".session:"+sessionID).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 WITHIN THIS SESSION/TASK. Key
// format ("destructive_window.agent:<id>.target:<slug>.session:<id>") must
// match cmd/nomos/store.go's openDestructiveWindow — both processes
// read/write the same autonomy_settings row. Scoped to one target AND one
// session so a typed confirmation for destroying container A in task X can
// never be read as authorizing anything against container A from a
// different, concurrently-running task Y.
func destructiveWindowActive(ctx context.Context, pool *db.Pool, agentID uuid.UUID, targetSlug, sessionID string) bool {
	if agentID == uuid.Nil || targetSlug == "" || sessionID == "" {
		return false
	}
	var expiresStr string
	err := pool.QueryRow(ctx,
		"SELECT value FROM autonomy_settings WHERE key = $1",
		"destructive_window.agent:"+agentID.String()+".target:"+targetSlug+".session:"+sessionID).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)
	tagsRaw, _ := args["tags"].(string)
	kind, _ := args["kind"].(string)

	// Normalize about: accept a single string slug or an array of slugs.
	var aboutSlugs []string
	switch v := args["about"].(type) {
	case string:
		if s := strings.TrimSpace(v); s != "" {
			aboutSlugs = []string{s}
		}
	case []interface{}:
		for _, item := range v {
			if s, ok := item.(string); ok {
				if s = strings.TrimSpace(s); s != "" {
					aboutSlugs = append(aboutSlugs, s)
				}
			}
		}
	}

	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(s) it's about, if given and not already linked.
	linked := ""
	if len(aboutSlugs) > 0 {
		var linkedSlugs []string
		for _, slug := range aboutSlugs {
			var targetID uuid.UUID
			if qerr := pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", slug).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)
				linkedSlugs = append(linkedSlugs, slug)
			}
		}
		if len(linkedSlugs) == 1 {
			linked = " and linked to " + linkedSlugs[0]
		} else if len(linkedSlugs) > 1 {
			linked = fmt.Sprintf(" and linked to %d entities", len(linkedSlugs))
		}
	}

	_ = 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})
}