feat(ft-recursive-002): the supervised retrain controller + operator-confirm promote (Epic 5)

ftloop.Controller: a supervised loop (default-off behind FT_LOOP_ENABLED) that
picks up a triggered cycle from the ledger and walks it curate->train->gate as
ctx-cancellable Python subprocesses, holding the compute lease + quiescing RSIC,
updating the ledger + the 6a ft-loop:<stage> jobhealth surface at each step.
FAIL -> archived + one ft-loop alert; lease-expiry/disk-floor -> class-4 high
alert-and-halt; PASS -> promote_pending (halts; promotion is operator-gated).
Started via StartSupervisedBackground (after the supervisor is set). The gate
now OPENS a cycle on a trigger decision (the executor logs, no alert — the
ledger is the controller's out-of-band signal, SPEC fork 7). mdemg ft-loop
promote --cycle-id [--reject] records the operator decision to the ledger.
AMD-1 epoch cap passed to train_ft. New config: FT_LOOP_POLL_INTERVAL_SEC,
FT_LOOP_LEASE_MAX_HOURS/PATH, FT_LOOP_MIN_FREE_DISK_GB, FT_LOOP_PYTHON_BIN,
FT_LOOP_MODEL_VERSION, FT_LORA_EPOCHS_CAP, FT_EARLY_STOP_VAL_LOSS_FACTOR.
Controller state-machine tests (stub runStage): all-stages, stage-failure halt,
disk-floor. Config guard 744/744. Subprocess arg sets validated live in Epic 6.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: rhenley1958

internal/ape/task_dispatch.go

@@ -1175,15 +1175,13 @@ func (d *Dispatcher) executeTriggerTrainingPipeline(ctx context.Context, spec RS
 		slog.Debug("RSIC trigger gate: suppressed", "decision", decision, "space", spec.TargetSpace)
 		return map[string]any{"action": spec.ActionType, "suppressed": true, "decision": decision}, nil
 	}
-	// decision == trigger: actuator enabled + fresh + clear. Until Epic 5's
-	// controller owns this path, surface the actionable signal once (the gate's
-	// interval/open-cycle checks already prevent spam).
-	out, aerr := d.executeAlertLog(ctx, spec, "Training data ready — launching retrain cycle")
-	if out != nil {
-		out["decision"] = decision
-		out["suppressed"] = false
-	}
-	return out, aerr
+	// decision == trigger: the gate has opened a cycle in the ledger; the
+	// supervised controller (a separate loop) consumes it out-of-band and
+	// alerts on the outcome (promote_pending / failed). No alert here — the
+	// ledger is the signal, and alerting on every trigger would re-introduce
+	// spam.
+	slog.Info("RSIC trigger gate: retrain cycle opened", "space", spec.TargetSpace)
+	return map[string]any{"action": spec.ActionType, "suppressed": false, "decision": decision, "cycle_opened": true}, nil
 }
 
 // executeAlertLog is a generic alert/log handler for RSIC-DATA and gap-fix actions.

internal/ape/task_dispatch_test.go

@@ -177,14 +177,19 @@ func TestExecuteTriggerTrainingPipeline_SF2(t *testing.T) {
 		t.Errorf("expected suppressed deliverables, got %v", out)
 	}
 
-	// Trigger → alert fires (under the distinct rsic-<action> service from SF-3).
+	// Trigger → the gate opened a cycle; the executor does NOT alert (the
+	// controller owns outcome alerts) — cycle_opened=true, no spam.
 	capTrig := &captureAlertDispatcher{}
 	d2 := &Dispatcher{alertDispatcher: capTrig, trainingTriggerGate: fakeTriggerGate{decision: "trigger", suppressed: false}}
-	if _, err := d2.executeTriggerTrainingPipeline(context.Background(), spec); err != nil {
+	out2, err := d2.executeTriggerTrainingPipeline(context.Background(), spec)
+	if err != nil {
 		t.Fatal(err)
 	}
-	if len(capTrig.services) != 1 || capTrig.services[0] != "rsic-trigger_training_pipeline" {
-		t.Errorf("trigger should alert under rsic-trigger_training_pipeline, got %v", capTrig.services)
+	if len(capTrig.services) != 0 {
+		t.Errorf("trigger must NOT alert (ledger is the signal), got %v", capTrig.services)
+	}
+	if out2["cycle_opened"] != true {
+		t.Errorf("trigger should report cycle_opened, got %v", out2)
 	}
 
 	// Nil gate → legacy alert (backward compat).

internal/api/server.go

@@ -2129,6 +2129,34 @@ func (s *Server) StartSupervisedBackground() {
 		s.tsdbBackupScheduler.SetSupervise(s.superviseFn)
 		s.tsdbBackupScheduler.Start()
 	}
+	// FT-RECURSIVE-002 Epic 5: the recursive-retrain controller (supervised,
+	// default-off behind FT_LOOP_ENABLED — Run returns immediately when off).
+	if s.tsdbClient != nil {
+		repoDir, _ := os.Getwd()
+		leasePath := s.cfg.FtLoopLeasePath
+		if leasePath == "" {
+			if home, err := os.UserHomeDir(); err == nil {
+				leasePath = filepath.Join(home, ".mdemg", "ft-loop.lease")
+			} else {
+				leasePath = filepath.Join(os.TempDir(), "mdemg-ft-loop.lease")
+			}
+		}
+		ctrl := ftloop.NewController(s.tsdbClient.Pool(), s.orchestrationPolicy, s.alertDispatcher,
+			ftloop.ControllerConfig{
+				Enabled:         s.cfg.FtLoopEnabled,
+				PollInterval:    time.Duration(s.cfg.FtLoopPollIntervalSec) * time.Second,
+				LeasePath:       leasePath,
+				LeaseMax:        time.Duration(s.cfg.FtLoopLeaseMaxHours) * time.Hour,
+				MinFreeDiskGB:   s.cfg.FtLoopMinFreeDiskGB,
+				PythonBin:       s.cfg.FtLoopPythonBin,
+				ModelVersion:    s.cfg.FtLoopModelVersion,
+				EpochsCap:       s.cfg.FtLoraEpochsCap,
+				EarlyStopFactor: s.cfg.FtEarlyStopValLossFactor,
+				RepoDir:         repoDir,
+				InstanceID:      s.cfg.InstanceID,
+			})
+		s.goSupervised("ft-loop-controller", ctrl.Run)
+	}
 }
 
 // StartPeriodicConsolidation starts a background goroutine that consolidates conversation memory

internal/cli/ft_loop.go

@@ -45,6 +45,84 @@ high-severity alert under the distinct "ft-loop" service). The autonomous
 actuator that calls these is a later phase (FT-RECURSIVE-002).`,
 	}
 	cmd.AddCommand(newFtLoopReportStageCmd())
+	cmd.AddCommand(newFtLoopPromoteCmd())
+	return cmd
+}
+
+func newFtLoopPromoteCmd() *cobra.Command {
+	var (
+		cycleID string
+		reject  bool
+		reason  string
+	)
+	cmd := &cobra.Command{
+		Use:   "promote",
+		Short: "Operator-confirm (or reject) a promote_pending retrain cycle",
+		Long: `Confirm or reject a cycle the controller left at promote_pending.
+
+The controller halts every cycle at promote_pending — promotion is operator-
+gated in Phase 6b (auto-promote + canary are Phase 7). --confirm records the
+cycle promoted; --reject records it rolled_back (with --reason). The actual
+GGUF symlink swap + llama-server restart is performed in the live promotion
+flow; this records the operator decision in the ft_training_cycles ledger.`,
+		RunE: func(_ *cobra.Command, _ []string) error {
+			if cycleID == "" {
+				return fmt.Errorf("--cycle-id is required")
+			}
+			ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second)
+			defer cancel()
+
+			client, err := tsdb.NewClient(ctx, tsdbConfigFromEnv())
+			if err != nil {
+				return fmt.Errorf("connect TSDB: %w", err)
+			}
+			defer client.Close()
+			pool := client.Pool()
+
+			status, found, err := tsdb.CycleStatus(ctx, pool, cycleID)
+			if err != nil {
+				return fmt.Errorf("query cycle: %w", err)
+			}
+			if !found {
+				return fmt.Errorf("cycle %q not found in the ledger", cycleID)
+			}
+			if status != tsdb.FtCyclePromotePending {
+				return fmt.Errorf("cycle %q is %q, not promote_pending — nothing to confirm/reject", cycleID, status)
+			}
+
+			modelVersion := strings.TrimSpace(os.Getenv("FT_LOOP_MODEL_VERSION"))
+			if modelVersion == "" {
+				modelVersion = "mdemg-llm-v1"
+			}
+			ev := tsdb.FtCycleEvent{
+				CycleID:      cycleID,
+				ModelVersion: modelVersion,
+				EvalResults:  map[string]any{"operator_decision": "confirm", "reason": reason},
+			}
+			if reject {
+				ev.Status = tsdb.FtCycleRolledBack
+				ev.Stage = "operator_reject"
+				ev.Error = reason
+				ev.EvalResults["operator_decision"] = "reject"
+			} else {
+				ev.Status = tsdb.FtCyclePromoted
+				ev.Stage = "operator_confirm"
+			}
+			if err := tsdb.RecordCycleEvent(ctx, pool, ev); err != nil {
+				return fmt.Errorf("record decision: %w", err)
+			}
+			if reject {
+				fmt.Printf("cycle %s rejected (rolled_back). reason: %s\n", cycleID, reason)
+			} else {
+				fmt.Printf("cycle %s confirmed (promoted). Perform the GGUF symlink swap + llama-server restart to serve the new model.\n", cycleID)
+			}
+			return nil
+		},
+	}
+	cmd.Flags().StringVar(&cycleID, "cycle-id", "", "the promote_pending cycle to act on")
+	cmd.Flags().BoolVar(&reject, "reject", false, "reject the cycle (rolled_back) instead of confirming")
+	cmd.Flags().StringVar(&reason, "reason", "", "operator reasoning (recorded; required-ish for --reject)")
+	_ = cmd.MarkFlagRequired("cycle-id")
 	return cmd
 }
 

internal/config/config.go

@@ -1228,6 +1228,16 @@ type Config struct {
 	FtLoopMinRetrainIntervalHours int     // FT_LOOP_MIN_RETRAIN_INTERVAL_HOURS — suppress a new trigger if a cycle started within this window (single-flight at retrain scale, not the RSIC seconds cooldown) (default: 168 = 7 days)
 	FtLoopMinFreshFraction        float64 // FT_LOOP_MIN_FRESH_FRACTION — minimum fraction of interactions newer than the last cycle before retraining (retrain on new signal, never the same corpus) (default: 0.30)
 
+	// FT recursive-retrain controller (FT-RECURSIVE-002 Phase 6b, Epic 5)
+	FtLoopPollIntervalSec    int     // FT_LOOP_POLL_INTERVAL_SEC — controller poll cadence for a triggered cycle (default: 60)
+	FtLoopLeaseMaxHours      int     // FT_LOOP_LEASE_MAX_HOURS — compute-lease expiry (≈1.5× the ~9h actual) so a crashed trainer can't wedge RSIC (default: 14)
+	FtLoopLeasePath          string  // FT_LOOP_LEASE_PATH — compute-lease lockfile path (default: $HOME/.mdemg/ft-loop.lease)
+	FtLoopMinFreeDiskGB      float64 // FT_LOOP_MIN_FREE_DISK_GB — disk-floor preflight before TRAIN (~85GB transient) (default: 100)
+	FtLoopPythonBin          string  // FT_LOOP_PYTHON_BIN — python interpreter for the training subprocesses (default: python3)
+	FtLoopModelVersion       string  // FT_LOOP_MODEL_VERSION — model_version recorded on cycles (default: mdemg-llm-v1)
+	FtLoraEpochsCap          int     // FT_LORA_EPOCHS_CAP — hard epoch cap passed to train_ft.py (AMD-1; the `auto` rejection is never weakened) (default: 3)
+	FtEarlyStopValLossFactor float64 // FT_EARLY_STOP_VAL_LOSS_FACTOR — SFT early-stop trips when val_loss > best × this (AMD-1) (default: 1.05)
+
 	// MAINT-LIVE-001 — maintenance liveness (only-ever-dry-runs detection).
 	MaintLiveAlertEnabled bool // MAINT_LIVE_ALERT_ENABLED — enable the maintenance_no_live_run rule (default: true)
 	MaintLiveLookbackDays int  // MAINT_LIVE_LOOKBACK_DAYS — window in which at least one live (dry_run=false) maintenance run must appear when any maintenance runs exist (default: 8)
@@ -4650,6 +4660,44 @@ func FromEnv() (Config, error) {
 	if ftLoopMinFreshFraction < 0 || ftLoopMinFreshFraction > 1 {
 		return Config{}, errors.New("FT_LOOP_MIN_FRESH_FRACTION must be in [0,1]")
 	}
+	ftLoopPollIntervalSec, err := atoi("FT_LOOP_POLL_INTERVAL_SEC", 60)
+	if err != nil {
+		return Config{}, err
+	}
+	if ftLoopPollIntervalSec < 1 {
+		return Config{}, errors.New("FT_LOOP_POLL_INTERVAL_SEC must be >= 1")
+	}
+	ftLoopLeaseMaxHours, err := atoi("FT_LOOP_LEASE_MAX_HOURS", 14)
+	if err != nil {
+		return Config{}, err
+	}
+	if ftLoopLeaseMaxHours < 1 {
+		return Config{}, errors.New("FT_LOOP_LEASE_MAX_HOURS must be >= 1")
+	}
+	ftLoopLeasePath := strings.TrimSpace(os.Getenv("FT_LOOP_LEASE_PATH"))
+	ftLoopMinFreeDiskGB, err := atof("FT_LOOP_MIN_FREE_DISK_GB", 100)
+	if err != nil {
+		return Config{}, err
+	}
+	if ftLoopMinFreeDiskGB < 0 {
+		return Config{}, errors.New("FT_LOOP_MIN_FREE_DISK_GB must be >= 0")
+	}
+	ftLoopPythonBin := get("FT_LOOP_PYTHON_BIN", "python3")
+	ftLoopModelVersion := get("FT_LOOP_MODEL_VERSION", "mdemg-llm-v1")
+	ftLoraEpochsCap, err := atoi("FT_LORA_EPOCHS_CAP", 3)
+	if err != nil {
+		return Config{}, err
+	}
+	if ftLoraEpochsCap < 1 {
+		return Config{}, errors.New("FT_LORA_EPOCHS_CAP must be >= 1")
+	}
+	ftEarlyStopValLossFactor, err := atof("FT_EARLY_STOP_VAL_LOSS_FACTOR", 1.05)
+	if err != nil {
+		return Config{}, err
+	}
+	if ftEarlyStopValLossFactor <= 1.0 {
+		return Config{}, errors.New("FT_EARLY_STOP_VAL_LOSS_FACTOR must be > 1.0")
+	}
 	maintLiveAlertEnabled := getBool("MAINT_LIVE_ALERT_ENABLED", true)
 	maintLiveLookbackDays, err := atoi("MAINT_LIVE_LOOKBACK_DAYS", 8)
 	if err != nil {
@@ -5490,6 +5538,14 @@ func FromEnv() (Config, error) {
 		FtLoopEnabled:                       ftLoopEnabled,
 		FtLoopMinRetrainIntervalHours:       ftLoopMinRetrainIntervalHours,
 		FtLoopMinFreshFraction:              ftLoopMinFreshFraction,
+		FtLoopPollIntervalSec:               ftLoopPollIntervalSec,
+		FtLoopLeaseMaxHours:                 ftLoopLeaseMaxHours,
+		FtLoopLeasePath:                     ftLoopLeasePath,
+		FtLoopMinFreeDiskGB:                 ftLoopMinFreeDiskGB,
+		FtLoopPythonBin:                     ftLoopPythonBin,
+		FtLoopModelVersion:                  ftLoopModelVersion,
+		FtLoraEpochsCap:                     ftLoraEpochsCap,
+		FtEarlyStopValLossFactor:            ftEarlyStopValLossFactor,
 		MaintLiveAlertEnabled:        maintLiveAlertEnabled,
 		MaintLiveLookbackDays:        maintLiveLookbackDays,
 		JobBackupStalenessHours:      jobBackupStalenessHours,