// 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/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/execlog"
	"github.com/dtoro/oikos/internal/observability"
	"github.com/dtoro/oikos/internal/policy"
	"github.com/dtoro/oikos/internal/remote"
	"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))
	}

	// Resource templates: let MCP clients browse and attach entities,
	// knowledge entries, and executions as conversation resources.
	s.AddResourceTemplate(&mcp.ResourceTemplate{
		URITemplate: "oikos://entity/{slug}",
		Name:        "Entity",
		Description: "Oikos entity by slug (e.g. host:hubris, lxc:jellyfin)",
		MIMEType:    "application/json",
	}, resourceHandler(pool, func(ctx context.Context, matches map[string]string) (string, error) {
		slug := matches["slug"]
		var id uuid.UUID
		if u, err := uuid.Parse(slug); err == nil {
			id = u
		} else {
			pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", slug).Scan(&id)
		}
		if id == uuid.Nil {
			return "", fmt.Errorf("entity not found: %s", slug)
		}
		result := queryEntity(ctx, pool, slug)
		return result.Content[0].(*mcp.TextContent).Text, nil
	}))

	s.AddResourceTemplate(&mcp.ResourceTemplate{
		URITemplate: "oikos://knowledge/{id}",
		Name:        "Knowledge",
		Description: "Knowledge entry by entity slug or UUID",
		MIMEType:    "application/json",
	}, resourceHandler(pool, func(ctx context.Context, matches map[string]string) (string, error) {
		idOrSlug := matches["id"]
		var entityID uuid.UUID
		if u, err := uuid.Parse(idOrSlug); err == nil {
			entityID = u
		} else {
			pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", idOrSlug).Scan(&entityID)
		}
		if entityID == uuid.Nil {
			return "", fmt.Errorf("knowledge not found: %s", idOrSlug)
		}
		result := queryRows(ctx, pool, `
			SELECT ke.title, ke.content, ke.tags::text, e.slug, e.type AS kind,
			       ke.updated_at::text
			FROM knowledge_entities ke
			JOIN entities e ON e.id = ke.entity_id
			WHERE ke.entity_id = $1`, entityID)
		return result.Content[0].(*mcp.TextContent).Text, nil
	}))

	s.AddResourceTemplate(&mcp.ResourceTemplate{
		URITemplate: "oikos://execution/{id}",
		Name:        "Execution",
		Description: "Execution by UUID (returns status, result, timing)",
		MIMEType:    "application/json",
	}, resourceHandler(pool, func(ctx context.Context, matches map[string]string) (string, error) {
		result := queryRows(ctx, pool, `
			SELECT e.entity_id, te.slug AS target, e.action, e.risk_class,
			       e.status, e.result::text, e.duration_ms,
			       e.started_at::text, e.completed_at::text
			FROM executions e
			JOIN entities te ON te.id = e.target_entity_id
			WHERE e.entity_id = $1`, matches["id"])
		return result.Content[0].(*mcp.TextContent).Text, nil
	}))

	return s
}

// resourceHandler adapts a simple func(ctx, params) → (string, error) into
// an MCP ResourceHandler, reading the URI matched by a ResourceTemplate.
func resourceHandler(pool *db.Pool, fn func(ctx context.Context, matches map[string]string) (string, error)) mcp.ResourceHandler {
	return func(ctx context.Context, req *mcp.ReadResourceRequest) (*mcp.ReadResourceResult, error) {
		uri := req.Params.URI
		matches := matchURITemplate(uri)
		if matches == nil {
			return nil, mcp.ResourceNotFoundError(uri)
		}

		text, err := fn(ctx, matches)
		if err != nil {
			return nil, mcp.ResourceNotFoundError(uri)
		}

		result, err := json.MarshalIndent(json.RawMessage(text), "", "  ")
		if err != nil {
			result = []byte(text)
		}

		return &mcp.ReadResourceResult{
			Contents: []*mcp.ResourceContents{{
				URI:      uri,
				MIMEType: "application/json",
				Text:     string(result),
			}},
		}, nil
	}
}

// matchURITemplate extracts parameters from a URI that matches one of the
// oikos:// resource templates. Returns nil if the URI doesn't match.
func matchURITemplate(uri string) map[string]string {
	// oikos://entity/{slug}
	if rest, ok := strings.CutPrefix(uri, "oikos://entity/"); ok && rest != "" {
		return map[string]string{"slug": rest}
	}
	// oikos://knowledge/{id}
	if rest, ok := strings.CutPrefix(uri, "oikos://knowledge/"); ok && rest != "" {
		return map[string]string{"id": rest}
	}
	// oikos://execution/{id}
	if rest, ok := strings.CutPrefix(uri, "oikos://execution/"); ok && rest != "" {
		return map[string]string{"id": rest}
	}
	return nil
}

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

// streamWriter buffers everything it is given while forwarding each write to a
// sink. Assigning one to session.Stdout and another (sharing the same buffer)
// to session.Stderr reproduces CombinedOutput's interleaving exactly, in the
// order the remote end actually produced it — which reading from StdoutPipe
// and StderrPipe separately would not guarantee.
type streamWriter struct {
	mu     *sync.Mutex
	buf    *bytes.Buffer
	stream string
	sink   execlog.Sink
}

func (w *streamWriter) Write(p []byte) (int, error) {
	w.mu.Lock()
	w.buf.Write(p)
	w.mu.Unlock()
	if w.sink != nil {
		// Copy: the ssh library reuses p after Write returns, and the sink
		// hands the bytes to a DB call that may outlive this frame.
		w.sink(w.stream, append([]byte(nil), p...))
	}
	return len(p), nil
}

func sshExec(ctx context.Context, host, user, command string) (string, error) {
	return sshExecStream(ctx, host, user, command, nil)
}

// sshExecStream runs a command and reports its combined output, forwarding
// each chunk to sink as it arrives. A nil sink behaves exactly as before.
func sshExecStream(ctx context.Context, host, user, command string, sink execlog.Sink) (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()

	var (
		mu  sync.Mutex
		buf bytes.Buffer
	)
	session.Stdout = &streamWriter{mu: &mu, buf: &buf, stream: "stdout", sink: sink}
	session.Stderr = &streamWriter{mu: &mu, buf: &buf, stream: "stderr", sink: sink}

	// collected returns whatever output has arrived so far. Callable while the
	// command is still running, which is what makes partial output on timeout
	// possible.
	collected := func() string {
		mu.Lock()
		defer mu.Unlock()
		return strings.TrimSpace(buf.String())
	}

	done := make(chan error, 1)
	go func() {
		// Recovers a panic in the 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 <- fmt.Errorf("panic in ssh exec: %v", r)
			}
		}()
		// Run rather than CombinedOutput so the assigned writers are used;
		// Run returns only after both streams have been fully drained.
		done <- session.Run(command)
	}()

	select {
	case err := <-done:
		text := collected()
		// 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 err != nil {
			if text != "" {
				return text, fmt.Errorf("%w: %s", err, text)
			}
			return text, fmt.Errorf("exec: %w", err)
		}
		return text, nil
	case <-time.After(sshExecTimeout):
		session.Close()
		client.Close()
		// Return what the command managed to print before it hung. This used
		// to return "", discarding everything — so a hung command, the case
		// where the output matters most, was the one case that left no trace.
		return collected(), fmt.Errorf("timed out after %s waiting for command to finish on %s", sshExecTimeout, host)
	case <-ctx.Done():
		session.Close()
		client.Close()
		return collected(), ctx.Err()
	}
}

// resolveHost resolves a host:<slug> to its reachable IP and SSH user. A thin
// wrapper over the shared resolver (internal/remote), kept so slug-based
// callers keep working; the shared resolver also prefers public_ipv4 over
// mesh and honors a per-entity ssh.user.
func resolveHost(ctx context.Context, pool *db.Pool, entitySlug string) (hostIP string, sshUserOut string, err error) {
	return remote.ResolveHost(ctx, pool, entitySlug, sshUser)
}

// 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.
//
// Delegates to the shared resolver (internal/remote), the single path used by
// both the MCP `run` tool and the scheduler's checks. The historical notes
// (host attr without prefix, vm host-resolution chain, nested-quoting
// handling via base64) all still hold — they now live in remote.guestWrap.
func resolveExecTarget(ctx context.Context, pool *db.Pool, targetSlug string) (host, user string, wrap func(cmd string) string, err error) {
	et, err := remote.ResolveExecTarget(ctx, pool, targetSlug, sshUser)
	if err != nil {
		return "", "", nil, err
	}
	return et.Host, et.User, et.Wrap, nil
}

// resolveProxmoxHostSlug resolves the Proxmox host slug that owns a given
// LXC/VM target (see internal/remote.ResolveProxmoxHostSlug for the chain).
// This slug-based wrapper looks up the entity id so slug callers keep working;
// the shared resolver takes an id directly.
func resolveProxmoxHostSlug(ctx context.Context, pool *db.Pool, entitySlug, hostAttr string) string {
	var id uuid.UUID
	if err := pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", entitySlug).Scan(&id); err != nil {
		id = uuid.Nil
	}
	return remote.ResolveProxmoxHostSlug(ctx, pool, id, hostAttr)
}

// 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.
// autoRun resolves a target, runs the command, and finalizes the execution
// with full timing.
//
// The three auto-run windows (read-only, assent, destructive) each carried
// their own copy of this logic, and none of them wrote duration_ms, started_at
// or completed_at — so every auto-run execution landed in the ledger with no
// timing at all, and the Ops "Duration" column was empty for exactly the
// executions that run most often.
func autoRun(ctx context.Context, pool *db.Pool, id uuid.UUID, targetSlug, command string) (string, error) {
	startedAt := time.Now()
	if _, err := pool.Exec(ctx,
		`UPDATE executions SET status='running', started_at=$2 WHERE entity_id=$1`,
		id, startedAt); err != nil {
		slog.Error("mcp: mark execution running", "error", err, "execution_id", id)
	}

	finalize := func(status string, result []byte) {
		if _, err := pool.Exec(ctx,
			`UPDATE executions SET status=$2, result=$3::jsonb, duration_ms=$4,
			        started_at=$5, completed_at=now()
			 WHERE entity_id=$1`,
			id, status, result, int(time.Since(startedAt).Milliseconds()), startedAt); err != nil {
			slog.Error("mcp: finalize execution", "error", err, "execution_id", id)
		}
	}

	host, user, wrap, err := resolveExecTarget(ctx, pool, targetSlug)
	if err != nil {
		finalize("failed", jsonErr("%s", err.Error()))
		return "", err
	}

	var correlationID string
	if qerr := pool.QueryRow(ctx,
		`SELECT correlation_id FROM executions WHERE entity_id = $1`, id).Scan(&correlationID); qerr != nil {
		correlationID = ""
	}
	sink, flush := execlog.New(ctx, pool, id, correlationID)

	out, err := sshExecStream(ctx, host, user, wrap(command), sink)
	flush()
	if err != nil {
		finalize("failed", jsonErr("%s: %s", err.Error(), out))
		return out, err
	}

	finalize("completed", jsonOut(out))
	return out, nil
}

// autoRunAsync starts a command in a goroutine, marking it running and returning
// immediately. The caller gets an execution_id to poll with get_execution_status.
// Used for commands containing sleep/wait/poll loops that would exceed the MCP
// client timeout (120s) — the execution continues server-side.
func autoRunAsync(ctx context.Context, pool *db.Pool, id uuid.UUID, targetSlug, command string) {
	startedAt := time.Now()
	if _, err := pool.Exec(ctx,
		`UPDATE executions SET status='running', started_at=$2 WHERE entity_id=$1`,
		id, startedAt); err != nil {
		slog.Error("mcp: mark execution running (async)", "error", err, "execution_id", id)
	}

	host, user, wrap, err := resolveExecTarget(ctx, pool, targetSlug)
	if err != nil {
		pool.Exec(ctx,
			`UPDATE executions SET status='failed', result=$2::jsonb, duration_ms=$3, completed_at=now() WHERE entity_id=$1`,
			id, jsonErr("%s", err.Error()), int(time.Since(startedAt).Milliseconds()))
		slog.Error("mcp: async run resolve target", "error", err, "execution_id", id, "target", targetSlug)
		return
	}

	var correlationID string
	if qerr := pool.QueryRow(ctx,
		`SELECT correlation_id FROM executions WHERE entity_id = $1`, id).Scan(&correlationID); qerr != nil {
		correlationID = ""
	}

	go func() {
		defer func() {
			if r := recover(); r != nil {
				slog.Error("mcp: async run panic", "panic", r, "execution_id", id)
				pool.Exec(context.Background(),
					`UPDATE executions SET status='failed', result=$2::jsonb, duration_ms=$3, completed_at=now() WHERE entity_id=$1`,
					id, jsonErr("panic: %v", r), int(time.Since(startedAt).Milliseconds()))
			}
		}()

		sink, flush := execlog.New(context.Background(), pool, id, correlationID)
		out, execErr := sshExecStream(context.Background(), host, user, wrap(command), sink)
		flush()
		if execErr != nil {
			pool.Exec(context.Background(),
				`UPDATE executions SET status='failed', result=$2::jsonb, duration_ms=$3, completed_at=now() WHERE entity_id=$1`,
				id, jsonErr("%s: %s", execErr.Error(), out), int(time.Since(startedAt).Milliseconds()))
			slog.Error("mcp: async run failed", "error", execErr, "execution_id", id, "output", out)
		} else {
			pool.Exec(context.Background(),
				`UPDATE executions SET status='completed', result=$2::jsonb, duration_ms=$3, completed_at=now() WHERE entity_id=$1`,
				id, jsonOut(out), int(time.Since(startedAt).Milliseconds()))
		}
	}()
}

// isLongRunningCommand detects shell commands containing sleep, wait, or poll
// loops that indicate the command will exceed the MCP client timeout (120s).
// These commands should use autoRunAsync to avoid the client timing out while
// the command continues server-side.
func isLongRunningCommand(cmd string) bool {
	cmd = strings.TrimSpace(cmd)

	// sleep with duration — `sleep 30`, `sleep 1m`, etc.
	if sleepRe.MatchString(cmd) {
		return true
	}

	// while/shell poll loops with sleep: `while ...; do ... sleep; done`
	if pollRe.MatchString(cmd) {
		return true
	}

	// standalone wait command
	if waitRe.MatchString(cmd) {
		return true
	}

	return false
}

var (
	sleepRe = regexp.MustCompile(`\bsleep\s+\d`)
	pollRe  = regexp.MustCompile(`\bwhile\b.*\bsleep\b`)
	waitRe  = regexp.MustCompile(`\bwait\s+\d|[&;]\s*wait\b`)
)

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

	// Target validation: host-only commands (qm, pct, pvesh, iptables) must
	// not be dispatched against lxc:/vm: targets — those aren't Proxmox hosts
	// and don't have these tools. Caught live 2026-08-04: the agent ran
	// `qm stop 100` against lxc:dns, wasting a turn.

	// Command syntax validation: catch LLM-generated bash bugs before they
	// hit the shell. The model sometimes inserts literal \n between commands
	// or puts spaces inside flags — these always fail, so reject early.
	if syntaxErr := validateCommandSyntax(command); syntaxErr != "" {
		return textResult(syntaxErr)
	}

	if cmdPrefix, hostOnly := hostOnlyCommand(command); hostOnly && !strings.HasPrefix(targetSlug, "host:") {
		hostSuggestion := resolveProxmoxHostSlug(ctx, pool, targetSlug, "")
		if hostSuggestion == "" {
			hostSuggestion = "host:hubris or host:strong"
		}
		return textResult(fmt.Sprintf("Cannot run %q on %s — %s is a Proxmox host command. Use target %s instead.",
			cmdPrefix, targetSlug, cmdPrefix, hostSuggestion))
	}

	// systemctl and docker work on hosts and LXCs, but not VMs.
	if cmdPrefix, hostLxc := hostLxcCommand(command); hostLxc {
		if !strings.HasPrefix(targetSlug, "host:") && !strings.HasPrefix(targetSlug, "lxc:") {
			return textResult(fmt.Sprintf("Cannot run %q on %s — %s only works on host:* or lxc:* targets.",
				cmdPrefix, targetSlug, cmdPrefix))
		}
	}

	// VM transport pre-flight: qm guest exec requires the QEMU guest agent
	// to be running inside the VM. If it's not, the execution would queue
	// for approval and never execute — the agent has no way to learn it's
	// stuck (spotted live 2026-08-05: vm:zimaos had qemu_guest_agent=not_running,
	// the run queued forever, and the agent fell back to unsafe raw SSH).
	if strings.HasPrefix(targetSlug, "vm:") {
		var rawAttrs []byte
		if err := pool.QueryRow(ctx, `SELECT attributes FROM entities WHERE id = $1`, targetID).Scan(&rawAttrs); err == nil {
			var attrs map[string]any
			if json.Unmarshal(rawAttrs, &attrs) == nil {
				if qga, ok := attrs["qemu_guest_agent"]; ok {
					qgaStr, _ := qga.(string)
					if qgaStr == "not_running" || qgaStr == "" {
						return textResult(fmt.Sprintf(
							"run on %s blocked: QEMU guest agent is not running (%s). qm guest exec cannot reach this VM. Start the agent inside the guest first (e.g. via SSH/systemctl start qemu-guest-agent), then re-run. If the agent is running but the entity attribute is stale, update it with update_entity_attributes(slug=%s, attributes={\"qemu_guest_agent\":\"running\"}).",
							targetSlug, qgaStr, targetSlug))
					}
				}
			}
		}
	}

	// 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()
	// Correlate the execution to the chat session that asked for it. This was
	// a fresh random UUID per execution, which correlated nothing — every row
	// had a unique value, so the correlation_id column and the
	// ?correlation_id= filter could only ever match one execution.
	//
	// Using the session id makes the field mean what it says ("what did this
	// session do?") and is what lets the chat tail live output: execution
	// events carry correlation_id, so the UI can match them to the session on
	// screen without a lookup. Falls back to a random id when there is no
	// session to scope to, keeping the column non-empty.
	correlationID := sessionID
	if correlationID == "" || correlationID == "ephemeral" {
		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)
		// Link execution to session for auto-continuation (nomos_plan_executions
		// was always empty — executions were never traceable back to sessions).
		if sid, serr := uuid.Parse(sessionID); serr == nil {
			pool.Exec(ctx, `
				INSERT INTO nomos_plan_executions (execution_id, session_id)
				VALUES ($1, $2) ON CONFLICT (execution_id) DO NOTHING`, id, sid)
		}
	}

	// Auto-classify: write the classification decision to the classifications
	// table (was always empty — 0 rows despite 1,884 executions). The route
	// matches the auto-run vs queue-for-approval decision below.
	classRoute := "escalate"
	if riskClass == policy.RiskReadOnly || riskClass == policy.RiskReversibleLow {
		classRoute = "auto-act"
	} else if riskClass == policy.RiskConfigMutation && assentWindowActive(ctx, pool, agentID, sessionID) {
		classRoute = "auto-act"
	} else if riskClass == policy.RiskDestructive && destructiveWindowActive(ctx, pool, agentID, targetSlug, sessionID) {
		classRoute = "auto-act"
	}
	classReason, _ := json.Marshal(map[string]string{
		"command": command, "purpose": purpose, "target": targetSlug, "declared_risk": declaredRisk,
	})
	classID, _ := uuid.NewV7()
	pool.Exec(ctx, `INSERT INTO entities (id, slug, type, name, attributes) VALUES ($1, $2, 'classification', $3, '{}')`,
		classID, "classification:"+classID.String(), "classification for "+execSlug)
	pool.Exec(ctx, `INSERT INTO classifications (entity_id, action, risk_class, route, reasoning, correlation_id)
		VALUES ($1, $2, $3, $4, $5, $6)`,
		classID, actionCol, riskClass, classRoute, classReason, correlationID)
	// Link classification to execution.
	pool.Exec(ctx, `UPDATE executions SET classification_id = $2 WHERE entity_id = $1`, id, classID)

	// Audit: record the execution creation with session_id for traceability.
	// Every run call, whether auto-run or queued-for-approval, gets an audit
	// entry so the agent's activity is traceable back to the originating session.
	var auditSessionID *uuid.UUID
	if sessionID != "" && sessionID != "ephemeral" {
		if sid, serr := uuid.Parse(sessionID); serr == nil {
			auditSessionID = &sid
		}
	}
	_ = observability.Audit(ctx, sqlcgen.New(pool), "agent", "nomos", "run",
		&id, "POST", "/mcp", correlationID, auditSessionID,
		map[string]any{"command": command, "target": targetSlug, "risk_class": riskClass, "purpose": purpose})

	// read_only and reversible_low both run unattended, as seeds/policy.yaml
	// and .agents/OIKOS.md declare ("reversible_low — restart, cache clear,
	// sync pull. Unattended + ledger.").
	//
	// reversible_low had no branch here, so it fell through to the gate. That
	// looked stricter but was actually perverse: computeCommandRisk never
	// returns reversible_low — the class can ONLY arise when the agent
	// declares it on a command the classifier already scored read_only
	// (ClassifyCommand keeps the higher of the two). So an agent that
	// honestly flagged "this restarts something" got gated, while the same
	// command with no declaration auto-ran. That penalised candor and gave
	// the agent a reason to stay quiet.
	//
	// Auto-running it is no more permissive than the read_only branch above,
	// because read_only is the only computed class it can accompany. An
	// agent still cannot talk a command DOWN: declaring reversible_low on
	// something computed as config_mutation keeps config_mutation.
	if riskClass == policy.RiskReadOnly || riskClass == policy.RiskReversibleLow {
		if isLongRunningCommand(command) {
			autoRunAsync(ctx, pool, id, targetSlug, command)
			return textResult(fmt.Sprintf("run on %s (%s, async): started — execution %s. Poll with get_execution_status(%s) for result.",
				targetSlug, riskClass, id, id))
		}
		out, xerr := autoRun(ctx, pool, id, targetSlug, command)
		if xerr != nil {
			return textResult(fmt.Sprintf("run on %s: ERROR %v\n%s", targetSlug, xerr, out))
		}
		return textResult(fmt.Sprintf("run on %s (%s, auto): %s", targetSlug, riskClass, 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) {
		if isLongRunningCommand(command) {
			autoRunAsync(ctx, pool, id, targetSlug, command)
			slog.Info("mcp: run async via assent window", "target", targetSlug, "execution_id", id)
			return textResult(fmt.Sprintf("run on %s (config_mutation, async via assent window): started — execution %s. Poll with get_execution_status(%s) for result.",
				targetSlug, id, id))
		}
		out, xerr := autoRun(ctx, pool, id, targetSlug, command)
		if xerr != nil {
			return textResult(fmt.Sprintf("run on %s: ERROR %v\n%s", targetSlug, xerr, 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) {
		if isLongRunningCommand(command) {
			autoRunAsync(ctx, pool, id, targetSlug, command)
			slog.Info("mcp: run async via destructive window", "target", targetSlug, "execution_id", id)
			return textResult(fmt.Sprintf("run on %s (destructive, async via confirmed-target window): started — execution %s. Poll with get_execution_status(%s) for result.",
				targetSlug, id, id))
		}
		out, xerr := autoRun(ctx, pool, id, targetSlug, command)
		if xerr != nil {
			return textResult(fmt.Sprintf("run on %s: ERROR %v\n%s", targetSlug, xerr, 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.

// hostOnlyCommands maps command prefixes that are only valid on Proxmox host
// targets (not LXCs or VMs). Running these against an lxc: or vm: target
// always fails with "command not found" and wastes a turn.
var hostOnlyCommands = map[string]bool{
	"qm":      true,
	"pct":     true,
	"pvesh":   true,
	"iptables": true,
}

// hostLxcCommands maps command prefixes valid on host:* and lxc:* but not vm:*.
var hostLxcCommands = map[string]bool{
	"systemctl": true,
	"docker":    true,
}

// hostOnlyCommand checks whether the leading word of cmd is a host-only
// command. Returns the command word and true if the command can only run on
// a host: target.
func hostOnlyCommand(cmd string) (string, bool) {
	trimmed := strings.TrimSpace(cmd)
	parts := strings.Fields(trimmed)
	if len(parts) == 0 {
		return "", false
	}
	first := parts[0]
	// Check for shell wrappers: bash -c 'actual_cmd', sh -c 'actual_cmd'
	if (first == "bash" || first == "sh") && len(parts) >= 3 && parts[1] == "-c" {
		// The actual command is inside the -c argument; extract the first word.
		// This handles `bash -c 'qm stop 100'` but not deeply nested wrappers.
		actual := strings.Trim(strings.Join(parts[2:], " "), "'\"")
		if inner := strings.Fields(actual); len(inner) > 0 {
			first = inner[0]
		}
	}
	// Strip path: /usr/sbin/qm → qm
	if idx := strings.LastIndexByte(first, '/'); idx >= 0 {
		first = first[idx+1:]
	}
	return first, hostOnlyCommands[first]
}

// hostLxcCommand checks whether the leading word of cmd is a command valid on
// host:* and lxc:* targets but not vm:*. Returns the command word and true if
// the command is restricted to host/lxc.
func hostLxcCommand(cmd string) (string, bool) {
	trimmed := strings.TrimSpace(cmd)
	parts := strings.Fields(trimmed)
	if len(parts) == 0 {
		return "", false
	}
	first := parts[0]
	if idx := strings.LastIndexByte(first, '/'); idx >= 0 {
		first = first[idx+1:]
	}
	return first, hostLxcCommands[first]
}

// validateCommandSyntax checks for common LLM-generated bash errors that always
// fail at the shell. Returns an error message or "" if the command looks valid.
func validateCommandSyntax(cmd string) string {
	// Reject literal \n (the LLM sometimes writes `echo "---" && \n curl ...`
	// — the \n is literal in the command string, not an actual newline).
	if strings.Contains(cmd, "\\n") {
		return fmt.Sprintf("Command contains literal '\\n' — use ';' or '&&' between commands, not a literal backslash-n. Command: %q", cmd)
	}

	// Reject `&& \n` patterns (the LLM writes `cmd1 && \n cmd2` — the \n is
	// a literal newline that bash interprets as a command separator, but the
	// leading backslash makes it a syntax error).
	if andBackslashRe.MatchString(cmd) {
		return fmt.Sprintf("Command contains '&&' followed by a literal backslash-newline — remove the backslash or use ';' instead. Command: %q", cmd)
	}

	// Reject `\` at end of command with no continuation (last line ends with
	// backslash but there's nothing after it).
	trimmed := strings.TrimSpace(cmd)
	if strings.HasSuffix(trimmed, "\\") {
		return fmt.Sprintf("Command ends with a backslash but has nothing after it to continue. Remove the trailing '\\'. Command: %q", cmd)
	}

	// Warn on common flag typos: `head - n`, `grep - i`, `tail - n`, etc.
	// These are space-between-flag-and-value errors the LLM produces.
	if flagSpaceRe.MatchString(cmd) {
		return fmt.Sprintf("Command has a space between a flag and its value (e.g. 'head - n' instead of 'head -n'). Remove the space. Command: %q", cmd)
	}

	return ""
}

var andBackslashRe = regexp.MustCompile(`&&\s*\\\s*\n`)
var flagSpaceRe = regexp.MustCompile(`\b(head|tail|grep|sed|awk|sort|uniq|wc)\s+(-\w)\s+\w`)

// sessionHasPlan reports whether this nomos session has any plan step on
// record that isn't `replaced`. Replaced steps (from session reopen via
// store.reopenSession) don't count — the agent must propose fresh plan before
// any `run`. 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 AND status <> 'replaced'`,
		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}
}