Plan 2: Situation controller and Acute Triage coordination

cmd/alertint/main.go

@@ -170,6 +170,16 @@ func runServe(args []string, _ io.Writer, stderr io.Writer) error {
 
 	logConfigWarnings(logger, cfg)
 
+	// Operator-configured observability boundary: an OTLP trace exporter
+	// only when telemetry.otlp is enabled, otherwise nothing is installed
+	// and every span stays a no-op. Deferred so the final flush runs after
+	// the foundation stop sequence below has ended its last span.
+	stopTelemetry, err := startTelemetry(ctx, cfg, resolveVersion(), logger)
+	if err != nil {
+		return err
+	}
+	defer stopTelemetry()
+
 	applyServeOverrides(cfg, *receiversAddr, *mcpAddr)
 
 	st, stagedInfo, err := openStoreWithStagedRestore(ctx, cfg.Storage.SQLitePath, logger)
@@ -377,17 +387,18 @@ func runServe(args []string, _ io.Writer, stderr io.Writer) error {
 		OccurrenceCap:   cfg.Memory.OccurrenceCap,
 		Lookback:        time.Duration(cfg.Memory.LookbackDays) * 24 * time.Hour,
 	}
-	cor := correlator.New(corCfg, st, incidentSink{skill: skill}, logger)
-
-	// SetRejudger/SetTriageFailureNotifier are safe to wire here, before
-	// reconstruction: neither is reachable from ApplyDelivery's
-	// durable-dispatch path (Rejudge is deliberately never called
-	// synchronously from a delivery commit, and the triage-exhausted
-	// notifier fires only from the correlator's own internal ticker loop,
-	// which isn't running during reconstruction). SetAuditor,
-	// SetResolutionNotifier, and SetOccurrenceNotifier are NOT wired here —
-	// all three ARE reachable from ApplyDelivery — see startCorrelator below.
-	cor.SetRejudger(skill)
+	cor := correlator.New(corCfg, st, productionIncidentSink(), logger)
+
+	// SetTriageFailureNotifier is safe to wire here, before reconstruction:
+	// it is not reachable from ApplyDelivery's durable-dispatch path (the
+	// triage-exhausted notifier fires only from the correlator's own
+	// internal ticker loop, which isn't running during reconstruction).
+	// The Correlator has no analyzer/LLM seam at all — no IncidentSink
+	// beyond the no-op one and no re-judgment runner (Plan 2 Task 7) —
+	// see TestProductionCorrelatorHasNoAcuteTriageDispatchDependency.
+	// SetAuditor, SetResolutionNotifier, and SetOccurrenceNotifier are NOT
+	// wired here — all three ARE reachable from ApplyDelivery — see
+	// startCorrelator below.
 	cor.SetTriageFailureNotifier(notifier)
 
 	// stopCorrelator is called exactly once, however runServe exits: inline,
@@ -403,10 +414,37 @@ func runServe(args []string, _ io.Writer, stderr io.Writer) error {
 	// The durable Situation foundation runtime (Task 8): the dispatch and
 	// input workers Tasks 6-7 built, plus the Reconstructor that converges
 	// durable state before either worker, or Receivers, ever runs. owner is
-	// this process's own lease identity, minted once.
+	// this process's own lease identity, minted once — every worker this
+	// process runs (foundation's dispatch/input, and the Situation
+	// controller/Triage workers below) derives its own stable lease-owner
+	// suffix from this SAME identity, so no two workers from the same
+	// process can ever fence each other's claims.
 	owner := "alertint-" + uuid.NewString()
 	rt := newFoundationRuntime(st, cor, owner, logger)
 
+	// The Situation controller runtime (Task 9): the Situation controller
+	// worker (Reconcile) and the Acute Triage worker Tasks 7-8 built, plus
+	// the startup-only recovery/backfill pass both depend on having already
+	// run. assessClient is the controller's own one-shot, provider-neutral
+	// L2 boundary — the SAME configured provider client Acute Triage uses
+	// (buildLLMClient/llmClient), NOT wrapped in llmClient's own hidden-retry
+	// Complete method, and wrapped with the installation LLM-health observer
+	// (buildAssessmentClient). skill (already constructed above) structurally
+	// satisfies situation.AcuteAnalyzer/AfterCommitter/ExhaustionNotifier all
+	// three, so the Triage worker dispatches through the SAME refactored
+	// Acute Triage skill instance — never a second, separately-constructed
+	// one — while the unmodified `skill` variable used by incidentSink above
+	// keeps driving the Correlator's own tick-triggered compatibility path
+	// unchanged. Existing Acute Triage still receives the ordinary
+	// retry-capable llmClient (acutetriage.Config's own Config, unchanged);
+	// the Correlator itself (cor, constructed above) receives no LLM
+	// dependency of its own — corCfg/correlator.New's signature carries none,
+	// and its only path to Acute Triage is via incidentSink{skill: skill}.
+	crt, err := buildControllerRuntime(st, llmClient, llmHealth, skill, cfg.Situations, owner, auditor, logger)
+	if err != nil {
+		return err
+	}
+
 	// Probe enabled integrations in the background: quickly (with backoff)
 	// while one is failing — at startup a co-deployed dependency may still
 	// be booting — then at a steady pace, logging losses and recoveries.
@@ -418,12 +456,15 @@ func runServe(args []string, _ io.Writer, stderr io.Writer) error {
 	var recvErrCh <-chan error
 	startupSeq := foundationSequence{
 		reconstruct: func(ctx context.Context) error {
-			report, err := rt.Reconstruct(ctx)
-			if err != nil {
-				return fmt.Errorf("situation foundation reconstruction: %w", err)
-			}
-			logReconstructionReport(logger, report)
-			return nil
+			return runFoundationReconstruction(ctx, rt, logger)
+		},
+		// Task 9: Triage migration backfill, interrupted Assessment-call/
+		// Triage-attempt recovery, and the one-hour startup horizon — all
+		// startup-only, zero-outward-effect primitives that must run before
+		// any worker resumes claiming controller/Triage work, exactly like
+		// foundation reconstruction above.
+		backfillAndRecoverControllerWork: func(ctx context.Context) error {
+			return runControllerRecovery(ctx, crt, logger)
 		},
 		// SetAuditor/SetResolutionNotifier/SetOccurrenceNotifier are wired
 		// here — between reconstruct and cor.Start, never before — because
@@ -446,7 +487,8 @@ func runServe(args []string, _ io.Writer, stderr io.Writer) error {
 			cor.SetOccurrenceNotifier(notifier)
 			return cor.Start(ctx)
 		},
-		startWorkers: rt.Start,
+		startWorkers:           rt.Start,
+		startControllerWorkers: crt.Start,
 		startReceivers: func() error {
 			var err error
 			recvSrv, recvErrCh, err = startReceivers(cfg, st, auditor, healthReg, llmHealth, rt.WakeDispatch, logger)
@@ -480,24 +522,23 @@ func runServe(args []string, _ io.Writer, stderr io.Writer) error {
 	shutdownCtx, cancel := context.WithTimeout(context.Background(), ingress.DefaultShutdownTimeout)
 	defer cancel()
 
-	// Receivers, then the foundation workers (rt.Stop: dispatch, then
-	// input), then the Correlator: Receivers stopping first means no new
-	// inbound work can be durably accepted; the queue workers stopping next
-	// means nothing already durably queued goes unclaimed mid-drain; the
-	// Correlator stopping last — via stopCorrelator, the same
-	// sync.Once-guarded call the defer above falls back to on every other
-	// exit path — means it keeps serving fixed-window expiry and Triage
-	// retry for exactly as long as anything upstream could still be handing
-	// it work.
+	// Receivers; stop/flush the Correlator's fixed-window production; drain
+	// foundation delivery/input work and due controller/Triage work — and
+	// the inputs the latter produces — in rounds until a round handles
+	// nothing; then stop the controller/Triage workers (crt.Stop) and the
+	// foundation workers (rt.Stop: dispatch, then input). See
+	// foundationStopSequence's own doc comment for why the Correlator stops
+	// before the drain (its ticker is the one producer that could otherwise
+	// commit fresh durable work after the drain has finished) and why the
+	// drain is a loop. stopCorrelator is the same sync.Once-guarded call the
+	// defer above falls back to on every other exit path.
 	stopSeq := foundationStopSequence{
-		stopReceivers: func() error {
-			if recvSrv == nil {
-				return nil
-			}
-			return recvSrv.Shutdown(shutdownCtx)
-		},
-		stopWorkers:    rt.Stop,
-		stopCorrelator: stopCorrelator,
+		stopReceivers:         receiverShutdown(shutdownCtx, recvSrv),
+		stopCorrelator:        stopCorrelator,
+		drainFoundationWork:   rt.Drain,
+		drainControllerWork:   crt.Drain,
+		stopControllerWorkers: crt.Stop,
+		stopWorkers:           rt.Stop,
 	}
 	if err := stopSeq.run(shutdownCtx); err != nil {
 		logger.Error("situation foundation shutdown failed", slog.String("err", err.Error()))
@@ -512,6 +553,18 @@ func runServe(args []string, _ io.Writer, stderr io.Writer) error {
 	return nil
 }
 
+// receiverShutdown is foundationStopSequence's stop-receivers step: a
+// graceful Shutdown of the receivers HTTP server bounded by ctx, or a no-op
+// when no receiver was started (nil server).
+func receiverShutdown(ctx context.Context, srv *http.Server) func() error {
+	return func() error {
+		if srv == nil {
+			return nil
+		}
+		return srv.Shutdown(ctx)
+	}
+}
+
 // stopLLMHealthRunner stops the LLM health runner — its final delivery pass
 // acknowledges the settled state — and waits, bounded by
 // llmhealth.DrainTimeout, for it to finish. Call only after every capability
@@ -1187,13 +1240,19 @@ func llmanthropicCfg(cfg *config.Config) llmanthropic.Config {
 	}
 }
 
-// incidentSink wraps an acutetriage.Skill as a correlator.IncidentSink.
-type incidentSink struct {
-	skill *acutetriage.Skill
-}
-
-func (s incidentSink) OnIncidentReady(ctx context.Context, inc store.Incident) error {
-	return s.skill.Run(ctx, inc)
+// productionIncidentSink is the only IncidentSink production ever hands the
+// Correlator: a no-op. Acute Triage dispatch belongs to the Triage worker
+// (internal/situation.TriageWorker) polling the gated incident_triage
+// schedule; the Correlator owns grouping, readiness, and attachment only
+// and must carry no analyzer/LLM dispatch dependency at all — not even a
+// dormant one. An earlier wiring passed a Skill-backed sink here (whose
+// callback called Skill.Run); the Correlator never invoked it after Task 7,
+// but a live LLM dependency handed to a component that must have none is a
+// wiring bug regardless of whether it is reachable, and one refactor away
+// from becoming a second dispatch path. TestProductionCorrelatorHasNoAcuteTriageDispatchDependency
+// pins this.
+func productionIncidentSink() correlator.IncidentSink {
+	return correlator.NopIncidentSink{}
 }
 
 // buildLogger constructs the runtime logger applying precedence

cmd/alertint/main_test.go

@@ -4,8 +4,109 @@ package main
 
 import (
 	"bytes"
+	"go/parser"
+	"go/token"
+	"os"
+	"path/filepath"
+	"reflect"
 	"strings"
 	"testing"
+
+	"github.com/alertint/alertint-agent/internal/correlator"
+	"github.com/alertint/alertint-agent/internal/situation"
+	"github.com/alertint/alertint-agent/skills/acutetriage"
+)
+
+// ----------------------------------------------------------------------
+// Task 9 production wiring proofs.
+// ----------------------------------------------------------------------
+
+// TestProductionCorrelatorHasNoAcuteTriageDispatchDependency proves the
+// Correlator production wires carries no analyzer/LLM dispatch dependency
+// at all, on two surfaces:
+//
+//  1. Wiring: the one IncidentSink runServe hands correlator.New is the
+//     no-op sink, never a Skill-backed wrapper, and the Correlator exposes
+//     no other seam (no Rejudger/SetRejudger — removed with Plan 2 Task 7)
+//     through which a Skill could be handed in.
+//  2. Structure: no non-test source file of internal/correlator imports
+//     internal/llm, internal/llmhealth, or any skills/* package, so the
+//     package cannot dispatch a model call even if a future seam were
+//     added by mistake.
+//
+// Acute Triage dispatch belongs exclusively to the Triage worker polling
+// the gated incident_triage schedule.
+func TestProductionCorrelatorHasNoAcuteTriageDispatchDependency(t *testing.T) {
+	sink := productionIncidentSink()
+	if _, ok := sink.(correlator.NopIncidentSink); !ok {
+		t.Fatalf("production IncidentSink = %T, want correlator.NopIncidentSink — any other sink hands the Correlator a dispatch dependency it must not own", sink)
+	}
+	if reflect.TypeOf(sink).NumField() != 0 {
+		t.Fatalf("production IncidentSink %T carries %d fields, want 0 — it must hold no Skill, client, or store", sink, reflect.TypeOf(sink).NumField())
+	}
+	corType := reflect.TypeOf(correlator.Correlator{})
+	for i := 0; i < corType.NumField(); i++ {
+		f := corType.Field(i)
+		if strings.Contains(strings.ToLower(f.Name), "rejudg") || strings.Contains(f.Type.String(), "Rejudger") {
+			t.Fatalf("correlator.Correlator carries field %s %s — a re-judgment seam is an analyzer/LLM dispatch dependency the Correlator must not own", f.Name, f.Type)
+		}
+	}
+	if _, ok := corType.MethodByName("SetRejudger"); ok {
+		t.Fatal("correlator.Correlator has SetRejudger — a re-judgment seam is an analyzer/LLM dispatch dependency the Correlator must not own")
+	}
+	if _, ok := reflect.PointerTo(corType).MethodByName("SetRejudger"); ok {
+		t.Fatal("*correlator.Correlator has SetRejudger — a re-judgment seam is an analyzer/LLM dispatch dependency the Correlator must not own")
+	}
+
+	forbidden := []string{
+		"github.com/alertint/alertint-agent/internal/llm",
+		"github.com/alertint/alertint-agent/internal/llmhealth",
+		"github.com/alertint/alertint-agent/skills/",
+	}
+	dir := filepath.Join("..", "..", "internal", "correlator")
+	entries, err := os.ReadDir(dir)
+	if err != nil {
+		t.Fatalf("read %s: %v", dir, err)
+	}
+	fset := token.NewFileSet()
+	checked := 0
+	for _, e := range entries {
+		name := e.Name()
+		if e.IsDir() || !strings.HasSuffix(name, ".go") || strings.HasSuffix(name, "_test.go") {
+			continue
+		}
+		f, err := parser.ParseFile(fset, filepath.Join(dir, name), nil, parser.ImportsOnly)
+		if err != nil {
+			t.Fatalf("parse %s: %v", name, err)
+		}
+		checked++
+		for _, imp := range f.Imports {
+			path := strings.Trim(imp.Path.Value, `"`)
+			for _, bad := range forbidden {
+				if strings.HasPrefix(path, bad) {
+					t.Errorf("internal/correlator/%s imports %s — the Correlator must carry no analyzer/LLM dispatch dependency", name, path)
+				}
+			}
+		}
+	}
+	if checked == 0 {
+		t.Fatal("no non-test Go files found under internal/correlator — the structural check ran against nothing")
+	}
+}
+
+// The refactored Acute Triage skill structurally satisfies every interface
+// the Situation controller runtime's Triage worker needs — proven at
+// compile time, so a future signature drift on either side fails the build
+// long before any test runs. skills/acutetriage.Skill.Analyze/AfterCommit/
+// OnTriageExhausted (Task 7) are what newControllerRuntime passes as
+// situation.AcuteAnalyzer/AfterCommitter/ExhaustionNotifier respectively.
+var (
+	_ situation.AcuteAnalyzer      = (*acutetriage.Skill)(nil)
+	_ situation.AfterCommitter     = (*acutetriage.Skill)(nil)
+	_ situation.ExhaustionNotifier = (*acutetriage.Skill)(nil)
+	// MinimumMemberAlertsPolicy is what lets the Triage worker resolve a
+	// below-minimum clean skip BEFORE claiming, so it consumes no attempt.
+	_ situation.MinimumMemberAlertsPolicy = (*acutetriage.Skill)(nil)
 )
 
 func TestRun_VersionFlagPrintsVersionAndExitsCleanly(t *testing.T) {