// 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(),
	})
	for _, t := range allTools(pool, agentID) {
		s.AddTool(t.tool, withActivityLogging(pool, agentID, t.tool.Name, t.handler))
	}
	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:, lxc:, or vm:) 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,
// `qm guest exec <pve_id> -- ...` for a VM.
//
// 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.
//
// vm: support (2026-07-18): VMs in inventory.yaml carry `pve_id` and a `host`
// attribute (or a `hosts` relationship) just like LXCs, but they're reached
// via `qm guest exec` instead of `pct exec`. Previously the agent had to
// SSH-hop via `host:hubris` to reach a VM (e.g. `ssh root@<vm_ip> '...'`),
// which broke on nested shell quoting and forced manual escaping workarounds
// — see plans/2026-07-18-session-review-three-sessions.md P1.6. A VM's
// `host` attribute is optional: if absent, fall back to looking up the
// `hosts` relationship on the VM entity, then to hubris (the documented
// default Proxmox host) — same fallback chain as LXCs.
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 := resolveProxmoxHostSlug(ctx, pool, targetSlug, hostAttr)
		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
	}
	if strings.HasPrefix(targetSlug, "vm:") {
		// VMs: same host-resolution chain as LXCs (attributes.host →
		// `hosts` relationship → hubris default), but reached via
		// `qm guest exec` instead of `pct exec`. Requires the QEMU
		// guest agent running inside the VM (the standard Proxmox
		// setup; ZimaOS/HAOS in this fleet already have it).
		var pveID, hostAttr string
		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("VM not found or missing pve_id: %s", targetSlug)
		}
		hostSlug := resolveProxmoxHostSlug(ctx, pool, targetSlug, hostAttr)
		host, user, err = resolveHost(ctx, pool, hostSlug)
		id := pveID
		return host, user, func(cmd string) string {
			b64 := base64.StdEncoding.EncodeToString([]byte(cmd))
			// `qm guest exec <id> -- /bin/bash -c '...'` returns JSON by
			// default; pipe through `jq -r .out` if available, else cat.
			// The base64 round-trip mirrors the LXC path so nested quoting
			// (the original VM-target pain point — session 55927f0a) is
			// handled identically to LXC dispatch.
			return fmt.Sprintf(
				"qm guest exec %s -- /bin/bash -c 'echo %s | base64 -d | bash' | jq -r '.out // .err // empty' 2>/dev/null || qm guest exec %s -- /bin/bash -c 'echo %s | base64 -d | bash'",
				id, b64, id, b64)
		}, err
	}
	return "", "", nil, fmt.Errorf("unsupported target %q: must be host:<slug>, lxc:<slug>, or vm:<slug>", targetSlug)
}

// resolveProxmoxHostSlug resolves the Proxmox host slug that owns a given
// LXC/VM target. Resolution order:
//  1. hostAttr if non-empty (the entity's attributes.host — stored without
//     "host:" prefix in inventory.yaml and pct_create).
//  2. the `hosts` relationship on the entity (e.g. host:hubris → vm:zimaos),
//     looked up in the relationships table — the canonical graph source.
//  3. "hubris" as a documented default Proxmox host fallback.
//
// Returns a slug with the "host:" prefix attached, ready for resolveHost.
// Extracted from the inline LXC path (2026-07-18) so the VM path shares the
// same chain — see plans/2026-07-18-session-review-three-sessions.md P1.6.
func resolveProxmoxHostSlug(ctx context.Context, pool *db.Pool, entitySlug, hostAttr string) string {
	hostSlug := strings.TrimSpace(hostAttr)
	if hostSlug == "" {
		// Fall back to the `hosts` relationship — the graph edge from
		// the Proxmox host to this LXC/VM. This is the canonical source
		// for "who owns this VM" in inventory.yaml; the `host` attribute
		// is a denormalized shortcut that not every entity has.
		var relHostSlug string
		// hosts relationship: source=host, target=lxc/vm. Look up the
		// source slug given the target.
		if err := pool.QueryRow(ctx, `
			SELECT e.slug FROM relationships r
			JOIN entities e ON e.id = r.source_id
			WHERE r.target_id = (SELECT id FROM entities WHERE slug = $1)
			  AND r.type = 'hosts' AND r.valid_to IS NULL
			LIMIT 1`, entitySlug).Scan(&relHostSlug); err == nil && relHostSlug != "" {
			hostSlug = relHostSlug
		}
	}
	if hostSlug == "" {
		hostSlug = "hubris" // documented default Proxmox host when unset
	}
	if !strings.HasPrefix(hostSlug, "host:") {
		hostSlug = "host:" + hostSlug
	}
	return hostSlug
}

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

	// P1 plan-first gate: every task must propose a plan before any `run`,
	// read-only or not. The only carve-out is a pure-DB Q&A that calls no
	// `run` at all (those never reach this code path). Without this gate the
	// SOUL.md "MANDATORY TASK FLOW" is unenforceable prose — weaker models
	// skip propose_plan and go straight to run, leaving the operator with
	// 23 individual approvals and no plan to approve (the original
	// anti-pattern the flow exists to prevent). Mirrors D.1's structural
	// refusal pattern in complete_task. sessionID == "" means a direct MCP
	// call with no nomos session (e.g. an external script) — gate is a
	// no-op there, since there's no session to hold a plan.
	if sessionID != "" && !sessionHasPlan(ctx, pool, sessionID) {
		return textResult("No plan for this session. Call set_goal then propose_plan before run — even read-only tasks require a one-step plan. A one-step plan (\"Inspect X, report, write back\") is fine for trivial questions; the gate is about ordering, not approval. Read-only commands still auto-execute once a plan exists.")
	}

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

	// P5: if this is a config_mutation command, no assent window is active,
	// and there's already a pending_approval for this session, refuse —
	// don't queue a second approval. The operator should see ONE approval
	// (the plan), approve it (which opens the assent window), and then all
	// subsequent config_mutation commands auto-run. Without this gate, the
	// agent queues N individual approvals before the operator can respond,
	// flooding the chat with approval cards — confirmed in session 20757eb9
	// (WhatsApp bridge: two approvals for what should have been one plan).
	if riskClass == policy.RiskConfigMutation && sessionID != "" && !assentWindowActive(ctx, pool, agentID, sessionID) {
		var anyPending int
		pool.QueryRow(ctx, `
			SELECT COUNT(*) FROM nomos_plan_executions pe
			JOIN executions ex ON ex.entity_id = pe.execution_id
			WHERE pe.session_id = $1 AND ex.status = 'pending_approval'`,
			sessionID).Scan(&anyPending)
		if anyPending > 0 {
			return textResult("An approval is already pending for this plan. Present the plan and its steps to the operator, then STOP and wait for their approval (\"approved\", \"yes\", \"go ahead\"). Do not call run again until the operator responds — after approval, all config_mutation commands will auto-run.")
		}
	}

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

	// 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. (The old plan-window
	// path that opened on set_goal/propose_plan was removed — it opened
	// before approval, letting config_mutation auto-run with zero operator
	// consent. The assent window, opened only on operator approval, is the
	// sole gate for config_mutation auto-run.)
	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)
	markSessionAwaitingApproval(ctx, pool, sessionID)
	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 was removed 2026-07-15: it opened on set_goal and
// propose_plan, letting config_mutation auto-run before operator approval.
// The assent window (opened only on approval in agent.go) is the sole gate
// for config_mutation auto-run now. See sessionHasPlan for the plan-existence
// check used by the P1 plan-first gate.

// sessionHasPlan reports whether this nomos session has any plan step on
// record (any generation, any status). Used by the P1 plan-first gate in
// classifyAndGate to refuse `run` before `propose_plan` has been called.
// A `replaced` step (from a prior plan generation that was superseded by a
// follow-up sub-task — see store.reopenSession) still counts: it proves the
// agent once framed a plan for this session, and the reopen path guarantees a
// fresh `propose_plan` will run before the next `run` anyway. Fails closed
// (returns true) when the query errors so a transient DB issue doesn't block
// an otherwise-valid run.
func sessionHasPlan(ctx context.Context, pool *db.Pool, sessionID string) bool {
	if sessionID == "" {
		return true // no session → no gate (direct MCP call from a script)
	}
	var count int
	if err := pool.QueryRow(ctx,
		`SELECT COUNT(*) FROM session_plan_steps WHERE session_id = $1`,
		sessionID).Scan(&count); err != nil {
		return true // fail open on DB error — don't block work over a flake
	}
	return count > 0
}

// 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 := sqlcgen.New(pool).InsertApproval(ctx, sqlcgen.InsertApprovalParams{
		EntityID:        execID,
		SubjectEntityID: &targetID,
		Action:          action,
		RiskClass:       riskClass,
		Kind:            "execution",
		Payload:         payload,
		TokenHash:       nil,
		ExpiresAt:       time.Now().Add(time.Hour),
	}); 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})
}

// markSessionAwaitingApproval flips a session to awaiting_input the moment
// one of its gated executions is queued for approval — mirrors what
// askOperator does for session_questions (cmd/nomos/store.go's askOperator),
// so a pending execution approval reads as "needs input" to both the
// frontend's Overview board (which only checks agent_sessions.status) and
// the idle-sweep safety net (staleGoalSessions, cmd/nomos/store.go, which
// already excludes awaiting_input from its stale-task sweep). Before this, a
// task blocked on a config_mutation/destructive approval just sat at
// 'executing' — indistinguishable from a task still genuinely working — so
// the idle sweep would eventually nudge it and then auto-close it with
// outcome=partial while the approval was still sitting there undecided.
// The httpapi package's DecideApproval flips the session back out once the
// approval is approved/denied/revoked (internal/httpapi/approvals.go).
//
// No-op for sessionID=="" (a direct MCP call with no nomos session) or a
// session that's already terminal/already awaiting_input — the status IN
// guard makes this safe to call unconditionally from classifyAndGate.
func markSessionAwaitingApproval(ctx context.Context, pool *db.Pool, sessionID string) {
	if sessionID == "" || sessionID == "ephemeral" {
		return
	}
	tag, err := pool.Exec(ctx, `
		UPDATE agent_sessions SET status = 'awaiting_input', last_active_at = now()
		WHERE id = $1 AND status IN ('active', 'planning', 'executing')`, sessionID)
	if err != nil || tag.RowsAffected() == 0 {
		return
	}
	_ = observability.Event(ctx, sqlcgen.New(pool), "task.status", sessionTaskEntity(ctx, pool, sessionID),
		"info", "nomos", sessionID, map[string]any{"status": "awaiting_input", "reason": "execution_pending_approval"})
}

// sessionTaskEntity resolves a session's own task-entity id, for anchoring
// events to the right node in the graph — mirrors cmd/nomos/store.go's
// (unexported) taskEntityPtr; duplicated here since that's a different
// package's private method.
func sessionTaskEntity(ctx context.Context, pool *db.Pool, sessionID string) *uuid.UUID {
	var id uuid.UUID
	if err := pool.QueryRow(ctx, `SELECT entity_id FROM agent_sessions WHERE id = $1`, sessionID).Scan(&id); err != nil || id == uuid.Nil {
		return nil
	}
	return &id
}

// inspectPathAcrossTargets is the bulk fact-gathering helper behind the
// inspect_path MCP tool (plans/2026-07-18-session-review-three-sessions.md
// P1.5). For each target slug, it runs a single read-only shell command
// producing mount/df/ls/stat output for the given path, and returns the
// results as a map keyed by target slug.
//
// Why this exists: sessions 1e9c7691 and 55927f0a each spent ~15 `run`
// calls gathering identical facts (`mount | grep`, `df`, `ls -la`, `stat`)
// across hosts and LXCs to trace where a path lives, who mounts it, and
// what permissions it has. One call here replaces that fan-out. All
// commands are read-only — the tool bypasses classifyAndGate and runs
// directly via sshExec against resolveExecTarget's host/wrap. Failures
// (unresolvable target, SSH error) are reported per-target in the result
// map, not as a single tool-level error, so one bad target doesn't lose
// the others.
//
// The per-target command is intentionally compact: one combined shell
// invocation that prints mount source/dest, df, ls -la of the path's
// parent + the path itself, and stat. Output is truncated to 4KB per
// target to keep the total result reasonable for an 8-target call.
func inspectPathAcrossTargets(ctx context.Context, pool *db.Pool, path string, targets []string) map[string]any {
	results := make(map[string]any, len(targets))
	path = strings.TrimSpace(path)

	var wg sync.WaitGroup
	var mu sync.Mutex
	wg.Add(len(targets))

	for _, tgt := range targets {
		go func(target string) {
			defer wg.Done()
			entry := inspectOneTarget(ctx, pool, path, target)
			mu.Lock()
			results[target] = entry
			mu.Unlock()
		}(tgt)
	}
	wg.Wait()
	return results
}

// inspectOneTarget runs the read-only inspection for one target. Returns a
// map with keys: "ok" (bool), "output" (string, on success), "error"
// (string, on failure). Kept small so the JSON shape is stable across the
// parallel-call path.
func inspectOneTarget(ctx context.Context, pool *db.Pool, path, target string) map[string]any {
	host, user, wrap, rerr := resolveExecTarget(ctx, pool, target)
	if rerr != nil {
		return map[string]any{"ok": false, "error": fmt.Sprintf("resolve target: %v", rerr)}
	}
	// One shell invocation, four sections, each guarded by `2>&1 || true`
	// so a missing path doesn't kill the rest. Stat with -c gives a
	// stable machine-readable line for ownership/perms; ls -la gives the
	// human-readable listing of the path and its parent (so we can see
	// both "what's in here" and "how the parent is laid out" — useful for
	// NFS-root-vs-subdir permission mismatches, the exact issue in
	// session 1e9c7691).
	cmd := fmt.Sprintf(
		`echo "=== mount ==="; mount 2>/dev/null | grep -- "%[1]s" || echo "(not a mount point)";
echo "=== df ==="; df -h "%[1]s" 2>&1 || true;
echo "=== stat ==="; stat -c '%%a %%U:%%G (size=%%s, type=%%F)' "%[1]s" 2>&1 || true;
echo "=== ls -la path ==="; ls -la "%[1]s" 2>&1 | head -40 || true;
echo "=== ls -la parent ==="; ls -la "$(dirname "%[1]s")" 2>&1 | head -20 || true`,
		path)
	out, xerr := sshExec(ctx, host, user, wrap(cmd))
	if xerr != nil {
		return map[string]any{"ok": false, "error": fmt.Sprintf("ssh: %v: %s", xerr, out)}
	}
	// Truncate per-target output to keep an 8-target call's total under
	// ~32KB. 4KB per target is enough for the head -40/head -20 listings
	// above; if a directory is enormous, the truncation keeps the result
	// usable without flooding the model's context.
	const maxPerTarget = 4096
	if len(out) > maxPerTarget {
		out = out[:maxPerTarget] + fmt.Sprintf("\n...truncated (%d bytes total)", len(out))
	}
	return map[string]any{"ok": true, "output": out}
}