[M4] Rule-based no-progress detector baseline

README.md

@@ -197,6 +197,7 @@ First experiment results: [`experiments/results/tool-semantic-drift-001.md`](exp
 | Confidence | No | Absent in scripted scaffold |
 | Raw telemetry | No | No latency/retry spikes |
 | **Semantic mismatch** | **Yes (step 4)** | Expected vs observed `plan_id_meaning` |
+| No progress | No | No repeated actions / flat goal progress in this case |
 | Recovery (`refresh_tool_contract`) | Success | Task recovers after remapping |
 
 Full specification: [`docs/roadmap.md`](docs/roadmap.md)  

baselines/rules/__init__.py

@@ -5,6 +5,7 @@
     ConfidenceDetector,
     DetectionResult,
     ExceptionDetector,
+    NoProgressDetector,
     RawTelemetryDetector,
     SemanticMismatchDetector,
     all_detectors,
@@ -15,6 +16,7 @@
     "ConfidenceDetector",
     "DetectionResult",
     "ExceptionDetector",
+    "NoProgressDetector",
     "RawTelemetryDetector",
     "SemanticMismatchDetector",
     "all_detectors",

baselines/rules/detectors.py

@@ -5,6 +5,7 @@
 from dataclasses import dataclass
 
 from runtime.features.mismatch import first_mismatch_step, semantic_mismatch
+from runtime.features.progress import flat_progress_step, repeated_action_step
 from runtime.schemas.trace import TraceEvent
 
 
@@ -112,10 +113,46 @@ def detect(self, events: list[TraceEvent]) -> DetectionResult:
         )
 
 
+class NoProgressDetector(BaseDetector):
+    """Rule-based planning-loop baseline: repeated actions or flat goal progress.
+
+    Flags episodes where the agent either (a) issues the same action (name +
+    arguments) several steps in a row, or (b) makes no forward goal progress for
+    several consecutive steps. Both are cheap, model-independent proxies for
+    planning-loop / stagnation failures (see docs/roadmap.md, Baseline B).
+    """
+
+    name = "no_progress"
+
+    def __init__(self, min_repeats: int = 2, min_flat_steps: int = 3) -> None:
+        self.min_repeats = min_repeats
+        self.min_flat_steps = min_flat_steps
+
+    def detect(self, events: list[TraceEvent]) -> DetectionResult:
+        repeat_step = repeated_action_step(events, min_repeats=self.min_repeats)
+        if repeat_step is not None:
+            return DetectionResult(
+                detector_name=self.name,
+                detected=True,
+                detection_step=repeat_step,
+                reason="repeated_action",
+            )
+        flat_step = flat_progress_step(events, min_flat_steps=self.min_flat_steps)
+        if flat_step is not None:
+            return DetectionResult(
+                detector_name=self.name,
+                detected=True,
+                detection_step=flat_step,
+                reason="flat_goal_progress",
+            )
+        return DetectionResult(detector_name=self.name, detected=False, reason="progress_nominal")
+
+
 def all_detectors() -> list[BaseDetector]:
     return [
         ExceptionDetector(),
         ConfidenceDetector(),
         RawTelemetryDetector(),
         SemanticMismatchDetector(),
+        NoProgressDetector(),
     ]

experiments/results/tool-semantic-drift-001-report.json

@@ -36,13 +36,21 @@
       "detection_step": 4,
       "reason": "plan_id_meaning_or_contract_divergence",
       "meets_first_detectable": true
+    },
+    {
+      "detector": "no_progress",
+      "detected": false,
+      "detection_step": null,
+      "reason": "progress_nominal",
+      "meets_first_detectable": false
     }
   ],
   "false_alarms_on_clean_run": {
     "exception": false,
     "confidence": false,
     "raw_telemetry": false,
-    "semantic_mismatch": false
+    "semantic_mismatch": false,
+    "no_progress": false
   },
   "ground_truth": {
     "root_cause": "plan_identifier_semantics_changed",

experiments/results/tool-semantic-drift-001.md

@@ -23,6 +23,7 @@
 | `confidence` | False | None | False | confidence_unavailable_in_scripted_scaffold |
 | `raw_telemetry` | False | None | False | telemetry_nominal |
 | `semantic_mismatch` | True | 4 | True | plan_id_meaning_or_contract_divergence |
+| `no_progress` | False | None | False | progress_nominal |
 
 ## Outcomes
 
@@ -31,4 +32,4 @@
 - Diagnosis: `plan_identifier_semantics_changed`
 - Recovery actions: `['refresh_tool_contract', 'validate_identifier_mapping', 'replan_pending_action']`
 - Recovery success: **True**
-- False alarms on clean run: `{'exception': False, 'confidence': False, 'raw_telemetry': False, 'semantic_mismatch': False}`
+- False alarms on clean run: `{'exception': False, 'confidence': False, 'raw_telemetry': False, 'semantic_mismatch': False, 'no_progress': False}`

runtime/features/__init__.py

@@ -1,5 +1,11 @@
 """Feature extractors."""
 
 from runtime.features.mismatch import first_mismatch_step, semantic_mismatch
+from runtime.features.progress import flat_progress_step, repeated_action_step
 
-__all__ = ["first_mismatch_step", "semantic_mismatch"]
+__all__ = [
+    "first_mismatch_step",
+    "semantic_mismatch",
+    "flat_progress_step",
+    "repeated_action_step",
+]

runtime/features/progress.py

@@ -0,0 +1,60 @@
+"""Repeated-action and flat goal-progress features (planning-loop / stagnation signals)."""
+
+from __future__ import annotations
+
+from runtime.schemas.trace import TraceEvent
+
+
+def _action_signature(event: TraceEvent) -> tuple[str, tuple[tuple[str, object], ...]]:
+    """Identity of an action call: its name plus sorted arguments."""
+    arguments = event.attributes.get("arguments", {})
+    if not isinstance(arguments, dict):
+        arguments = {}
+    return event.name, tuple(sorted(arguments.items()))
+
+
+def repeated_action_step(events: list[TraceEvent], min_repeats: int = 2) -> int | None:
+    """Return the step of the first action repeated `min_repeats` times in a row.
+
+    A "repeat" is a consecutive event with the same action name and arguments as the
+    previous event — evidence the agent is stuck retrying the same move without
+    revising its plan.
+    """
+    if min_repeats < 2:
+        raise ValueError("min_repeats must be >= 2")
+    streak = 1
+    previous: tuple[str, tuple[tuple[str, object], ...]] | None = None
+    for event in events:
+        signature = _action_signature(event)
+        streak = streak + 1 if signature == previous else 1
+        previous = signature
+        if streak >= min_repeats:
+            step = event.attributes.get("step")
+            return int(step) if step is not None else None
+    return None
+
+
+def flat_progress_step(events: list[TraceEvent], min_flat_steps: int = 3) -> int | None:
+    """Return the step where goal progress has stalled for `min_flat_steps` events.
+
+    Reads the `goal_progress` value recorded on each event's attributes by
+    `TraceCollector`. Events without a recorded progress value are skipped rather
+    than treated as stalled.
+    """
+    if min_flat_steps < 2:
+        raise ValueError("min_flat_steps must be >= 2")
+    streak = 1
+    previous_progress: float | None = None
+    for event in events:
+        raw_progress = event.attributes.get("goal_progress")
+        if raw_progress is None:
+            continue
+        progress = float(raw_progress)
+        streak = (
+            streak + 1 if previous_progress is not None and progress <= previous_progress else 1
+        )
+        previous_progress = progress
+        if streak >= min_flat_steps:
+            step = event.attributes.get("step")
+            return int(step) if step is not None else None
+    return None

runtime/tracing/collector.py

@@ -52,7 +52,11 @@ def record(
                 success=observation.success,
                 attributes=obs_attrs,
             ),
-            attributes={"step": action.step, "arguments": action.arguments},
+            attributes={
+                "step": action.step,
+                "arguments": action.arguments,
+                "goal_progress": goal_progress,
+            },
         )
         self.events.append(event)
         self.goal_progress.append(goal_progress)

tests/unit/test_no_progress_detector.py

@@ -0,0 +1,105 @@
+"""Unit tests for the rule-based no-progress detector baseline."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+from agentfailbench.runners.episode import _run_traced, load_case
+from baselines.rules.detectors import NoProgressDetector, all_detectors
+from runtime.schemas.trace import TraceEvent
+
+
+def _event(
+    name: str,
+    step: int,
+    *,
+    arguments: dict[str, Any] | None = None,
+    goal_progress: float | None = None,
+) -> TraceEvent:
+    return TraceEvent(
+        event_id=f"evt-{step}",
+        timestamp=datetime.now(UTC),
+        task_id="test-task",
+        scaffold="test-scaffold",
+        entity="tool",
+        name=name,
+        attributes={
+            "step": step,
+            "arguments": arguments or {},
+            "goal_progress": goal_progress,
+        },
+    )
+
+
+def test_no_progress_detector_flags_repeated_action() -> None:
+    events = [
+        _event("get_customer", 1, arguments={"customer_id": "c1"}, goal_progress=0.2),
+        _event("get_plan", 2, arguments={"plan_id": "p1"}, goal_progress=0.4),
+        _event("get_plan", 3, arguments={"plan_id": "p1"}, goal_progress=0.5),
+        _event("get_plan", 4, arguments={"plan_id": "p1"}, goal_progress=0.6),
+    ]
+
+    result = NoProgressDetector(min_repeats=2).detect(events)
+
+    assert result.detected is True
+    assert result.reason == "repeated_action"
+    assert result.detection_step == 3
+
+
+def test_no_progress_detector_flags_flat_goal_progress() -> None:
+    events = [
+        _event("get_customer", 1, arguments={"customer_id": "c1"}, goal_progress=0.2),
+        _event("get_subscription", 2, arguments={"customer_id": "c1"}, goal_progress=0.4),
+        _event("replan", 3, arguments={"attempt": 1}, goal_progress=0.4),
+        _event("replan", 4, arguments={"attempt": 2}, goal_progress=0.4),
+        _event("replan", 5, arguments={"attempt": 3}, goal_progress=0.3),
+    ]
+
+    result = NoProgressDetector(min_flat_steps=3).detect(events)
+
+    assert result.detected is True
+    assert result.reason == "flat_goal_progress"
+    assert result.detection_step == 4
+
+
+def test_no_progress_detector_ignores_non_consecutive_repeats() -> None:
+    events = [
+        _event("get_customer", 1, arguments={"customer_id": "c1"}, goal_progress=0.125),
+        _event("get_subscription", 2, arguments={"customer_id": "c1"}, goal_progress=0.25),
+        _event("get_plan", 3, arguments={"plan_id": "p1"}, goal_progress=0.375),
+        _event("get_subscription", 4, arguments={"customer_id": "c1"}, goal_progress=0.5),
+        _event("get_plan", 5, arguments={"plan_id": "p1"}, goal_progress=0.625),
+        _event("update_subscription", 6, arguments={"plan_id": "p1"}, goal_progress=0.75),
+        _event("get_subscription", 7, arguments={"customer_id": "c1"}, goal_progress=0.875),
+        _event("get_plan", 8, arguments={"plan_id": "p1"}, goal_progress=1.0),
+    ]
+
+    result = NoProgressDetector().detect(events)
+
+    assert result.detected is False
+    assert result.reason == "progress_nominal"
+
+
+def test_no_progress_detector_is_registered_in_all_detectors() -> None:
+    names = [detector.name for detector in all_detectors()]
+    assert "no_progress" in names
+
+
+def test_no_progress_detector_no_false_alarm_on_clean_scripted_episode() -> None:
+    case = load_case("tool-semantic-drift-001")
+    _env, _agent, collector, _injector = _run_traced(case, inject_failure=False)
+
+    result = NoProgressDetector().detect(collector.events)
+
+    assert result.detected is False
+
+
+def test_no_progress_detector_no_false_alarm_on_drifted_scripted_episode() -> None:
+    case = load_case("tool-semantic-drift-001")
+    _env, _agent, collector, injector = _run_traced(case, inject_failure=True)
+    assert injector.triggered
+
+    result = NoProgressDetector().detect(collector.events)
+
+    assert result.detected is False