fix(observe): broaden reaper poll set + always report probe fact
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).
This commit is contained in:
harshitsinghbhandari
parent
11475fbc72
commit
1eaaa4ce1d
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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"}},
|
||||
},
|
||||
|
|
|
|||
Loading…
Reference in New Issue