fix(experiments): wire progress + context-request heartbeats into the thrash signal

Addresses two PR #704 review findings: the `activity` tally the
spawn-thrash detector reads leaned on signals that never moved during the
window that matters.

1. context_requests was dead. get_task_context emitted a
   "context_requested" event but never called
   LiveExperimentMonitor.record_context_request, so the counter stayed 0.
   Wire the call (mirrors log_decision).

2. report_task_progress (the 25/50/75% heartbeat) was not counted at all.
   get_experiment_status exposes only task counts + logged-work counters,
   so an agent advancing via progress reports left `activity` flat. Add a
   progress_updates counter + record_progress() to the monitor, expose it
   in get_status (plus the mlflow metrics + text summary), and call it
   from report_task_progress.

progress_updates is the earliest of these signals -- it fires as soon as a
claimed agent reports, well before the first artifact -- closing the
claim->first-artifact gap that could otherwise fast-fail a healthy run.
Both are folded into the runner's activity tally.

Tests: monitor heartbeat counters + get_status exposure.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: lwgray

dev-tools/experiments/runners/spawn_agents.py

@@ -1985,13 +1985,19 @@ def _emit(message: str) -> None:
                     # "Real progress" signal for the spawn-thrash detector:
                     # a monotonic, cumulative tally of the signals an agent
                     # emits while genuinely working a task — completions
-                    # plus logged work (context requests, artifacts,
-                    # decisions, blockers). It rises on real work and stays
-                    # flat on claim-and-exit churn; unlike in_progress it
-                    # never flickers back down, so it cannot mask thrash.
-                    # See SpawnThrashDetector.observe.
+                    # plus the report_task_progress heartbeat (25/50/75%)
+                    # and logged work (context requests, artifacts,
+                    # decisions, blockers). progress_updates is the earliest
+                    # of these: it fires as soon as a claimed agent reports,
+                    # well before the first artifact lands, closing the
+                    # claim->first-artifact gap that would otherwise let the
+                    # detector fast-fail a healthy run. It rises on real work
+                    # and stays flat on claim-and-exit churn; unlike
+                    # in_progress it never flickers back down, so it cannot
+                    # mask thrash. See SpawnThrashDetector.observe.
                     activity = (
                         done
+                        + int(status.get("progress_updates", 0))
                         + int(status.get("context_requests", 0))
                         + int(status.get("artifacts_created", 0))
                         + int(status.get("decisions_logged", 0))

src/experiments/live_experiment_monitor.py

@@ -96,6 +96,7 @@ def __init__(
         self.artifacts_created = 0
         self.decisions_logged = 0
         self.context_requests = 0
+        self.progress_updates = 0
 
         # Previous completed-task count, used to derive per-interval velocity
         self._last_completed_count = 0
@@ -227,6 +228,7 @@ async def stop(self) -> Dict[str, Any]:
             "total_artifacts": float(self.artifacts_created),
             "total_decisions": float(self.decisions_logged),
             "total_context_requests": float(self.context_requests),
+            "total_progress_updates": float(self.progress_updates),
         }
 
         # Generate summary
@@ -556,6 +558,29 @@ def record_context_request(
 
         logger.debug(f"Recorded context request: {agent_id} for {task_id}")
 
+    def record_progress(
+        self,
+        agent_id: str,
+        task_id: str,
+        progress: int = 0,
+        status: str = "in_progress",
+    ) -> None:
+        """Record a task-progress heartbeat (a report_task_progress call).
+
+        Counts every progress report — including intermediate 25/50/75%
+        updates — as a sign the agent is alive and working its claimed
+        task. The runner's spawn-thrash detector reads this (via
+        ``progress_updates`` in :meth:`get_status`) so it does not tear
+        down a healthy-but-slow run that is advancing without a
+        completion yet.
+        """
+        self.progress_updates += 1
+
+        logger.debug(
+            f"Recorded progress update: {agent_id} on {task_id} "
+            f"({progress}% {status})"
+        )
+
     async def get_status(self) -> Dict[str, Any]:
         """
         Get current experiment status.
@@ -598,6 +623,7 @@ async def get_status(self) -> Dict[str, Any]:
             "artifacts_created": self.artifacts_created,
             "decisions_logged": self.decisions_logged,
             "context_requests": self.context_requests,
+            "progress_updates": self.progress_updates,
         }
 
         # Ground truth from the kanban backend. This is what
@@ -657,6 +683,7 @@ def _generate_summary(self) -> str:
 - Artifacts Created: {self.artifacts_created}
 - Decisions Logged: {self.decisions_logged}
 - Context Requests: {self.context_requests}
+- Progress Updates: {self.progress_updates}
 
 Top Agents by Completions:
 """

src/marcus_mcp/tools/context.py

@@ -181,6 +181,16 @@ async def get_task_context(task_id: str, state: Any) -> Dict[str, Any]:
         },
     )
 
+    # Record in the active experiment if one is running (mirrors
+    # log_decision above). Without this, context_requests stays 0 and the
+    # runner's spawn-thrash detector cannot see a claimed agent pulling
+    # context as a sign of forward progress.
+    from src.experiments.live_experiment_monitor import get_active_monitor
+
+    _monitor = get_active_monitor()
+    if _monitor and _monitor.is_running:
+        _monitor.record_context_request(agent_id=agent_id, task_id=task_id)
+
     try:
         # Check if this is a subtask
         if hasattr(state, "subtask_manager") and state.subtask_manager:

src/marcus_mcp/tools/task.py

@@ -3732,6 +3732,22 @@ async def report_task_progress(
         },
     )
 
+    # Record the progress heartbeat in the active experiment monitor so
+    # the runner's spawn-thrash detector sees a claimed task advancing
+    # (report_task_progress at 25/50/75%) as real progress, not only
+    # completions. Counted on every report — a report arriving at all
+    # proves the agent is alive and working its task.
+    from src.experiments.live_experiment_monitor import get_active_monitor
+
+    _progress_monitor = get_active_monitor()
+    if _progress_monitor and _progress_monitor.is_running:
+        _progress_monitor.record_progress(
+            agent_id=agent_id,
+            task_id=task_id,
+            progress=progress,
+            status=status,
+        )
+
     # Escalation flag — set when the retry ceiling is hit during
     # validation. The task is routed through the normal completion
     # path (kanban update, lease cleanup, branch merge) but the

tests/unit/experiments/test_live_experiment_monitor.py

@@ -43,9 +43,44 @@ def _make_monitor(kanban_client: Any = None) -> LiveExperimentMonitor:
     monitor.artifacts_created = 0
     monitor.decisions_logged = 0
     monitor.context_requests = 0
+    monitor.progress_updates = 0
     return monitor
 
 
+class TestProgressHeartbeat:
+    """record_progress + progress_updates surface the report_task_progress
+    heartbeat (PR #704 review).
+
+    The spawn-thrash detector reads ``progress_updates`` so a claimed
+    agent reporting 25/50/75% counts as forward progress before any task
+    completes — closing the claim->first-artifact gap that could
+    otherwise fast-fail a healthy run.
+    """
+
+    def test_record_progress_increments_counter(self) -> None:
+        """Each report_task_progress call bumps the counter."""
+        monitor = _make_monitor()
+        assert monitor.progress_updates == 0
+        monitor.record_progress(agent_id="a1", task_id="t1", progress=25)
+        monitor.record_progress(agent_id="a1", task_id="t1", progress=50)
+        assert monitor.progress_updates == 2
+
+    @pytest.mark.asyncio
+    async def test_get_status_exposes_progress_updates(self) -> None:
+        """get_status surfaces progress_updates for the runner to read."""
+        monitor = _make_monitor()
+        monitor.record_progress(agent_id="a1", task_id="t1", progress=75)
+        status = await monitor.get_status()
+        assert status["progress_updates"] == 1
+
+    def test_record_context_request_increments_counter(self) -> None:
+        """record_context_request (now wired into get_task_context) counts."""
+        monitor = _make_monitor()
+        assert monitor.context_requests == 0
+        monitor.record_context_request(agent_id="a1", task_id="t1")
+        assert monitor.context_requests == 1
+
+
 class TestGetStatusKanbanTruth:
     """get_status must expose kanban-truth task counts."""