diff --git a/orchestrator/events.py b/orchestrator/events.py index 1fd081f082..097cea9793 100644 --- a/orchestrator/events.py +++ b/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" diff --git a/orchestrator/health_monitor.py b/orchestrator/health_monitor.py index 1719a57357..18fdbb6947 100644 --- a/orchestrator/health_monitor.py +++ b/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. diff --git a/orchestrator/models.py b/orchestrator/models.py index 91e002da6c..639f9bc2a3 100644 --- a/orchestrator/models.py +++ b/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" ) diff --git a/orchestrator/overseer/decision_maker.py b/orchestrator/overseer/decision_maker.py index ae3a66ff73..53e6c363e8 100644 --- a/orchestrator/overseer/decision_maker.py +++ b/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,7 +104,11 @@ 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. @@ -105,6 +116,11 @@ async def decide_corrective_action( 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 ` 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 diff --git a/orchestrator/overseer/monitor.py b/orchestrator/overseer/monitor.py index 51dbdf8f20..98320b5af7 100644 --- a/orchestrator/overseer/monitor.py +++ b/orchestrator/overseer/monitor.py @@ -10,6 +10,7 @@ import asyncio import hashlib +import inspect import json import logging import os @@ -47,6 +48,26 @@ _ACTION_WORDS = ("intervention", "attention", "review", "required", "needed", "escalat") +def _accepts_kwarg(func: Any, name: str) -> bool: + """Return True if *func* accepts a keyword argument named *name*. + + Uses :func:`inspect.signature` to inspect the callable. ``True`` is + returned when the parameter is declared explicitly or absorbed by a + ``**kwargs`` catch-all. Callables whose signature can't be + introspected (e.g. some C-implemented builtins) default to ``True`` + on the assumption that they accept arbitrary kwargs — matching how + :class:`unittest.mock.AsyncMock` and friends behave at the call site. + """ + try: + sig = inspect.signature(func) + except TypeError, ValueError: + return True + params = sig.parameters + if name in params: + return True + return any(p.kind is inspect.Parameter.VAR_KEYWORD for p in params.values()) + + class _DefaultConfig: """Fallback config when no PipelineConfig is provided.""" @@ -228,13 +249,35 @@ async def _check_decision_consistency_cls( return await self._classifier.check_decision_consistency(phase_output, prior_decisions) return await check_decision_consistency(phase_output, prior_decisions) - async def _decide_corrective_action(self, classification: dict, context: dict) -> dict: + async def _decide_corrective_action( + self, + classification: dict, + context: dict, + *, + redirect_history: list[dict] | None = None, + ) -> dict: model = getattr(self.config, "overseer_decision_maker_model", "sonnet") if self._decision_maker and hasattr(self._decision_maker, "decide_corrective_action"): - return await self._decision_maker.decide_corrective_action( - classification, context, model=model - ) - return await decide_corrective_action(classification, context, model=model) + method = self._decision_maker.decide_corrective_action + if _accepts_kwarg(method, "redirect_history"): + return await method( + classification, + context, + model=model, + redirect_history=redirect_history, + ) + # Test doubles with explicit signatures that pre-date the + # redirect_history kwarg fall through here; the guard + # downstream (_enforce_no_first_stall_restart) is bypassed + # in that path, which is fine for tests that don't exercise + # it. + return await method(classification, context, model=model) + return await decide_corrective_action( + classification, + context, + model=model, + redirect_history=redirect_history, + ) async def _compose_redirect_message(self, agent_role: str, issue: str, context: dict) -> str: model = getattr(self.config, "overseer_decision_maker_model", "sonnet") @@ -417,6 +460,7 @@ async def _poll_cycle(self) -> None: decision = await self._decide_corrective_action( classification, action_context, + redirect_history=list(self._escalation_history.get(agent_role, [])), ) await self._execute_action(decision, agent_role, container_logs=container_logs) await self._resolve_alert( @@ -529,6 +573,7 @@ async def handle_escalation( decision = await self._decide_corrective_action( classification, action_context, + redirect_history=history, ) await self._execute_action(decision, agent_role, container_logs=container_logs) diff --git a/orchestrator/routes/commit_authorship.py b/orchestrator/routes/commit_authorship.py index ce35beb840..95da1814a0 100644 --- a/orchestrator/routes/commit_authorship.py +++ b/orchestrator/routes/commit_authorship.py @@ -48,6 +48,36 @@ def get_logger(name: str, **kwargs: Any): # type: ignore[misc] logger = get_logger("orchestrator.commit_authorship") + +def _publish_container_activity(pipeline_id: str | None, role: str, kind: str) -> None: + """Best-effort publish of a CONTAINER_ACTIVITY event. + + The event lets HealthMonitor suppress heartbeat/progress stall alerts + against agents that are demonstrably alive (mid-commit) but not + emitting bus-level HEARTBEATs. See issue #2190. + + Failure to publish must not affect the registration response. + """ + if not isinstance(pipeline_id, str) or not pipeline_id.strip(): + return + try: + try: + from events import Event, EventType, get_event_bus # type: ignore[import-not-found] + except ImportError: + from ..events import Event, EventType, get_event_bus # type: ignore[no-redef] + + get_event_bus().publish( + Event( + event_type=EventType.CONTAINER_ACTIVITY, + pipeline_id=pipeline_id, + data={"agent_role": role, "kind": kind}, + source="commit_authorship", + ) + ) + except Exception as exc: # pragma: no cover - defensive + logger.warning("container_activity_publish_failed", error=str(exc)) + + commit_authorship_bp = Blueprint( "commit_authorship", __name__, url_prefix="/api/v1/commit-authorship" ) @@ -165,6 +195,8 @@ def register_commit() -> tuple[Response, int] | Response: ) return _json_error("Failed to register commit authorship", 500) + _publish_container_activity(pipeline_id, role.strip().lower(), "git_commit") + return ( jsonify( { @@ -206,6 +238,7 @@ def register_bulk() -> tuple[Response, int] | Response: return _json_error("Commit-authorship store unavailable", 500) results: list[dict[str, Any]] = [] + activity_seen: set[tuple[str, str]] = set() for item in items: if not isinstance(item, dict): results.append({"success": False, "message": "Item must be an object"}) @@ -251,6 +284,19 @@ def register_bulk() -> tuple[Response, int] | Response: } ) + item_pipeline = item.get("pipeline_id") + item_role = item.get("role", "") + if ( + isinstance(item_pipeline, str) + and item_pipeline + and isinstance(item_role, str) + and item_role + ): + key = (item_pipeline, item_role.strip().lower()) + if key not in activity_seen: + activity_seen.add(key) + _publish_container_activity(item_pipeline, key[1], "git_commit") + return jsonify({"success": True, "results": results}), 200 diff --git a/orchestrator/tests/test_commit_authorship_routes.py b/orchestrator/tests/test_commit_authorship_routes.py index 513ffc5273..8997453a0a 100644 --- a/orchestrator/tests/test_commit_authorship_routes.py +++ b/orchestrator/tests/test_commit_authorship_routes.py @@ -242,6 +242,54 @@ def register(self, *_a, **_kw): # Generic message — no internal details. assert "simulated backing-store failure" not in body["message"] + def test_register_publishes_container_activity(self, client, monkeypatch): + """A successful registration publishes a CONTAINER_ACTIVITY event so + HealthMonitor can suppress heartbeat/progress alerts against an + agent that is demonstrably alive (#2190).""" + # Reroute get_event_bus inside the route module's helper. The route + # imports get_event_bus lazily inside _publish_container_activity, + # so patching the events module is sufficient. + import events as events_mod # type: ignore[import-not-found] + from events import EventBus, EventType # type: ignore[import-not-found] + + bus = EventBus(async_delivery=False) + monkeypatch.setattr(events_mod, "get_event_bus", lambda: bus) + + captured: list = [] + bus.subscribe(EventType.CONTAINER_ACTIVITY, lambda e: captured.append(e)) + + response = client.post( + "/api/v1/commit-authorship/register", + data=json.dumps({"sha": _VALID_SHA, "role": "coder", "pipeline_id": "issue-1882"}), + content_type="application/json", + ) + assert response.status_code == 200 + assert len(captured) == 1 + evt = captured[0] + assert evt.pipeline_id == "issue-1882" + assert evt.data["agent_role"] == "coder" + assert evt.data["kind"] == "git_commit" + + def test_register_without_pipeline_id_does_not_publish(self, client, monkeypatch): + """Orphan registrations (no pipeline_id) do not publish CONTAINER_ACTIVITY — + the event has no pipeline scope to attach to.""" + import events as events_mod # type: ignore[import-not-found] + from events import EventBus, EventType # type: ignore[import-not-found] + + bus = EventBus(async_delivery=False) + monkeypatch.setattr(events_mod, "get_event_bus", lambda: bus) + + captured: list = [] + bus.subscribe(EventType.CONTAINER_ACTIVITY, lambda e: captured.append(e)) + + response = client.post( + "/api/v1/commit-authorship/register", + data=json.dumps({"sha": _VALID_SHA, "role": "coder"}), + content_type="application/json", + ) + assert response.status_code == 200 + assert captured == [] + # --------------------------------------------------------------------------- # /lookup — batch attribution diff --git a/orchestrator/tests/test_health_monitor.py b/orchestrator/tests/test_health_monitor.py index 751711c8dc..8a7c4a6c05 100644 --- a/orchestrator/tests/test_health_monitor.py +++ b/orchestrator/tests/test_health_monitor.py @@ -195,6 +195,199 @@ def test_heartbeat_within_threshold_no_alert(self): assert len(actions) == 0 +# --------------------------------------------------------------------------- +# Tests: Container-activity suppression (issue #2190) +# --------------------------------------------------------------------------- + + +def _emit_container_activity( + event_bus: EventBus, + agent_id: str = AGENT_ID, + pipeline_id: str = PIPELINE_ID, + kind: str = "git_commit", +) -> Event: + """Emit a CONTAINER_ACTIVITY event for an agent.""" + return event_bus.emit( + EventType.CONTAINER_ACTIVITY, + pipeline_id=pipeline_id, + data={"agent_role": agent_id, "kind": kind}, + ) + + +class TestContainerActivitySuppression: + """Issue #2190: a fresh CONTAINER_ACTIVITY event suppresses heartbeat + and progress stall alerts even when bus-level HEARTBEATs are absent. + + Repro from the issue: a coder mid-pytest is making commits but not + emitting heartbeats during a 10-minute blocking ``TaskOutput`` call; + the detector previously fired ``agent-heartbeat-stall`` and + recommended container restart, which would destroy in-flight work. + """ + + def test_recent_activity_suppresses_heartbeat_alert(self): + """Activity within the quiet window defers the heartbeat alert.""" + bus = _make_event_bus() + config = _make_config( + orchestrator_heartbeat_timeout_seconds=60, + orchestrator_activity_quiet_seconds=120, + ) + monitor = _make_monitor(bus, config) + + _emit_heartbeat(bus, agent_id=AGENT_ID) + + # Heartbeat is stale (61s old) but a fresh activity event arrives. + with patch("health_monitor.time") as mock_time: + base = time.time() + mock_time.time.return_value = base + 61 + _emit_container_activity(bus, agent_id=AGENT_ID) + + actions = monitor.check_heartbeats() + + assert actions == [] + + # And the agent is not flagged escalated, so a later poll + # (after activity has gone stale) still escalates. + with patch("health_monitor.time") as mock_time: + base = time.time() + mock_time.time.return_value = base + 1000 + actions = monitor.check_heartbeats() + + assert len(actions) == 1 + assert actions[0]["agent_id"] == AGENT_ID + + def test_recent_activity_suppresses_progress_alert(self): + """Same suppression applies to the progress stall detector.""" + bus = _make_event_bus() + config = _make_config( + orchestrator_heartbeat_timeout_seconds=60, + orchestrator_activity_quiet_seconds=120, + ) + monitor = _make_monitor(bus, config) + + _emit_progress(bus, agent_id=AGENT_ID) + + with patch("health_monitor.time") as mock_time: + base = time.time() + mock_time.time.return_value = base + 61 + _emit_container_activity(bus, agent_id=AGENT_ID) + + actions = monitor.check_progress() + + assert actions == [] + + def test_stale_activity_does_not_suppress(self): + """An activity event older than the quiet window does not suppress.""" + bus = _make_event_bus() + config = _make_config( + orchestrator_heartbeat_timeout_seconds=60, + orchestrator_activity_quiet_seconds=120, + ) + monitor = _make_monitor(bus, config) + + _emit_heartbeat(bus, agent_id=AGENT_ID) + _emit_container_activity(bus, agent_id=AGENT_ID) + + # Both heartbeat and activity are now > heartbeat threshold (60s) + # AND > activity quiet window (120s). + with patch("health_monitor.time") as mock_time: + mock_time.time.return_value = time.time() + 200 + actions = monitor.check_heartbeats() + + assert len(actions) == 1 + + def test_no_activity_event_does_not_suppress(self): + """An agent that has never emitted CONTAINER_ACTIVITY still escalates + on the heartbeat anchor — last_activity defaults to 0.0.""" + bus = _make_event_bus() + config = _make_config( + orchestrator_heartbeat_timeout_seconds=60, + orchestrator_activity_quiet_seconds=120, + ) + monitor = _make_monitor(bus, config) + + _emit_heartbeat(bus, agent_id=AGENT_ID) + + with patch("health_monitor.time") as mock_time: + mock_time.time.return_value = time.time() + 61 + actions = monitor.check_heartbeats() + + assert len(actions) == 1 + + def test_disabled_gate_no_suppression(self): + """``orchestrator_activity_quiet_seconds=0`` disables the gate. + + Even a CONTAINER_ACTIVITY event one second old must NOT suppress + a heartbeat alert — operators set the threshold to 0 as an + escape hatch when the gate is producing false negatives. + """ + bus = _make_event_bus() + config = _make_config( + orchestrator_heartbeat_timeout_seconds=60, + orchestrator_activity_quiet_seconds=0, + ) + monitor = _make_monitor(bus, config) + + _emit_heartbeat(bus, agent_id=AGENT_ID) + + with patch("health_monitor.time") as mock_time: + base = time.time() + # Heartbeat is stale (61s old) and a brand-new activity event + # arrives — but the gate is disabled, so the alert still fires. + mock_time.time.return_value = base + 61 + _emit_container_activity(bus, agent_id=AGENT_ID) + actions = monitor.check_heartbeats() + + assert len(actions) == 1 + assert actions[0]["agent_id"] == AGENT_ID + + def test_activity_event_for_other_pipeline_ignored(self): + """CONTAINER_ACTIVITY for a different pipeline does not suppress.""" + bus = _make_event_bus() + config = _make_config( + orchestrator_heartbeat_timeout_seconds=60, + orchestrator_activity_quiet_seconds=120, + ) + monitor = _make_monitor(bus, config) + + _emit_heartbeat(bus, agent_id=AGENT_ID) + _emit_container_activity(bus, agent_id=AGENT_ID, pipeline_id="some-other-pipeline") + + with patch("health_monitor.time") as mock_time: + mock_time.time.return_value = time.time() + 61 + actions = monitor.check_heartbeats() + + assert len(actions) == 1 + + def test_activity_event_for_other_agent_does_not_suppress_focal(self): + """Activity from a peer agent does not suppress an agent's own alert + via the focal-agent gate (per-agent isolation property of #2190). + + The peer-progress gate (#2242) is OR'd in at the alert sites and + WILL defer on a peer's recent heartbeat, so we disable it here + (``orchestrator_alert_progress_gate_seconds=0``) to isolate the + focal-agent gate's behavior under test. + """ + bus = _make_event_bus() + config = _make_config( + orchestrator_heartbeat_timeout_seconds=60, + orchestrator_activity_quiet_seconds=120, + orchestrator_alert_progress_gate_seconds=0, + ) + monitor = _make_monitor(bus, config) + + _emit_heartbeat(bus, agent_id=AGENT_ID) + _emit_heartbeat(bus, agent_id=AGENT_ID_2) + _emit_container_activity(bus, agent_id=AGENT_ID_2) + + with patch("health_monitor.time") as mock_time: + mock_time.time.return_value = time.time() + 61 + actions = monitor.check_heartbeats() + + # AGENT_ID has no activity of its own → escalates. + agent_ids = {a["agent_id"] for a in actions} + assert AGENT_ID in agent_ids + + # --------------------------------------------------------------------------- # Tests: reset_agent (issue #2084) # --------------------------------------------------------------------------- diff --git a/orchestrator/tests/test_overseer_decision_maker.py b/orchestrator/tests/test_overseer_decision_maker.py index df93616eac..89b1f508b9 100644 --- a/orchestrator/tests/test_overseer_decision_maker.py +++ b/orchestrator/tests/test_overseer_decision_maker.py @@ -153,6 +153,161 @@ def test_decide_corrective_action_escalate(self, mock_agent: AsyncMock) -> None: mock_agent.assert_awaited_once() +class TestDecideCorrectiveActionFirstStallDowngrade: + """Issue #2190: a `restart_agent` recommendation on a first-occurrence + `stuck` classification must be overridden to `hitl` to avoid + destroying in-flight commits from agents mid-tool-call. Routing + through HITL (rather than `nudge`) keeps operator-targeted text out + of the agent's inbox and gives the operator a real decision surface. + + The decision-maker prompt asks the model not to recommend + `restart_agent` for a first stall alert, but prompts are advisory. + The post-hoc guard in `_enforce_no_first_stall_restart` is the + load-bearing enforcement. + """ + + @patch(_AGENT_PATCH, new_callable=AsyncMock) + def test_first_stall_restart_overridden_to_hitl(self, mock_agent: AsyncMock) -> None: + # Model disregards the prompt and emits restart_agent on a first + # stall — the guard must rewrite the action. + mock_agent.return_value = _make_result( + json.dumps( + { + "action": "restart_agent", + "message": "Agent is stuck. Restart to recover.", + "priority": "high", + } + ) + ) + classification = { + "classification": "stuck", + "confidence": 0.9, + "reasoning": "No heartbeat for 5 minutes", + } + + result = _run(decide_corrective_action(classification, {}, redirect_history=[])) + + assert result["action"] == "hitl" + # Original recommendation is preserved in the message body for the + # operator to see what the model wanted to do. + assert "Agent is stuck. Restart to recover." in result["message"] + # Inspection-first guidance is included for the operator. + assert "mcp__egg__get_container_logs" in result["message"] + + @patch(_AGENT_PATCH, new_callable=AsyncMock) + def test_first_stall_restart_with_empty_message_no_dangling_colon( + self, mock_agent: AsyncMock + ) -> None: + # When the model returns an empty message, the override must not + # leave a dangling ``Model's recommendation:`` suffix. + mock_agent.return_value = _make_result( + json.dumps( + { + "action": "restart_agent", + "message": "", + "priority": "high", + } + ) + ) + classification = {"classification": "stuck", "confidence": 0.9, "reasoning": ""} + + result = _run(decide_corrective_action(classification, {}, redirect_history=[])) + + assert result["action"] == "hitl" + assert "Model's recommendation:" not in result["message"] + # And no trailing colon at the end of the body. + assert not result["message"].rstrip().endswith(":") + + @patch(_AGENT_PATCH, new_callable=AsyncMock) + def test_restart_allowed_after_prior_redirect(self, mock_agent: AsyncMock) -> None: + # Once a nudge or redirect has already been sent, restart_agent + # is permitted (the agent has had a chance to recover). + mock_agent.return_value = _make_result( + json.dumps( + { + "action": "restart_agent", + "message": "Restart after no response to redirect.", + "priority": "high", + } + ) + ) + classification = {"classification": "stuck", "confidence": 0.95, "reasoning": ""} + history = [{"action": "redirect", "timestamp": 1000}] + + result = _run(decide_corrective_action(classification, {}, redirect_history=history)) + + assert result["action"] == "restart_agent" + + @pytest.mark.parametrize( + "prior_action", + ["issue", "slack", "restart_phase"], + ) + @patch(_AGENT_PATCH, new_callable=AsyncMock) + def test_restart_allowed_after_other_intervention_types( + self, mock_agent: AsyncMock, prior_action: str + ) -> None: + # Any prior intervention bypasses the guard — the docstring + # promises every action in the decision-maker vocabulary counts, + # not just nudge/redirect/restart_agent/hitl. + mock_agent.return_value = _make_result( + json.dumps( + { + "action": "restart_agent", + "message": f"Restart after {prior_action} didn't help.", + "priority": "high", + } + ) + ) + classification = {"classification": "stuck", "confidence": 0.95, "reasoning": ""} + history = [{"action": prior_action, "timestamp": 1000}] + + result = _run(decide_corrective_action(classification, {}, redirect_history=history)) + + assert result["action"] == "restart_agent" + + @patch(_AGENT_PATCH, new_callable=AsyncMock) + def test_restart_allowed_after_prior_restart(self, mock_agent: AsyncMock) -> None: + # Prior `restart_agent` history also bypasses the guard — the + # "first occurrence" check spans every intervention type, so a + # second restart isn't blocked once any corrective action has + # already fired. + mock_agent.return_value = _make_result( + json.dumps( + { + "action": "restart_agent", + "message": "Second restart after first didn't help.", + "priority": "high", + } + ) + ) + classification = {"classification": "stuck", "confidence": 0.95, "reasoning": ""} + history = [{"action": "restart_agent", "timestamp": 1000}] + + result = _run(decide_corrective_action(classification, {}, redirect_history=history)) + + assert result["action"] == "restart_agent" + + @patch(_AGENT_PATCH, new_callable=AsyncMock) + def test_non_stall_classification_unchanged(self, mock_agent: AsyncMock) -> None: + # The guard only triggers on stuck / needs_help. A "working" + # classification (which does flow through the LLM path) leaves + # the model's restart_agent recommendation alone. + mock_agent.return_value = _make_result( + json.dumps( + { + "action": "restart_agent", + "message": "Agent crashed.", + "priority": "critical", + } + ) + ) + classification = {"classification": "working", "confidence": 0.7, "reasoning": ""} + + result = _run(decide_corrective_action(classification, {}, redirect_history=[])) + + assert result["action"] == "restart_agent" + + # =================================================================== # compose_redirect_message # =================================================================== diff --git a/orchestrator/tests/test_overseer_monitor.py b/orchestrator/tests/test_overseer_monitor.py index 3068310dfb..a3b2993d7a 100644 --- a/orchestrator/tests/test_overseer_monitor.py +++ b/orchestrator/tests/test_overseer_monitor.py @@ -2836,3 +2836,46 @@ async def counting_query(agent_role: str, tail: int = 200) -> str: assert len(classify_calls) == 2 for call in classify_calls: assert call.kwargs.get("container_logs") == "some logs" + + +# =================================================================== +# _accepts_kwarg +# =================================================================== + + +class TestAcceptsKwarg: + """Tests for ``_accepts_kwarg`` signature-introspection helper.""" + + def test_explicit_kwarg(self) -> None: + from overseer.monitor import _accepts_kwarg + + def f(a, redirect_history=None): + pass + + assert _accepts_kwarg(f, "redirect_history") is True + + def test_var_keyword(self) -> None: + from overseer.monitor import _accepts_kwarg + + def f(*args, **kwargs): + pass + + assert _accepts_kwarg(f, "redirect_history") is True + + def test_legacy_signature_rejects_kwarg(self) -> None: + from overseer.monitor import _accepts_kwarg + + def f(a, b): + pass + + assert _accepts_kwarg(f, "redirect_history") is False + + def test_uninspectable_builtin_falls_through_to_true(self) -> None: + # ``inspect.signature(int)`` raises ``ValueError`` on CPython + # because the C-implemented type does not expose a Python-level + # signature. The fallback should assume kwargs are accepted so a + # genuinely callable target isn't excluded just because its + # signature can't be introspected. + from overseer.monitor import _accepts_kwarg + + assert _accepts_kwarg(int, "redirect_history") is True