feat(observe): reaper for liveness probe + TickEscalations heartbeat (#9)

backend/internal/lifecycle/manager.go

@@ -52,6 +52,14 @@ type Manager struct {
 	trackers  map[trackerKey]*reactionTracker
 	trackerMu sync.Mutex
 	clock     func() time.Time
+
+	// sessionLister returns every session known to persistence so RunningSessions
+	// can filter by runtime axis without coupling the LCM to a cross-project
+	// store API the Tom-store does not yet expose. The daemon (lane #10) injects
+	// the production lister via WithSessionLister; until then, the call returns
+	// no sessions so a reaper attached to an unwired Manager is a clean no-op
+	// rather than a panic.
+	sessionLister func(ctx context.Context) ([]domain.SessionRecord, error)
 }
 
 var _ ports.LifecycleManager = (*Manager)(nil)
@@ -67,6 +75,16 @@ func New(store ports.LifecycleStore, notifier ports.Notifier, messenger ports.Ag
 	}
 }
 
+// WithSessionLister injects the function the LCM uses to enumerate all
+// persisted sessions for RunningSessions. The daemon wires this against the
+// store at startup; it must be called BEFORE any reaper attached to this
+// Manager starts running, since concurrent calls would race the bare-field
+// read in RunningSessions. Calling it more than once replaces the previous
+// lister.
+func (m *Manager) WithSessionLister(fn func(ctx context.Context) ([]domain.SessionRecord, error)) {
+	m.sessionLister = fn
+}
+
 // ---- per-session serialisation ----
 
 // keyedMutex hands out one lock per session id so the load->decide->persist
@@ -409,6 +427,47 @@ func (m *Manager) OnKillRequested(ctx context.Context, id domain.SessionID, r po
 	return nil
 }
 
+// ---- read-snapshot helpers ----
+
+// RunningSessions returns a snapshot of every persisted session worth probing
+// in the next reaper tick. "Worth probing" is wider than "runtime axis alive":
+// it includes sessions in the Detecting quarantine, because a fresh probe is
+// the only fact that can recover them (back to working) or escalate them
+// (terminal killed). Filtering to runtime-axis-alive would silently park every
+// Detecting session — a single failed probe would never get a second chance
+// and recovery via runtime probe would be unreachable.
+//
+// The predicate is "not a final session state". Terminal session states (done,
+// terminated) are excluded because Restore is the only path back; observations
+// must not reopen them (#1 invariant). Sessions in earlier states — not_started,
+// working, idle, needs_input, stuck, detecting — are all included. Those that
+// lack runtime handle metadata (e.g. not_started before OnSpawnCompleted) are
+// returned and harmlessly skipped by the reaper's per-session handle guard.
+//
+// The call only reads and copies, so it does not break the single-writer
+// invariant; concurrent Apply* calls may move sessions in or out of the probe
+// set between snapshots, which is correct — the next tick re-reads.
+//
+// When no lister has been wired (e.g. tests construct a bare Manager), the
+// method returns nil so a goroutine attached to such a Manager degrades to a
+// no-op rather than panicking.
+func (m *Manager) RunningSessions(ctx context.Context) ([]domain.SessionRecord, error) {
+	if m.sessionLister == nil {
+		return nil, nil
+	}
+	all, err := m.sessionLister(ctx)
+	if err != nil {
+		return nil, err
+	}
+	out := make([]domain.SessionRecord, 0, len(all))
+	for _, rec := range all {
+		if !isTerminal(rec.Lifecycle.Session.State) {
+			out = append(out, rec)
+		}
+	}
+	return out, nil
+}
+
 // ---- diff helpers ----
 
 // setSessionIfChanged sets next.Session only when the decided sub-state differs

backend/internal/lifecycle/manager_test.go

@@ -2,6 +2,7 @@ package lifecycle
 
 import (
 	"context"
+	"errors"
 	"sync"
 	"testing"
 	"time"
@@ -525,6 +526,78 @@ func TestPerSessionSerialization(t *testing.T) {
 	}
 }
 
+// ---- RunningSessions (reaper poll-set) ----
+
+func TestRunningSessions_NoListerWired_ReturnsEmpty(t *testing.T) {
+	m, _ := newManager()
+	got, err := m.RunningSessions(context.Background())
+	if err != nil {
+		t.Fatalf("RunningSessions: %v", err)
+	}
+	if len(got) != 0 {
+		t.Fatalf("expected empty slice when no lister wired, got %d records", len(got))
+	}
+}
+
+func TestRunningSessions_ListerErrorPropagates(t *testing.T) {
+	m, _ := newManager()
+	wantErr := errors.New("boom")
+	m.WithSessionLister(func(_ context.Context) ([]domain.SessionRecord, error) {
+		return nil, wantErr
+	})
+	_, err := m.RunningSessions(context.Background())
+	if !errors.Is(err, wantErr) {
+		t.Fatalf("expected lister error to propagate, got %v", err)
+	}
+}
+
+// TestRunningSessions_FilterIncludesProbableExcludesTerminal locks in the
+// reaper poll-set predicate. The bug we are guarding against is filtering to
+// "runtime.State == RuntimeAlive": detecting sessions (RuntimeMissing /
+// RuntimeProbeFailed) would be silently parked, breaking the probe-driven
+// recovery path proved by manager_test.go:59 and the dead+dead -> killed path
+// proved by manager_test.go:79.
+func TestRunningSessions_FilterIncludesProbableExcludesTerminal(t *testing.T) {
+	m, _ := newManager()
+	records := []domain.SessionRecord{
+		{ID: "working-alive", Lifecycle: lc(domain.SessionWorking, domain.ReasonTaskInProgress, domain.RuntimeAlive)},
+		{ID: "detecting-probefailed", Lifecycle: lc(domain.SessionDetecting, domain.ReasonProbeFailure, domain.RuntimeProbeFailed)},
+		{ID: "detecting-missing", Lifecycle: lc(domain.SessionDetecting, domain.ReasonRuntimeLost, domain.RuntimeMissing)},
+		{ID: "idle-alive", Lifecycle: lc(domain.SessionIdle, domain.ReasonResearchComplete, domain.RuntimeAlive)},
+		{ID: "needs-input-alive", Lifecycle: lc(domain.SessionNeedsInput, domain.ReasonAwaitingUserInput, domain.RuntimeAlive)},
+		{ID: "not-started", Lifecycle: lc(domain.SessionNotStarted, domain.ReasonSpawnRequested, domain.RuntimeUnknown)},
+		{ID: "terminated", Lifecycle: lc(domain.SessionTerminated, domain.ReasonManuallyKilled, domain.RuntimeExited)},
+		{ID: "done", Lifecycle: lc(domain.SessionDone, domain.ReasonPRMerged, domain.RuntimeExited)},
+	}
+	m.WithSessionLister(func(_ context.Context) ([]domain.SessionRecord, error) {
+		return records, nil
+	})
+
+	got, err := m.RunningSessions(context.Background())
+	if err != nil {
+		t.Fatalf("RunningSessions: %v", err)
+	}
+	gotIDs := map[domain.SessionID]bool{}
+	for _, r := range got {
+		gotIDs[r.ID] = true
+	}
+	wantIncluded := []domain.SessionID{
+		"working-alive", "detecting-probefailed", "detecting-missing",
+		"idle-alive", "needs-input-alive", "not-started",
+	}
+	for _, id := range wantIncluded {
+		if !gotIDs[id] {
+			t.Errorf("expected %q in poll set, missing", id)
+		}
+	}
+	wantExcluded := []domain.SessionID{"terminated", "done"}
+	for _, id := range wantExcluded {
+		if gotIDs[id] {
+			t.Errorf("expected %q NOT in poll set, found", id)
+		}
+	}
+}
+
 // ---- helpers ----
 
 func lc(state domain.SessionState, reason domain.SessionReason, rt domain.RuntimeState) domain.CanonicalSessionLifecycle {

backend/internal/observe/reaper/reaper.go

@@ -0,0 +1,215 @@
+// Package reaper implements the OBSERVE-layer polling timer that supplies the
+// LCM with the two facts the LCM cannot wake itself to discover: a periodic
+// duration-based escalation heartbeat, and per-session runtime liveness probes.
+//
+// The reaper sits OUTSIDE the LCM's per-session serial loop. It only REPORTS
+// facts — it never decides whether a session is "truly" dead. The decider
+// (anti-flap Detecting quarantine, terminal-session rules) is owned by the LCM
+// and consumes these facts through the regular ApplyRuntimeObservation entry
+// point. A probe error is reported as a probe-failure fact, never collapsed to
+// "alive" or "dead", so the LCM's failed-probe ≠ dead invariant holds.
+package reaper
+
+import (
+	"context"
+	"log/slog"
+	"time"
+
+	"github.com/aoagents/agent-orchestrator/backend/internal/domain"
+	"github.com/aoagents/agent-orchestrator/backend/internal/lifecycle"
+	"github.com/aoagents/agent-orchestrator/backend/internal/ports"
+)
+
+// DefaultTickInterval is the cadence used when Config.Tick is zero. It mirrors
+// the design doc's 5s sampling window for runtime liveness.
+const DefaultTickInterval = 5 * time.Second
+
+// RuntimeRegistry resolves a runtime adapter by the RuntimeName recorded in a
+// session's RuntimeHandle. The reaper looks the runtime up per-session so a
+// single reaper instance can probe tmux- and zellij-backed sessions side by
+// side without knowing about either at construction.
+type RuntimeRegistry interface {
+	Runtime(name string) (ports.Runtime, bool)
+}
+
+// MapRegistry is the trivial RuntimeRegistry: a name->runtime map. Callers
+// that need dynamic registration can implement RuntimeRegistry themselves.
+type MapRegistry map[string]ports.Runtime
+
+// Runtime implements RuntimeRegistry.
+func (m MapRegistry) Runtime(name string) (ports.Runtime, bool) {
+	rt, ok := m[name]
+	return rt, ok
+}
+
+// Config holds the externally-tunable knobs for a Reaper. Every field is
+// optional; zero values fall back to safe defaults so production wiring (which
+// only needs to inject the LCM and registry) and tests (which inject a clock
+// plus a fast tick) can both stay terse.
+type Config struct {
+	// Tick is the interval between ticks. <=0 means DefaultTickInterval.
+	Tick time.Duration
+	// Clock supplies ObservedAt and TickEscalations now stamps. nil means
+	// time.Now. Injected in tests so assertions don't race wallclock.
+	Clock func() time.Time
+	// Logger receives operational diagnostics (probe errors, skipped sessions,
+	// LCM call failures). The reaper logs but does not propagate these errors
+	// because a single failed probe must not kill the loop. nil means
+	// slog.Default.
+	Logger *slog.Logger
+}
+
+// Reaper is the polling timer. Construct it with New; start the background
+// goroutine with Start, or drive a single cycle synchronously with Tick.
+type Reaper struct {
+	lcm      ports.LifecycleManager
+	registry RuntimeRegistry
+	tick     time.Duration
+	clock    func() time.Time
+	logger   *slog.Logger
+}
+
+// New constructs a Reaper. The LCM is the sole writer destination (the reaper
+// reports facts via ApplyRuntimeObservation and TickEscalations); the registry
+// resolves the runtime adapter to use per session.
+func New(lcm ports.LifecycleManager, registry RuntimeRegistry, cfg Config) *Reaper {
+	r := &Reaper{
+		lcm:      lcm,
+		registry: registry,
+		tick:     cfg.Tick,
+		clock:    cfg.Clock,
+		logger:   cfg.Logger,
+	}
+	if r.tick <= 0 {
+		r.tick = DefaultTickInterval
+	}
+	if r.clock == nil {
+		r.clock = time.Now
+	}
+	if r.logger == nil {
+		r.logger = slog.Default()
+	}
+	return r
+}
+
+// Start launches the background goroutine and returns a channel that closes
+// once the loop has exited. The loop exits on ctx cancellation; the channel
+// gives the daemon a clean shutdown hook (wait on it after cancel to confirm
+// the reaper has stopped before tearing down dependencies).
+func (r *Reaper) Start(ctx context.Context) <-chan struct{} {
+	done := make(chan struct{})
+	go r.loop(ctx, done)
+	return done
+}
+
+func (r *Reaper) loop(ctx context.Context, done chan<- struct{}) {
+	defer close(done)
+	t := time.NewTicker(r.tick)
+	defer t.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-t.C:
+			if err := r.Tick(ctx); err != nil {
+				r.logger.Error("reaper: tick failed", "err", err)
+			}
+		}
+	}
+}
+
+// Tick runs one observation cycle: it always fires TickEscalations first (the
+// duration-based escalation heartbeat, which the synchronous LCM cannot wake
+// itself to drive), then enumerates the LCM's running sessions, probes each
+// one's runtime, and reports any non-alive result back as a fact.
+//
+// Tick is exported so the daemon (and tests) can drive cycles synchronously,
+// and so the Start goroutine has a single chokepoint to log against.
+//
+// Errors: only the RunningSessions failure is propagated, since it short-
+// circuits the rest of the cycle. TickEscalations and per-session
+// ApplyRuntimeObservation failures are logged but never propagated — one
+// failed call must not bring down the loop.
+func (r *Reaper) Tick(ctx context.Context) error {
+	now := r.clock()
+
+	// Heartbeat is best-effort and runs before enumeration so duration-based
+	// escalations still fire if the running-set lookup is the thing that
+	// errored. The LCM's TickEscalations is itself idempotent (no canonical
+	// writes) — at worst we miss escalating once and pick it up next tick.
+	if err := r.lcm.TickEscalations(ctx, now); err != nil {
+		r.logger.Error("reaper: TickEscalations failed", "err", err)
+	}
+
+	sessions, err := r.lcm.RunningSessions(ctx)
+	if err != nil {
+		return err
+	}
+
+	for _, sess := range sessions {
+		r.probeOne(ctx, sess, now)
+	}
+	return nil
+}
+
+// probeOne handles a single session's probe + fact-report. Every probe result —
+// alive, dead, or failed — is reported as a fact to the LCM. The reaper does
+// not optimize away the "alive" case, because a session in Detecting (whose
+// runtime axis is NOT alive) is included in the running set and needs the
+// alive probe to recover; the reaper has no business deciding what counts as
+// a no-op. The LCM's ApplyRuntimeObservation diffs against canonical and
+// only Upserts on actual change, so steady-state alive is already cheap.
+func (r *Reaper) probeOne(ctx context.Context, sess domain.SessionRecord, now time.Time) {
+	handle, ok := handleFromRecord(sess)
+	if !ok {
+		// A session in the running-set without a handle is an anomaly worth
+		// surfacing (OnSpawnCompleted should have set both keys). Warn rather
+		// than Debug so it doesn't hide behind a noisy log level.
+		r.logger.Warn("reaper: session has no runtime handle metadata, skipping",
+			"session", sess.ID)
+		return
+	}
+	rt, ok := r.registry.Runtime(handle.RuntimeName)
+	if !ok {
+		r.logger.Warn("reaper: no runtime registered for session, skipping",
+			"session", sess.ID, "runtime", handle.RuntimeName)
+		return
+	}
+
+	alive, probeErr := rt.IsAlive(ctx, handle)
+	facts := ports.RuntimeFacts{ObservedAt: now}
+	switch {
+	case probeErr != nil:
+		// Failed probe must NOT be collapsed to alive — that would let a
+		// transient tmux/zellij outage hide a really-dead session, and a
+		// transient adapter bug terminate a really-alive one. Report failed
+		// and let the LCM's detecting quarantine arbitrate.
+		facts.RuntimeState = ports.RuntimeProbeFailed
+		facts.ProcessState = ports.ProcessProbeFailed
+		r.logger.Debug("reaper: probe error reported as failed fact",
+			"session", sess.ID, "runtime", handle.RuntimeName, "err", probeErr)
+	case alive:
+		facts.RuntimeState = ports.RuntimeProbeAlive
+		facts.ProcessState = ports.ProcessProbeAlive
+	default:
+		facts.RuntimeState = ports.RuntimeProbeDead
+		facts.ProcessState = ports.ProcessProbeDead
+	}
+
+	if err := r.lcm.ApplyRuntimeObservation(ctx, sess.ID, facts); err != nil {
+		r.logger.Error("reaper: ApplyRuntimeObservation failed",
+			"session", sess.ID, "err", err)
+	}
+}
+
+// handleFromRecord reconstructs the RuntimeHandle stored on the session by
+// OnSpawnCompleted. Both keys are required; either being empty is the
+// "session lacks a probable handle" signal that probeOne uses to skip.
+func handleFromRecord(rec domain.SessionRecord) (ports.RuntimeHandle, bool) {
+	id := rec.Metadata[lifecycle.MetaRuntimeHandleID]
+	name := rec.Metadata[lifecycle.MetaRuntimeName]
+	if id == "" || name == "" {
+		return ports.RuntimeHandle{}, false
+	}
+	return ports.RuntimeHandle{ID: id, RuntimeName: name}, true
+}