bkinnightskytw
/
agent-orchestrator
mirror of https://github.com/ComposioHQ/agent-orchestrator.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat(lifecycle): activity resolves detecting + review polish 

Address Harshit's PR #5 review (approve w/ design confirms + polish):

- #1 (design decision): a valid activity signal is proof of life, so it now
  resolves a detecting session — writes the activity-mapped session state and
  clears the quarantine memory. Scoped to detecting only; a liveness-escalated
  stuck stays the probe pipeline's to resolve. Terminal still never reopens.
- #2: document why a merged/closed PR parks the session axis even over an
  activity-owned needs_input/blocked (a merge is a milestone), unlike the
  open-PR path that defers to activity.
- #3: map plain idle activity to a neutral session reason instead of the
  misleading research_complete (kept for ready, which implies completion).
- #6: cover all three kill kinds (manual/cleanup/error), the open-PR review
  branches (changes_requested/mergeable/review_pending), and the neutral idle
  reason. Coverage 86.5% -> 88.6%.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
harshitsinghbhandari 2026-05-27 01:49:28 +05:30
parent 3945b10f20
commit 9eb5348604
3 changed files with 115 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
backend/internal/lifecycle/decide_bridge.go
View File

@ -121,8 +121,13 @@ func activityToSession(a domain.ActivityState) (domain.SessionState, domain.Sess
switch a {
case domain.ActivityActive:
return domain.SessionWorking, domain.ReasonTaskInProgress, true
case domain.ActivityReady, domain.ActivityIdle:
case domain.ActivityReady:
// ready = the agent finished a unit and is waiting for more work.
return domain.SessionIdle, domain.ReasonResearchComplete, true
case domain.ActivityIdle:
// plain inactivity carries no completion claim, so no specific reason
// (research_complete here would read misleadingly in diagnostics).
return domain.SessionIdle, "", true
case domain.ActivityWaitingInput:
return domain.SessionNeedsInput, domain.ReasonAwaitingUserInput, true
case domain.ActivityBlocked:
@ -175,11 +180,19 @@ func shouldWriteSessionRuntime(d decide.LifecycleDecision, cur domain.CanonicalS
}
// 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).
// activity axis owns working/idle/waiting. A valid activity signal is direct
// proof of life, so it is allowed to RESOLVE a detecting session (pull it out of
// the liveness quarantine) — but it must not resurrect a terminal session, and
// it leaves a liveness-escalated stuck state to the probe pipeline (stuck is a
// deliberate human-facing escalation, not a transient quarantine).
func shouldWriteSessionActivity(cur domain.CanonicalSessionLifecycle) bool {
return !isTerminal(cur.Session.State) && !isLivenessOwned(cur.Session)
if isTerminal(cur.Session.State) {
return false
}
if cur.Session.State == domain.SessionDetecting {
return true
}
return !isLivenessOwned(cur.Session)
}
// ---- explicit-kill mapping (SM's terminal-write authority) ----

17
backend/internal/lifecycle/manager.go
View File

@ -181,6 +181,11 @@ func (m *Manager) ApplySCMObservation(ctx context.Context, id domain.SessionID,
d := decide.ResolveTerminalPRStateDecision(f.PRState)
var patch ports.LifecyclePatch
changed := setPRIfChanged(&patch, cur, d, f)
// A merge/close is a milestone that ends the work, so it parks the
// session axis (idle / merged_waiting_decision) even over an
// activity-owned needs_input/blocked — unlike the open-PR path,
// which leaves the session axis to activity. A terminal session is
// still never reopened.
if !isTerminal(cur.Session.State) {
changed = setSessionIfChanged(&patch, cur, d.SessionState, d.SessionReason) || changed
}
@ -195,8 +200,9 @@ func (m *Manager) ApplySCMObservation(ctx context.Context, id domain.SessionID,
// 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.
// and maps the classification onto the session axis. A valid signal is proof of
// life, so it may resolve a detecting session — clearing the quarantine memory
// so a later probe doesn't resume counting from a stale prior.
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 {
@ -213,6 +219,13 @@ func (m *Manager) ApplyActivitySignal(ctx context.Context, id domain.SessionID,
}
if st, rs, ok := activityToSession(s.Activity); ok && shouldWriteSessionActivity(cur) {
changed = setSessionIfChanged(&patch, cur, st, rs) || changed
// Proof of life that pulls the session out of detecting must also
// drop the quarantine memory (detecting memory only exists while
// detecting, so this is a no-op otherwise).
if cur.Detecting != nil {
patch.ClearDetecting = true
changed = true
}
}
return patch, changed, nil

104
backend/internal/lifecycle/manager_test.go
View File

@ -150,6 +150,8 @@ func TestApplyActivitySignal(t *testing.T) {
seed domain.CanonicalSessionLifecycle
signal ports.ActivitySignal
wantSession domain.SessionState
wantReason domain.SessionReason
checkReason bool
wantActivity domain.ActivityState
wantChanged bool
}{
@ -169,6 +171,16 @@ func TestApplyActivitySignal(t *testing.T) {
wantActivity: domain.ActivityActive,
wantChanged: true,
},
{
name: "valid idle maps to idle with a neutral reason",
seed: lc(domain.SessionWorking, domain.ReasonTaskInProgress, domain.RuntimeAlive),
signal: ports.ActivitySignal{State: ports.SignalValid, Activity: domain.ActivityIdle, Timestamp: t0, Source: domain.SourceHook},
wantSession: domain.SessionIdle,
wantReason: "",
checkReason: true,
wantActivity: domain.ActivityIdle,
wantChanged: true,
},
{
name: "low-confidence signal is dropped (no idleness inferred)",
seed: lc(domain.SessionWorking, domain.ReasonTaskInProgress, domain.RuntimeAlive),
@ -177,12 +189,12 @@ func TestApplyActivitySignal(t *testing.T) {
wantChanged: false,
},
{
name: "activity does not touch a liveness-owned detecting session",
name: "valid activity resolves a detecting session (proof of life)",
seed: detectingLC(),
signal: ports.ActivitySignal{State: ports.SignalValid, Activity: domain.ActivityActive, Timestamp: t0, Source: domain.SourceHook},
wantSession: domain.SessionDetecting,
wantSession: domain.SessionWorking,
wantActivity: domain.ActivityActive,
wantChanged: true, // activity sub-state still updates
wantChanged: true,
},
}
@ -199,6 +211,9 @@ func TestApplyActivitySignal(t *testing.T) {
if l.Session.State != tt.wantSession {
t.Errorf("session = %v, want %v", l.Session.State, tt.wantSession)
}
if tt.checkReason && l.Session.Reason != tt.wantReason {
t.Errorf("session reason = %q, want %q", l.Session.Reason, tt.wantReason)
}
if tt.wantChanged && l.Revision != 1 {
t.Errorf("revision = %d, want 1 (expected a write)", l.Revision)
}
@ -208,8 +223,8 @@ func TestApplyActivitySignal(t *testing.T) {
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")
if tt.name == "valid activity resolves a detecting session (proof of life)" && l.Detecting != nil {
t.Errorf("resolving detecting must clear the quarantine memory, got %+v", l.Detecting)
}
})
}
@ -266,6 +281,35 @@ func TestApplySCMObservation(t *testing.T) {
}
})
t.Run("open-PR review branches map to the PR axis", func(t *testing.T) {
cases := []struct {
name string
facts ports.SCMFacts
wantReason domain.PRReason
wantStatus domain.SessionStatus
}{
{"changes requested", ports.SCMFacts{Fetched: true, PRState: domain.PROpen, ReviewDecision: ports.ReviewChangesRequested}, domain.PRReasonChangesRequested, domain.StatusChangesRequested},
{"approved + mergeable", ports.SCMFacts{Fetched: true, PRState: domain.PROpen, ReviewDecision: ports.ReviewApproved, Mergeability: ports.Mergeability{Mergeable: true}}, domain.PRReasonMergeReady, domain.StatusMergeable},
{"review pending", ports.SCMFacts{Fetched: true, PRState: domain.PROpen, ReviewDecision: ports.ReviewPending}, domain.PRReasonReviewPending, domain.StatusReviewPending},
}
for _, c := range cases {
t.Run(c.name, func(t *testing.T) {
mgr, store := newManager()
store.seed(sid, lc(domain.SessionWorking, domain.ReasonTaskInProgress, domain.RuntimeAlive))
if err := mgr.ApplySCMObservation(context.Background(), sid, c.facts); err != nil {
t.Fatalf("apply: %v", err)
}
l := mustLoad(t, store)
if l.PR.State != domain.PROpen || l.PR.Reason != c.wantReason {
t.Errorf("pr = %v/%v, want open/%v", l.PR.State, l.PR.Reason, c.wantReason)
}
if got := domain.DeriveLegacyStatus(l); got != c.wantStatus {
t.Errorf("display = %v, want %v", got, c.wantStatus)
}
})
}
})
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))
@ -311,25 +355,41 @@ func TestOnSpawnCompleted(t *testing.T) {
}
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)
tests := []struct {
name string
kind ports.LifecycleKillReason
wantReason domain.SessionReason
wantRuntime domain.RuntimeReason
wantDisplay domain.SessionStatus
}{
{"manual", ports.KillManual, domain.ReasonManuallyKilled, domain.RuntimeReasonManualKillRequested, domain.StatusKilled},
{"cleanup", ports.KillCleanup, domain.ReasonAutoCleanup, domain.RuntimeReasonAutoCleanup, domain.StatusCleanup},
{"error", ports.KillError, domain.ReasonErrorInProcess, domain.RuntimeReasonProbeError, domain.StatusErrored},
}
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)
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
mgr, store := newManager()
store.seed(sid, detectingLC())
if err := mgr.OnKillRequested(context.Background(), sid, ports.KillReason{Kind: tt.kind, Detail: "x"}); err != nil {
t.Fatalf("apply: %v", err)
}
l := mustLoad(t, store)
if l.Session.State != domain.SessionTerminated || l.Session.Reason != tt.wantReason {
t.Errorf("session = %v/%v, want terminated/%v", l.Session.State, l.Session.Reason, tt.wantReason)
}
if l.Runtime.Reason != tt.wantRuntime {
t.Errorf("runtime reason = %v, want %v", l.Runtime.Reason, tt.wantRuntime)
}
if l.Detecting != nil {
t.Errorf("kill must clear detecting memory, got %+v", l.Detecting)
}
if got := domain.DeriveLegacyStatus(l); got != tt.wantDisplay {
t.Errorf("display = %v, want %v", got, tt.wantDisplay)
}
})
}
}