Fix #2190: focal-agent activity gate on heartbeat/progress stall alerts

orchestrator/events.py

@@ -81,6 +81,12 @@ class EventType(StrEnum):
 
     # Progress monitoring
     PROGRESS_EMITTED = "progress.emitted"
+    # Container-side activity (e.g. successful mcp__task__add_commit) that
+    # demonstrates an agent is alive even when no HEARTBEAT message-bus
+    # traffic exists. Consumed by HealthMonitor to suppress
+    # heartbeat/progress stall alerts against agents legitimately blocked
+    # in long tool calls. See issue #2190.
+    CONTAINER_ACTIVITY = "container.activity"
 
     # System events — health check framework (see health_checks/runner.py)
     HEALTH_CHECK = "system.health_check"

orchestrator/health_monitor.py

@@ -67,6 +67,11 @@ def get_logger(name: str, **kwargs) -> logging.Logger:  # type: ignore[misc]
 EscalationCallback = Callable[[dict[str, Any]], None]
 ThrottleCallback = Callable[[dict[str, Any]], None]
 
+# Sentinel for AgentState.last_activity meaning "no CONTAINER_ACTIVITY event
+# has ever arrived for this agent." Compared with `<=` so any non-positive
+# float counts as never-seen. See `_has_recent_activity` (#2190).
+_NEVER_SEEN_ACTIVITY: float = 0.0
+
 
 @dataclass
 class AgentState:
@@ -75,6 +80,14 @@ class AgentState:
     agent_id: str
     last_heartbeat: float = field(default_factory=time.time)
     last_progress: float = field(default_factory=time.time)
+    # Wall-clock timestamp of the most recent CONTAINER_ACTIVITY event for
+    # this agent (e.g. successful git commit registration).  Used to
+    # suppress heartbeat/progress stall alerts against agents that are
+    # legitimately blocked in long tool calls but still making real
+    # progress (issue #2190).  Defaults to ``_NEVER_SEEN_ACTIVITY`` so a
+    # freshly spawned agent's silence is governed by the heartbeat anchor
+    # alone.
+    last_activity: float = _NEVER_SEEN_ACTIVITY
     error_counts: dict[str, int] = field(default_factory=lambda: defaultdict(int))
     message_timestamps: list[float] = field(default_factory=list)
     heartbeat_escalated: bool = False
@@ -131,6 +144,7 @@ def __init__(
         self._event_bus.subscribe(EventType.ERROR, self._on_error)
         self._event_bus.subscribe(EventType.CONTAINER_STOPPED, self._on_container_stopped)
         self._event_bus.subscribe(EventType.MESSAGE_SENT, self._on_message_sent)
+        self._event_bus.subscribe(EventType.CONTAINER_ACTIVITY, self._on_container_activity)
 
     # -----------------------------------------------------------------
     # Callback registration
@@ -227,6 +241,48 @@ def _get_post_ack_confirmation_timeout(self) -> int:
             return self._config.orchestrator_plan_post_ack_confirmation_timeout_seconds
         return self._config.orchestrator_post_ack_confirmation_timeout_seconds
 
+    def _has_recent_activity(self, agent_id: str, now: float) -> tuple[bool, str | None]:
+        """Return (defer, reason) for the focal-agent activity gate (#2190).
+
+        Suppresses ``heartbeat_timeout`` / ``progress_stall`` alerts against
+        an agent that is demonstrably alive — i.e. has emitted a
+        :class:`EventType.CONTAINER_ACTIVITY` event within
+        ``orchestrator_activity_quiet_seconds`` — even if no
+        bus-level ``HEARTBEAT`` has arrived. The repro from #2190 was a
+        coder mid-pytest with multi-minute blocking ``TaskOutput`` calls,
+        committing along the way; the bus saw silence but the agent was
+        making real progress.
+
+        Returns ``(False, None)`` when the agent has never emitted an
+        activity event (``last_activity == _NEVER_SEEN_ACTIVITY``) so a
+        freshly spawned but truly silent agent still escalates on the
+        heartbeat anchor.
+
+        Setting ``orchestrator_activity_quiet_seconds=0`` disables the
+        gate entirely — operator escape hatch if the gate produces
+        false negatives in production.
+
+        OR'd with :func:`_has_recent_peer_progress` (#2242) at the
+        per-agent alert sites: focal-agent activity OR peer progress
+        defers. The escalated flag is intentionally not set on defer so
+        the next poll re-checks once activity goes stale.
+        """
+        threshold = self._config.orchestrator_activity_quiet_seconds
+        if threshold <= 0:
+            return False, None
+
+        with self._lock:
+            agent = self._agents.get(agent_id)
+            last_activity = agent.last_activity if agent is not None else _NEVER_SEEN_ACTIVITY
+
+        if last_activity <= _NEVER_SEEN_ACTIVITY:
+            return False, None
+
+        age = now - last_activity
+        if 0 <= age < threshold:
+            return True, f"container activity {int(age)}s ago"
+        return False, None
+
     def _has_recent_peer_progress(
         self, exclude_agent_id: str, now: float
     ) -> tuple[bool, str | None]:
@@ -443,6 +499,26 @@ def _on_progress(self, event: Event) -> None:
                 if new_state != "blocked" or new_blocker != old_blocker:
                     agent.infra_error_escalated = False
 
+    def _on_container_activity(self, event: Event) -> None:
+        """Handle CONTAINER_ACTIVITY event — record focal-agent activity (#2190).
+
+        Activity events fire on demonstrable signs of life that don't go
+        through the bus-level HEARTBEAT path: today, successful commit
+        registrations from the gateway commit observer. Used by
+        :func:`_has_recent_activity` to suppress heartbeat/progress
+        alerts against agents legitimately blocked in long tool calls.
+        """
+        if event.pipeline_id != self._pipeline_id:
+            return
+
+        agent_id = event.data.get("agent_id") or event.data.get("agent_role")
+        if not agent_id:
+            return
+
+        with self._lock:
+            agent = self._get_or_create_agent(agent_id)
+            agent.last_activity = time.time()
+
     def _on_error(self, event: Event) -> None:
         """Handle ERROR event — track repeated identical errors."""
         if event.pipeline_id != self._pipeline_id:
@@ -617,12 +693,20 @@ def check_heartbeats(self) -> list[dict[str, Any]]:
             if self._is_brc_idle(agent_id):
                 continue
 
-            # Alive-signal gate (#2242): defer if peers are still moving.
-            # Don't set heartbeat_escalated — the next poll re-checks.
-            defer, gate_reason = self._has_recent_peer_progress(agent_id, now)
+            # Focal-agent activity gate (#2190) OR alive-signal peer-progress
+            # gate (#2242): defer if either fires. The activity gate catches
+            # an agent legitimately blocked in a long tool call (e.g. a
+            # multi-minute background pytest) that is still committing but
+            # not emitting bus-level HEARTBEAT. The peer-progress gate
+            # catches the case where the broader pipeline is clearly alive
+            # via BRC bus signals or peer heartbeats. Don't set
+            # heartbeat_escalated on defer — the next poll re-checks.
+            defer, gate_reason = self._has_recent_activity(agent_id, now)
+            if not defer:
+                defer, gate_reason = self._has_recent_peer_progress(agent_id, now)
             if defer:
                 logger.info(
-                    "Heartbeat alert deferred by progress gate",
+                    "Heartbeat alert deferred by alive-signal gate",
                     pipeline_id=self._pipeline_id,
                     agent_id=agent_id,
                     elapsed_seconds=int(elapsed),
@@ -713,11 +797,15 @@ def check_progress(self) -> list[dict[str, Any]]:
             if self._is_brc_idle(agent_id):
                 continue
 
-            # Alive-signal gate (#2242): defer if peers are still moving.
-            defer, gate_reason = self._has_recent_peer_progress(agent_id, now)
+            # Focal-agent activity gate (#2190) OR alive-signal peer-progress
+            # gate (#2242): same OR pattern as check_heartbeats — defer when
+            # either fires.
+            defer, gate_reason = self._has_recent_activity(agent_id, now)
+            if not defer:
+                defer, gate_reason = self._has_recent_peer_progress(agent_id, now)
             if defer:
                 logger.info(
-                    "Progress alert deferred by progress gate",
+                    "Progress alert deferred by alive-signal gate",
                     pipeline_id=self._pipeline_id,
                     agent_id=agent_id,
                     elapsed_seconds=int(elapsed),
@@ -845,6 +933,7 @@ def stop(self, pipeline_id: str | None = None) -> None:
         self._event_bus.unsubscribe(EventType.ERROR, self._on_error)
         self._event_bus.unsubscribe(EventType.CONTAINER_STOPPED, self._on_container_stopped)
         self._event_bus.unsubscribe(EventType.MESSAGE_SENT, self._on_message_sent)
+        self._event_bus.unsubscribe(EventType.CONTAINER_ACTIVITY, self._on_container_activity)
 
     def _check_infra_errors(self) -> list[dict[str, Any]]:
         """Detect blocked progress events with infrastructure error keywords.

orchestrator/models.py

@@ -495,6 +495,16 @@ class PipelineConfig(BaseModel):
     orchestrator_message_rate_limit: int = Field(
         default=20, ge=1, description="Max messages per minute before auto-throttle"
     )
+    orchestrator_activity_quiet_seconds: int = Field(
+        default=120,
+        ge=0,
+        description=(
+            "Seconds since the last CONTAINER_ACTIVITY event below which an "
+            "agent is considered alive — suppresses heartbeat/progress stall "
+            "alerts even when bus-level HEARTBEATs are absent (issue #2190). "
+            "Set to 0 to disable the gate entirely (operator escape hatch)."
+        ),
+    )
     overseer_poll_interval_seconds: int = Field(
         default=30, ge=5, description="Overseer polling interval in seconds"
     )

orchestrator/overseer/decision_maker.py

@@ -60,6 +60,13 @@
     re.IGNORECASE,
 )
 
+# Spans every action in the decision-maker's vocabulary so any prior
+# corrective intervention (destructive or not) bypasses the
+# first-stall restart guard. See ``_enforce_no_first_stall_restart``.
+_PRIOR_INTERVENTIONS: frozenset[str] = frozenset(
+    {"nudge", "redirect", "issue", "slack", "restart_agent", "restart_phase", "hitl"}
+)
+
 
 def _is_restartable(error_text: str) -> bool:
     """Return True if *error_text* describes a transient, auto-restartable error.
@@ -97,14 +104,23 @@ async def _call_decision_maker(prompt: str, context: str, *, model: str | None =
 
 
 async def decide_corrective_action(
-    classification: dict, context: dict, *, model: str | None = None
+    classification: dict,
+    context: dict,
+    *,
+    model: str | None = None,
+    redirect_history: list[dict] | None = None,
 ) -> dict:
     """Decide what corrective action to take based on a classification.
 
     Args:
         classification: Output from a classifier function (e.g. classify_stall).
         context: Additional context (pipeline state, agent history, etc.).
         model: Override the default decision model.
+        redirect_history: Prior corrective actions sent to this agent.
+            Used by the deterministic post-hoc guard (#2190) to downgrade
+            ``restart_agent`` recommendations on first-occurrence
+            ``stuck`` classifications when the model disregards the
+            prompt's no-restart-on-first-stall guidance.
 
     Returns:
         A dict with keys:
@@ -141,6 +157,22 @@ async def decide_corrective_action(
         '  "restart_phase" - Restart all agents in the current phase (requires HITL approval)\n'
         '  "issue" - File a diagnostic GitHub issue\n'
         '  "slack" - Send urgent Slack notification\n\n'
+        "Recommendation ladder for stall / silent-agent classifications "
+        "(issue #2190): inspecting agent state must come before any "
+        "container-restart recommendation. An apparently-silent agent is "
+        "often mid-tool-call (e.g. a multi-minute pytest). Restarting "
+        "would destroy in-flight commits.\n"
+        "  - First response: `nudge` or `redirect` whose message body "
+        'leads with "Inspect container logs via '
+        "`mcp__egg__get_container_logs(task_id=…, agent_role=…)` before "
+        'taking destructive action."\n'
+        "  - Do NOT recommend `restart_agent` for a first stall alert. "
+        "Reserve `restart_agent` for follow-up alerts that fire after a "
+        "log inspection has confirmed the agent is genuinely inactive "
+        "(no recent commits, no pushes, no tool-call results), or for "
+        "infrastructure errors classified separately.\n"
+        "  - Never embed `egg-orch container restart <id>` as a first-line "
+        "operator action in the message body.\n\n"
         "Respond with ONLY a JSON object (no markdown fences) with these keys:\n"
         '  "action": one of the actions above\n'
         '  "message": string describing the action or message to send\n'
@@ -149,11 +181,70 @@ async def decide_corrective_action(
     ctx = json.dumps({"classification": classification, "context": context}, default=str)
 
     raw = await _call_decision_maker(prompt, ctx, model=model)
-    return _parse_json_or_fallback(
+    decision = _parse_json_or_fallback(
         raw,
         {"action": "nudge", "message": raw, "priority": "medium"},
     )
 
+    return _enforce_no_first_stall_restart(decision, classification, redirect_history)
+
+
+def _enforce_no_first_stall_restart(
+    decision: dict,
+    classification: dict,
+    redirect_history: list[dict] | None,
+) -> dict:
+    """Deterministically downgrade ``restart_agent`` on first-stall alerts.
+
+    The decision-maker prompt instructs the model not to recommend
+    ``restart_agent`` for a first-occurrence stall classification (issue
+    #2190 — restarting destroys in-flight commits from agents mid-pytest).
+    Prompts are advisory; this guard is the load-bearing enforcement.
+
+    Trigger: ``action == "restart_agent"`` AND classification is
+    ``stuck``/``needs_help`` AND no prior intervention of any kind appears
+    in ``redirect_history``. In that state we rewrite the decision to a
+    ``hitl`` so the operator gets a real decision surface (the original
+    recommendation and the model's reasoning are preserved in the
+    question text).
+
+    The "no prior intervention" check spans every action in the
+    decision-maker's vocabulary — ``nudge``, ``redirect``, ``issue``,
+    ``slack``, ``restart_agent``, ``restart_phase``, and ``hitl``. The
+    intent is "ensure at least one corrective action of any kind has
+    fired before destruction": if the operator (or the overseer) has
+    already had a chance to respond to the agent's state via any
+    intervention type, the guard yields and the model's recommendation
+    stands. A previous ``restart_agent`` (which may itself have been
+    the wrong call) likewise bypasses the guard rather than fast-track
+    the next restart through it.
+    """
+    if decision.get("action") != "restart_agent":
+        return decision
+
+    cls = classification.get("classification")
+    if cls not in {"stuck", "needs_help"}:
+        return decision
+
+    history = redirect_history or []
+    if any(h.get("action") in _PRIOR_INTERVENTIONS for h in history):
+        return decision
+
+    original_msg = decision.get("message", "")
+    original_suffix = f" Model's recommendation: {original_msg}" if original_msg else ""
+    return {
+        "action": "hitl",
+        "message": (
+            "Overseer overrode a `restart_agent` recommendation on a "
+            "first-occurrence stall. The agent may be mid-tool-call (e.g. "
+            "a multi-minute pytest) rather than genuinely stuck; restart "
+            "would destroy in-flight commits. Inspect container logs via "
+            "`mcp__egg__get_container_logs(task_id=…, agent_role=…)` "
+            "before approving a restart." + original_suffix
+        ).strip(),
+        "priority": decision.get("priority", "medium"),
+    }
+
 
 async def compose_redirect_message(
     agent_role: str, issue: str, context: dict, *, model: str | None = None