feat(lifecycle): implement LCM Apply* pipeline (split A)
Implements ports.LifecycleManager as a synchronous observe->decide->persist reducer. Every entrypoint runs the shared pipeline under a per-session lock: load canonical -> run the matching pure decider -> diff into a sparse merge-patch -> persist. Never polls, never writes the display status. - ApplyRuntimeObservation -> probe decider; always writes the runtime axis. - ApplySCMObservation -> open-PR / terminal-PR deciders (failed fetch is a no-op: failed probe != "no PR"). Open PRs write only the PR axis. - ApplyActivitySignal -> updates the activity axis + maps onto the session axis; only valid-confidence signals are authoritative. - OnSpawnCompleted -> runtime alive + handles to metadata; session stays not_started (display: spawning). - OnKillRequested -> SM's explicit terminal-write authority. - TickEscalations -> no-op stub (reaction/escalation engine is split B). Composition rule (#1): liveness owns the runtime + death axis; activity owns the working/idle/waiting axis. A healthy probe verdict writes the session axis only to recover a liveness-owned state, so it never clobbers an activity-owned needs_input/blocked. Activity is the mirror: it stays off the death axis. Detecting clear (#3): a non-detecting probe verdict clears stale detecting memory so the next probe reads no phantom Prior. Built/tested against in-memory fakes (LifecycleStore with full merge-patch + ExpectedRevision, recording Notifier/AgentMessenger). Per-session serialisation verified under -race. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
harshitsinghbhandari
parent
217c50eeb3
commit
1420bb9493
|
|
@ -0,0 +1,204 @@
|
|||
package lifecycle
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"github.com/aoagents/agent-orchestrator/backend/internal/domain"
|
||||
"github.com/aoagents/agent-orchestrator/backend/internal/domain/decide"
|
||||
"github.com/aoagents/agent-orchestrator/backend/internal/ports"
|
||||
)
|
||||
|
||||
// defaultRecentActivityWindow is how fresh the last activity signal must be for
|
||||
// the probe decider to treat the agent as "recently active" (which keeps an
|
||||
// ambiguous dead-runtime probe in detecting instead of concluding death).
|
||||
const defaultRecentActivityWindow = 60 * time.Second
|
||||
|
||||
// ---- fact translation: ports DTOs -> pure decide inputs ----
|
||||
|
||||
// runtimeFactsToProbeInput maps a raw RuntimeFacts (plus the prior detecting
|
||||
// memory and last-known activity read back from canonical) into the probe
|
||||
// decider's input. KillRequested is always false here: the inferred-death path
|
||||
// never carries an explicit kill — that arrives via OnKillRequested.
|
||||
func runtimeFactsToProbeInput(f ports.RuntimeFacts, cur domain.CanonicalSessionLifecycle, window time.Duration) decide.ProbeInput {
|
||||
rt, rtFailed := runtimeProbeToState(f.RuntimeState)
|
||||
proc, procFailed := processProbeToLiveness(f.ProcessState)
|
||||
now := nowOr(f.ObservedAt)
|
||||
return decide.ProbeInput{
|
||||
Runtime: rt,
|
||||
RuntimeFailed: rtFailed,
|
||||
Process: proc,
|
||||
ProcessFailed: procFailed,
|
||||
RecentActivity: hasRecentActivity(cur.Activity, now, window),
|
||||
Prior: cur.Detecting,
|
||||
Now: now,
|
||||
}
|
||||
}
|
||||
|
||||
func runtimeProbeToState(p ports.RuntimeProbe) (domain.RuntimeState, bool) {
|
||||
switch p {
|
||||
case ports.RuntimeProbeAlive:
|
||||
return domain.RuntimeAlive, false
|
||||
case ports.RuntimeProbeDead:
|
||||
return domain.RuntimeExited, false
|
||||
case ports.RuntimeProbeFailed:
|
||||
return domain.RuntimeProbeFailed, true
|
||||
default: // indeterminate / unset: ambiguous, never a death conclusion
|
||||
return domain.RuntimeUnknown, false
|
||||
}
|
||||
}
|
||||
|
||||
func processProbeToLiveness(p ports.ProcessProbe) (decide.ProcessLiveness, bool) {
|
||||
switch p {
|
||||
case ports.ProcessProbeAlive:
|
||||
return decide.ProcessAlive, false
|
||||
case ports.ProcessProbeDead:
|
||||
return decide.ProcessDead, false
|
||||
case ports.ProcessProbeFailed:
|
||||
return decide.ProcessIndeterminate, true
|
||||
default: // indeterminate / unset
|
||||
return decide.ProcessIndeterminate, false
|
||||
}
|
||||
}
|
||||
|
||||
// runtimeSubstateFromFacts derives the runtime sub-state to persist. Liveness
|
||||
// always owns this axis, so it is written on every runtime observation
|
||||
// regardless of what the session axis does.
|
||||
func runtimeSubstateFromFacts(f ports.RuntimeFacts) domain.RuntimeSubstate {
|
||||
switch f.RuntimeState {
|
||||
case ports.RuntimeProbeAlive:
|
||||
return domain.RuntimeSubstate{State: domain.RuntimeAlive, Reason: domain.RuntimeReasonProcessRunning}
|
||||
case ports.RuntimeProbeDead:
|
||||
return domain.RuntimeSubstate{State: domain.RuntimeExited, Reason: domain.RuntimeReasonTmuxMissing}
|
||||
case ports.RuntimeProbeFailed:
|
||||
return domain.RuntimeSubstate{State: domain.RuntimeProbeFailed, Reason: domain.RuntimeReasonProbeError}
|
||||
default:
|
||||
return domain.RuntimeSubstate{State: domain.RuntimeUnknown, Reason: domain.RuntimeReasonProbeError}
|
||||
}
|
||||
}
|
||||
|
||||
// hasRecentActivity answers the probe decider's "was the agent heard from
|
||||
// recently?" question. Sticky states (waiting_input/blocked) count as recent
|
||||
// because they mean a live-but-paused agent; an explicit exited signal never
|
||||
// counts; otherwise we age the last-activity timestamp against the window.
|
||||
func hasRecentActivity(a domain.ActivitySubstate, now time.Time, window time.Duration) bool {
|
||||
if a.State == domain.ActivityExited {
|
||||
return false
|
||||
}
|
||||
if a.State.IsSticky() {
|
||||
return true
|
||||
}
|
||||
if a.LastActivityAt.IsZero() {
|
||||
return false
|
||||
}
|
||||
return now.Sub(a.LastActivityAt) <= window
|
||||
}
|
||||
|
||||
// openPRInput maps SCM facts onto the open-PR ladder. IdleBeyond is always false
|
||||
// in split A — the idle-duration signal is owned by the escalation engine
|
||||
// (split B); the synchronous LCM has no clock of its own here.
|
||||
func openPRInput(f ports.SCMFacts) decide.OpenPRInput {
|
||||
return decide.OpenPRInput{
|
||||
CIFailing: f.CISummary == ports.CIFailing,
|
||||
ChangesRequested: f.ReviewDecision == ports.ReviewChangesRequested,
|
||||
Approved: f.ReviewDecision == ports.ReviewApproved,
|
||||
Mergeable: f.Mergeability.Mergeable,
|
||||
ReviewPending: f.ReviewDecision == ports.ReviewPending,
|
||||
Number: f.PRNumber,
|
||||
URL: f.PRURL,
|
||||
}
|
||||
}
|
||||
|
||||
// ---- activity -> session axis mapping (activity owns working/idle/waiting) ----
|
||||
|
||||
// activityToSession maps an activity classification onto the session sub-state.
|
||||
// exited returns ok=false: an exit signal must NOT write a terminal session
|
||||
// state — only the probe pipeline (via detecting) may conclude inferred death.
|
||||
func activityToSession(a domain.ActivityState) (domain.SessionState, domain.SessionReason, bool) {
|
||||
switch a {
|
||||
case domain.ActivityActive:
|
||||
return domain.SessionWorking, domain.ReasonTaskInProgress, true
|
||||
case domain.ActivityReady, domain.ActivityIdle:
|
||||
return domain.SessionIdle, domain.ReasonResearchComplete, true
|
||||
case domain.ActivityWaitingInput:
|
||||
return domain.SessionNeedsInput, domain.ReasonAwaitingUserInput, true
|
||||
case domain.ActivityBlocked:
|
||||
return domain.SessionStuck, domain.ReasonAwaitingUserInput, true
|
||||
default: // exited / unset
|
||||
return "", "", false
|
||||
}
|
||||
}
|
||||
|
||||
// ---- composition predicates: who may write the session axis ----
|
||||
|
||||
// isTerminal reports a final session state that must not be resurrected by an
|
||||
// observation (only an explicit Restore reopens a terminal session).
|
||||
func isTerminal(s domain.SessionState) bool {
|
||||
return s == domain.SessionDone || s == domain.SessionTerminated
|
||||
}
|
||||
|
||||
// isLivenessOwned reports whether the current session sub-state was set by the
|
||||
// liveness/death axis (the probe pipeline) and may therefore be recovered by a
|
||||
// later healthy probe. detecting is always liveness-owned; a stuck/terminated
|
||||
// state is liveness-owned only when its reason came from a death inference.
|
||||
func isLivenessOwned(s domain.SessionSubstate) bool {
|
||||
if s.State == domain.SessionDetecting {
|
||||
return true
|
||||
}
|
||||
switch s.Reason {
|
||||
case domain.ReasonRuntimeLost, domain.ReasonAgentProcessExited, domain.ReasonProbeFailure:
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// shouldWriteSessionRuntime is the #1 composition rule for ApplyRuntimeObservation.
|
||||
// A death-axis verdict (detecting/stuck/terminal) always writes — it overrides
|
||||
// activity because a (maybe) dead agent can't be working/waiting. A healthy
|
||||
// "working" verdict only writes when it is recovering a liveness-owned state
|
||||
// (e.g. detecting -> working); it must NOT clobber an activity-owned
|
||||
// needs_input/blocked/idle the activity axis is responsible for.
|
||||
func shouldWriteSessionRuntime(d decide.LifecycleDecision, cur domain.CanonicalSessionLifecycle) bool {
|
||||
if d.SessionState == domain.SessionWorking {
|
||||
return !isTerminal(cur.Session.State) && isLivenessOwned(cur.Session)
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// shouldWriteSessionActivity is the mirror rule for ApplyActivitySignal: the
|
||||
// activity axis owns working/idle/waiting, but it must not touch the death axis.
|
||||
// It writes unless the session is terminal or currently liveness-owned (let the
|
||||
// probe pipeline resolve detecting / death-inferred states instead).
|
||||
func shouldWriteSessionActivity(cur domain.CanonicalSessionLifecycle) bool {
|
||||
return !isTerminal(cur.Session.State) && !isLivenessOwned(cur.Session)
|
||||
}
|
||||
|
||||
// ---- explicit-kill mapping (SM's terminal-write authority) ----
|
||||
|
||||
func killSession(k ports.LifecycleKillReason) domain.SessionSubstate {
|
||||
switch k {
|
||||
case ports.KillManual:
|
||||
return domain.SessionSubstate{State: domain.SessionTerminated, Reason: domain.ReasonManuallyKilled}
|
||||
case ports.KillCleanup:
|
||||
return domain.SessionSubstate{State: domain.SessionTerminated, Reason: domain.ReasonAutoCleanup}
|
||||
default: // error
|
||||
return domain.SessionSubstate{State: domain.SessionTerminated, Reason: domain.ReasonErrorInProcess}
|
||||
}
|
||||
}
|
||||
|
||||
func killRuntime(k ports.LifecycleKillReason) domain.RuntimeSubstate {
|
||||
switch k {
|
||||
case ports.KillManual:
|
||||
return domain.RuntimeSubstate{State: domain.RuntimeExited, Reason: domain.RuntimeReasonManualKillRequested}
|
||||
case ports.KillCleanup:
|
||||
return domain.RuntimeSubstate{State: domain.RuntimeExited, Reason: domain.RuntimeReasonAutoCleanup}
|
||||
default: // error
|
||||
return domain.RuntimeSubstate{State: domain.RuntimeExited, Reason: domain.RuntimeReasonProbeError}
|
||||
}
|
||||
}
|
||||
|
||||
func nowOr(t time.Time) time.Time {
|
||||
if t.IsZero() {
|
||||
return time.Now()
|
||||
}
|
||||
return t
|
||||
}
|
||||
|
|
@ -0,0 +1,161 @@
|
|||
package lifecycle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/aoagents/agent-orchestrator/backend/internal/domain"
|
||||
"github.com/aoagents/agent-orchestrator/backend/internal/ports"
|
||||
)
|
||||
|
||||
// fakeStore is an in-memory LifecycleStore that faithfully applies merge-patch
|
||||
// semantics (sparse field writes, the three-way Detecting/ClearDetecting rule,
|
||||
// ExpectedRevision optimistic-concurrency check, monotonic Revision bump) so
|
||||
// tests assert against the real persisted canonical.
|
||||
type fakeStore struct {
|
||||
mu sync.Mutex
|
||||
records map[domain.SessionID]*domain.SessionRecord
|
||||
metadata map[domain.SessionID]map[string]string
|
||||
}
|
||||
|
||||
var _ ports.LifecycleStore = (*fakeStore)(nil)
|
||||
|
||||
func newFakeStore() *fakeStore {
|
||||
return &fakeStore{
|
||||
records: map[domain.SessionID]*domain.SessionRecord{},
|
||||
metadata: map[domain.SessionID]map[string]string{},
|
||||
}
|
||||
}
|
||||
|
||||
// seed installs a starting lifecycle for a session id (bypassing the patch path).
|
||||
func (s *fakeStore) seed(id domain.SessionID, l domain.CanonicalSessionLifecycle) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if l.Version == 0 {
|
||||
l.Version = domain.LifecycleVersion
|
||||
}
|
||||
s.records[id] = &domain.SessionRecord{ID: id, Lifecycle: l}
|
||||
}
|
||||
|
||||
func (s *fakeStore) Load(_ context.Context, id domain.SessionID) (domain.CanonicalSessionLifecycle, bool, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
rec, ok := s.records[id]
|
||||
if !ok {
|
||||
return domain.CanonicalSessionLifecycle{}, false, nil
|
||||
}
|
||||
return rec.Lifecycle, true, nil
|
||||
}
|
||||
|
||||
func (s *fakeStore) PatchLifecycle(_ context.Context, id domain.SessionID, p ports.LifecyclePatch) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
|
||||
rec, ok := s.records[id]
|
||||
if !ok {
|
||||
rec = &domain.SessionRecord{ID: id, Lifecycle: domain.CanonicalSessionLifecycle{Version: domain.LifecycleVersion}}
|
||||
s.records[id] = rec
|
||||
}
|
||||
l := &rec.Lifecycle
|
||||
|
||||
if p.ExpectedRevision != nil && *p.ExpectedRevision != l.Revision {
|
||||
return fmt.Errorf("revision mismatch for %s: have %d, expected %d", id, l.Revision, *p.ExpectedRevision)
|
||||
}
|
||||
|
||||
if p.Session != nil {
|
||||
l.Session = *p.Session
|
||||
}
|
||||
if p.PR != nil {
|
||||
l.PR = *p.PR
|
||||
}
|
||||
if p.Runtime != nil {
|
||||
l.Runtime = *p.Runtime
|
||||
}
|
||||
if p.Activity != nil {
|
||||
l.Activity = *p.Activity
|
||||
}
|
||||
switch {
|
||||
case p.ClearDetecting:
|
||||
l.Detecting = nil
|
||||
case p.Detecting != nil:
|
||||
d := *p.Detecting
|
||||
l.Detecting = &d
|
||||
}
|
||||
|
||||
l.Version = domain.LifecycleVersion
|
||||
l.Revision++
|
||||
rec.UpdatedAt = time.Now()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *fakeStore) List(_ context.Context, project domain.ProjectID) ([]domain.SessionRecord, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
var out []domain.SessionRecord
|
||||
for _, rec := range s.records {
|
||||
if rec.ProjectID == project {
|
||||
out = append(out, *rec)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *fakeStore) GetMetadata(_ context.Context, id domain.SessionID) (map[string]string, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
out := map[string]string{}
|
||||
for k, v := range s.metadata[id] {
|
||||
out[k] = v
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (s *fakeStore) PatchMetadata(_ context.Context, id domain.SessionID, kv map[string]string) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if s.metadata[id] == nil {
|
||||
s.metadata[id] = map[string]string{}
|
||||
}
|
||||
for k, v := range kv {
|
||||
s.metadata[id][k] = v
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// recordingNotifier captures emitted events for assertions.
|
||||
type recordingNotifier struct {
|
||||
mu sync.Mutex
|
||||
events []ports.OrchestratorEvent
|
||||
}
|
||||
|
||||
var _ ports.Notifier = (*recordingNotifier)(nil)
|
||||
|
||||
func (n *recordingNotifier) Notify(_ context.Context, e ports.OrchestratorEvent) error {
|
||||
n.mu.Lock()
|
||||
defer n.mu.Unlock()
|
||||
n.events = append(n.events, e)
|
||||
return nil
|
||||
}
|
||||
|
||||
// recordingMessenger captures messages injected into agents.
|
||||
type recordingMessenger struct {
|
||||
mu sync.Mutex
|
||||
sent []struct {
|
||||
ID domain.SessionID
|
||||
Message string
|
||||
}
|
||||
}
|
||||
|
||||
var _ ports.AgentMessenger = (*recordingMessenger)(nil)
|
||||
|
||||
func (a *recordingMessenger) Send(_ context.Context, id domain.SessionID, message string) error {
|
||||
a.mu.Lock()
|
||||
defer a.mu.Unlock()
|
||||
a.sent = append(a.sent, struct {
|
||||
ID domain.SessionID
|
||||
Message string
|
||||
}{id, message})
|
||||
return nil
|
||||
}
|
||||
|
|
@ -0,0 +1,327 @@
|
|||
// Package lifecycle implements ports.LifecycleManager: the synchronous
|
||||
// observe->decide->persist reducer. Every Apply*/On* entrypoint runs the same
|
||||
// pipeline under a per-session lock — load canonical, run the matching pure
|
||||
// decider, diff the result into a sparse merge-patch, persist. The LCM never
|
||||
// polls and never writes the display status (that is derived on read).
|
||||
//
|
||||
// Split A scope is the Apply* pipeline only. The reaction table + escalation
|
||||
// engine (ACT) and the Session Manager land in later splits; TickEscalations is
|
||||
// a documented no-op here.
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/aoagents/agent-orchestrator/backend/internal/domain"
|
||||
"github.com/aoagents/agent-orchestrator/backend/internal/domain/decide"
|
||||
"github.com/aoagents/agent-orchestrator/backend/internal/ports"
|
||||
)
|
||||
|
||||
// Metadata keys OnSpawnCompleted records for the spawned session's handles.
|
||||
const (
|
||||
MetaBranch = "branch"
|
||||
MetaWorkspacePath = "workspacePath"
|
||||
MetaRuntimeHandleID = "runtimeHandleId"
|
||||
MetaRuntimeName = "runtimeName"
|
||||
MetaAgentSessionID = "agentSessionId"
|
||||
)
|
||||
|
||||
// Manager is the LCM. Notifier/AgentMessenger are held for the ACT lane (split
|
||||
// B); the Apply* pipeline does not fire reactions yet.
|
||||
type Manager struct {
|
||||
store ports.LifecycleStore
|
||||
notifier ports.Notifier
|
||||
messenger ports.AgentMessenger
|
||||
|
||||
recentActivityWindow time.Duration
|
||||
locks keyedMutex
|
||||
}
|
||||
|
||||
var _ ports.LifecycleManager = (*Manager)(nil)
|
||||
|
||||
func New(store ports.LifecycleStore, notifier ports.Notifier, messenger ports.AgentMessenger) *Manager {
|
||||
return &Manager{
|
||||
store: store,
|
||||
notifier: notifier,
|
||||
messenger: messenger,
|
||||
recentActivityWindow: defaultRecentActivityWindow,
|
||||
}
|
||||
}
|
||||
|
||||
// ---- per-session serialisation ----
|
||||
|
||||
// keyedMutex hands out one lock per session id so the load->decide->persist
|
||||
// read-modify-write is serial within a session but parallel across sessions.
|
||||
type keyedMutex struct {
|
||||
mu sync.Mutex
|
||||
locks map[domain.SessionID]*sync.Mutex
|
||||
}
|
||||
|
||||
func (k *keyedMutex) lock(id domain.SessionID) func() {
|
||||
k.mu.Lock()
|
||||
if k.locks == nil {
|
||||
k.locks = make(map[domain.SessionID]*sync.Mutex)
|
||||
}
|
||||
m, ok := k.locks[id]
|
||||
if !ok {
|
||||
m = &sync.Mutex{}
|
||||
k.locks[id] = m
|
||||
}
|
||||
k.mu.Unlock()
|
||||
|
||||
m.Lock()
|
||||
return m.Unlock
|
||||
}
|
||||
|
||||
func (m *Manager) withLock(id domain.SessionID, fn func() error) error {
|
||||
unlock := m.locks.lock(id)
|
||||
defer unlock()
|
||||
return fn()
|
||||
}
|
||||
|
||||
// mutate runs the shared pipeline: load -> build patch -> persist (only if the
|
||||
// patch changed something). decideFn returns the diffed patch and whether it
|
||||
// touches anything; a false "changed" is a clean no-op (no write, no revision
|
||||
// bump), which is how failed-probe / unknown-fact inputs are dropped.
|
||||
func (m *Manager) mutate(
|
||||
ctx context.Context,
|
||||
id domain.SessionID,
|
||||
decideFn func(cur domain.CanonicalSessionLifecycle, exists bool) (ports.LifecyclePatch, bool, error),
|
||||
) error {
|
||||
return m.withLock(id, func() error {
|
||||
cur, exists, err := m.store.Load(ctx, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
patch, changed, err := decideFn(cur, exists)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if !changed {
|
||||
return nil
|
||||
}
|
||||
return m.store.PatchLifecycle(ctx, id, patch)
|
||||
})
|
||||
}
|
||||
|
||||
// ---- OBSERVE entrypoints ----
|
||||
|
||||
// ApplyRuntimeObservation feeds the probe decider. Liveness always writes the
|
||||
// runtime axis; the session axis follows the #1 composition rule; and a
|
||||
// non-detecting verdict clears any stale detecting memory (#3) so the next
|
||||
// probe doesn't read a phantom prior.
|
||||
func (m *Manager) ApplyRuntimeObservation(ctx context.Context, id domain.SessionID, f ports.RuntimeFacts) error {
|
||||
return m.mutate(ctx, id, func(cur domain.CanonicalSessionLifecycle, exists bool) (ports.LifecyclePatch, bool, error) {
|
||||
if !exists {
|
||||
return ports.LifecyclePatch{}, false, nil // nothing seeded; ignore stray probe
|
||||
}
|
||||
|
||||
d := decide.ResolveProbeDecision(runtimeFactsToProbeInput(f, cur, m.recentActivityWindow))
|
||||
|
||||
var patch ports.LifecyclePatch
|
||||
changed := false
|
||||
|
||||
if rt := runtimeSubstateFromFacts(f); cur.Runtime != rt {
|
||||
patch.Runtime = &rt
|
||||
changed = true
|
||||
}
|
||||
if shouldWriteSessionRuntime(d, cur) {
|
||||
changed = setSessionIfChanged(&patch, cur, d.SessionState, d.SessionReason) || changed
|
||||
}
|
||||
changed = setDetecting(&patch, cur, d.Detecting) || changed
|
||||
|
||||
return patch, changed, nil
|
||||
})
|
||||
}
|
||||
|
||||
// ApplySCMObservation maps PR facts onto the PR axis. A failed fetch is dropped
|
||||
// (failed probe != "no PR"). An open PR writes only the PR sub-state — the
|
||||
// session axis stays owned by activity, and DeriveLegacyStatus surfaces the PR
|
||||
// reason for display. A terminal PR (merged/closed) also parks the session.
|
||||
func (m *Manager) ApplySCMObservation(ctx context.Context, id domain.SessionID, f ports.SCMFacts) error {
|
||||
return m.mutate(ctx, id, func(cur domain.CanonicalSessionLifecycle, exists bool) (ports.LifecyclePatch, bool, error) {
|
||||
if !exists || !f.Fetched {
|
||||
return ports.LifecyclePatch{}, false, nil
|
||||
}
|
||||
|
||||
switch f.PRState {
|
||||
case domain.PROpen:
|
||||
d := decide.ResolveOpenPRDecision(openPRInput(f))
|
||||
var patch ports.LifecyclePatch
|
||||
changed := setPRIfChanged(&patch, cur, d, f)
|
||||
return patch, changed, nil
|
||||
|
||||
case domain.PRMerged, domain.PRClosed:
|
||||
d := decide.ResolveTerminalPRStateDecision(f.PRState)
|
||||
var patch ports.LifecyclePatch
|
||||
changed := setPRIfChanged(&patch, cur, d, f)
|
||||
if !isTerminal(cur.Session.State) {
|
||||
changed = setSessionIfChanged(&patch, cur, d.SessionState, d.SessionReason) || changed
|
||||
}
|
||||
return patch, changed, nil
|
||||
|
||||
default: // none / unset: no PR-driven transition in split A
|
||||
return ports.LifecyclePatch{}, false, nil
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// ApplyActivitySignal updates the activity axis. Only a valid-confidence signal
|
||||
// is authoritative (stale/unavailable/probe_failure != idleness). It refreshes
|
||||
// the persisted activity sub-state (the probe decider's RecentActivity input)
|
||||
// and maps the classification onto the session axis, subject to the mirror
|
||||
// composition rule that keeps activity off the death axis.
|
||||
func (m *Manager) ApplyActivitySignal(ctx context.Context, id domain.SessionID, s ports.ActivitySignal) error {
|
||||
return m.mutate(ctx, id, func(cur domain.CanonicalSessionLifecycle, exists bool) (ports.LifecyclePatch, bool, error) {
|
||||
if !exists || s.State != ports.SignalValid {
|
||||
return ports.LifecyclePatch{}, false, nil
|
||||
}
|
||||
|
||||
var patch ports.LifecyclePatch
|
||||
changed := false
|
||||
|
||||
act := domain.ActivitySubstate{State: s.Activity, LastActivityAt: nowOr(s.Timestamp), Source: s.Source}
|
||||
if !sameActivity(cur.Activity, act) {
|
||||
patch.Activity = &act
|
||||
changed = true
|
||||
}
|
||||
if st, rs, ok := activityToSession(s.Activity); ok && shouldWriteSessionActivity(cur) {
|
||||
changed = setSessionIfChanged(&patch, cur, st, rs) || changed
|
||||
}
|
||||
|
||||
return patch, changed, nil
|
||||
})
|
||||
}
|
||||
|
||||
// ---- mutation outcomes reported by the Session Manager ----
|
||||
|
||||
// OnSpawnCompleted records that a spawn finished: the runtime is up and the
|
||||
// handles are known. Per the agreed rule it flips the runtime axis to alive and
|
||||
// stores the handles in metadata, but leaves the session at not_started
|
||||
// (display: spawning) — the agent "acknowledges" via the first activity signal.
|
||||
func (m *Manager) OnSpawnCompleted(ctx context.Context, id domain.SessionID, o ports.SpawnOutcome) error {
|
||||
return m.withLock(id, func() error {
|
||||
cur, _, err := m.store.Load(ctx, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
rt := domain.RuntimeSubstate{State: domain.RuntimeAlive, Reason: domain.RuntimeReasonProcessRunning}
|
||||
if cur.Runtime != rt {
|
||||
if err := m.store.PatchLifecycle(ctx, id, ports.LifecyclePatch{Runtime: &rt}); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if meta := spawnMetadata(o); len(meta) > 0 {
|
||||
if err := m.store.PatchMetadata(ctx, id, meta); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
// OnKillRequested is the SM's explicit terminal-write authority (the one
|
||||
// terminal path that does not go through the inferred-death decider). It writes
|
||||
// the terminal session/runtime sub-states for the kill kind and clears any
|
||||
// in-flight detecting memory.
|
||||
func (m *Manager) OnKillRequested(ctx context.Context, id domain.SessionID, r ports.KillReason) error {
|
||||
return m.mutate(ctx, id, func(cur domain.CanonicalSessionLifecycle, exists bool) (ports.LifecyclePatch, bool, error) {
|
||||
var patch ports.LifecyclePatch
|
||||
changed := false
|
||||
|
||||
if sess := killSession(r.Kind); cur.Session != sess {
|
||||
patch.Session = &sess
|
||||
changed = true
|
||||
}
|
||||
if rt := killRuntime(r.Kind); cur.Runtime != rt {
|
||||
patch.Runtime = &rt
|
||||
changed = true
|
||||
}
|
||||
if cur.Detecting != nil {
|
||||
patch.ClearDetecting = true
|
||||
changed = true
|
||||
}
|
||||
return patch, changed, nil
|
||||
})
|
||||
}
|
||||
|
||||
// TickEscalations is a no-op in split A. The reaper will call this to fire
|
||||
// duration-based escalations the synchronous LCM can't wake itself for, but the
|
||||
// reaction table + escalation engine that back it land in split B.
|
||||
func (m *Manager) TickEscalations(ctx context.Context, now time.Time) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// ---- patch helpers (diff -> sparse merge-patch) ----
|
||||
|
||||
// setSessionIfChanged sets patch.Session only when the decided sub-state
|
||||
// differs from current; an empty decided state means "decider does not address
|
||||
// the session axis" and is left untouched.
|
||||
func setSessionIfChanged(patch *ports.LifecyclePatch, cur domain.CanonicalSessionLifecycle, st domain.SessionState, rs domain.SessionReason) bool {
|
||||
if st == "" {
|
||||
return false
|
||||
}
|
||||
want := domain.SessionSubstate{State: st, Reason: rs}
|
||||
if cur.Session == want {
|
||||
return false
|
||||
}
|
||||
patch.Session = &want
|
||||
return true
|
||||
}
|
||||
|
||||
// setPRIfChanged folds the decided PR sub-state plus the fact-borne PR identity
|
||||
// (number/url) into the patch when it differs from current.
|
||||
func setPRIfChanged(patch *ports.LifecyclePatch, cur domain.CanonicalSessionLifecycle, d decide.LifecycleDecision, f ports.SCMFacts) bool {
|
||||
want := domain.PRSubstate{State: d.PRState, Reason: d.PRReason, Number: f.PRNumber, URL: f.PRURL}
|
||||
if cur.PR == want {
|
||||
return false
|
||||
}
|
||||
patch.PR = &want
|
||||
return true
|
||||
}
|
||||
|
||||
// setDetecting implements the three-way detecting semantics: set/replace when
|
||||
// the decision carries memory, clear (#3) when it doesn't but canonical still
|
||||
// holds stale memory, else leave untouched.
|
||||
func setDetecting(patch *ports.LifecyclePatch, cur domain.CanonicalSessionLifecycle, d *domain.DetectingState) bool {
|
||||
if d != nil {
|
||||
if cur.Detecting != nil && *cur.Detecting == *d {
|
||||
return false
|
||||
}
|
||||
patch.Detecting = d
|
||||
return true
|
||||
}
|
||||
if cur.Detecting != nil {
|
||||
patch.ClearDetecting = true
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// sameActivity compares activity sub-states with time-aware equality (== on
|
||||
// time.Time is monotonic-clock sensitive and would spuriously report changes).
|
||||
func sameActivity(a, b domain.ActivitySubstate) bool {
|
||||
return a.State == b.State && a.Source == b.Source && a.LastActivityAt.Equal(b.LastActivityAt)
|
||||
}
|
||||
|
||||
func spawnMetadata(o ports.SpawnOutcome) map[string]string {
|
||||
meta := map[string]string{}
|
||||
if o.Branch != "" {
|
||||
meta[MetaBranch] = o.Branch
|
||||
}
|
||||
if o.WorkspacePath != "" {
|
||||
meta[MetaWorkspacePath] = o.WorkspacePath
|
||||
}
|
||||
if o.RuntimeHandle.ID != "" {
|
||||
meta[MetaRuntimeHandleID] = o.RuntimeHandle.ID
|
||||
}
|
||||
if o.RuntimeHandle.RuntimeName != "" {
|
||||
meta[MetaRuntimeName] = o.RuntimeHandle.RuntimeName
|
||||
}
|
||||
if o.AgentSessionID != "" {
|
||||
meta[MetaAgentSessionID] = o.AgentSessionID
|
||||
}
|
||||
return meta
|
||||
}
|
||||
|
|
@ -0,0 +1,388 @@
|
|||
package lifecycle
|
||||
|
||||
import (
|
||||
"context"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/aoagents/agent-orchestrator/backend/internal/domain"
|
||||
"github.com/aoagents/agent-orchestrator/backend/internal/ports"
|
||||
)
|
||||
|
||||
var t0 = time.Date(2026, 5, 26, 12, 0, 0, 0, time.UTC)
|
||||
|
||||
const sid domain.SessionID = "s1"
|
||||
|
||||
func newManager() (*Manager, *fakeStore) {
|
||||
store := newFakeStore()
|
||||
return New(store, &recordingNotifier{}, &recordingMessenger{}), store
|
||||
}
|
||||
|
||||
func mustLoad(t *testing.T, store *fakeStore) domain.CanonicalSessionLifecycle {
|
||||
t.Helper()
|
||||
l, ok, err := store.Load(context.Background(), sid)
|
||||
if err != nil || !ok {
|
||||
t.Fatalf("load: ok=%v err=%v", ok, err)
|
||||
}
|
||||
return l
|
||||
}
|
||||
|
||||
// ---- ApplyRuntimeObservation + #1 composition + #3 detecting clear ----
|
||||
|
||||
func TestApplyRuntimeObservation(t *testing.T) {
|
||||
aliveProbe := ports.RuntimeFacts{RuntimeState: ports.RuntimeProbeAlive, ProcessState: ports.ProcessProbeAlive, ObservedAt: t0}
|
||||
failedProbe := ports.RuntimeFacts{RuntimeState: ports.RuntimeProbeFailed, ProcessState: ports.ProcessProbeAlive, ObservedAt: t0}
|
||||
deadProbe := ports.RuntimeFacts{RuntimeState: ports.RuntimeProbeDead, ProcessState: ports.ProcessProbeDead, ObservedAt: t0}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
seed domain.CanonicalSessionLifecycle
|
||||
facts ports.RuntimeFacts
|
||||
wantSession domain.SessionState
|
||||
wantReason domain.SessionReason
|
||||
wantRuntime domain.RuntimeState
|
||||
wantDisplay domain.SessionStatus
|
||||
wantDetecting bool // expect non-nil detecting memory persisted
|
||||
}{
|
||||
{
|
||||
name: "healthy probe must not clobber an activity-owned needs_input (#1)",
|
||||
seed: lc(domain.SessionNeedsInput, domain.ReasonAwaitingUserInput, domain.RuntimeAlive),
|
||||
facts: aliveProbe,
|
||||
wantSession: domain.SessionNeedsInput,
|
||||
wantReason: domain.ReasonAwaitingUserInput,
|
||||
wantRuntime: domain.RuntimeAlive,
|
||||
wantDisplay: domain.StatusNeedsInput,
|
||||
wantDetecting: false,
|
||||
},
|
||||
{
|
||||
name: "healthy probe recovers a liveness-owned detecting -> working and clears memory (#1 + #3)",
|
||||
seed: detectingLC(),
|
||||
facts: aliveProbe,
|
||||
wantSession: domain.SessionWorking,
|
||||
wantReason: domain.ReasonTaskInProgress,
|
||||
wantRuntime: domain.RuntimeAlive,
|
||||
wantDisplay: domain.StatusWorking,
|
||||
wantDetecting: false,
|
||||
},
|
||||
{
|
||||
name: "failed probe routes to detecting and records memory",
|
||||
seed: lc(domain.SessionWorking, domain.ReasonTaskInProgress, domain.RuntimeAlive),
|
||||
facts: failedProbe,
|
||||
wantSession: domain.SessionDetecting,
|
||||
wantReason: domain.ReasonProbeFailure,
|
||||
wantRuntime: domain.RuntimeProbeFailed,
|
||||
wantDisplay: domain.StatusDetecting,
|
||||
wantDetecting: true,
|
||||
},
|
||||
{
|
||||
name: "dead+dead with no recent activity concludes killed and clears detecting (#3)",
|
||||
seed: detectingLC(),
|
||||
facts: deadProbe,
|
||||
wantSession: domain.SessionTerminated,
|
||||
wantReason: domain.ReasonRuntimeLost,
|
||||
wantRuntime: domain.RuntimeExited,
|
||||
wantDisplay: domain.StatusKilled,
|
||||
wantDetecting: false,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
mgr, store := newManager()
|
||||
store.seed(sid, tt.seed)
|
||||
|
||||
if err := mgr.ApplyRuntimeObservation(context.Background(), sid, tt.facts); err != nil {
|
||||
t.Fatalf("apply: %v", err)
|
||||
}
|
||||
|
||||
l := mustLoad(t, store)
|
||||
if l.Session.State != tt.wantSession || l.Session.Reason != tt.wantReason {
|
||||
t.Errorf("session = %v/%v, want %v/%v", l.Session.State, l.Session.Reason, tt.wantSession, tt.wantReason)
|
||||
}
|
||||
if l.Runtime.State != tt.wantRuntime {
|
||||
t.Errorf("runtime = %v, want %v", l.Runtime.State, tt.wantRuntime)
|
||||
}
|
||||
if got := domain.DeriveLegacyStatus(l); got != tt.wantDisplay {
|
||||
t.Errorf("display = %v, want %v", got, tt.wantDisplay)
|
||||
}
|
||||
if (l.Detecting != nil) != tt.wantDetecting {
|
||||
t.Errorf("detecting present = %v, want %v (%+v)", l.Detecting != nil, tt.wantDetecting, l.Detecting)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestApplyRuntimeObservation_NoRecordIsNoOp(t *testing.T) {
|
||||
mgr, store := newManager()
|
||||
if err := mgr.ApplyRuntimeObservation(context.Background(), sid, ports.RuntimeFacts{RuntimeState: ports.RuntimeProbeAlive, ProcessState: ports.ProcessProbeAlive, ObservedAt: t0}); err != nil {
|
||||
t.Fatalf("apply: %v", err)
|
||||
}
|
||||
if _, ok, _ := store.Load(context.Background(), sid); ok {
|
||||
t.Error("a probe for an unseeded session must not fabricate a record")
|
||||
}
|
||||
}
|
||||
|
||||
// ---- ApplyActivitySignal ----
|
||||
|
||||
func TestApplyActivitySignal(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
seed domain.CanonicalSessionLifecycle
|
||||
signal ports.ActivitySignal
|
||||
wantSession domain.SessionState
|
||||
wantActivity domain.ActivityState
|
||||
wantChanged bool
|
||||
}{
|
||||
{
|
||||
name: "valid waiting_input maps to needs_input",
|
||||
seed: lc(domain.SessionWorking, domain.ReasonTaskInProgress, domain.RuntimeAlive),
|
||||
signal: ports.ActivitySignal{State: ports.SignalValid, Activity: domain.ActivityWaitingInput, Timestamp: t0, Source: domain.SourceHook},
|
||||
wantSession: domain.SessionNeedsInput,
|
||||
wantActivity: domain.ActivityWaitingInput,
|
||||
wantChanged: true,
|
||||
},
|
||||
{
|
||||
name: "valid active recovers needs_input -> working",
|
||||
seed: lc(domain.SessionNeedsInput, domain.ReasonAwaitingUserInput, domain.RuntimeAlive),
|
||||
signal: ports.ActivitySignal{State: ports.SignalValid, Activity: domain.ActivityActive, Timestamp: t0, Source: domain.SourceHook},
|
||||
wantSession: domain.SessionWorking,
|
||||
wantActivity: domain.ActivityActive,
|
||||
wantChanged: true,
|
||||
},
|
||||
{
|
||||
name: "low-confidence signal is dropped (no idleness inferred)",
|
||||
seed: lc(domain.SessionWorking, domain.ReasonTaskInProgress, domain.RuntimeAlive),
|
||||
signal: ports.ActivitySignal{State: ports.SignalProbeFailure, Activity: domain.ActivityIdle, Timestamp: t0, Source: domain.SourceHook},
|
||||
wantSession: domain.SessionWorking,
|
||||
wantChanged: false,
|
||||
},
|
||||
{
|
||||
name: "activity does not touch a liveness-owned detecting session",
|
||||
seed: detectingLC(),
|
||||
signal: ports.ActivitySignal{State: ports.SignalValid, Activity: domain.ActivityActive, Timestamp: t0, Source: domain.SourceHook},
|
||||
wantSession: domain.SessionDetecting,
|
||||
wantActivity: domain.ActivityActive,
|
||||
wantChanged: true, // activity sub-state still updates
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
mgr, store := newManager()
|
||||
store.seed(sid, tt.seed)
|
||||
|
||||
if err := mgr.ApplyActivitySignal(context.Background(), sid, tt.signal); err != nil {
|
||||
t.Fatalf("apply: %v", err)
|
||||
}
|
||||
|
||||
l := mustLoad(t, store)
|
||||
if l.Session.State != tt.wantSession {
|
||||
t.Errorf("session = %v, want %v", l.Session.State, tt.wantSession)
|
||||
}
|
||||
if tt.wantChanged && l.Revision != 1 {
|
||||
t.Errorf("revision = %d, want 1 (expected a write)", l.Revision)
|
||||
}
|
||||
if !tt.wantChanged && l.Revision != 0 {
|
||||
t.Errorf("revision = %d, want 0 (expected a no-op)", l.Revision)
|
||||
}
|
||||
if tt.wantChanged && tt.wantActivity != "" && l.Activity.State != tt.wantActivity {
|
||||
t.Errorf("activity = %v, want %v", l.Activity.State, tt.wantActivity)
|
||||
}
|
||||
if tt.name == "activity does not touch a liveness-owned detecting session" && l.Detecting == nil {
|
||||
t.Error("activity must leave detecting memory for the probe pipeline to resolve")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// ---- ApplySCMObservation ----
|
||||
|
||||
func TestApplySCMObservation(t *testing.T) {
|
||||
t.Run("failed fetch is a no-op (failed probe != no PR)", func(t *testing.T) {
|
||||
mgr, store := newManager()
|
||||
store.seed(sid, lc(domain.SessionWorking, domain.ReasonTaskInProgress, domain.RuntimeAlive))
|
||||
if err := mgr.ApplySCMObservation(context.Background(), sid, ports.SCMFacts{Fetched: false, PRState: domain.PROpen}); err != nil {
|
||||
t.Fatalf("apply: %v", err)
|
||||
}
|
||||
if l := mustLoad(t, store); l.Revision != 0 || l.PR.State != "" {
|
||||
t.Errorf("expected no-op, got revision=%d pr=%v", l.Revision, l.PR.State)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("open PR writes only the PR axis; session stays activity-owned", func(t *testing.T) {
|
||||
mgr, store := newManager()
|
||||
store.seed(sid, lc(domain.SessionWorking, domain.ReasonTaskInProgress, domain.RuntimeAlive))
|
||||
f := ports.SCMFacts{Fetched: true, PRState: domain.PROpen, CISummary: ports.CIFailing, PRNumber: 12, PRURL: "https://x/12"}
|
||||
if err := mgr.ApplySCMObservation(context.Background(), sid, f); err != nil {
|
||||
t.Fatalf("apply: %v", err)
|
||||
}
|
||||
l := mustLoad(t, store)
|
||||
if l.PR.State != domain.PROpen || l.PR.Reason != domain.PRReasonCIFailing || l.PR.Number != 12 {
|
||||
t.Errorf("pr = %+v, want open/ci_failing/#12", l.PR)
|
||||
}
|
||||
if l.Session.State != domain.SessionWorking {
|
||||
t.Errorf("session = %v, want working (untouched)", l.Session.State)
|
||||
}
|
||||
if got := domain.DeriveLegacyStatus(l); got != domain.StatusCIFailed {
|
||||
t.Errorf("display = %v, want ci_failed", got)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("merged PR parks the session and displays merged", func(t *testing.T) {
|
||||
mgr, store := newManager()
|
||||
seed := lc(domain.SessionWorking, domain.ReasonTaskInProgress, domain.RuntimeAlive)
|
||||
seed.PR = domain.PRSubstate{State: domain.PROpen, Reason: domain.PRReasonInProgress, Number: 12}
|
||||
store.seed(sid, seed)
|
||||
f := ports.SCMFacts{Fetched: true, PRState: domain.PRMerged, PRNumber: 12}
|
||||
if err := mgr.ApplySCMObservation(context.Background(), sid, f); err != nil {
|
||||
t.Fatalf("apply: %v", err)
|
||||
}
|
||||
l := mustLoad(t, store)
|
||||
if l.PR.State != domain.PRMerged || l.Session.Reason != domain.ReasonMergedWaitingDecision {
|
||||
t.Errorf("got pr=%v session=%v, want merged + merged_waiting_decision", l.PR.State, l.Session.Reason)
|
||||
}
|
||||
if got := domain.DeriveLegacyStatus(l); got != domain.StatusMerged {
|
||||
t.Errorf("display = %v, want merged", got)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("no PR is a no-op in split A", func(t *testing.T) {
|
||||
mgr, store := newManager()
|
||||
store.seed(sid, lc(domain.SessionWorking, domain.ReasonTaskInProgress, domain.RuntimeAlive))
|
||||
if err := mgr.ApplySCMObservation(context.Background(), sid, ports.SCMFacts{Fetched: true, PRState: domain.PRNone}); err != nil {
|
||||
t.Fatalf("apply: %v", err)
|
||||
}
|
||||
if l := mustLoad(t, store); l.Revision != 0 {
|
||||
t.Errorf("expected no-op, got revision=%d", l.Revision)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// ---- mutation outcomes ----
|
||||
|
||||
func TestOnSpawnCompleted(t *testing.T) {
|
||||
mgr, store := newManager()
|
||||
store.seed(sid, lc(domain.SessionNotStarted, domain.ReasonSpawnRequested, domain.RuntimeUnknown))
|
||||
|
||||
out := ports.SpawnOutcome{
|
||||
Branch: "feat/x",
|
||||
WorkspacePath: "/w/x",
|
||||
RuntimeHandle: ports.RuntimeHandle{ID: "tmux:1", RuntimeName: "tmux"},
|
||||
AgentSessionID: "agent-1",
|
||||
}
|
||||
if err := mgr.OnSpawnCompleted(context.Background(), sid, out); err != nil {
|
||||
t.Fatalf("apply: %v", err)
|
||||
}
|
||||
|
||||
l := mustLoad(t, store)
|
||||
if l.Runtime.State != domain.RuntimeAlive {
|
||||
t.Errorf("runtime = %v, want alive", l.Runtime.State)
|
||||
}
|
||||
if l.Session.State != domain.SessionNotStarted {
|
||||
t.Errorf("session = %v, want not_started (spawn does not assert acknowledgement)", l.Session.State)
|
||||
}
|
||||
if got := domain.DeriveLegacyStatus(l); got != domain.StatusSpawning {
|
||||
t.Errorf("display = %v, want spawning", got)
|
||||
}
|
||||
meta, _ := store.GetMetadata(context.Background(), sid)
|
||||
if meta[MetaBranch] != "feat/x" || meta[MetaAgentSessionID] != "agent-1" || meta[MetaRuntimeName] != "tmux" {
|
||||
t.Errorf("metadata not recorded: %+v", meta)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOnKillRequested(t *testing.T) {
|
||||
mgr, store := newManager()
|
||||
store.seed(sid, detectingLC())
|
||||
|
||||
if err := mgr.OnKillRequested(context.Background(), sid, ports.KillReason{Kind: ports.KillManual, Detail: "user"}); err != nil {
|
||||
t.Fatalf("apply: %v", err)
|
||||
}
|
||||
|
||||
l := mustLoad(t, store)
|
||||
if l.Session.State != domain.SessionTerminated || l.Session.Reason != domain.ReasonManuallyKilled {
|
||||
t.Errorf("session = %v/%v, want terminated/manually_killed", l.Session.State, l.Session.Reason)
|
||||
}
|
||||
if l.Runtime.Reason != domain.RuntimeReasonManualKillRequested {
|
||||
t.Errorf("runtime reason = %v, want manual_kill_requested", l.Runtime.Reason)
|
||||
}
|
||||
if l.Detecting != nil {
|
||||
t.Errorf("kill must clear detecting memory, got %+v", l.Detecting)
|
||||
}
|
||||
if got := domain.DeriveLegacyStatus(l); got != domain.StatusKilled {
|
||||
t.Errorf("display = %v, want killed", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestTickEscalationsIsNoOp(t *testing.T) {
|
||||
mgr, store := newManager()
|
||||
store.seed(sid, lc(domain.SessionWorking, domain.ReasonTaskInProgress, domain.RuntimeAlive))
|
||||
if err := mgr.TickEscalations(context.Background(), t0); err != nil {
|
||||
t.Fatalf("tick: %v", err)
|
||||
}
|
||||
if l := mustLoad(t, store); l.Revision != 0 {
|
||||
t.Errorf("TickEscalations must not write, got revision=%d", l.Revision)
|
||||
}
|
||||
}
|
||||
|
||||
// ---- fake store contract ----
|
||||
|
||||
func TestFakeStoreExpectedRevision(t *testing.T) {
|
||||
store := newFakeStore()
|
||||
store.seed(sid, lc(domain.SessionWorking, domain.ReasonTaskInProgress, domain.RuntimeAlive)) // revision 0
|
||||
rt := domain.RuntimeSubstate{State: domain.RuntimeExited}
|
||||
|
||||
bad := 99
|
||||
if err := store.PatchLifecycle(context.Background(), sid, ports.LifecyclePatch{Runtime: &rt, ExpectedRevision: &bad}); err == nil {
|
||||
t.Error("stale ExpectedRevision must be rejected")
|
||||
}
|
||||
good := 0
|
||||
if err := store.PatchLifecycle(context.Background(), sid, ports.LifecyclePatch{Runtime: &rt, ExpectedRevision: &good}); err != nil {
|
||||
t.Errorf("matching ExpectedRevision must succeed, got %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// ---- per-session serialisation under the race detector ----
|
||||
|
||||
func TestPerSessionSerialization(t *testing.T) {
|
||||
mgr, store := newManager()
|
||||
store.seed(sid, lc(domain.SessionWorking, domain.ReasonTaskInProgress, domain.RuntimeAlive))
|
||||
|
||||
const n = 50
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(n)
|
||||
for i := 0; i < n; i++ {
|
||||
go func(i int) {
|
||||
defer wg.Done()
|
||||
_ = mgr.ApplyActivitySignal(context.Background(), sid, ports.ActivitySignal{
|
||||
State: ports.SignalValid,
|
||||
Activity: domain.ActivityActive,
|
||||
Timestamp: t0.Add(time.Duration(i) * time.Second),
|
||||
Source: domain.SourceHook,
|
||||
})
|
||||
}(i)
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
// Each goroutine writes a distinct LastActivityAt, so every call is a real
|
||||
// change; with correct serialisation all n land without a lost update.
|
||||
if l := mustLoad(t, store); l.Revision != n {
|
||||
t.Errorf("revision = %d, want %d (lost update under concurrency)", l.Revision, n)
|
||||
}
|
||||
}
|
||||
|
||||
// ---- helpers ----
|
||||
|
||||
func lc(state domain.SessionState, reason domain.SessionReason, rt domain.RuntimeState) domain.CanonicalSessionLifecycle {
|
||||
return domain.CanonicalSessionLifecycle{
|
||||
Version: domain.LifecycleVersion,
|
||||
Session: domain.SessionSubstate{State: state, Reason: reason},
|
||||
Runtime: domain.RuntimeSubstate{State: rt},
|
||||
}
|
||||
}
|
||||
|
||||
func detectingLC() domain.CanonicalSessionLifecycle {
|
||||
l := lc(domain.SessionDetecting, domain.ReasonRuntimeLost, domain.RuntimeMissing)
|
||||
l.Detecting = &domain.DetectingState{Attempts: 1, StartedAt: t0, EvidenceHash: "abc"}
|
||||
return l
|
||||
}
|
||||
Loading…
Reference in New Issue