oikos/internal/scheduler/scheduler.go

// Package scheduler implements the Oikos observe + decide loop (Phase 3).
// It loads enabled check_defs, runs checks on schedule, manages signal
// lifecycle (dedup, flap suppression, maintenance mode), and writes metrics.
package scheduler

import (
	"context"
	"crypto/tls"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"net"
	"net/http"
	"os/exec"
	"regexp"
	"runtime"
	"strconv"
	"strings"
	"time"

	"github.com/dtoro/oikos/internal/config"
	"github.com/dtoro/oikos/internal/db"
	"github.com/dtoro/oikos/internal/db/sqlcgen"
	"github.com/dtoro/oikos/internal/observability"
	"github.com/google/uuid"
	"golang.org/x/sync/errgroup"
	"golang.org/x/sys/unix"
)

var (
	sshKeyPath string
	sshUser    string
)

// Run starts the scheduler loop. Blocks until ctx is cancelled.
func Run(ctx context.Context, pool *db.Pool, cfg config.Config) {
	slog.Info("scheduler: starting", "interval", cfg.SchedulerInterval)
	interval := cfg.SchedulerInterval
	if interval <= 0 {
		interval = 30 * time.Second
	}

	sshKeyPath = cfg.SSHKeyPath
	sshUser = cfg.SSHUser
	if sshUser == "" {
		sshUser = "root"
	}

	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	// Immediate first pass
	runCheckPass(ctx, pool)

	for {
		select {
		case <-ctx.Done():
			slog.Info("scheduler: shutting down")
			return
		case <-ticker.C:
			runCheckPass(ctx, pool)
		}
	}
}

// runCheckPass executes one full cycle of check evaluation.
func runCheckPass(ctx context.Context, pool *db.Pool) {
	q := sqlcgen.New(pool)

	defs, err := q.ListEnabledCheckDefs(ctx)
	if err != nil {
		slog.Error("scheduler: list check defs", "error", err)
		return
	}
	if len(defs) == 0 {
		// Still run housekeeping: a fleet with no enabled check_defs is
		// precisely the case coverageSweep exists to report, and returning
		// here would mean the one situation that most needs reporting is the
		// one situation that stays silent.
		slog.Debug("scheduler: no enabled check_defs")
		housekeeping(ctx, pool)
		return
	}

	g, gctx := errgroup.WithContext(ctx)
	g.SetLimit(10) // bounded worker pool

	for _, def := range defs {
		cd := def
		g.Go(func() error {
			runCheck(gctx, pool, cd)
			return nil
		})
	}
	g.Wait()

	// Housekeeping after each pass
	housekeeping(ctx, pool)
}

// runCheck executes a single check and processes the result.
//
// check_defs.entity_id identifies the *check* (probe) entity itself;
// check_defs.target_id identifies the entity actually being observed (the
// host/service/etc). Health, metrics, and events must attach to the target
// so the observed entity's own record reflects reality — not the internal
// probe. Signals stay keyed by the check entity (cd.EntityID), matching how
// they are created below and resolved elsewhere.
func runCheck(ctx context.Context, pool *db.Pool, cd sqlcgen.ListEnabledCheckDefsRow) {
	q := sqlcgen.New(pool)
	start := time.Now()

	result := executeCheck(ctx, cd)

	// Stamp the run before processing the result: due-ness must advance even
	// when a check fails, or a permanently failing check would be re-run on
	// every pass instead of at its declared interval. last_health records THIS
	// check's own verdict, which is what makes the aggregation below possible.
	checkHealth := result.health
	if checkHealth == "" {
		checkHealth = "healthy"
	}
	if err := q.MarkCheckRun(ctx, sqlcgen.MarkCheckRunParams{
		EntityID:   cd.EntityID,
		LastHealth: &checkHealth,
	}); err != nil {
		slog.Error("scheduler: mark check run", "entity", cd.EntitySlug, "error", err)
	}

	latency := time.Since(start).Milliseconds()

	if result.metrics == nil {
		result.metrics = make(map[string]float64)
	}
	result.metrics["probe_latency_ms"] = float64(latency)

	targetID := cd.EntityID
	if cd.TargetID != nil {
		targetID = *cd.TargetID
	}

	for metric, value := range result.metrics {
		_ = q.InsertMetricSample(ctx, sqlcgen.InsertMetricSampleParams{
			EntityID: targetID,
			Metric:   metric,
			Value:    value,
			Tags:     []byte(`{}`),
		})
	}

	if result.err != nil {
		slog.Warn("scheduler: check failed",
			"entity", cd.EntitySlug, "kind", cd.Kind, "error", result.err)
	}

	prevHealth := currentHealth(ctx, pool, targetID)

	if result.signalKind == "" || result.health == "healthy" {
		resolveSignal(ctx, pool, cd.EntityID, targetID, cd.EntitySlug)
		// NOT unconditionally "healthy": this check passing says nothing about
		// the entity's other checks. Writing healthy here is what let one
		// passing probe erase a genuine failure reported by another — and,
		// alternating with a failing probe, produced 226 health flips an hour
		// on a host that was fine throughout.
		applyAggregateHealth(ctx, pool, q, targetID, cd.EntitySlug, prevHealth)
		return
	}

	slog.Warn("scheduler: raising signal",
		"entity", cd.EntitySlug, "kind", result.signalKind, "evidence", result.evidence)

	severity := evaluateSeverity(cd.Kind, result.signalKind, cd.Config, result.metrics)

	sig, err := q.UpsertSignal(ctx, sqlcgen.UpsertSignalParams{
		EntityID:       cd.EntityID,
		Kind:           result.signalKind,
		Severity:       severity,
		TargetEntityID: cd.TargetID,
		Evidence:       &result.evidence,
	})
	if err != nil {
		slog.Error("scheduler: upsert signal", "error", err)
		return
	}

	_ = sig

	if prevHealth == "" || prevHealth == "healthy" {
		emitSchedulerEvent(ctx, pool, "signal.raised", targetID, severity,
			map[string]any{"slug": cd.EntitySlug, "kind": result.signalKind, "evidence": result.evidence})
	}
	applyAggregateHealth(ctx, pool, q, targetID, cd.EntitySlug, prevHealth)
}

// applyAggregateHealth sets the target's health to the worst verdict across
// all of its enabled checks, and emits health.changed only when that aggregate
// actually moves.
//
// Health is a property of the entity, but each check only ever observes one
// facet of it — reachability, disk, a systemd unit. Letting whichever check
// finished last overwrite the entity's health meant a host with six checks
// reported whichever facet was sampled most recently, so one failing probe and
// five passing ones oscillated forever instead of settling on "degraded".
func applyAggregateHealth(ctx context.Context, pool *db.Pool, q *sqlcgen.Queries,
	targetID uuid.UUID, checkSlug, prevHealth string) {

	health, err := q.WorstHealthForTarget(ctx, &targetID)
	if err != nil {
		slog.Error("scheduler: aggregate health", "entity", checkSlug, "error", err)
		return
	}

	_ = q.UpsertEntityStatus(ctx, sqlcgen.UpsertEntityStatusParams{
		EntityID:    targetID,
		Health:      health,
		LastCheckAt: &[]time.Time{time.Now()}[0],
		Details:     []byte(`{}`),
	})

	if prevHealth == health {
		return
	}
	severity := "info"
	switch health {
	case "down":
		severity = "critical"
	case "degraded", "stale":
		severity = "warning"
	}
	emitSchedulerEvent(ctx, pool, "health.changed", targetID, severity,
		map[string]any{"slug": checkSlug, "from": prevHealth, "to": health})
}

// currentHealth reads the last recorded health for an entity, or "" if none.
func currentHealth(ctx context.Context, pool *db.Pool, entityID uuid.UUID) string {
	var health string
	if err := pool.QueryRow(ctx,
		`SELECT health FROM entity_status WHERE entity_id = $1`, entityID).Scan(&health); err != nil {
		return ""
	}
	return health
}

// emitSchedulerEvent records a scheduler-sourced event for SSE fan-out.
func emitSchedulerEvent(ctx context.Context, pool *db.Pool, eventType string, entityID uuid.UUID, severity string, data map[string]any) {
	_ = observability.Event(ctx, sqlcgen.New(pool), eventType, &entityID, severity, "scheduler", "", data)
}

// resolveSignal resolves any open signal raised by the given check entity.
// checkID matches how signals are keyed (UpsertSignal uses the check's own
// entity id); targetID is the observed entity whose status this affects.
func resolveSignal(ctx context.Context, pool *db.Pool, checkID, targetID uuid.UUID, slug string) {
	// Check if there's an open signal on this entity
	tag, err := pool.Exec(ctx, `UPDATE signals SET state = 'resolved', updated_at = now()
		WHERE entity_id = $1 AND state = 'raised'`, checkID)
	if err != nil {
		return
	}
	if tag.RowsAffected() > 0 {
		emitSchedulerEvent(ctx, pool, "signal.resolved", targetID, "info",
			map[string]any{"slug": slug})
		slog.Info("scheduler: signal resolved", "entity", slug)
	}
	// Deliberately does NOT write health. Resolving THIS check's signal says
	// nothing about the target's other checks; the caller re-derives health
	// from all of them. Forcing "healthy" here was a second path by which one
	// passing probe erased another probe's genuine failure.
}

// checkResult bundles the outcome of a single check execution.
type checkResult struct {
	health     string
	signalKind string
	evidence   string
	metrics    map[string]float64
	err        error
}

// executeCheck dispatches to the appropriate checker by kind.
func executeCheck(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) checkResult {
	switch cd.Kind {
	case "http":
		return checkHTTP(ctx, cd)
	case "tcp":
		return checkTCP(ctx, cd)
	case "disk":
		return checkDisk(ctx, cd)
	case "cert-expiry":
		return checkCertExpiry(ctx, cd)
	case "ping":
		return checkPing(ctx, cd)
	case "ssh-script":
		return checkSSHScript(ctx, cd)
	case "backup-freshness":
		return checkBackupFreshness(ctx, cd)
	default:
		return checkResult{health: "unknown"}
	}
}

// housekeeping runs background maintenance tasks.
func housekeeping(ctx context.Context, pool *db.Pool) {

	// Prune expired idempotency keys (older than 24h)
	cutoff := time.Now().Add(-24 * time.Hour)
	_, err := pool.Exec(ctx,
		"DELETE FROM idempotency_keys WHERE created_at < $1", cutoff)
	if err != nil {
		slog.Error("scheduler: prune idempotency keys", "error", err)
	}

	staleSweep(ctx, pool)
	coverageSweep(ctx, pool)

	// Log housekeeping completion
	slog.Debug("scheduler: housekeeping done", "pruned_idempotency_before", cutoff.Format(time.RFC3339))
}

// staleMultiplier and staleFloor bound how long an entity can go unobserved
// before its last-known health is no longer trusted. An entity is stale once
// it has gone longer than staleMultiplier times its fastest enabled check's
// interval (or staleFloor, whichever is larger) without a fresh observation —
// covering both a stalled scheduler and a disabled/broken check_def.
const (
	staleMultiplier = 3
	staleFloor      = 5 * time.Minute
)

// staleSweep marks entities whose last observation has aged past their
// check's expected cadence as 'stale', so the system never reports an old
// health value as if it were current. Runs once per housekeeping pass.
func staleSweep(ctx context.Context, pool *db.Pool) {
	rows, err := pool.Query(ctx, `
		SELECT e.id, e.slug, st.health
		FROM entity_status st
		JOIN entities e ON e.id = st.entity_id
		JOIN (
			SELECT target_id, MIN(interval_s) AS min_interval
			FROM check_defs
			WHERE enabled AND target_id IS NOT NULL
			GROUP BY target_id
		) iv ON iv.target_id = st.entity_id
		WHERE st.health <> 'stale'
		  AND (st.last_check_at IS NULL
		       OR st.last_check_at < now() - make_interval(secs => GREATEST(iv.min_interval * $1, $2)))`,
		staleMultiplier, int(staleFloor.Seconds()))
	if err != nil {
		slog.Error("scheduler: stale sweep query", "error", err)
		return
	}
	defer rows.Close()

	type staleEntity struct {
		id     uuid.UUID
		slug   string
		health string
	}
	var stale []staleEntity
	for rows.Next() {
		var se staleEntity
		if err := rows.Scan(&se.id, &se.slug, &se.health); err != nil {
			continue
		}
		stale = append(stale, se)
	}
	rows.Close()

	for _, se := range stale {
		_, err := pool.Exec(ctx,
			`UPDATE entity_status SET health = 'stale', updated_at = now() WHERE entity_id = $1`, se.id)
		if err != nil {
			slog.Error("scheduler: mark stale", "entity", se.slug, "error", err)
			continue
		}
		slog.Warn("scheduler: entity stale", "entity", se.slug, "prev_health", se.health)
		emitSchedulerEvent(ctx, pool, "health.stale", se.id, "warning",
			map[string]any{"slug": se.slug, "from": se.health, "to": "stale"})
	}
}

// checkHTTP performs an HTTP health check.
func checkHTTP(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) checkResult {
	cfg := struct {
		URL            string `json:"url"`
		ExpectedStatus int    `json:"expected_status"`
		// MaxStatus accepts a range instead of one exact code. Most services
		// sit behind Authentik and answer 302 or 401 — a working service, but
		// an exact-match on 200 reports it degraded and raises a signal.
		// Unset expected_status means "any response below MaxStatus is fine".
		MaxStatus int  `json:"max_status"`
		Insecure  bool `json:"insecure"`
	}{
		MaxStatus: 500,
	}
	if len(cd.Config) > 0 {
		_ = json.Unmarshal(cd.Config, &cfg)
	}
	if cfg.URL == "" {
		return checkResult{health: "healthy"}
	}

	timeout := time.Duration(cd.TimeoutS) * time.Second
	if timeout <= 0 {
		timeout = 30 * time.Second
	}

	client := &http.Client{
		Timeout: timeout,
		Transport: &http.Transport{
			TLSClientConfig: &tls.Config{InsecureSkipVerify: cfg.Insecure},
		},
	}

	req, err := http.NewRequestWithContext(ctx, http.MethodGet, cfg.URL, nil)
	if err != nil {
		return checkResult{
			health: "down", signalKind: "http",
			evidence: fmt.Sprintf("invalid URL %q: %v", cfg.URL, err),
			err:      err,
		}
	}

	resp, err := client.Do(req)
	if err != nil {
		return checkResult{
			health: "down", signalKind: "http",
			evidence: fmt.Sprintf("GET %s: %v", cfg.URL, err),
			err:      err,
		}
	}
	defer resp.Body.Close()

	if cfg.ExpectedStatus != 0 {
		if resp.StatusCode != cfg.ExpectedStatus {
			return checkResult{
				health: "degraded", signalKind: "http",
				evidence: fmt.Sprintf("GET %s returned %d (expected %d)", cfg.URL, resp.StatusCode, cfg.ExpectedStatus),
			}
		}
	} else if resp.StatusCode >= cfg.MaxStatus {
		return checkResult{
			health: "degraded", signalKind: "http",
			evidence: fmt.Sprintf("GET %s returned %d (expected below %d)", cfg.URL, resp.StatusCode, cfg.MaxStatus),
		}
	}

	return checkResult{health: "healthy"}
}

// checkTCP performs a TCP dial check.
func checkTCP(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) checkResult {
	cfg := struct {
		Host string `json:"host"`
		Port int    `json:"port"`
	}{}
	if len(cd.Config) > 0 {
		_ = json.Unmarshal(cd.Config, &cfg)
	}
	if cfg.Host == "" || cfg.Port == 0 {
		return checkResult{health: "healthy"}
	}

	timeout := time.Duration(cd.TimeoutS) * time.Second
	if timeout <= 0 {
		timeout = 30 * time.Second
	}

	addr := net.JoinHostPort(cfg.Host, fmt.Sprintf("%d", cfg.Port))
	conn, err := net.DialTimeout("tcp", addr, timeout)
	if err != nil {
		return checkResult{
			health: "down", signalKind: "tcp",
			evidence: fmt.Sprintf("dial %s: %v", addr, err),
			err:      err,
		}
	}
	conn.Close()
	return checkResult{health: "healthy"}
}

// checkDisk performs a disk usage check.
func checkDisk(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) checkResult {
	cfg := struct {
		Path         string `json:"path"`
		ThresholdPct int    `json:"threshold_pct"`
	}{
		Path:         "/",
		ThresholdPct: 85,
	}
	if len(cd.Config) > 0 {
		_ = json.Unmarshal(cd.Config, &cfg)
	}

	var stat unix.Statfs_t
	if err := unix.Statfs(cfg.Path, &stat); err != nil {
		return checkResult{
			health: "down", signalKind: "disk",
			evidence: fmt.Sprintf("statfs %s: %v", cfg.Path, err),
			err:      err,
		}
	}

	total := stat.Blocks * uint64(stat.Bsize)
	free := stat.Bfree * uint64(stat.Bsize)
	if total == 0 {
		return checkResult{health: "healthy"}
	}

	usedPct := float64(total-free) / float64(total) * 100
	inodePct := 0.0
	if stat.Files > 0 {
		inodePct = float64(stat.Files-stat.Ffree) / float64(stat.Files) * 100
	}

	metrics := map[string]float64{
		"disk_used_pct":  usedPct,
		"disk_inode_pct": inodePct,
	}

	if usedPct > float64(cfg.ThresholdPct) {
		return checkResult{
			health: "degraded", signalKind: "disk",
			evidence: fmt.Sprintf("%s %.1f%% full (threshold %d%%)", cfg.Path, usedPct, cfg.ThresholdPct),
			metrics:  metrics,
		}
	}

	return checkResult{health: "healthy", metrics: metrics}
}

// checkCertExpiry checks TLS certificate expiry.
func checkCertExpiry(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) checkResult {
	cfg := struct {
		Host     string `json:"host"`
		Port     int    `json:"port"`
		WarnDays int    `json:"warn_days"`
		CritDays int    `json:"crit_days"`
	}{
		Port:     443,
		WarnDays: 30,
		CritDays: 7,
	}
	if len(cd.Config) > 0 {
		_ = json.Unmarshal(cd.Config, &cfg)
	}
	if cfg.Host == "" {
		return checkResult{health: "healthy"}
	}

	timeout := time.Duration(cd.TimeoutS) * time.Second
	if timeout <= 0 {
		timeout = 30 * time.Second
	}

	addr := net.JoinHostPort(cfg.Host, fmt.Sprintf("%d", cfg.Port))

	d := tls.Dialer{Config: &tls.Config{InsecureSkipVerify: true}}
	conn, err := d.DialContext(ctx, "tcp", addr)
	if err != nil {
		return checkResult{
			health: "down", signalKind: "cert-expiry",
			evidence: fmt.Sprintf("TLS dial %s: %v", addr, err),
			err:      err,
		}
	}
	defer conn.Close()

	tlsConn := conn.(*tls.Conn)
	cs := tlsConn.ConnectionState()
	if len(cs.PeerCertificates) == 0 {
		return checkResult{
			health: "down", signalKind: "cert-expiry",
			evidence: "no peer certificates",
		}
	}

	cert := cs.PeerCertificates[0]
	daysLeft := int(time.Until(cert.NotAfter).Hours() / 24)

	metrics := map[string]float64{
		"cert_days_left": float64(daysLeft),
	}

	if daysLeft <= cfg.CritDays {
		return checkResult{
			health: "down", signalKind: "cert-expiry",
			evidence: fmt.Sprintf("%s expires in %d days (crit=%d)", cfg.Host, daysLeft, cfg.CritDays),
			metrics:  metrics,
		}
	}
	if daysLeft <= cfg.WarnDays {
		return checkResult{
			health: "degraded", signalKind: "cert-expiry",
			evidence: fmt.Sprintf("%s expires in %d days (warn=%d)", cfg.Host, daysLeft, cfg.WarnDays),
			metrics:  metrics,
		}
	}

	return checkResult{health: "healthy", metrics: metrics}
}

// checkPing performs an ICMP ping check using the system ping command.
func checkPing(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) checkResult {
	cfg := struct {
		Host  string `json:"host"`
		Count int    `json:"count"`
		// Port for the TCP fallback below. Defaults to 22; set it for hosts
		// that answer on something else (a Home Assistant VM has no sshd).
		Port int `json:"port"`
	}{}
	if len(cd.Config) > 0 {
		_ = json.Unmarshal(cd.Config, &cfg)
	}
	if cfg.Host == "" {
		return checkResult{health: "healthy"}
	}
	if cfg.Count <= 0 {
		cfg.Count = 1
	}

	timeout := time.Duration(cd.TimeoutS) * time.Second
	if timeout <= 0 {
		timeout = 10 * time.Second
	}

	deadline := time.Duration(cfg.Count+1) * timeout
	ctx, cancel := context.WithTimeout(ctx, deadline)
	defer cancel()

	countStr := strconv.Itoa(cfg.Count)
	timeoutSec := strconv.Itoa(int(timeout.Seconds()))
	if timeoutSec == "0" {
		timeoutSec = "1"
	}

	cmd := exec.CommandContext(ctx, "ping", "-c", countStr, "-W", timeoutSec, cfg.Host)
	if runtime.GOOS == "darwin" {
		cmd = exec.CommandContext(ctx, "ping", "-c", countStr, "-t", timeoutSec, cfg.Host)
	}

	output, err := cmd.Output()
	if err != nil {
		// ICMP failing does not mean the host is down — it may mean ICMP is
		// simply unavailable from here. On this deployment the scheduler runs
		// in Docker on macOS, whose VM network stack does not route ICMP to
		// the LAN: loopback pings succeed, every LAN ping fails, and all seven
		// ping checks reported "down" for hosts that were demonstrably up
		// (including the Docker host itself). Under health aggregation that one
		// broken probe was enough to drag each entity to down.
		//
		// The question this check exists to answer is "is it reachable", and
		// ICMP is only one way to ask. Fall back to a TCP connect before
		// concluding anything.
		if tcpReachable(ctx, cfg.Host, cfg.Port, timeout) {
			return checkResult{
				health:  "healthy",
				metrics: map[string]float64{},
			}
		}
		return checkResult{
			health: "down", signalKind: "ping",
			evidence: fmt.Sprintf("no ICMP or TCP response from %s: %v", cfg.Host, err),
			err:      err,
		}
	}

	latency := parsePingLatency(output)
	metrics := map[string]float64{}
	if latency > 0 {
		metrics["ping_latency_ms"] = latency
	}

	return checkResult{health: "healthy", metrics: metrics}
}

// tcpReachable reports whether a TCP handshake completes, used as the
// reachability answer when ICMP is unavailable rather than unanswered.
func tcpReachable(ctx context.Context, host string, port int, timeout time.Duration) bool {
	if port == 0 {
		port = 22
	}
	if timeout <= 0 {
		timeout = 5 * time.Second
	}
	d := net.Dialer{Timeout: timeout}
	conn, err := d.DialContext(ctx, "tcp", net.JoinHostPort(host, strconv.Itoa(port)))
	if err != nil {
		return false
	}
	conn.Close()
	return true
}

var pingRttRe = regexp.MustCompile(`(?:rtt\s+min\/avg\/max\/mdev|round-trip\s+min\/avg\/max\/stddev)\s*=\s*[\d.]+\/([\d.]+)\/`)

func parsePingLatency(output []byte) float64 {
	matches := pingRttRe.FindSubmatch(output)
	if len(matches) < 2 {
		return 0
	}
	val, err := strconv.ParseFloat(string(matches[1]), 64)
	if err != nil {
		return 0
	}
	return val
}

// checkSSHScript executes an allowlisted script on a remote host via SSH.
func checkSSHScript(ctx context.Context, cd sqlcgen.ListEnabledCheckDefsRow) checkResult {
	cfg := struct {
		Host   string `json:"host"`
		Port   int    `json:"port"`
		User   string `json:"user"`
		Script string `json:"script"`
		// Args is a single positional argument for the script. checkdefaults
		// has always written it for process_check.sh, but nothing read it —
		// so every process check ran argument-less and process_check.sh
		// answered "no service name provided" with health unknown.
		Args string `json:"args"`
	}{}
	if len(cd.Config) > 0 {
		_ = json.Unmarshal(cd.Config, &cfg)
	}
	if cfg.Host == "" || cfg.Script == "" {
		return checkResult{health: "healthy"}
	}
	if cfg.Port == 0 {
		cfg.Port = 22
	}
	if cfg.User == "" {
		cfg.User = sshUser
	}

	if !allowlistedScript(cfg.Script) {
		return checkResult{
			health: "unknown", signalKind: "ssh-script",
			evidence: fmt.Sprintf("script %q not allowlisted", cfg.Script),
		}
	}

	timeout := time.Duration(cd.TimeoutS) * time.Second
	if timeout <= 0 {
		timeout = 10 * time.Second
	}

	ctx, cancel := context.WithTimeout(ctx, timeout)
	defer cancel()

	scriptPath := "/opt/oikos/checks/" + cfg.Script
	if cfg.Args != "" {
		// Single-quote the argument so an entity name can never break out of
		// the remote command. The script name itself is allowlisted above.
		scriptPath += " '" + strings.ReplaceAll(cfg.Args, "'", `'\''`) + "'"
	}
	port := strconv.Itoa(cfg.Port)

	output, err := sshExec(ctx, cfg.Host, port, cfg.User, scriptPath, timeout)
	if err != nil {
		return checkResult{
			health: "down", signalKind: "ssh-script",
			evidence: fmt.Sprintf("ssh %s:%s %s: %v", cfg.Host, strconv.Itoa(cfg.Port), cfg.Script, err),
			err:      err,
		}
	}

	type scriptOutput struct {
		Health     string             `json:"health"`
		SignalKind string             `json:"signalKind"`
		Evidence   string             `json:"evidence"`
		Metrics    map[string]float64 `json:"metrics"`
	}
	var so scriptOutput
	if err := json.Unmarshal(output, &so); err != nil {
		return checkResult{
			health: "down", signalKind: "ssh-script",
			evidence: fmt.Sprintf("invalid script output from %s: %v", cfg.Script, err),
			err:      err,
		}
	}

	health := so.Health
	if health == "" {
		health = "healthy"
	}

	metrics := so.Metrics
	if metrics == nil {
		metrics = make(map[string]float64)
	}

	return checkResult{
		health:     health,
		signalKind: so.SignalKind,
		evidence:   so.Evidence,
		metrics:    metrics,
	}
}

var scriptNameRe = regexp.MustCompile(`^[a-z][a-z0-9_-]+\.sh$`)

func allowlistedScript(name string) bool {
	return scriptNameRe.MatchString(name)
}

func sshExec(ctx context.Context, host, port, user, cmd string, timeout time.Duration) ([]byte, error) {
	args := []string{
		"-o", "ConnectTimeout=" + strconv.Itoa(int(timeout.Seconds())),
		"-o", "StrictHostKeyChecking=no",
		"-o", "BatchMode=yes",
		"-o", "UserKnownHostsFile=/dev/null",
		"-o", "LogLevel=ERROR",
	}
	if sshKeyPath != "" {
		args = append(args, "-i", sshKeyPath)
	}
	if port != "" && port != "22" {
		args = append(args, "-p", port)
	}
	args = append(args, "-l", user, host, cmd)
	c := exec.CommandContext(ctx, "ssh", args...)
	out, err := c.Output()
	if err != nil {
		var ee *exec.ExitError
		if errors.As(err, &ee) {
			return nil, fmt.Errorf("ssh %s: %v (stderr: %s)", host, err, string(ee.Stderr))
		}
		return nil, fmt.Errorf("ssh %s: %v", host, err)
	}
	return out, nil
}

// metricThreshold defines warn/crit thresholds for a single metric.
type metricThreshold struct {
	Warn float64 `json:"warn"`
	Crit float64 `json:"crit"`
}

// thresholdsConfig is parsed from check_defs.config.thresholds JSONB.
type thresholdsConfig map[string]metricThreshold

// evaluateSeverity determines signal severity from check result and thresholds.
func evaluateSeverity(kind string, signalKind string, config []byte, metrics map[string]float64) string {
	var thresholds thresholdsConfig
	if len(config) > 0 {
		_ = json.Unmarshal(config, &thresholds)
	}

	for metric, value := range metrics {
		t, ok := thresholds[metric]
		if !ok {
			continue
		}
		if t.Crit > 0 && value >= t.Crit {
			return "critical"
		}
		if t.Warn > 0 && value >= t.Warn {
			return "warning"
		}
	}

	if signalKind == "down" {
		return "critical"
	}
	return "warning"
}

var _ = uuid.UUID{} // ensure uuid import stays