oikos/internal/httpapi/actuator.go

package httpapi

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"log/slog"
	"os"
	"strconv"
	"strings"
	"time"

	"github.com/dtoro/oikos/internal/actuator"
	"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/google/uuid"
)

var (
	_sshUser string
	_sshKey  []byte
)

// flexBool accepts a JSON bool, number (0/1), or string ("true"/"1"/"yes").
// LLMs routinely emit `"privileged": 0` instead of `false`; a strict `bool`
// field made the approved pct_create execution fail to parse *after* the
// operator had already approved it — the container was never created and the
// operator saw "queued" with no result. This type tolerates the common shapes.
type flexBool bool

func (b *flexBool) UnmarshalJSON(data []byte) error {
	s := strings.TrimSpace(strings.Trim(string(data), `"`))
	switch strings.ToLower(s) {
	case "true", "1", "yes", "on":
		*b = true
	case "false", "0", "no", "off", "", "null":
		*b = false
	default:
		return fmt.Errorf("cannot parse %q as bool", s)
	}
	return nil
}

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

// sshExecTimeout bounds how long a single remote command may run. Without
// this, a hung remote command (e.g. a piped install script stuck retrying
// DNS against a misconfigured gateway) blocks the executing goroutine
// forever: the execution never leaves 'approved'/'running', the operator
// sees an unkillable spinner, and get_execution_status has nothing new to
// report. Generous enough for a real apt/docker install; not infinite.
const sshExecTimeout = 10 * time.Minute

// streamWriter buffers everything it is given while forwarding each write to a
// sink. One on session.Stdout and another sharing the same buffer on
// session.Stderr reproduces CombinedOutput's interleaving in the order the
// remote end produced it. Shared implementation lives in internal/actuator
// (actuator.streamWriter / actuator.RunStreaming).

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
	}

	signer, err := actuator.LoadSignerFromBytes(_sshKey)
	if err != nil {
		return "", fmt.Errorf("parse key: %w", err)
	}

	client, err := actuator.Dial(ctx, actuator.DialOptions{Host: host, User: user, Signer: signer})
	if err != nil {
		return "", err
	}
	defer client.Close()

	return actuator.RunStreaming(ctx, client, command, sink, sshExecTimeout)
}

func resolveHostSSH(ctx context.Context, pool *db.Pool, entitySlug string) (string, string, 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)
	}

	sshUser := _sshUser
	if sshUser == "" {
		sshUser = "root"
	}

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

// resolveRunTarget mirrors internal/mcp.resolveExecTarget for the approved-
// execution side: any target slug (host: or lxc:) resolves to the SSH
// endpoint that runs the command plus a wrap function that turns a plain
// shell command into what actually needs to be sent — identity for a host,
// `pct exec <pve_id>` for an LXC. Kept as a small duplicate rather than a
// cross-package import to avoid coupling httpapi to mcp for one helper.
func resolveRunTarget(ctx context.Context, pool *db.Pool, targetSlug string) (host, user string, wrap func(string) string, err error) {
	if strings.HasPrefix(targetSlug, "host:") {
		host, user, err = resolveHostSSH(ctx, pool, targetSlug)
		return host, user, func(cmd string) string { return cmd }, err
	}
	if strings.HasPrefix(targetSlug, "lxc:") {
		var pveID, hostAttr string
		// COALESCE the host column: many older LXC entities (seeded from
		// inventory, not provisioned by pct_create) have pve_id but no host
		// attribute at all. Scanning a SQL NULL into a plain string errors
		// the whole row, wrongly reporting "missing pve_id" even when it was
		// present — COALESCE avoids the NULL, "" is handled below.
		if qerr := pool.QueryRow(ctx, "SELECT attributes->>'pve_id', COALESCE(attributes->>'host', '') FROM entities WHERE slug = $1", targetSlug).Scan(&pveID, &hostAttr); qerr != nil || pveID == "" {
			return "", "", nil, fmt.Errorf("LXC not found or missing pve_id: %s", targetSlug)
		}
		hostSlug := hostAttr
		if hostSlug == "" {
			hostSlug = "hubris"
		}
		if !strings.HasPrefix(hostSlug, "host:") {
			hostSlug = "host:" + hostSlug
		}
		host, user, err = resolveHostSSH(ctx, pool, hostSlug)
		id := pveID
		return host, user, func(cmd string) string {
			b64 := base64.StdEncoding.EncodeToString([]byte(cmd))
			return fmt.Sprintf("pct exec %s -- bash -c 'echo %s | base64 -d | bash'", id, b64)
		}, err
	}
	return "", "", nil, fmt.Errorf("unsupported target %q: must be host:<slug> or lxc:<slug>", targetSlug)
}

// executeApprovedAction runs a gated action after operator approval.
// Runs in a background goroutine to not block the HTTP response.
// emitExecutionEvent records an execution lifecycle event for SSE fan-out so
// the control room can watch approved actions run to completion live.
func emitExecutionEvent(ctx context.Context, pool *db.Pool, execID uuid.UUID, status string, detail map[string]any) {
	severity := "info"
	if status == "failed" {
		severity = "warning"
	}
	// The correlation id was hardcoded to "", so execution events could not be
	// tied back to the session that caused them — the one join you want when
	// asking "what did this agent turn actually do?". It is already on the
	// execution row; read it rather than threading it through eleven callers.
	var correlationID string
	if err := pool.QueryRow(ctx,
		`SELECT correlation_id FROM executions WHERE entity_id = $1`, execID).Scan(&correlationID); err != nil {
		correlationID = ""
	}
	_ = observability.Event(ctx, sqlcgen.New(pool), "execution."+status, &execID, severity, "actuator", correlationID, detail)
	if status == "completed" || status == "failed" || status == "cancelled" {
		closePlanStepForExecution(ctx, pool, execID, status)
	}
}

// closePlanStepForExecution auto-closes a task plan step whose linked execution
// just reached a terminal state, so the task board advances even if the agent
// doesn't call update_plan_step itself (belt and suspenders — the agent links
// the step to the execution when it starts it; the api finishes it here). Emits
// plan.step.finished correlated to the step's session. No-op for the vast
// majority of executions, which aren't plan steps.
func closePlanStepForExecution(ctx context.Context, pool *db.Pool, execID uuid.UUID, execStatus string) {
	stepStatus := "done"
	if execStatus == "failed" || execStatus == "cancelled" {
		stepStatus = "failed"
	}
	var stepID, sessionID string
	var seq int
	if err := pool.QueryRow(ctx, `
		UPDATE session_plan_steps SET status = $2, finished_at = now()
		WHERE execution_id = $1 AND status NOT IN ('done', 'failed', 'skipped')
		RETURNING id::text, session_id::text, seq`, execID, stepStatus).Scan(&stepID, &sessionID, &seq); err != nil {
		return // no matching open step
	}
	_ = observability.Event(ctx, sqlcgen.New(pool), "plan.step.finished", &execID, "info", "actuator", sessionID,
		map[string]any{"step_id": stepID, "seq": seq, "status": stepStatus, "execution_id": execID.String()})
}

func executeApprovedAction(ctx context.Context, pool *db.Pool, execID uuid.UUID, targetSlug string, actionStr string) {
	slog.Info("httpapi: executing approved action", "execution_id", execID, "target", targetSlug, "action", actionStr)

	host, user, wrap, err := resolveRunTarget(ctx, pool, targetSlug)
	if err != nil {
		slog.Error("httpapi: resolve host for approved execution", "error", err, "target", targetSlug)
		pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`,
			execID, jsonErr("%s", err.Error()))
		emitExecutionEvent(ctx, pool, execID, "failed", map[string]any{"target": targetSlug, "error": err.Error()})
		return
	}

	idx := strings.Index(actionStr, ":")
	if idx < 0 {
		slog.Error("httpapi: malformed action string (no colon)", "action", actionStr)
		return
	}
	action, params := actionStr[:idx], actionStr[idx+1:]

	startedAt := time.Now()
	// Persist started_at now, not at the end. It was captured here but only
	// written in the terminal UPDATE, so a running execution reported
	// started_at = NULL for its entire life — the UI could not show how long
	// anything had been going, which is exactly when you want to know.
	if _, err := pool.Exec(ctx,
		`UPDATE executions SET status = 'running', started_at = $2 WHERE entity_id = $1`,
		execID, startedAt); err != nil {
		slog.Error("httpapi: mark execution running", "error", err, "execution_id", execID)
	}

	// Stream output for the actions whose output an operator actually watches:
	// a long apt upgrade, a pct create, an arbitrary approved `run`. The small
	// internal lookups further down (listing template cache, pvesh nextid) stay
	// unstreamed — they are plumbing, and logging them would bury the command
	// the operator approved.
	var correlationID string
	if qerr := pool.QueryRow(ctx,
		`SELECT correlation_id FROM executions WHERE entity_id = $1`, execID).Scan(&correlationID); qerr != nil {
		correlationID = ""
	}
	sink, flushLogs := execlog.New(ctx, pool, execID, correlationID)
	defer flushLogs()

	var output, cmd string

	switch action {
	case "systemctl":
		svc := strings.TrimPrefix(targetSlug, "lxc:")
		switch {
		case strings.HasPrefix(params, "enable:"):
			svc = strings.TrimPrefix(params, "enable:")
			cmd = fmt.Sprintf("systemctl enable %s --now 2>&1; sleep 1; systemctl is-active %s", svc, svc)
		case strings.HasPrefix(params, "disable:"):
			svc = strings.TrimPrefix(params, "disable:")
			cmd = fmt.Sprintf("systemctl disable %s --now 2>&1; sleep 1; systemctl is-active %s", svc, svc)
		default:
			cmd = fmt.Sprintf("systemctl %s %s 2>&1", params, svc)
		}
		output, err = sshExecStream(ctx, host, user, cmd, sink)

	case "apt_upgrade":
		svc := strings.TrimPrefix(targetSlug, "lxc:")
		cmd = fmt.Sprintf("apt update -qq 2>&1 >/dev/null && apt upgrade -y -qq 2>&1; echo '---'; systemctl is-active %s || true", svc)
		output, err = sshExecStream(ctx, host, user, cmd, sink)

	case "pct_create":
		var cfg struct {
			VMID         int      `json:"vmid"`
			Hostname     string   `json:"hostname"`
			Cores        int      `json:"cores"`
			Memory       int      `json:"memory"`
			DiskGB       int      `json:"disk_gb"`
			IP           string   `json:"ip"`
			GW           string   `json:"gw"`
			Bridge       string   `json:"bridge"` // e.g. vmbr0/vmbr1 — which bridge actually reaches the target subnet on this host varies per host, don't assume vmbr0
			Storage      string   `json:"storage"`
			Template     string   `json:"template"`
			Privileged   flexBool `json:"privileged"`
			Nesting      flexBool `json:"nesting"`
			Mounts       []string `json:"mounts"`
			Nameserver   string   `json:"nameserver"`
			Searchdomain string   `json:"searchdomain"`
			// No services/post_install here anymore — pct_create is atomic
			// (create + start + register only). Installing packages and
			// running setup scripts is the agent's job via follow-up `run`
			// calls against lxc:<hostname>, so each step is individually
			// observable and recoverable instead of one opaque multi-minute
			// black box. See the comment above the removed post-create block.
		}
		if err := json.Unmarshal([]byte(params), &cfg); err != nil {
			slog.Error("httpapi: pct_create parse params", "error", err, "params", params)
			pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`,
				execID, jsonErr("invalid pct_create params: %v", err))
			emitExecutionEvent(ctx, pool, execID, "failed", map[string]any{"target": targetSlug, "error": err.Error()})
			return
		}
		// Only hostname is required. vmid is optional — when 0 (or later found
		// to collide) the VMID guard below assigns a free cluster id.
		if cfg.Hostname == "" {
			pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`,
				execID, `{"error":"pct_create: hostname is required"}`)
			emitExecutionEvent(ctx, pool, execID, "failed", map[string]any{"target": targetSlug, "error": "missing hostname"})
			return
		}
		if cfg.Cores == 0 {
			cfg.Cores = 1
		}
		if cfg.Memory == 0 {
			cfg.Memory = 512
		}
		if cfg.DiskGB == 0 {
			cfg.DiskGB = 8
		}
		if cfg.Storage == "" {
			cfg.Storage = "local-lvm"
		}
		if cfg.GW == "" {
			cfg.GW = "192.168.8.2"
		}
		if cfg.Nameserver == "" {
			cfg.Nameserver = "192.168.8.2"
		}
		if cfg.Searchdomain == "" {
			cfg.Searchdomain = "hubris.network"
		}
		// Template pre-flight: resolve against what the host actually has
		// cached. A hardcoded name (e.g. debian-13) fails opaquely with a raw
		// `pct` error when that exact file isn't present. List the cache, then
		// either validate the requested template or auto-pick the newest
		// debian one; on miss, fail early with the available list so the
		// operator/agent can retry with a real name.
		cacheList, tplErr := sshExec(ctx, host, user, "ls -1 /var/lib/vz/template/cache/ 2>/dev/null | grep -E '\\.tar\\.(zst|gz|xz)$' || true")
		available := []string{}
		for _, l := range strings.Split(strings.TrimSpace(cacheList), "\n") {
			if l = strings.TrimSpace(l); l != "" {
				available = append(available, l)
			}
		}
		if tplErr != nil {
			pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`,
				execID, jsonErr("list templates on %s: %s", targetSlug, tplErr.Error()))
			emitExecutionEvent(ctx, pool, execID, "failed", map[string]any{"target": targetSlug, "error": tplErr.Error()})
			return
		}
		cfg.Template = resolveTemplate(cfg.Template, available)
		if cfg.Template == "" {
			msg := fmt.Sprintf("no usable LXC template on %s. Available: %v", targetSlug, available)
			pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`,
				execID, jsonErr("%s", msg))
			emitExecutionEvent(ctx, pool, execID, "failed", map[string]any{"target": targetSlug, "error": msg})
			return
		}

		// VMID collision guard. Proxmox VMIDs are cluster-wide, so the model's
		// guess (e.g. 132) can collide with a container on another node — pct
		// create then fails with "CT N already exists on node X". Fetch the set
		// of in-use VMIDs across the cluster; if the requested id is taken (or
		// absent), fall back to the cluster's next free id so provisioning
		// still succeeds instead of dead-ending on the operator's approval.
		usedRaw, _ := sshExec(ctx, host, user, `pvesh get /cluster/resources --type vm --output-format json 2>/dev/null | grep -o '"vmid":[0-9]*' | grep -o '[0-9]*' || true`)
		used := map[int]bool{}
		for _, l := range strings.Fields(usedRaw) {
			if n, e := strconv.Atoi(strings.TrimSpace(l)); e == nil {
				used[n] = true
			}
		}
		if cfg.VMID == 0 || used[cfg.VMID] {
			nextRaw, nerr := sshExec(ctx, host, user, `pvesh get /cluster/nextid 2>/dev/null`)
			nextID, cerr := strconv.Atoi(strings.TrimSpace(nextRaw))
			if nerr != nil || cerr != nil || nextID == 0 {
				msg := fmt.Sprintf("VMID %d is already in use on the cluster and could not resolve a free id", cfg.VMID)
				pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`,
					execID, jsonErr("%s", msg))
				emitExecutionEvent(ctx, pool, execID, "failed", map[string]any{"target": targetSlug, "error": msg})
				return
			}
			slog.Info("httpapi: pct_create VMID reassigned", "requested", cfg.VMID, "assigned", nextID)
			cfg.VMID = nextID
		}

		privFlag := "--unprivileged 1"
		if cfg.Privileged {
			privFlag = "--unprivileged 0"
		}

		nestingFlag := ""
		features := []string{}
		if cfg.Nesting {
			features = append(features, "nesting=1")
		}
		if cfg.Privileged {
			features = append(features, "keyctl=1")
		}
		if len(features) > 0 {
			nestingFlag = fmt.Sprintf(" --features %s", strings.Join(features, ","))
		}

		if cfg.Bridge == "" {
			cfg.Bridge = "vmbr0"
		}

		// net0: DHCP when no static IP is given (or ip=="dhcp"). Proxmox
		// rejects a gateway alongside ip=dhcp, so only add gw for a static IP.
		net0 := "name=eth0,bridge=" + cfg.Bridge + ","
		isStatic := cfg.IP != "" && !strings.EqualFold(cfg.IP, "dhcp")
		if !isStatic {
			net0 += "ip=dhcp"
		} else {
			net0 += "ip=" + cfg.IP
			if cfg.GW != "" {
				net0 += ",gw=" + cfg.GW
			}
		}

		// Pre-flight: for a static config, ping the gateway from the target
		// HOST, on the SPECIFIC BRIDGE being requested, before spending 5+
		// minutes creating the container. This is the check that would have
		// caught the real TypeType failure immediately instead of after a
		// full provision attempt.
		//
		// Binding to the bridge (`ping -I <bridge>`) matters and was found
		// live: a plain unqualified `ping <gw>` from the host can succeed via
		// the host's own routing table (multiple routes, possibly through an
		// upstream router) even when the *container* — which only gets a
		// naive on-link default route via its bridge's veth — can never ARP
		// that gateway at all. Confirmed on `strong`: bare `ping 192.168.8.2`
		// succeeded (via the host's default route), but a container actually
		// attached to vmbr0 showed 100% packet loss trying to reach the same
		// address, because vmbr0 doesn't carry that subnet's L2 segment.
		// Binding to the bridge interface reproduces what the container will
		// actually experience, not what the host's broader routing table can
		// reach.
		if isStatic && cfg.GW != "" {
			pingOut, pingErr := sshExec(ctx, host, user, fmt.Sprintf("ping -I %s -c1 -W2 %s >/dev/null 2>&1 && echo PREFLIGHT_OK || echo PREFLIGHT_FAIL", cfg.Bridge, cfg.GW))
			if pingErr != nil || !gatewayPreflightPassed(pingOut) {
				msg := fmt.Sprintf(
					"gateway %s is not reachable from %s on bridge %s — this almost always means the bridge doesn't carry that subnet on this host (each bridge only reaches the network it's physically wired to). "+
						"Do not retry with a different gateway guess in the same subnet: find an existing LXC on this host with an IP in the same /28 and copy its exact bridge+gateway, or use DHCP instead.",
					cfg.GW, targetSlug, cfg.Bridge)
				pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`,
					execID, jsonErr("%s", msg))
				emitExecutionEvent(ctx, pool, execID, "failed", map[string]any{"target": targetSlug, "error": msg})
				return
			}
		}

		templatePath := fmt.Sprintf("/var/lib/vz/template/cache/%s", cfg.Template)
		createCmd := fmt.Sprintf(
			"pct create %d %s --hostname %s --cores %d --memory %d --rootfs %s:%d %s --net0 %s%s --start 1",
			cfg.VMID, templatePath, cfg.Hostname, cfg.Cores, cfg.Memory,
			cfg.Storage, cfg.DiskGB, privFlag, net0, nestingFlag)

		if cfg.Nameserver != "" {
			createCmd += fmt.Sprintf(" --nameserver %s", cfg.Nameserver)
		}
		if cfg.Searchdomain != "" {
			createCmd += fmt.Sprintf(" --searchdomain %s", cfg.Searchdomain)
		}

		// Add mount points
		for i, mp := range cfg.Mounts {
			if i < 10 { // pct supports up to mp9
				createCmd += fmt.Sprintf(" --mp%d %s", i, mp)
			}
		}

		slog.Info("httpapi: pct_create running", "vmid", cfg.VMID, "hostname", cfg.Hostname, "cmd", createCmd)
		output, err = sshExecStream(ctx, host, user, createCmd, sink)

		// pct_create is now DELIBERATELY ATOMIC: create + start + register,
		// nothing else. It used to also run apt installs and a post_install
		// script inline as one black-box multi-minute SSH call — the agent
		// got back a single opaque success/fail for the whole thing with no
		// way to see (or fix) which step actually broke. That's the opposite
		// of what makes an agent able to recover from errors.
		//
		// Installing packages, running post_install, and verifying the
		// service now happen as the agent's OWN follow-up `run` calls against
		// the new lxc:<hostname> target — each one is synchronous (in an
		// active assent window) or individually gated, so the agent observes
		// every step's real output and can diagnose + retry the exact thing
		// that failed instead of re-doing the whole container. See SOUL.md
		// "After pct_create: you drive the install" and provisionScript's
		// surviving role (DNS self-heal) is now something the agent invokes
		// itself via `run`, not something baked into this handler.
		//
		// cfg.Services/cfg.PostInstall are intentionally no longer read here.

		// On success, register the entity in the DB with proper relationships
		if err == nil {
			slug := "lxc:" + cfg.Hostname
			var lxcID uuid.UUID
			lxcID, _ = uuid.NewV7()
			attrs := map[string]any{
				"pve_id": fmt.Sprintf("%d", cfg.VMID),
				"host":   strings.TrimPrefix(targetSlug, "host:"),
				"ip":     cfg.IP,
			}
			attrsJSON, _ := json.Marshal(attrs)
			_, insErr := pool.Exec(ctx, `INSERT INTO entities (id, slug, type, name, state, attributes, enrolled_at)
				VALUES ($1, $2, 'lxc', $3, 'provisioning', $4, now()) ON CONFLICT (slug) DO NOTHING`, lxcID, slug, cfg.Hostname, attrsJSON)
			if insErr != nil {
				slog.Error("httpapi: pct_create entity insert", "error", insErr, "slug", slug)
			}

			// Create hosts relationship: Proxmox host → LXC
			var hostID uuid.UUID
			if err := pool.QueryRow(ctx, "SELECT id FROM entities WHERE slug = $1", targetSlug).Scan(&hostID); err == nil {
				_, relErr := pool.Exec(ctx, `INSERT INTO relationships (source_id, target_id, type, attributes, valid_from)
					VALUES ($1, $2, 'hosts', '{"provisioned_by":"nomos"}'::jsonb, now())`, hostID, lxcID)
				if relErr != nil {
					slog.Error("httpapi: pct_create relationship insert", "error", relErr, "host", targetSlug, "lxc", slug)
				}
			}

			// Create entity_status row for health tracking
			pool.Exec(ctx, `INSERT INTO entity_status (entity_id, health, last_check_at)
				VALUES ($1, 'unknown', now()) ON CONFLICT (entity_id) DO NOTHING`, lxcID)

			emitExecutionEvent(ctx, pool, execID, "executing", map[string]any{
				"lxc_slug": slug, "vmid": cfg.VMID, "host": targetSlug,
			})

			slog.Info("httpapi: pct_create entity registered", "slug", slug, "vmid", cfg.VMID, "host", targetSlug)
		}

	case "run":
		// The general gated primitive: arbitrary shell against any host or
		// LXC, approved and classified by internal/policy.ClassifyCommand at
		// request time (see mcp/server.go's "run" tool). No fixed action
		// enum — new capability doesn't require new Go code here.
		var cfg struct {
			Command string `json:"command"`
			Purpose string `json:"purpose"`
		}
		if perr := json.Unmarshal([]byte(params), &cfg); perr != nil {
			pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`,
				execID, jsonErr("invalid run params: %v", perr))
			emitExecutionEvent(ctx, pool, execID, "failed", map[string]any{"target": targetSlug, "error": perr.Error()})
			return
		}
		cmd = wrap(cfg.Command)
		output, err = sshExecStream(ctx, host, user, cmd, sink)

	default:
		slog.Error("httpapi: unknown gated action for approved execution", "action", action, "execution_id", execID)
		pool.Exec(ctx, `UPDATE executions SET status='failed', result=$2::jsonb WHERE entity_id=$1`,
			execID, jsonErr("unknown action: %s", action))
		return
	}

	durationMs := int(time.Since(startedAt).Milliseconds())
	status := "completed"
	verified := true
	// Build result via json.Marshal, not string interpolation. Command output
	// (apt/pct) contains quotes, backslashes and control chars; the old
	// fmt.Sprintf only escaped "\n", producing invalid JSON that failed the
	// ::jsonb cast — so this UPDATE was silently discarded and the execution
	// was stuck at "approved" forever even though provisioning succeeded.
	resMap := map[string]any{"output": output}
	if err != nil {
		resMap["error"] = err.Error()
		status = "failed"
		verified = false
	}
	resultJSON, _ := json.Marshal(resMap)

	if _, uerr := pool.Exec(ctx, `UPDATE executions SET status=$2, result=$3::jsonb, duration_ms=$4, verified=$5, started_at=$6, completed_at=$7 WHERE entity_id=$1`,
		execID, status, resultJSON, durationMs, verified, startedAt, time.Now()); uerr != nil {
		slog.Error("httpapi: finalize execution status", "error", uerr, "execution_id", execID, "intended_status", status)
	}

	emitExecutionEvent(ctx, pool, execID, status, map[string]any{
		"action": action, "target": targetSlug, "duration_ms": durationMs,
	})

	slog.Info("httpapi: approved action executed",
		"execution_id", execID, "action", action, "status", status, "duration_ms", durationMs)
}

// jsonErr builds a valid {"error": "..."} JSON payload for an execution's
// result column. Always use this instead of fmt.Sprintf'ing JSON by hand —
// error text and command output routinely contain quotes/backslashes that
// break a hand-built string and fail the ::jsonb cast.
func jsonErr(format string, args ...any) []byte {
	b, _ := json.Marshal(map[string]any{"error": fmt.Sprintf(format, args...)})
	return b
}

// resolveTemplate maps a requested template name to one actually present in
// the host's template cache. Exact match wins; a bare distro hint (e.g.
// "debian-13" or "debian") matches by prefix; empty picks the newest debian
// (falling back to any) template available. Returns "" when nothing fits.
// gatewayPreflightPassed interprets the PREFLIGHT_OK/PREFLIGHT_FAIL markers
// from the pct_create gateway pre-flight check. Pulled out as its own
// function (rather than an inline strings.Contains at the call site) so it's
// unit-testable: a prior version checked for "REACHABLE", which is a
// substring of "UNREACHABLE" — the check could never actually fail, and it
// took a live deployment to notice. Exact-match markers plus a test make
// that specific bug class structurally unable to recur silently.
func gatewayPreflightPassed(out string) bool {
	return strings.TrimSpace(out) == "PREFLIGHT_OK"
}

func resolveTemplate(requested string, available []string) string {
	if len(available) == 0 {
		return ""
	}
	if requested != "" {
		for _, a := range available {
			if a == requested {
				return a
			}
		}
		for _, a := range available {
			if strings.HasPrefix(a, requested) {
				return a
			}
		}
	}
	// Auto-pick: prefer debian, then the lexically-greatest (newest version).
	best := ""
	for _, a := range available {
		if strings.Contains(a, "debian") && a > best {
			best = a
		}
	}
	if best != "" {
		return best
	}
	for _, a := range available {
		if a > best {
			best = a
		}
	}
	return best
}