Merge pull request #29 from aoagents/feat/observe-reaper
feat(observe): reaper for liveness probe + TickEscalations heartbeat (#9)
This commit is contained in:
Harshit Singh Bhandari
GitHub
commit
527d9c8303
|
|
@ -52,6 +52,14 @@ type Manager struct {
|
|||
trackers map[trackerKey]*reactionTracker
|
||||
trackerMu sync.Mutex
|
||||
clock func() time.Time
|
||||
|
||||
// sessionLister returns every session known to persistence so RunningSessions
|
||||
// can filter by runtime axis without coupling the LCM to a cross-project
|
||||
// store API the Tom-store does not yet expose. The daemon (lane #10) injects
|
||||
// the production lister via WithSessionLister; until then, the call returns
|
||||
// no sessions so a reaper attached to an unwired Manager is a clean no-op
|
||||
// rather than a panic.
|
||||
sessionLister func(ctx context.Context) ([]domain.SessionRecord, error)
|
||||
}
|
||||
|
||||
var _ ports.LifecycleManager = (*Manager)(nil)
|
||||
|
|
@ -67,6 +75,16 @@ func New(store ports.LifecycleStore, notifier ports.Notifier, messenger ports.Ag
|
|||
}
|
||||
}
|
||||
|
||||
// WithSessionLister injects the function the LCM uses to enumerate all
|
||||
// persisted sessions for RunningSessions. The daemon wires this against the
|
||||
// store at startup; it must be called BEFORE any reaper attached to this
|
||||
// Manager starts running, since concurrent calls would race the bare-field
|
||||
// read in RunningSessions. Calling it more than once replaces the previous
|
||||
// lister.
|
||||
func (m *Manager) WithSessionLister(fn func(ctx context.Context) ([]domain.SessionRecord, error)) {
|
||||
m.sessionLister = fn
|
||||
}
|
||||
|
||||
// ---- per-session serialisation ----
|
||||
|
||||
// keyedMutex hands out one lock per session id so the load->decide->persist
|
||||
|
|
@ -409,6 +427,47 @@ func (m *Manager) OnKillRequested(ctx context.Context, id domain.SessionID, r po
|
|||
return nil
|
||||
}
|
||||
|
||||
// ---- read-snapshot helpers ----
|
||||
|
||||
// RunningSessions returns a snapshot of every persisted session worth probing
|
||||
// in the next reaper tick. "Worth probing" is wider than "runtime axis alive":
|
||||
// it includes sessions in the Detecting quarantine, because a fresh probe is
|
||||
// the only fact that can recover them (back to working) or escalate them
|
||||
// (terminal killed). Filtering to runtime-axis-alive would silently park every
|
||||
// Detecting session — a single failed probe would never get a second chance
|
||||
// and recovery via runtime probe would be unreachable.
|
||||
//
|
||||
// The predicate is "not a final session state". Terminal session states (done,
|
||||
// terminated) are excluded because Restore is the only path back; observations
|
||||
// must not reopen them (#1 invariant). Sessions in earlier states — not_started,
|
||||
// working, idle, needs_input, stuck, detecting — are all included. Those that
|
||||
// lack runtime handle metadata (e.g. not_started before OnSpawnCompleted) are
|
||||
// returned and harmlessly skipped by the reaper's per-session handle guard.
|
||||
//
|
||||
// The call only reads and copies, so it does not break the single-writer
|
||||
// invariant; concurrent Apply* calls may move sessions in or out of the probe
|
||||
// set between snapshots, which is correct — the next tick re-reads.
|
||||
//
|
||||
// When no lister has been wired (e.g. tests construct a bare Manager), the
|
||||
// method returns nil so a goroutine attached to such a Manager degrades to a
|
||||
// no-op rather than panicking.
|
||||
func (m *Manager) RunningSessions(ctx context.Context) ([]domain.SessionRecord, error) {
|
||||
if m.sessionLister == nil {
|
||||
return nil, nil
|
||||
}
|
||||
all, err := m.sessionLister(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out := make([]domain.SessionRecord, 0, len(all))
|
||||
for _, rec := range all {
|
||||
if !isTerminal(rec.Lifecycle.Session.State) {
|
||||
out = append(out, rec)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// ---- diff helpers ----
|
||||
|
||||
// setSessionIfChanged sets next.Session only when the decided sub-state differs
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -0,0 +1,215 @@
|
|||
// Package reaper implements the OBSERVE-layer polling timer that supplies the
|
||||
// LCM with the two facts the LCM cannot wake itself to discover: a periodic
|
||||
// duration-based escalation heartbeat, and per-session runtime liveness probes.
|
||||
//
|
||||
// The reaper sits OUTSIDE the LCM's per-session serial loop. It only REPORTS
|
||||
// facts — it never decides whether a session is "truly" dead. The decider
|
||||
// (anti-flap Detecting quarantine, terminal-session rules) is owned by the LCM
|
||||
// and consumes these facts through the regular ApplyRuntimeObservation entry
|
||||
// point. A probe error is reported as a probe-failure fact, never collapsed to
|
||||
// "alive" or "dead", so the LCM's failed-probe ≠ dead invariant holds.
|
||||
package reaper
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
"github.com/aoagents/agent-orchestrator/backend/internal/domain"
|
||||
"github.com/aoagents/agent-orchestrator/backend/internal/lifecycle"
|
||||
"github.com/aoagents/agent-orchestrator/backend/internal/ports"
|
||||
)
|
||||
|
||||
// DefaultTickInterval is the cadence used when Config.Tick is zero. It mirrors
|
||||
// the design doc's 5s sampling window for runtime liveness.
|
||||
const DefaultTickInterval = 5 * time.Second
|
||||
|
||||
// RuntimeRegistry resolves a runtime adapter by the RuntimeName recorded in a
|
||||
// session's RuntimeHandle. The reaper looks the runtime up per-session so a
|
||||
// single reaper instance can probe tmux- and zellij-backed sessions side by
|
||||
// side without knowing about either at construction.
|
||||
type RuntimeRegistry interface {
|
||||
Runtime(name string) (ports.Runtime, bool)
|
||||
}
|
||||
|
||||
// MapRegistry is the trivial RuntimeRegistry: a name->runtime map. Callers
|
||||
// that need dynamic registration can implement RuntimeRegistry themselves.
|
||||
type MapRegistry map[string]ports.Runtime
|
||||
|
||||
// Runtime implements RuntimeRegistry.
|
||||
func (m MapRegistry) Runtime(name string) (ports.Runtime, bool) {
|
||||
rt, ok := m[name]
|
||||
return rt, ok
|
||||
}
|
||||
|
||||
// Config holds the externally-tunable knobs for a Reaper. Every field is
|
||||
// optional; zero values fall back to safe defaults so production wiring (which
|
||||
// only needs to inject the LCM and registry) and tests (which inject a clock
|
||||
// plus a fast tick) can both stay terse.
|
||||
type Config struct {
|
||||
// Tick is the interval between ticks. <=0 means DefaultTickInterval.
|
||||
Tick time.Duration
|
||||
// Clock supplies ObservedAt and TickEscalations now stamps. nil means
|
||||
// time.Now. Injected in tests so assertions don't race wallclock.
|
||||
Clock func() time.Time
|
||||
// Logger receives operational diagnostics (probe errors, skipped sessions,
|
||||
// LCM call failures). The reaper logs but does not propagate these errors
|
||||
// because a single failed probe must not kill the loop. nil means
|
||||
// slog.Default.
|
||||
Logger *slog.Logger
|
||||
}
|
||||
|
||||
// Reaper is the polling timer. Construct it with New; start the background
|
||||
// goroutine with Start, or drive a single cycle synchronously with Tick.
|
||||
type Reaper struct {
|
||||
lcm ports.LifecycleManager
|
||||
registry RuntimeRegistry
|
||||
tick time.Duration
|
||||
clock func() time.Time
|
||||
logger *slog.Logger
|
||||
}
|
||||
|
||||
// New constructs a Reaper. The LCM is the sole writer destination (the reaper
|
||||
// reports facts via ApplyRuntimeObservation and TickEscalations); the registry
|
||||
// resolves the runtime adapter to use per session.
|
||||
func New(lcm ports.LifecycleManager, registry RuntimeRegistry, cfg Config) *Reaper {
|
||||
r := &Reaper{
|
||||
lcm: lcm,
|
||||
registry: registry,
|
||||
tick: cfg.Tick,
|
||||
clock: cfg.Clock,
|
||||
logger: cfg.Logger,
|
||||
}
|
||||
if r.tick <= 0 {
|
||||
r.tick = DefaultTickInterval
|
||||
}
|
||||
if r.clock == nil {
|
||||
r.clock = time.Now
|
||||
}
|
||||
if r.logger == nil {
|
||||
r.logger = slog.Default()
|
||||
}
|
||||
return r
|
||||
}
|
||||
|
||||
// Start launches the background goroutine and returns a channel that closes
|
||||
// once the loop has exited. The loop exits on ctx cancellation; the channel
|
||||
// gives the daemon a clean shutdown hook (wait on it after cancel to confirm
|
||||
// the reaper has stopped before tearing down dependencies).
|
||||
func (r *Reaper) Start(ctx context.Context) <-chan struct{} {
|
||||
done := make(chan struct{})
|
||||
go r.loop(ctx, done)
|
||||
return done
|
||||
}
|
||||
|
||||
func (r *Reaper) loop(ctx context.Context, done chan<- struct{}) {
|
||||
defer close(done)
|
||||
t := time.NewTicker(r.tick)
|
||||
defer t.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-t.C:
|
||||
if err := r.Tick(ctx); err != nil {
|
||||
r.logger.Error("reaper: tick failed", "err", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Tick runs one observation cycle: it always fires TickEscalations first (the
|
||||
// duration-based escalation heartbeat, which the synchronous LCM cannot wake
|
||||
// itself to drive), then enumerates the LCM's running sessions, probes each
|
||||
// one's runtime, and reports any non-alive result back as a fact.
|
||||
//
|
||||
// Tick is exported so the daemon (and tests) can drive cycles synchronously,
|
||||
// and so the Start goroutine has a single chokepoint to log against.
|
||||
//
|
||||
// Errors: only the RunningSessions failure is propagated, since it short-
|
||||
// circuits the rest of the cycle. TickEscalations and per-session
|
||||
// ApplyRuntimeObservation failures are logged but never propagated — one
|
||||
// failed call must not bring down the loop.
|
||||
func (r *Reaper) Tick(ctx context.Context) error {
|
||||
now := r.clock()
|
||||
|
||||
// Heartbeat is best-effort and runs before enumeration so duration-based
|
||||
// escalations still fire if the running-set lookup is the thing that
|
||||
// errored. The LCM's TickEscalations is itself idempotent (no canonical
|
||||
// writes) — at worst we miss escalating once and pick it up next tick.
|
||||
if err := r.lcm.TickEscalations(ctx, now); err != nil {
|
||||
r.logger.Error("reaper: TickEscalations failed", "err", err)
|
||||
}
|
||||
|
||||
sessions, err := r.lcm.RunningSessions(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for _, sess := range sessions {
|
||||
r.probeOne(ctx, sess, now)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// probeOne handles a single session's probe + fact-report. Every probe result —
|
||||
// alive, dead, or failed — is reported as a fact to the LCM. The reaper does
|
||||
// not optimize away the "alive" case, because a session in Detecting (whose
|
||||
// runtime axis is NOT alive) is included in the running set and needs the
|
||||
// alive probe to recover; the reaper has no business deciding what counts as
|
||||
// a no-op. The LCM's ApplyRuntimeObservation diffs against canonical and
|
||||
// only Upserts on actual change, so steady-state alive is already cheap.
|
||||
func (r *Reaper) probeOne(ctx context.Context, sess domain.SessionRecord, now time.Time) {
|
||||
handle, ok := handleFromRecord(sess)
|
||||
if !ok {
|
||||
// A session in the running-set without a handle is an anomaly worth
|
||||
// surfacing (OnSpawnCompleted should have set both keys). Warn rather
|
||||
// than Debug so it doesn't hide behind a noisy log level.
|
||||
r.logger.Warn("reaper: session has no runtime handle metadata, skipping",
|
||||
"session", sess.ID)
|
||||
return
|
||||
}
|
||||
rt, ok := r.registry.Runtime(handle.RuntimeName)
|
||||
if !ok {
|
||||
r.logger.Warn("reaper: no runtime registered for session, skipping",
|
||||
"session", sess.ID, "runtime", handle.RuntimeName)
|
||||
return
|
||||
}
|
||||
|
||||
alive, probeErr := rt.IsAlive(ctx, handle)
|
||||
facts := ports.RuntimeFacts{ObservedAt: now}
|
||||
switch {
|
||||
case probeErr != nil:
|
||||
// Failed probe must NOT be collapsed to alive — that would let a
|
||||
// transient tmux/zellij outage hide a really-dead session, and a
|
||||
// transient adapter bug terminate a really-alive one. Report failed
|
||||
// and let the LCM's detecting quarantine arbitrate.
|
||||
facts.RuntimeState = ports.RuntimeProbeFailed
|
||||
facts.ProcessState = ports.ProcessProbeFailed
|
||||
r.logger.Debug("reaper: probe error reported as failed fact",
|
||||
"session", sess.ID, "runtime", handle.RuntimeName, "err", probeErr)
|
||||
case alive:
|
||||
facts.RuntimeState = ports.RuntimeProbeAlive
|
||||
facts.ProcessState = ports.ProcessProbeAlive
|
||||
default:
|
||||
facts.RuntimeState = ports.RuntimeProbeDead
|
||||
facts.ProcessState = ports.ProcessProbeDead
|
||||
}
|
||||
|
||||
if err := r.lcm.ApplyRuntimeObservation(ctx, sess.ID, facts); err != nil {
|
||||
r.logger.Error("reaper: ApplyRuntimeObservation failed",
|
||||
"session", sess.ID, "err", err)
|
||||
}
|
||||
}
|
||||
|
||||
// handleFromRecord reconstructs the RuntimeHandle stored on the session by
|
||||
// OnSpawnCompleted. Both keys are required; either being empty is the
|
||||
// "session lacks a probable handle" signal that probeOne uses to skip.
|
||||
func handleFromRecord(rec domain.SessionRecord) (ports.RuntimeHandle, bool) {
|
||||
id := rec.Metadata[lifecycle.MetaRuntimeHandleID]
|
||||
name := rec.Metadata[lifecycle.MetaRuntimeName]
|
||||
if id == "" || name == "" {
|
||||
return ports.RuntimeHandle{}, false
|
||||
}
|
||||
return ports.RuntimeHandle{ID: id, RuntimeName: name}, true
|
||||
}
|
||||
|
|
@ -0,0 +1,386 @@
|
|||
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{
|
||||
Session: domain.SessionSubstate{State: domain.SessionWorking, Reason: domain.ReasonTaskInProgress},
|
||||
Runtime: domain.RuntimeSubstate{State: domain.RuntimeAlive, Reason: domain.RuntimeReasonProcessRunning},
|
||||
},
|
||||
Metadata: map[string]string{
|
||||
lifecycle.MetaRuntimeHandleID: handleID,
|
||||
lifecycle.MetaRuntimeName: runtimeName,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// 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) {
|
||||
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
|
||||
}{
|
||||
{
|
||||
// "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"}},
|
||||
},
|
||||
{
|
||||
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},
|
||||
},
|
||||
{
|
||||
Kind: "ApplyRuntimeObservation",
|
||||
Session: "s2",
|
||||
Facts: ports.RuntimeFacts{ObservedAt: now, RuntimeState: ports.RuntimeProbeAlive, ProcessState: ports.ProcessProbeAlive},
|
||||
},
|
||||
},
|
||||
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