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"}}, },