feat(observe): reaper for liveness probe + TickEscalations heartbeat (#9)
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.
This commit is contained in:
harshitsinghbhandari
parent
878190602d
commit
11475fbc72
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
Loading…
Reference in New Issue