diff --git a/backend/internal/lifecycle/reactions.go b/backend/internal/lifecycle/reactions.go index c9ee7f5e8..df9904f3b 100644 --- a/backend/internal/lifecycle/reactions.go +++ b/backend/internal/lifecycle/reactions.go @@ -99,6 +99,10 @@ var defaultReactions = map[reactionKey]reactionConfig{ eventType: "reaction.approved-and-green", }, reactionAgentStuck: { + // §4.2 lists a threshold: 10m here; it is intentionally not gated — entry + // into stuck is already debounced upstream by the detecting->stuck + // quarantine (DETECTING_MAX_ATTEMPTS/DURATION), so a second timer would be + // redundant. action: actionNotify, priority: ports.PriorityUrgent, message: "Agent is stuck and needs attention.", eventType: "reaction.agent-stuck", @@ -187,6 +191,15 @@ type reactionTracker struct { // the reaction we left, then dispatch the reaction for the one we entered. It // fires only on a genuine reaction change, so re-persisting the same state does // not re-dispatch. Synchronous by design (see file header). +// +// Integration-time caveat: react runs AFTER withLock releases (deliberately, so +// a busy-waiting send-to-agent never holds the per-session mutex). Under a live +// daemon with concurrent observers (SCM poller + reaper + activity ingest) the +// afterLC snapshot can be stale by dispatch time — e.g. a ci-failed send firing +// after the session already moved to approved. Tests are single-threaded so it +// is not observable yet; when the daemon lands, give react a per-session +// ordering (a small react queue) or re-check the triggering state before +// dispatching. func (m *Manager) react(ctx context.Context, id domain.SessionID, tr *transition, rc reactionContext) error { if tr == nil { return nil @@ -196,25 +209,55 @@ func (m *Manager) react(ctx context.Context, id domain.SessionID, tr *transition changed := beforeKey != afterKey - if hadBefore && (!hasAfter || changed) { - // A persistent tracker survives oscillation within an open PR; it only - // resets once the incident is over. - if !defaultReactions[beforeKey].persistent || incidentOver(tr.afterLC) { + switch { + case incidentOver(tr.afterLC) || recovered(tr.afterLC): + // The PR-pipeline incident has ended — the PR resolved (merged/closed), + // the session went terminal, or it reached an approved/green state. Every + // tracker for this session is now stale, including a persistent ci-failed + // one. This is keyed on the state REACHED, not the one left: the recovery + // transition is typically review_pending->approved (beforeKey empty), so + // clearing only beforeKey would leak the ci-failed tracker and leave its + // escalated=true to silence a future regression. Clear them all. + m.clearSessionTrackers(id) + case hadBefore && (!hasAfter || changed): + // Within an unresolved open PR: a normal tracker resets when its state is + // left. A persistent one (ci-failed) is NOT cleared here — it must survive + // the ambiguous review_pending limbo (the fail->pending->fail flap, §4.2); + // it only resets via the recovery/incident-over branch above. + if !defaultReactions[beforeKey].persistent { m.clearTracker(id, beforeKey) } } + if hasAfter && (!hadBefore || changed) { return m.executeReaction(ctx, id, afterKey, rc) } return nil } -// incidentOver reports that a PR-pipeline incident has truly ended, so even a -// persistent tracker (ci-failed) may reset. +// incidentOver reports that a PR-pipeline incident has truly ended (PR no longer +// open, or the session terminal), so all trackers for the session may reset. func incidentOver(l domain.CanonicalSessionLifecycle) bool { return l.PR.State != domain.PROpen || isTerminal(l.Session.State) } +// recovered reports a genuinely-green open PR: an approved/mergeable state, which +// unambiguously means CI is no longer failing (the open-PR ladder ranks ci_failing +// above approved, so an approved display cannot coexist with failing CI). Unlike +// the ambiguous review_pending state — which may just be CI re-running — reaching +// this ends a ci-failed incident and re-arms its budget. +func recovered(l domain.CanonicalSessionLifecycle) bool { + if l.PR.State != domain.PROpen { + return false + } + switch l.PR.Reason { + case domain.PRReasonApproved, domain.PRReasonMergeReady: + return true + default: + return false + } +} + func (m *Manager) executeReaction(ctx context.Context, id domain.SessionID, key reactionKey, rc reactionContext) error { cfg := defaultReactions[key] switch cfg.action { @@ -312,6 +355,19 @@ func (m *Manager) clearTracker(id domain.SessionID, key reactionKey) { m.trackerMu.Unlock() } +// clearSessionTrackers drops every tracker for a session — used when its +// incident is over, so no budget (and no stale escalated=true) survives into a +// later unrelated incident. +func (m *Manager) clearSessionTrackers(id domain.SessionID) { + m.trackerMu.Lock() + for k := range m.trackers { + if k.id == id { + delete(m.trackers, k) + } + } + m.trackerMu.Unlock() +} + // TickEscalations fires the duration-based escalations the synchronous LCM // cannot wake itself for. The reaper calls it on a timer; it escalates any // not-yet-escalated tracker whose escalateAfter has elapsed. Notifications are diff --git a/backend/internal/lifecycle/reactions_test.go b/backend/internal/lifecycle/reactions_test.go index cf00309d5..a1bbc3c65 100644 --- a/backend/internal/lifecycle/reactions_test.go +++ b/backend/internal/lifecycle/reactions_test.go @@ -247,6 +247,68 @@ func TestReaction_NonPersistentTrackerClearsOnLeave(t *testing.T) { } } +func TestReaction_CIFailedRearmsOnGenuineRecovery(t *testing.T) { + m, store, notf, msgr := newReactive() + store.seed(sid, lcOpenPR(domain.PRReasonReviewPending)) + + // Drain the ci-failed budget to escalation (silenced thereafter). + for i := 0; i < 4; i++ { + failCI(t, m) + pendingCI(t, m) + } + if notifyCount(notf, "reaction.escalated") != 1 { + t.Fatalf("precondition: want one escalation, got %d", notifyCount(notf, "reaction.escalated")) + } + sentBefore := len(msgr.sent) + + // A genuine recovery (approved + green) ends the incident and re-arms the + // budget; a later regression must re-nudge the agent, not stay silenced. + if err := m.ApplySCMObservation(ctx(), sid, ports.SCMFacts{ + Fetched: true, PRState: domain.PROpen, ReviewDecision: ports.ReviewApproved, + Mergeability: ports.Mergeability{Mergeable: true}, PRNumber: 7, + }); err != nil { + t.Fatalf("recover: %v", err) + } + failCI(t, m) + + if len(msgr.sent) != sentBefore+1 { + t.Errorf("regression after recovery must re-nudge the agent: sends %d -> %d", sentBefore, len(msgr.sent)) + } +} + +func TestReaction_IncidentOverClearsAllSessionTrackers(t *testing.T) { + m, store, _, _ := newReactive() + store.seed(sid, lcOpenPR(domain.PRReasonReviewPending)) + + failCI(t, m) // creates a persistent ci-failed tracker + if sessionTrackerCount(m, sid) == 0 { + t.Fatalf("precondition: expected a ci-failed tracker") + } + + // Merging ends the incident; no tracker (and no stale escalated=true) may + // survive for the session. + if err := m.ApplySCMObservation(ctx(), sid, ports.SCMFacts{ + Fetched: true, PRState: domain.PRMerged, PRNumber: 7, + }); err != nil { + t.Fatalf("merge: %v", err) + } + if n := sessionTrackerCount(m, sid); n != 0 { + t.Errorf("incident over must clear all trackers, %d left", n) + } +} + +func sessionTrackerCount(m *Manager, id domain.SessionID) int { + m.trackerMu.Lock() + defer m.trackerMu.Unlock() + c := 0 + for k := range m.trackers { + if k.id == id { + c++ + } + } + return c +} + // ---- TickEscalations never writes canonical state ---- func TestTickEscalations_DoesNotPersist(t *testing.T) {