glpi-neural-brain/internal/sourceagent/controller_store.go

package sourceagent

import (
	"context"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"path/filepath"
	"sort"
	"strings"
	"time"
)

const controllerPolicyMetaKey = "docker_controller_policy_v1"

func defaultControllerPolicy() ControllerPolicy {
	return ControllerPolicy{
		Enabled: false, AutonomousEnabled: false, DryRun: true, AllowDestructive: false,
		MaxConcurrentJobs: 1, MaxJobDurationText: "10m",
		AllowedImages:       []string{"curlimages/curl:"},
		ProtectedContainers: []string{"brain", "*brain*", "source-agent", "*source-agent*"},
	}
}

func normalizeControllerPolicy(in ControllerPolicy) (ControllerPolicy, error) {
	if strings.TrimSpace(in.MaxJobDurationText) == "" {
		in.MaxJobDurationText = "10m"
	}
	d, err := time.ParseDuration(in.MaxJobDurationText)
	if err != nil || d < 5*time.Second || d > 2*time.Hour {
		return in, errors.New("max_job_duration must be between 5s and 2h")
	}
	in.MaxJobDuration = d
	if in.MaxConcurrentJobs < 1 {
		in.MaxConcurrentJobs = 1
	}
	if in.MaxConcurrentJobs > 8 {
		return in, errors.New("max_concurrent_jobs must be between 1 and 8")
	}
	in.AllowedImages = uniqueStrings(in.AllowedImages)
	roots := make([]string, 0, len(in.AllowedComposeRoots))
	for _, root := range uniqueStrings(in.AllowedComposeRoots) {
		abs, err := filepath.Abs(strings.TrimSpace(root))
		if err != nil {
			return in, fmt.Errorf("invalid compose root %q: %w", root, err)
		}
		roots = append(roots, filepath.Clean(abs))
	}
	in.AllowedComposeRoots = uniqueStrings(roots)
	in.ProtectedContainers = uniqueStrings(in.ProtectedContainers)
	in.ProtectedNetworks = uniqueStrings(in.ProtectedNetworks)
	in.ProtectedVolumes = uniqueStrings(in.ProtectedVolumes)
	return in, nil
}

func (s *Store) initController(ctx context.Context) error {
	stmts := []string{
		`CREATE TABLE IF NOT EXISTS controller_profiles (
			id TEXT PRIMARY KEY, name TEXT NOT NULL, purpose TEXT NOT NULL, kind TEXT NOT NULL,
			agent_id TEXT NOT NULL DEFAULT '', enabled INTEGER NOT NULL DEFAULT 1, autonomous INTEGER NOT NULL DEFAULT 0,
			config_json TEXT NOT NULL DEFAULT '{}', created_at_ns INTEGER NOT NULL, updated_at_ns INTEGER NOT NULL,
			last_run_at_ns INTEGER NOT NULL DEFAULT 0
		) WITHOUT ROWID`,
		`CREATE TABLE IF NOT EXISTS controller_jobs (
			id TEXT PRIMARY KEY, agent_id TEXT NOT NULL DEFAULT '', profile_id TEXT NOT NULL DEFAULT '', kind TEXT NOT NULL,
			purpose TEXT NOT NULL DEFAULT '', status TEXT NOT NULL, autonomous INTEGER NOT NULL DEFAULT 0,
			dry_run INTEGER NOT NULL DEFAULT 1, parameters_json TEXT NOT NULL DEFAULT '{}', result_json TEXT NOT NULL DEFAULT '{}',
			error TEXT NOT NULL DEFAULT '', created_at_ns INTEGER NOT NULL, updated_at_ns INTEGER NOT NULL,
			started_at_ns INTEGER NOT NULL DEFAULT 0, completed_at_ns INTEGER NOT NULL DEFAULT 0,
			lease_until_ns INTEGER NOT NULL DEFAULT 0, claimed_by TEXT NOT NULL DEFAULT ''
		) WITHOUT ROWID`,
		`CREATE INDEX IF NOT EXISTS idx_controller_jobs_status_created ON controller_jobs(status, created_at_ns)`,
		`CREATE INDEX IF NOT EXISTS idx_controller_profiles_autonomous ON controller_profiles(enabled, autonomous, kind)`,
	}
	for _, stmt := range stmts {
		if _, err := s.db.ExecContext(ctx, stmt); err != nil {
			return err
		}
	}
	if err := s.ensureAgentColumn(ctx, "controller_json", `TEXT NOT NULL DEFAULT '{}'`); err != nil {
		return err
	}
	// Claims cannot survive a Brain restart because the in-flight HTTP action and
	// authorization watcher disappeared with the previous process.
	now := time.Now().UTC().UnixNano()
	_, err := s.db.ExecContext(ctx, `UPDATE controller_jobs SET status='queued',claimed_by='',lease_until_ns=0,updated_at_ns=? WHERE status='claimed'`, now)
	return err
}

func (s *Store) ControllerPolicy(ctx context.Context) (ControllerPolicy, error) {
	p := defaultControllerPolicy()
	var raw string
	err := s.db.QueryRowContext(ctx, `SELECT value FROM source_meta WHERE key=?`, controllerPolicyMetaKey).Scan(&raw)
	if errors.Is(err, sql.ErrNoRows) {
		return normalizeControllerPolicy(p)
	}
	if err != nil {
		return p, err
	}
	if err := json.Unmarshal([]byte(raw), &p); err != nil {
		return p, err
	}
	return normalizeControllerPolicy(p)
}

func (s *Store) SetControllerPolicy(ctx context.Context, p ControllerPolicy) (ControllerPolicy, error) {
	p, err := normalizeControllerPolicy(p)
	if err != nil {
		return p, err
	}
	raw, _ := json.Marshal(p)
	if _, err := s.db.ExecContext(ctx, `INSERT INTO source_meta(key,value) VALUES(?,?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, controllerPolicyMetaKey, string(raw)); err != nil {
		return p, err
	}
	if !p.Enabled {
		now := time.Now().UTC().UnixNano()
		_, _ = s.db.ExecContext(ctx, `UPDATE controller_jobs SET status='canceled',error='controller master switch disabled',completed_at_ns=?,updated_at_ns=? WHERE status='queued'`, now, now)
	} else if !p.AutonomousEnabled {
		now := time.Now().UTC().UnixNano()
		_, _ = s.db.ExecContext(ctx, `UPDATE controller_jobs SET status='canceled',error='autonomous controller switch disabled',completed_at_ns=?,updated_at_ns=? WHERE status='queued' AND autonomous=1`, now, now)
	}
	if p.Enabled && !p.AllowDestructive {
		now := time.Now().UTC().UnixNano()
		_, _ = s.db.ExecContext(ctx, `UPDATE controller_jobs SET status='canceled',error='destructive controller actions disabled',completed_at_ns=?,updated_at_ns=? WHERE status='queued' AND kind IN ('container_remove','network_remove','volume_remove','compose_down')`, now, now)
	}
	return p, nil
}

func normalizeControllerProfile(p ControllerProfile) (ControllerProfile, error) {
	p.ID = normalizeID(p.ID, "controller-profile")
	p.Name = strings.TrimSpace(p.Name)
	p.Purpose = strings.ToLower(strings.TrimSpace(p.Purpose))
	p.Kind = strings.TrimSpace(p.Kind)
	p.AgentID = strings.TrimSpace(p.AgentID)
	if p.Name == "" || p.Kind == "" {
		return p, errors.New("name and kind are required")
	}
	switch p.Purpose {
	case "research", "test", "performance", "validation", "recovery", "operations":
	default:
		return p, errors.New("purpose must be research, test, performance, validation, recovery or operations")
	}
	if !validControllerKind(p.Kind) {
		return p, fmt.Errorf("unsupported controller kind %q", p.Kind)
	}
	if p.Config == nil {
		p.Config = map[string]any{}
	}
	return p, nil
}

func validControllerKind(kind string) bool {
	switch strings.TrimSpace(kind) {
	case "evidence_http_probe", "health_recovery", "compute_capacity_compose", "compose_smoke_test",
		"container_start", "container_stop", "container_restart", "container_remove", "container_create",
		"network_create", "network_remove", "volume_create", "volume_remove",
		"compose_up", "compose_down", "compose_restart", "compose_pull", "compose_ps", "inventory_refresh":
		return true
	default:
		return false
	}
}

func validateControllerParametersForStorage(job ControllerJob) error {
	// Controller jobs and profiles are deliberately auditable in SQLite and the
	// dashboard. Do not accept obvious inline credentials that would turn the
	// audit trail into a secret store. Long-lived services should use Docker/
	// Compose secrets or environment files referenced by an operator-approved
	// Compose file instead.
	if job.Kind != "container_create" {
		return nil
	}
	for _, entry := range paramStrings(job.Parameters, "env", 64, 4096) {
		key := entry
		if idx := strings.IndexByte(key, '='); idx >= 0 {
			key = key[:idx]
		}
		upper := strings.ToUpper(strings.TrimSpace(key))
		for _, marker := range []string{"PASSWORD", "PASSWD", "TOKEN", "SECRET", "API_KEY", "APIKEY", "PRIVATE_KEY", "ACCESS_KEY", "CREDENTIAL"} {
			if strings.Contains(upper, marker) {
				return fmt.Errorf("inline secret-like environment variable %q is not allowed in controller job history; use an approved Compose secret/env file", key)
			}
		}
	}
	return nil
}

func (s *Store) UpsertControllerProfile(ctx context.Context, p ControllerProfile) (ControllerProfile, error) {
	var err error
	p, err = normalizeControllerProfile(p)
	if err != nil {
		return p, err
	}
	now := time.Now().UTC()
	if p.CreatedAt.IsZero() {
		var created int64
		if err := s.db.QueryRowContext(ctx, `SELECT created_at_ns FROM controller_profiles WHERE id=?`, p.ID).Scan(&created); err == nil && created > 0 {
			p.CreatedAt = time.Unix(0, created).UTC()
		} else {
			p.CreatedAt = now
		}
	}
	p.UpdatedAt = now
	raw, _ := json.Marshal(p.Config)
	_, err = s.db.ExecContext(ctx, `INSERT INTO controller_profiles(id,name,purpose,kind,agent_id,enabled,autonomous,config_json,created_at_ns,updated_at_ns,last_run_at_ns)
		VALUES(?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(id) DO UPDATE SET name=excluded.name,purpose=excluded.purpose,kind=excluded.kind,agent_id=excluded.agent_id,enabled=excluded.enabled,autonomous=excluded.autonomous,config_json=excluded.config_json,updated_at_ns=excluded.updated_at_ns`,
		p.ID, p.Name, p.Purpose, p.Kind, p.AgentID, boolInt(p.Enabled), boolInt(p.Autonomous), string(raw), p.CreatedAt.UnixNano(), now.UnixNano(), timeToNS(p.LastRunAt))
	return p, err
}

func (s *Store) DeleteControllerProfile(ctx context.Context, id string) error {
	_, err := s.db.ExecContext(ctx, `DELETE FROM controller_profiles WHERE id=?`, strings.TrimSpace(id))
	return err
}

func (s *Store) ListControllerProfiles(ctx context.Context) ([]ControllerProfile, error) {
	rows, err := s.db.QueryContext(ctx, `SELECT id,name,purpose,kind,agent_id,enabled,autonomous,config_json,created_at_ns,updated_at_ns,last_run_at_ns FROM controller_profiles ORDER BY purpose,name,id`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []ControllerProfile
	for rows.Next() {
		var p ControllerProfile
		var en, au int
		var raw string
		var c, u, l int64
		if err := rows.Scan(&p.ID, &p.Name, &p.Purpose, &p.Kind, &p.AgentID, &en, &au, &raw, &c, &u, &l); err != nil {
			return nil, err
		}
		p.Enabled = en != 0
		p.Autonomous = au != 0
		p.CreatedAt = nsTime(c)
		p.UpdatedAt = nsTime(u)
		p.LastRunAt = nsTime(l)
		_ = json.Unmarshal([]byte(raw), &p.Config)
		out = append(out, p)
	}
	return out, rows.Err()
}

func (s *Store) ControllerProfile(ctx context.Context, id string) (ControllerProfile, error) {
	var p ControllerProfile
	var en, au int
	var raw string
	var c, u, l int64
	err := s.db.QueryRowContext(ctx, `SELECT id,name,purpose,kind,agent_id,enabled,autonomous,config_json,created_at_ns,updated_at_ns,last_run_at_ns FROM controller_profiles WHERE id=?`, strings.TrimSpace(id)).Scan(&p.ID, &p.Name, &p.Purpose, &p.Kind, &p.AgentID, &en, &au, &raw, &c, &u, &l)
	if err != nil {
		return p, err
	}
	p.Enabled = en != 0
	p.Autonomous = au != 0
	p.CreatedAt = nsTime(c)
	p.UpdatedAt = nsTime(u)
	p.LastRunAt = nsTime(l)
	_ = json.Unmarshal([]byte(raw), &p.Config)
	return p, nil
}

func (s *Store) QueueControllerProfile(ctx context.Context, profileID string, autonomous bool, overrides map[string]any) (ControllerJob, error) {
	p, err := s.ControllerProfile(ctx, profileID)
	if err != nil {
		return ControllerJob{}, err
	}
	if !p.Enabled {
		return ControllerJob{}, errors.New("controller profile disabled")
	}
	if autonomous && !p.Autonomous {
		return ControllerJob{}, errors.New("controller profile is not approved for autonomous execution")
	}
	params := cloneAnyMap(p.Config)
	for k, v := range overrides {
		params[k] = v
	}
	job := ControllerJob{AgentID: p.AgentID, ProfileID: p.ID, Kind: p.Kind, Purpose: p.Purpose, Autonomous: autonomous, Parameters: params}
	return s.QueueControllerJob(ctx, job)
}

func (s *Store) QueueControllerJob(ctx context.Context, job ControllerJob) (ControllerJob, error) {
	policy, err := s.ControllerPolicy(ctx)
	if err != nil {
		return job, err
	}
	if !policy.Enabled {
		return job, errors.New("docker controller master switch is disabled")
	}
	if job.Autonomous && !policy.AutonomousEnabled {
		return job, errors.New("autonomous docker controller actions are disabled")
	}
	if !validControllerKind(job.Kind) {
		return job, errors.New("unsupported controller job kind")
	}
	if err := validateControllerParametersForStorage(job); err != nil {
		return job, err
	}
	job.DryRun = policy.DryRun || job.DryRun
	if !job.DryRun {
		if err := validateControllerJob(job, policy); err != nil {
			return job, err
		}
	}
	job.ID = randomID("controller")
	job.Status = ControllerJobQueued
	job.CreatedAt = time.Now().UTC()
	job.UpdatedAt = job.CreatedAt
	if job.Parameters == nil {
		job.Parameters = map[string]any{}
	}
	params, _ := json.Marshal(job.Parameters)
	result, _ := json.Marshal(map[string]any{})
	_, err = s.db.ExecContext(ctx, `INSERT INTO controller_jobs(id,agent_id,profile_id,kind,purpose,status,autonomous,dry_run,parameters_json,result_json,error,created_at_ns,updated_at_ns,started_at_ns,completed_at_ns,lease_until_ns,claimed_by) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, job.ID, job.AgentID, job.ProfileID, job.Kind, job.Purpose, job.Status, boolInt(job.Autonomous), boolInt(job.DryRun), string(params), string(result), "", job.CreatedAt.UnixNano(), job.UpdatedAt.UnixNano(), 0, 0, 0, "")
	return job, err
}

func (s *Store) ListControllerJobs(ctx context.Context, limit int) ([]ControllerJob, error) {
	if limit < 1 {
		limit = 100
	}
	if limit > 1000 {
		limit = 1000
	}
	rows, err := s.db.QueryContext(ctx, `SELECT id,agent_id,profile_id,kind,purpose,status,autonomous,dry_run,parameters_json,result_json,error,created_at_ns,updated_at_ns,started_at_ns,completed_at_ns,lease_until_ns,claimed_by FROM controller_jobs ORDER BY created_at_ns DESC LIMIT ?`, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []ControllerJob
	for rows.Next() {
		j, err := scanControllerJob(rows.Scan)
		if err != nil {
			return nil, err
		}
		out = append(out, j)
	}
	return out, rows.Err()
}

func (s *Store) ControllerJob(ctx context.Context, id string) (ControllerJob, error) {
	row := s.db.QueryRowContext(ctx, `SELECT id,agent_id,profile_id,kind,purpose,status,autonomous,dry_run,parameters_json,result_json,error,created_at_ns,updated_at_ns,started_at_ns,completed_at_ns,lease_until_ns,claimed_by FROM controller_jobs WHERE id=?`, strings.TrimSpace(id))
	return scanControllerJob(row.Scan)
}

func scanControllerJob(scan func(...any) error) (ControllerJob, error) {
	var j ControllerJob
	var au, dr int
	var p, r string
	var c, u, st, co, l int64
	err := scan(&j.ID, &j.AgentID, &j.ProfileID, &j.Kind, &j.Purpose, &j.Status, &au, &dr, &p, &r, &j.Error, &c, &u, &st, &co, &l, &j.ClaimedBy)
	if err != nil {
		return j, err
	}
	j.Autonomous = au != 0
	j.DryRun = dr != 0
	j.CreatedAt = nsTime(c)
	j.UpdatedAt = nsTime(u)
	j.StartedAt = nsTime(st)
	j.CompletedAt = nsTime(co)
	j.LeaseUntil = nsTime(l)
	_ = json.Unmarshal([]byte(p), &j.Parameters)
	_ = json.Unmarshal([]byte(r), &j.Result)
	return j, nil
}

func (s *Store) ClaimControllerJob(ctx context.Context, agentID string, lease time.Duration, canCompose bool) (ControllerJob, bool, error) {
	policy, err := s.ControllerPolicy(ctx)
	if err != nil {
		return ControllerJob{}, false, err
	}
	if !policy.Enabled {
		return ControllerJob{}, false, nil
	}
	if lease <= 0 || lease > policy.MaxJobDuration {
		lease = policy.MaxJobDuration
	}
	now := time.Now().UTC()
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return ControllerJob{}, false, err
	}
	defer tx.Rollback()
	var active int
	if err := tx.QueryRowContext(ctx, `SELECT COUNT(*) FROM controller_jobs WHERE status='claimed' AND lease_until_ns>?`, now.UnixNano()).Scan(&active); err != nil {
		return ControllerJob{}, false, err
	}
	if active >= policy.MaxConcurrentJobs {
		return ControllerJob{}, false, nil
	}
	// Requeue expired claims first.
	_, _ = tx.ExecContext(ctx, `UPDATE controller_jobs SET status='queued',claimed_by='',lease_until_ns=0,updated_at_ns=? WHERE status='claimed' AND lease_until_ns>0 AND lease_until_ns<?`, now.UnixNano(), now.UnixNano())
	var id string
	err = tx.QueryRowContext(ctx, `SELECT id FROM controller_jobs WHERE status='queued' AND (agent_id='' OR agent_id=?) AND (?=1 OR kind NOT IN ('compose_up','compose_down','compose_restart','compose_pull','compose_ps','compute_capacity_compose','compose_smoke_test')) ORDER BY created_at_ns LIMIT 1`, agentID, boolInt(canCompose)).Scan(&id)
	if errors.Is(err, sql.ErrNoRows) {
		return ControllerJob{}, false, nil
	}
	if err != nil {
		return ControllerJob{}, false, err
	}
	res, err := tx.ExecContext(ctx, `UPDATE controller_jobs SET status='claimed',claimed_by=?,started_at_ns=CASE WHEN started_at_ns=0 THEN ? ELSE started_at_ns END,lease_until_ns=?,updated_at_ns=? WHERE id=? AND status='queued'`, agentID, now.UnixNano(), now.Add(lease).UnixNano(), now.UnixNano(), id)
	if err != nil {
		return ControllerJob{}, false, err
	}
	n, _ := res.RowsAffected()
	if n != 1 {
		return ControllerJob{}, false, nil
	}
	if err := tx.Commit(); err != nil {
		return ControllerJob{}, false, err
	}
	var job ControllerJob
	row := s.db.QueryRowContext(ctx, `SELECT id,agent_id,profile_id,kind,purpose,status,autonomous,dry_run,parameters_json,result_json,error,created_at_ns,updated_at_ns,started_at_ns,completed_at_ns,lease_until_ns,claimed_by FROM controller_jobs WHERE id=?`, id)
	job, err = scanControllerJob(row.Scan)
	return job, err == nil, err
}

func (s *Store) CompleteControllerJob(ctx context.Context, agentID string, result ControllerJobResult) error {
	if strings.TrimSpace(result.JobID) == "" {
		return errors.New("job_id required")
	}
	status := ControllerJobSucceeded
	if result.Error != "" || result.Status == ControllerJobFailed {
		status = ControllerJobFailed
	}
	if result.Status == ControllerJobCanceled {
		status = ControllerJobCanceled
	}
	now := time.Now().UTC()
	raw, _ := json.Marshal(result.Result)
	res, err := s.db.ExecContext(ctx, `UPDATE controller_jobs SET status=?,result_json=?,error=?,completed_at_ns=?,updated_at_ns=?,lease_until_ns=0 WHERE id=? AND status='claimed' AND claimed_by=?`, status, string(raw), strings.TrimSpace(result.Error), now.UnixNano(), now.UnixNano(), result.JobID, agentID)
	if err != nil {
		return err
	}
	n, _ := res.RowsAffected()
	if n != 1 {
		return errors.New("controller job not found, canceled, or claimed by another agent")
	}
	_, _ = s.db.ExecContext(ctx, `UPDATE controller_profiles SET last_run_at_ns=?,updated_at_ns=? WHERE id=(SELECT profile_id FROM controller_jobs WHERE id=?) AND profile_id<>''`, now.UnixNano(), now.UnixNano(), result.JobID)
	return nil
}

func (s *Store) CancelControllerJob(ctx context.Context, id string) error {
	now := time.Now().UTC()
	res, err := s.db.ExecContext(ctx, `UPDATE controller_jobs SET status='canceled',error='canceled by Brain operator',completed_at_ns=?,updated_at_ns=? WHERE id=? AND status IN ('queued','claimed')`, now.UnixNano(), now.UnixNano(), id)
	if err != nil {
		return err
	}
	n, _ := res.RowsAffected()
	if n == 0 {
		return errors.New("controller job is not active")
	}
	return nil
}

func (s *Store) ControllerJobAuthorized(ctx context.Context, agentID, jobID string) (bool, string, error) {
	p, err := s.ControllerPolicy(ctx)
	if err != nil {
		return false, "", err
	}
	if !p.Enabled {
		return false, "controller master switch disabled", nil
	}
	var status, claimed, kind string
	var autonomous, dryRun int
	var lease int64
	err = s.db.QueryRowContext(ctx, `SELECT status,claimed_by,lease_until_ns,kind,autonomous,dry_run FROM controller_jobs WHERE id=?`, jobID).Scan(&status, &claimed, &lease, &kind, &autonomous, &dryRun)
	if errors.Is(err, sql.ErrNoRows) {
		return false, "job not found", nil
	}
	if err != nil {
		return false, "", err
	}
	if status != ControllerJobClaimed || claimed != agentID {
		return false, "job is no longer claimed by this agent", nil
	}
	if lease > 0 && time.Now().UTC().After(time.Unix(0, lease)) {
		return false, "job lease expired", nil
	}
	if err := validateControllerJob(ControllerJob{Kind: kind, Autonomous: autonomous != 0, DryRun: dryRun != 0}, p); err != nil {
		return false, err.Error(), nil
	}
	return true, "", nil
}

func (s *Store) ControllerStats(ctx context.Context) map[string]any {
	out := map[string]any{"queued": 0, "claimed": 0, "succeeded": 0, "failed": 0, "canceled": 0, "profiles": 0, "autonomous_profiles": 0}
	for _, st := range []string{ControllerJobQueued, ControllerJobClaimed, ControllerJobSucceeded, ControllerJobFailed, ControllerJobCanceled} {
		var n int
		_ = s.db.QueryRowContext(ctx, `SELECT COUNT(*) FROM controller_jobs WHERE status=?`, st).Scan(&n)
		out[st] = n
	}
	var p, a int
	_ = s.db.QueryRowContext(ctx, `SELECT COUNT(*),COALESCE(SUM(CASE WHEN enabled=1 AND autonomous=1 THEN 1 ELSE 0 END),0) FROM controller_profiles`).Scan(&p, &a)
	out["profiles"] = p
	out["autonomous_profiles"] = a
	if policy, err := s.ControllerPolicy(ctx); err == nil {
		out["policy"] = policy
	}
	return out
}

func (s *Store) AutonomousControllerProfiles(ctx context.Context, kind string) ([]ControllerProfile, error) {
	all, err := s.ListControllerProfiles(ctx)
	if err != nil {
		return nil, err
	}
	var out []ControllerProfile
	for _, p := range all {
		if p.Enabled && p.Autonomous && (kind == "" || p.Kind == kind) {
			out = append(out, p)
		}
	}
	sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID })
	return out, nil
}

func cloneAnyMap(in map[string]any) map[string]any {
	out := map[string]any{}
	for k, v := range in {
		out[k] = v
	}
	return out
}
func timeToNS(v time.Time) int64 {
	if v.IsZero() {
		return 0
	}
	return v.UTC().UnixNano()
}

func (s *Store) queueAutonomousControllerProfile(ctx context.Context, p ControllerProfile, overrides map[string]any, dedupeKey string, minInterval time.Duration) (ControllerJob, bool, error) {
	policy, err := s.ControllerPolicy(ctx)
	if err != nil || !policy.Enabled || !policy.AutonomousEnabled {
		return ControllerJob{}, false, err
	}
	if !p.Enabled || !p.Autonomous {
		return ControllerJob{}, false, nil
	}
	if minInterval <= 0 {
		minInterval = 10 * time.Minute
	}
	if strings.TrimSpace(dedupeKey) != "" {
		jobs, _ := s.ListControllerJobs(ctx, 500)
		cutoff := time.Now().UTC().Add(-minInterval)
		for _, j := range jobs {
			if j.CreatedAt.Before(cutoff) {
				continue
			}
			if key := paramString(j.Parameters, "dedupe_key"); key == dedupeKey && j.Status != ControllerJobFailed && j.Status != ControllerJobCanceled {
				return ControllerJob{}, false, nil
			}
		}
	}
	params := cloneAnyMap(p.Config)
	for k, v := range overrides {
		params[k] = v
	}
	if dedupeKey != "" {
		params["dedupe_key"] = dedupeKey
	}
	job := ControllerJob{AgentID: p.AgentID, ProfileID: p.ID, Kind: p.Kind, Purpose: p.Purpose, Autonomous: true, Parameters: params}
	job, err = s.QueueControllerJob(ctx, job)
	return job, err == nil, err
}

func (s *Store) QueueFirstAutonomousControllerProfile(ctx context.Context, kind string, overrides map[string]any, dedupeKey string, minInterval time.Duration) (ControllerJob, bool, error) {
	profiles, err := s.AutonomousControllerProfiles(ctx, kind)
	if err != nil || len(profiles) == 0 {
		return ControllerJob{}, false, err
	}
	return s.queueAutonomousControllerProfile(ctx, profiles[0], overrides, dedupeKey, minInterval)
}

func (s *Store) QueueAutonomousEvidenceProbe(ctx context.Context, targetURL, title, sourceQuality string, sourceNodeIDs []string) (ControllerJob, bool, error) {
	targetURL = strings.TrimSpace(targetURL)
	if targetURL == "" {
		return ControllerJob{}, false, nil
	}
	sum := sha256.Sum256([]byte(strings.ToLower(targetURL)))
	key := "evidence:" + hex.EncodeToString(sum[:8])
	return s.QueueFirstAutonomousControllerProfile(ctx, "evidence_http_probe", map[string]any{"url": targetURL, "title": strings.TrimSpace(title), "source_quality": strings.TrimSpace(sourceQuality), "source_node_ids": uniqueStrings(sourceNodeIDs)}, key, 6*time.Hour)
}

func (s *Store) QueueComputeCapacityController(ctx context.Context, computeKind string) (ControllerJob, bool, error) {
	key := "compute-capacity:" + strings.TrimSpace(computeKind)
	return s.QueueFirstAutonomousControllerProfile(ctx, "compute_capacity_compose", map[string]any{"compute_kind": computeKind}, key, 20*time.Minute)
}

func (s *Store) RunControllerHealthAutomation(ctx context.Context) ([]ControllerJob, error) {
	policy, err := s.ControllerPolicy(ctx)
	if err != nil || !policy.Enabled || !policy.AutonomousEnabled {
		return nil, err
	}
	profiles, err := s.AutonomousControllerProfiles(ctx, "health_recovery")
	if err != nil {
		return nil, err
	}
	agents, err := s.ListAgents(ctx)
	if err != nil {
		return nil, err
	}
	now := time.Now().UTC()
	var queued []ControllerJob
	for _, p := range profiles {
		target := paramString(p.Config, "container")
		if target == "" || protectedName(target, policy.ProtectedContainers) {
			continue
		}
		cooldown := 30 * time.Minute
		if raw := paramString(p.Config, "cooldown"); raw != "" {
			if d, e := time.ParseDuration(raw); e == nil && d >= time.Minute {
				cooldown = d
			}
		}
		if !p.LastRunAt.IsZero() && now.Sub(p.LastRunAt) < cooldown {
			continue
		}
		unhealthy := false
		observedAgent := ""
		for _, a := range agents {
			if !a.Enabled || (!p.AgentIDIsEmptyOr(a.ID)) || a.LastSeen.IsZero() || now.Sub(a.LastSeen) > 3*time.Minute {
				continue
			}
			for _, c := range a.Controller.Inventory.Containers {
				for _, n := range c.Names {
					if strings.EqualFold(strings.TrimPrefix(n, "/"), target) && c.Health == "unhealthy" {
						unhealthy = true
						observedAgent = a.ID
						break
					}
				}
				if unhealthy {
					break
				}
			}
			if unhealthy {
				break
			}
		}
		if !unhealthy {
			continue
		}
		selected := p
		if selected.AgentID == "" {
			selected.AgentID = observedAgent
		}
		job, ok, e := s.queueAutonomousControllerProfile(ctx, selected, map[string]any{"container": target}, "recovery:"+strings.ToLower(target)+":"+strings.ToLower(observedAgent), cooldown)
		if e != nil {
			return queued, e
		}
		if ok {
			queued = append(queued, job)
		}
	}
	return queued, nil
}

func (p ControllerProfile) AgentIDIsEmptyOr(id string) bool {
	return p.AgentID == "" || p.AgentID == id
}

func (s *Store) RunControllerScheduledTests(ctx context.Context) ([]ControllerJob, error) {
	policy, err := s.ControllerPolicy(ctx)
	if err != nil || !policy.Enabled || !policy.AutonomousEnabled {
		return nil, err
	}
	profiles, err := s.AutonomousControllerProfiles(ctx, "compose_smoke_test")
	if err != nil {
		return nil, err
	}
	now := time.Now().UTC()
	var queued []ControllerJob
	for _, p := range profiles {
		interval := 6 * time.Hour
		if raw := paramString(p.Config, "interval"); raw != "" {
			if d, e := time.ParseDuration(raw); e == nil && d >= 10*time.Minute {
				interval = d
			}
		}
		if !p.LastRunAt.IsZero() && now.Sub(p.LastRunAt) < interval {
			continue
		}
		job, ok, e := s.queueAutonomousControllerProfile(ctx, p, nil, "smoke:"+p.ID, interval/2)
		if e != nil {
			return queued, e
		}
		if ok {
			queued = append(queued, job)
		}
	}
	return queued, nil
}