From 11475fbc720b8ea1fd184f92257d1b4b3a7713fa Mon Sep 17 00:00:00 2001 From: harshitsinghbhandari <24b4506@iitb.ac.in> Date: Fri, 29 May 2026 22:10:29 +0530 Subject: [PATCH 1/2] feat(observe): reaper for liveness probe + TickEscalations heartbeat (#9) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The reaper sits OUTSIDE the LCM's per-session serial loop. On every tick it: 1. Fires lcm.TickEscalations(now) — the duration-based escalation heartbeat a non-polling LCM cannot wake itself to drive. 2. Asks lcm.RunningSessions for the snapshot of sessions whose runtime axis is alive, then calls runtime.IsAlive(handle) per session via a RuntimeRegistry that dispatches by RuntimeHandle.RuntimeName (so a single reaper covers tmux + zellij side by side). 3. Reports any non-alive result back as a fact via ApplyRuntimeObservation — dead -> RuntimeProbeDead, probe error -> RuntimeProbeFailed (never collapsed to alive: failed probe ≠ dead, but it ≠ alive either). Steady- state alive is skipped so we don't churn the LCM with no-op load/diff work. The reaper REPORTS facts; the LCM owns DECIDE (anti-flap Detecting quarantine, terminal-session rules). The reaper never writes. Open-question resolution: add RunningSessions(ctx) to ports.LifecycleManager (option a). The Manager implements it via an injectable session lister (Manager.WithSessionLister) so the LCM itself does not require a new LifecycleStore method — Tom's store contract is untouched, daemon wiring (lane #10) will inject the production lister at startup. Scope: reaper goroutine + the minimum LCM seam. No activity ingest, no FS watcher, no daemon wiring, no new schema fields, no store changes. --- backend/internal/lifecycle/manager.go | 44 +++ backend/internal/observe/reaper/reaper.go | 212 +++++++++++ .../internal/observe/reaper/reaper_test.go | 334 ++++++++++++++++++ backend/internal/ports/inbound.go | 8 + backend/internal/session/fakes_test.go | 4 + 5 files changed, 602 insertions(+) create mode 100644 backend/internal/observe/reaper/reaper.go create mode 100644 backend/internal/observe/reaper/reaper_test.go diff --git a/backend/internal/lifecycle/manager.go b/backend/internal/lifecycle/manager.go index 23d6a9b90..defa3b643 100644 --- a/backend/internal/lifecycle/manager.go +++ b/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,13 @@ 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. 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 +424,35 @@ 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 whose runtime +// axis is alive. It exists so the reaper (OBSERVE) can decide whom to probe +// without taking on a LifecycleStore dependency or knowing the LCM's internal +// state. Because the call only reads and copies, it does not break the +// single-writer invariant; concurrent Apply* calls may move sessions in or out +// of "alive" 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 an empty slice 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 rec.Lifecycle.Runtime.State == domain.RuntimeAlive { + out = append(out, rec) + } + } + return out, nil +} + // ---- diff helpers ---- // setSessionIfChanged sets next.Session only when the decided sub-state differs diff --git a/backend/internal/observe/reaper/reaper.go b/backend/internal/observe/reaper/reaper.go new file mode 100644 index 000000000..5bd4b0d70 --- /dev/null +++ b/backend/internal/observe/reaper/reaper.go @@ -0,0 +1,212 @@ +// 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. It is intentionally +// silent on the alive case: a probe that confirms the steady "alive" state is +// not a fact worth re-reporting (the runtime axis is already alive, so the LCM +// would diff to a no-op anyway). Dead and probe-failure ARE reported. +func (r *Reaper) probeOne(ctx context.Context, sess domain.SessionRecord, now time.Time) { + handle, ok := handleFromRecord(sess) + if !ok { + r.logger.Debug("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: + // Steady-state alive carries no new information — skip the call so we + // don't churn the LCM with no-op load/diff/persist work on every tick. + // Recovery from detecting via probe still flows through this path + // whenever the probe is NOT alive (failure or death). + return + 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 +} diff --git a/backend/internal/observe/reaper/reaper_test.go b/backend/internal/observe/reaper/reaper_test.go new file mode 100644 index 000000000..783fbad73 --- /dev/null +++ b/backend/internal/observe/reaper/reaper_test.go @@ -0,0 +1,334 @@ +package reaper_test + +import ( + "context" + "errors" + "reflect" + "sync" + "testing" + "time" + + "github.com/aoagents/agent-orchestrator/backend/internal/domain" + "github.com/aoagents/agent-orchestrator/backend/internal/lifecycle" + "github.com/aoagents/agent-orchestrator/backend/internal/observe/reaper" + "github.com/aoagents/agent-orchestrator/backend/internal/ports" +) + +// ---- fakes ---- + +type aliveResult struct { + alive bool + err error +} + +// fakeRuntime is a programmable ports.Runtime. The reaper only calls IsAlive, +// but the interface requires the other methods so we stub them. +type fakeRuntime struct { + mu sync.Mutex + results map[string]aliveResult + probed []string +} + +var _ ports.Runtime = (*fakeRuntime)(nil) + +func (f *fakeRuntime) IsAlive(_ context.Context, h ports.RuntimeHandle) (bool, error) { + f.mu.Lock() + f.probed = append(f.probed, h.ID) + f.mu.Unlock() + r, ok := f.results[h.ID] + if !ok { + return false, errors.New("fakeRuntime: no programmed response for " + h.ID) + } + return r.alive, r.err +} + +func (f *fakeRuntime) Create(context.Context, ports.RuntimeConfig) (ports.RuntimeHandle, error) { + return ports.RuntimeHandle{}, nil +} +func (f *fakeRuntime) Destroy(context.Context, ports.RuntimeHandle) error { return nil } +func (f *fakeRuntime) SendMessage(context.Context, ports.RuntimeHandle, string) error { + return nil +} +func (f *fakeRuntime) GetOutput(context.Context, ports.RuntimeHandle, int) (string, error) { + return "", nil +} + +// fakeLCM records every reaper-facing call in order so tests can assert the +// exact sequence (TickEscalations -> RunningSessions -> ApplyRuntimeObservation). +type fakeLCM struct { + mu sync.Mutex + sessions []domain.SessionRecord + calls []call + + runErr error + tickErr error + obsErr error +} + +type call struct { + Kind string + Now time.Time + Session domain.SessionID + Facts ports.RuntimeFacts +} + +var _ ports.LifecycleManager = (*fakeLCM)(nil) + +func (l *fakeLCM) RunningSessions(_ context.Context) ([]domain.SessionRecord, error) { + l.mu.Lock() + defer l.mu.Unlock() + l.calls = append(l.calls, call{Kind: "RunningSessions"}) + if l.runErr != nil { + return nil, l.runErr + } + out := make([]domain.SessionRecord, len(l.sessions)) + copy(out, l.sessions) + return out, nil +} + +func (l *fakeLCM) TickEscalations(_ context.Context, now time.Time) error { + l.mu.Lock() + defer l.mu.Unlock() + l.calls = append(l.calls, call{Kind: "TickEscalations", Now: now}) + return l.tickErr +} + +func (l *fakeLCM) ApplyRuntimeObservation(_ context.Context, id domain.SessionID, f ports.RuntimeFacts) error { + l.mu.Lock() + defer l.mu.Unlock() + l.calls = append(l.calls, call{Kind: "ApplyRuntimeObservation", Session: id, Facts: f}) + return l.obsErr +} + +// unused methods on the LCM port — the reaper never invokes them. +func (l *fakeLCM) ApplySCMObservation(context.Context, domain.SessionID, ports.SCMFacts) error { + return nil +} +func (l *fakeLCM) ApplyActivitySignal(context.Context, domain.SessionID, ports.ActivitySignal) error { + return nil +} +func (l *fakeLCM) OnSpawnInitiated(context.Context, domain.SessionRecord) error { return nil } +func (l *fakeLCM) OnSpawnCompleted(context.Context, domain.SessionID, ports.SpawnOutcome) error { + return nil +} +func (l *fakeLCM) OnKillRequested(context.Context, domain.SessionID, ports.KillReason) error { + return nil +} + +// ---- helpers ---- + +func aliveSessionWith(id domain.SessionID, runtimeName, handleID string) domain.SessionRecord { + return domain.SessionRecord{ + ID: id, + Lifecycle: domain.CanonicalSessionLifecycle{ + Runtime: domain.RuntimeSubstate{State: domain.RuntimeAlive, Reason: domain.RuntimeReasonProcessRunning}, + }, + Metadata: map[string]string{ + lifecycle.MetaRuntimeHandleID: handleID, + lifecycle.MetaRuntimeName: runtimeName, + }, + } +} + +// ---- tests ---- + +func TestReaper_Tick(t *testing.T) { + now := time.Date(2026, 5, 28, 12, 0, 0, 0, time.UTC) + clock := func() time.Time { return now } + + type runtimeProbes struct { + name string + results map[string]aliveResult + } + + tests := []struct { + name string + sessions []domain.SessionRecord + runtimes []runtimeProbes + wantCalls []call + wantProbe map[string][]string // runtime name -> handle IDs probed, in order + }{ + { + name: "alive session: no death applied, but tick still fires", + sessions: []domain.SessionRecord{aliveSessionWith("s1", "tmux", "h1")}, + runtimes: []runtimeProbes{{name: "tmux", results: map[string]aliveResult{"h1": {alive: true}}}}, + wantCalls: []call{ + {Kind: "TickEscalations", Now: now}, + {Kind: "RunningSessions"}, + }, + wantProbe: map[string][]string{"tmux": {"h1"}}, + }, + { + name: "dead session: exactly one ApplyRuntimeObservation with Dead facts", + sessions: []domain.SessionRecord{aliveSessionWith("s1", "tmux", "h1")}, + runtimes: []runtimeProbes{{name: "tmux", results: map[string]aliveResult{"h1": {alive: false}}}}, + wantCalls: []call{ + {Kind: "TickEscalations", Now: now}, + {Kind: "RunningSessions"}, + { + Kind: "ApplyRuntimeObservation", + Session: "s1", + Facts: ports.RuntimeFacts{ObservedAt: now, RuntimeState: ports.RuntimeProbeDead, ProcessState: ports.ProcessProbeDead}, + }, + }, + wantProbe: map[string][]string{"tmux": {"h1"}}, + }, + { + name: "probe error: reported as failed fact, NOT collapsed to alive", + sessions: []domain.SessionRecord{aliveSessionWith("s1", "tmux", "h1")}, + runtimes: []runtimeProbes{{name: "tmux", results: map[string]aliveResult{"h1": {err: errors.New("boom")}}}}, + wantCalls: []call{ + {Kind: "TickEscalations", Now: now}, + {Kind: "RunningSessions"}, + { + Kind: "ApplyRuntimeObservation", + Session: "s1", + Facts: ports.RuntimeFacts{ObservedAt: now, RuntimeState: ports.RuntimeProbeFailed, ProcessState: ports.ProcessProbeFailed}, + }, + }, + wantProbe: map[string][]string{"tmux": {"h1"}}, + }, + { + name: "multi-runtime dispatch: tmux + zellij in same tick", + sessions: []domain.SessionRecord{ + aliveSessionWith("s1", "tmux", "ht"), + aliveSessionWith("s2", "zellij", "hz"), + }, + runtimes: []runtimeProbes{ + {name: "tmux", results: map[string]aliveResult{"ht": {alive: false}}}, + {name: "zellij", results: map[string]aliveResult{"hz": {alive: true}}}, + }, + wantCalls: []call{ + {Kind: "TickEscalations", Now: now}, + {Kind: "RunningSessions"}, + { + Kind: "ApplyRuntimeObservation", + Session: "s1", + Facts: ports.RuntimeFacts{ObservedAt: now, RuntimeState: ports.RuntimeProbeDead, ProcessState: ports.ProcessProbeDead}, + }, + }, + wantProbe: map[string][]string{"tmux": {"ht"}, "zellij": {"hz"}}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + lcm := &fakeLCM{sessions: tc.sessions} + registry := reaper.MapRegistry{} + byName := map[string]*fakeRuntime{} + for _, r := range tc.runtimes { + rt := &fakeRuntime{results: r.results} + registry[r.name] = rt + byName[r.name] = rt + } + rp := reaper.New(lcm, registry, reaper.Config{Clock: clock, Tick: time.Hour}) + + if err := rp.Tick(context.Background()); err != nil { + t.Fatalf("Tick error: %v", err) + } + + if !reflect.DeepEqual(lcm.calls, tc.wantCalls) { + t.Errorf("LCM call log mismatch:\n got %#v\n want %#v", lcm.calls, tc.wantCalls) + } + + for name, want := range tc.wantProbe { + got := byName[name].probed + if !reflect.DeepEqual(got, want) { + t.Errorf("runtime %q probed handles mismatch: got %v want %v", name, got, want) + } + } + }) + } +} + +// TestReaper_Loop verifies the background goroutine actually drives ticks and +// exits on context cancel without leaking. +func TestReaper_Loop(t *testing.T) { + now := time.Date(2026, 5, 28, 12, 0, 0, 0, time.UTC) + clock := func() time.Time { return now } + lcm := &fakeLCM{} + rp := reaper.New(lcm, reaper.MapRegistry{}, reaper.Config{Clock: clock, Tick: 5 * time.Millisecond}) + + ctx, cancel := context.WithCancel(context.Background()) + done := rp.Start(ctx) + + // Wait for at least two ticks so we know the loop is actually firing. + deadline := time.Now().Add(500 * time.Millisecond) + for time.Now().Before(deadline) { + lcm.mu.Lock() + n := countKind(lcm.calls, "TickEscalations") + lcm.mu.Unlock() + if n >= 2 { + break + } + time.Sleep(2 * time.Millisecond) + } + cancel() + + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("reaper goroutine did not exit within 1s of ctx cancel") + } + + lcm.mu.Lock() + defer lcm.mu.Unlock() + if got := countKind(lcm.calls, "TickEscalations"); got < 2 { + t.Errorf("expected at least 2 TickEscalations calls during loop, got %d", got) + } +} + +func countKind(calls []call, kind string) int { + n := 0 + for _, c := range calls { + if c.Kind == kind { + n++ + } + } + return n +} + +// TestReaper_SkipsUnknownRuntime verifies the reaper does not panic and does not +// report a fact when a session references an unregistered runtime — the reaper +// only reports what it actually probed. +func TestReaper_SkipsUnknownRuntime(t *testing.T) { + now := time.Date(2026, 5, 28, 12, 0, 0, 0, time.UTC) + clock := func() time.Time { return now } + lcm := &fakeLCM{sessions: []domain.SessionRecord{aliveSessionWith("s1", "ghost", "h1")}} + rp := reaper.New(lcm, reaper.MapRegistry{}, reaper.Config{Clock: clock, Tick: time.Hour}) + + if err := rp.Tick(context.Background()); err != nil { + t.Fatalf("Tick error: %v", err) + } + + for _, c := range lcm.calls { + if c.Kind == "ApplyRuntimeObservation" { + t.Fatalf("unexpected ApplyRuntimeObservation for unknown-runtime session: %+v", c) + } + } +} + +// TestReaper_SkipsMissingHandle verifies the reaper does not probe (and does not +// report) for sessions whose runtime handle metadata is missing — probing +// nothing returns no fact. +func TestReaper_SkipsMissingHandle(t *testing.T) { + now := time.Date(2026, 5, 28, 12, 0, 0, 0, time.UTC) + clock := func() time.Time { return now } + sess := aliveSessionWith("s1", "tmux", "h1") + delete(sess.Metadata, lifecycle.MetaRuntimeHandleID) + lcm := &fakeLCM{sessions: []domain.SessionRecord{sess}} + rt := &fakeRuntime{results: map[string]aliveResult{}} + rp := reaper.New(lcm, reaper.MapRegistry{"tmux": rt}, reaper.Config{Clock: clock, Tick: time.Hour}) + + if err := rp.Tick(context.Background()); err != nil { + t.Fatalf("Tick error: %v", err) + } + if len(rt.probed) != 0 { + t.Errorf("expected no probes for session without handle id, got %v", rt.probed) + } + for _, c := range lcm.calls { + if c.Kind == "ApplyRuntimeObservation" { + t.Fatalf("unexpected ApplyRuntimeObservation: %+v", c) + } + } +} diff --git a/backend/internal/ports/inbound.go b/backend/internal/ports/inbound.go index 6508845c0..58ec2015f 100644 --- a/backend/internal/ports/inbound.go +++ b/backend/internal/ports/inbound.go @@ -28,6 +28,14 @@ type LifecycleManager interface { // Reaper heartbeat that drives duration-based escalation (a non-polling // LCM can't wake itself to fire a "30m elapsed" escalation). TickEscalations(ctx context.Context, now time.Time) error + + // RunningSessions returns a snapshot of every session whose runtime axis is + // alive. The reaper calls it once per tick to decide whom to probe. It is a + // read snapshot — the slice and its elements are safe for the caller to + // iterate without holding any LCM lock — and does not violate the + // single-writer invariant (the reaper never writes; it reports facts back + // through ApplyRuntimeObservation). + RunningSessions(ctx context.Context) ([]domain.SessionRecord, error) } // SessionManager is the inbound contract called by the API layer and CLI. It diff --git a/backend/internal/session/fakes_test.go b/backend/internal/session/fakes_test.go index 1796e509e..71eaa4afd 100644 --- a/backend/internal/session/fakes_test.go +++ b/backend/internal/session/fakes_test.go @@ -328,6 +328,10 @@ func (l *recordingLCM) TickEscalations(ctx context.Context, now time.Time) error return l.inner.TickEscalations(ctx, now) } +func (l *recordingLCM) RunningSessions(ctx context.Context) ([]domain.SessionRecord, error) { + return l.inner.RunningSessions(ctx) +} + // ---- harness: wires the SM against the fakes + the real LCM ---- type harness struct { From 1eaaa4ce1d88702ede82d2fe8bb4aae507dc6912 Mon Sep 17 00:00:00 2001 From: harshitsinghbhandari <24b4506@iitb.ac.in> Date: Fri, 29 May 2026 22:18:54 +0530 Subject: [PATCH 2/2] fix(observe): broaden reaper poll set + always report probe fact MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address blocker found in self-review (B1 + I1): - Manager.RunningSessions previously filtered to runtime.State == RuntimeAlive, but a session enters Detecting with runtime axis = RuntimeProbeFailed (failed probe path, decide_bridge.go:72) or RuntimeMissing (detectingLC in manager_test.go:539). The filter silently parked every Detecting session, so the recovery path proved by manager_test.go:59 ("healthy probe recovers liveness-owned detecting -> working") and the terminal path proved by manager_test.go:79 ("dead+dead with no recent activity concludes killed") were both unreachable through the reaper. Broaden the predicate to "session is not in a terminal state" (mirrors the LCM's existing isTerminal helper) and document the wider semantics. - reaper.probeOne now reports every probe result — including alive — back to the LCM as ApplyRuntimeObservation facts. The previous skip-alive optimization was a layering violation: the reaper has no business deciding what counts as a no-op. The LCM's ApplyRuntimeObservation already diffs against canonical and only Upserts on actual change, so steady-state alive stays cheap. With the broadened poll set, an alive probe for a Detecting session IS the recovery fact. - Add unit tests for Manager.RunningSessions covering: nil-lister no-op, lister error propagation, and the full canonical state matrix (working/idle/ needs_input/detecting-probefailed/detecting-missing/not_started included; terminated/done excluded). - Update reaper tests: alive case now asserts the alive fact is reported; new "detecting session: alive probe reported so LCM can recover from quarantine" case locks in the recovery path; multi-runtime case now asserts both runtime facts flow through. - Bump "session in poll set without handle metadata" log from Debug to Warn — it is an anomaly (OnSpawnCompleted should have written both keys), not a routine event. - Document WithSessionLister must be called before any reaper attached to the Manager starts running (it is a bare field read; concurrent re-injection is meaningless anyway). --- backend/internal/lifecycle/manager.go | 35 ++++++--- backend/internal/lifecycle/manager_test.go | 73 +++++++++++++++++++ backend/internal/observe/reaper/reaper.go | 23 +++--- .../internal/observe/reaper/reaper_test.go | 54 +++++++++++++- 4 files changed, 164 insertions(+), 21 deletions(-) diff --git a/backend/internal/lifecycle/manager.go b/backend/internal/lifecycle/manager.go index defa3b643..b5751e867 100644 --- a/backend/internal/lifecycle/manager.go +++ b/backend/internal/lifecycle/manager.go @@ -77,7 +77,10 @@ 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. Calling it more than once replaces the previous lister. +// 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 } @@ -426,16 +429,28 @@ func (m *Manager) OnKillRequested(ctx context.Context, id domain.SessionID, r po // ---- read-snapshot helpers ---- -// RunningSessions returns a snapshot of every persisted session whose runtime -// axis is alive. It exists so the reaper (OBSERVE) can decide whom to probe -// without taking on a LifecycleStore dependency or knowing the LCM's internal -// state. Because the call only reads and copies, it does not break the -// single-writer invariant; concurrent Apply* calls may move sessions in or out -// of "alive" between snapshots, which is correct — the next tick re-reads. +// 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 an empty slice so a goroutine attached to such a Manager -// degrades to a no-op rather than panicking. +// 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 @@ -446,7 +461,7 @@ func (m *Manager) RunningSessions(ctx context.Context) ([]domain.SessionRecord, } out := make([]domain.SessionRecord, 0, len(all)) for _, rec := range all { - if rec.Lifecycle.Runtime.State == domain.RuntimeAlive { + if !isTerminal(rec.Lifecycle.Session.State) { out = append(out, rec) } } diff --git a/backend/internal/lifecycle/manager_test.go b/backend/internal/lifecycle/manager_test.go index e93a33c5d..6a2cc1d1a 100644 --- a/backend/internal/lifecycle/manager_test.go +++ b/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 { diff --git a/backend/internal/observe/reaper/reaper.go b/backend/internal/observe/reaper/reaper.go index 5bd4b0d70..66456ea6e 100644 --- a/backend/internal/observe/reaper/reaper.go +++ b/backend/internal/observe/reaper/reaper.go @@ -152,14 +152,20 @@ func (r *Reaper) Tick(ctx context.Context) error { return nil } -// probeOne handles a single session's probe + fact-report. It is intentionally -// silent on the alive case: a probe that confirms the steady "alive" state is -// not a fact worth re-reporting (the runtime axis is already alive, so the LCM -// would diff to a no-op anyway). Dead and probe-failure ARE reported. +// 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 { - r.logger.Debug("reaper: session has no runtime handle metadata, skipping", + // 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 } @@ -183,11 +189,8 @@ func (r *Reaper) probeOne(ctx context.Context, sess domain.SessionRecord, now ti r.logger.Debug("reaper: probe error reported as failed fact", "session", sess.ID, "runtime", handle.RuntimeName, "err", probeErr) case alive: - // Steady-state alive carries no new information — skip the call so we - // don't churn the LCM with no-op load/diff/persist work on every tick. - // Recovery from detecting via probe still flows through this path - // whenever the probe is NOT alive (failure or death). - return + facts.RuntimeState = ports.RuntimeProbeAlive + facts.ProcessState = ports.ProcessProbeAlive default: facts.RuntimeState = ports.RuntimeProbeDead facts.ProcessState = ports.ProcessProbeDead diff --git a/backend/internal/observe/reaper/reaper_test.go b/backend/internal/observe/reaper/reaper_test.go index 783fbad73..d6b88efdf 100644 --- a/backend/internal/observe/reaper/reaper_test.go +++ b/backend/internal/observe/reaper/reaper_test.go @@ -121,6 +121,7 @@ func aliveSessionWith(id domain.SessionID, runtimeName, handleID string) domain. return domain.SessionRecord{ ID: id, Lifecycle: domain.CanonicalSessionLifecycle{ + Session: domain.SessionSubstate{State: domain.SessionWorking, Reason: domain.ReasonTaskInProgress}, Runtime: domain.RuntimeSubstate{State: domain.RuntimeAlive, Reason: domain.RuntimeReasonProcessRunning}, }, Metadata: map[string]string{ @@ -130,6 +131,23 @@ func aliveSessionWith(id domain.SessionID, runtimeName, handleID string) domain. } } +// detectingSessionWith returns a session in the Detecting quarantine, the +// shape `Manager.RunningSessions` MUST include so a probe-alive can recover it +// (otherwise the reaper traps every session that hiccups once in detecting). +func detectingSessionWith(id domain.SessionID, runtimeName, handleID string) domain.SessionRecord { + return domain.SessionRecord{ + ID: id, + Lifecycle: domain.CanonicalSessionLifecycle{ + Session: domain.SessionSubstate{State: domain.SessionDetecting, Reason: domain.ReasonProbeFailure}, + Runtime: domain.RuntimeSubstate{State: domain.RuntimeProbeFailed, Reason: domain.RuntimeReasonProbeError}, + }, + Metadata: map[string]string{ + lifecycle.MetaRuntimeHandleID: handleID, + lifecycle.MetaRuntimeName: runtimeName, + }, + } +} + // ---- tests ---- func TestReaper_Tick(t *testing.T) { @@ -149,12 +167,41 @@ func TestReaper_Tick(t *testing.T) { wantProbe map[string][]string // runtime name -> handle IDs probed, in order }{ { - name: "alive session: no death applied, but tick still fires", + // "No death applied" per the spec: the LCM does not receive a + // death-causing fact. It still receives the alive fact, because + // the reaper reports what it probed and the LCM is the one that + // diffs against canonical (a no-op when runtime is already alive, + // a recovery when the session was in Detecting). + name: "alive session: alive fact reported, no death applied, tick still fires", sessions: []domain.SessionRecord{aliveSessionWith("s1", "tmux", "h1")}, runtimes: []runtimeProbes{{name: "tmux", results: map[string]aliveResult{"h1": {alive: true}}}}, wantCalls: []call{ {Kind: "TickEscalations", Now: now}, {Kind: "RunningSessions"}, + { + Kind: "ApplyRuntimeObservation", + Session: "s1", + Facts: ports.RuntimeFacts{ObservedAt: now, RuntimeState: ports.RuntimeProbeAlive, ProcessState: ports.ProcessProbeAlive}, + }, + }, + wantProbe: map[string][]string{"tmux": {"h1"}}, + }, + { + // Recovery path: a session in Detecting+probe_failed must be in + // the poll set so an alive probe can flow through and recover it. + // If the reaper filtered to runtime-axis-alive only, this session + // would be trapped in Detecting forever. + name: "detecting session: alive probe reported so LCM can recover from quarantine", + sessions: []domain.SessionRecord{detectingSessionWith("s1", "tmux", "h1")}, + runtimes: []runtimeProbes{{name: "tmux", results: map[string]aliveResult{"h1": {alive: true}}}}, + wantCalls: []call{ + {Kind: "TickEscalations", Now: now}, + {Kind: "RunningSessions"}, + { + Kind: "ApplyRuntimeObservation", + Session: "s1", + Facts: ports.RuntimeFacts{ObservedAt: now, RuntimeState: ports.RuntimeProbeAlive, ProcessState: ports.ProcessProbeAlive}, + }, }, wantProbe: map[string][]string{"tmux": {"h1"}}, }, @@ -206,6 +253,11 @@ func TestReaper_Tick(t *testing.T) { Session: "s1", Facts: ports.RuntimeFacts{ObservedAt: now, RuntimeState: ports.RuntimeProbeDead, ProcessState: ports.ProcessProbeDead}, }, + { + Kind: "ApplyRuntimeObservation", + Session: "s2", + Facts: ports.RuntimeFacts{ObservedAt: now, RuntimeState: ports.RuntimeProbeAlive, ProcessState: ports.ProcessProbeAlive}, + }, }, wantProbe: map[string][]string{"tmux": {"ht"}, "zellij": {"hz"}}, },