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