fix(swarm): unified reconcile + MCP keepalive + heartbeat coverage (#132)

Symptoms reported in #132 (WorkBuddy + DeepSeek MCP):
- MCP run_swarm blocked 30 min then errored, losing run_id
- get_swarm_status / get_run_result showed stale snapshot while tasks/*.json
  had real progress
- Host crash left runs "running" forever with no recovery path
- list_swarm_runs / SSE rendered the same stale snapshot

Architecture:
- store.reconcile_run(write=True): single read-side entry point. Hydrates
  live task files, recovers terminal-but-not-finalized state (host crashed
  between last layer sync and finalize), and reaps stale running runs whose
  events.jsonl has gone silent past compute_stale_threshold.
- Threshold is per-run: max(60, min(heartbeat_interval*10, retry_ceiling))
  — fast detection when heartbeats work, retry-budget-clamped when not.
- HeartbeatTimer wraps registry.execute, llm.stream_chat, AND grounding
  fetch so no normal execution phase can produce >heartbeat_interval of
  silence. Slow first-token / reasoning-mode / pure tool-call responses
  no longer trip the reaper.
- run_swarm emits ctx.report_progress every poll with elapsed seconds, so
  MCP transport stays alive via progress notifications; first frame is
  fixed-format ``swarm_started run_id=<id>`` so transport-drop clients can
  still recover the handle.

Endpoints updated to use reconcile_run:
- MCP: get_swarm_status, get_run_result, list_runs, run_swarm polling
- API: /swarm/runs, /swarm/runs/{id}, /swarm/runs/{id}/events SSE
- SwarmTool: polling + budget-out path (no longer cancels on timeout)

Other fixes:
- SWARM_HEARTBEAT_INTERVAL_S parses with ValueError fallback in worker
  (was crashing import on bad config)
- reaper derives run terminal status from task statuses (no more
  hardcoded "failed" on a run whose tasks all completed)
- serialize_task surfaces started_at / completed_at / depends_on / blocked_by
- reap_stale_runs() MCP tool for explicit user-triggered cleanup

Tests: 22 new regressions in tests/test_swarm_status_hydration.py covering
hydrate, reconcile, terminal recovery, stale reap, MCP keepalive cadence,
auto-recovery on read, threshold math, env var robustness, heartbeat
wiring around LLM + tool execution.

Verification:
- targeted suite: 22 passed
- full swarm/mcp suite: 169 passed, 4 skipped, 0 failed
- compileall + import smoke clean

Author: warren618

agent/api_server.py

@@ -1687,46 +1687,45 @@ async def create_swarm_run(payload: dict, http_request: Request):
 
 @app.get("/swarm/runs", dependencies=[Depends(require_auth)])
 async def list_swarm_runs(limit: int = Query(20, ge=1, le=100)):
-    """List swarm runs (newest first)."""
+    """List swarm runs (newest first), reconciled."""
     runtime = _get_swarm_runtime()
     runs = runtime._store.list_runs(limit=limit)
-    return [
-        {
-            "id": r.id,
-            "preset_name": r.preset_name,
-            "status": r.status.value,
-            "created_at": r.created_at,
-            "task_count": len(r.tasks),
-            "completed_count": sum(1 for t in r.tasks if t.status.value == "completed"),
-        }
-        for r in runs
-    ]
+    items = []
+    for r in runs:
+        # Reconcile each row: a zombie running run will be auto-finalized so
+        # the dashboard never shows a permanent "running" stuck row.
+        reconciled = runtime._store.reconcile_run(r, write=True)
+        items.append(
+            {
+                "id": reconciled.id,
+                "preset_name": reconciled.preset_name,
+                "status": reconciled.status.value,
+                "is_stale": runtime._store.is_run_stale(reconciled),
+                "created_at": reconciled.created_at,
+                "completed_at": reconciled.completed_at,
+                "task_count": len(reconciled.tasks),
+                "completed_count": sum(1 for t in reconciled.tasks if t.status.value == "completed"),
+            }
+        )
+    return items
 
 
 @app.get("/swarm/runs/{run_id}", dependencies=[Depends(require_auth)])
 async def get_swarm_run(run_id: str):
-    """Swarm run detail including task statuses."""
-    from src.swarm.task_store import TaskStore
-
+    """Swarm run detail including task statuses (reconciled)."""
     _validate_path_param(run_id, "run_id")
     runtime = _get_swarm_runtime()
-    run = runtime._store.load_run(run_id)
-    if not run:
+    loaded = runtime._store.load_run(run_id)
+    if not loaded:
         raise HTTPException(status_code=404, detail=f"Run {run_id} not found")
 
-    # Merge real-time task statuses from task_store (updated during execution)
-    run_dir = runtime._store.run_dir(run_id)
-    tasks_dir = run_dir / "tasks"
-    if tasks_dir.exists():
-        task_store = TaskStore(run_dir)
-        live_tasks = task_store.load_all()
-        if live_tasks:
-            run.tasks = live_tasks
+    run = runtime._store.reconcile_run(loaded, write=True)
 
     return {
         "id": run.id,
         "preset_name": run.preset_name,
         "status": run.status.value,
+        "is_stale": runtime._store.is_run_stale(run),
         "user_vars": run.user_vars,
         "agents": [a.model_dump() for a in run.agents],
         "tasks": [t.model_dump() for t in run.tasks],
@@ -1754,9 +1753,14 @@ async def event_stream():
                 idx += 1
                 yield f"id: {idx}\nevent: {evt.type}\ndata: {json.dumps(evt.model_dump(), ensure_ascii=False)}\n\n"
             run = runtime._store.load_run(run_id)
-            if run and run.status.value in ("completed", "failed", "cancelled"):
-                yield f"event: done\ndata: {{\"status\": \"{run.status.value}\"}}\n\n"
-                break
+            if run:
+                # Reconcile so a zombie running run can still close this SSE
+                # stream cleanly — without it, a dead host would keep the
+                # stream open forever and block the dashboard's "done" state.
+                reconciled = runtime._store.reconcile_run(run, write=True)
+                if reconciled.status.value in ("completed", "failed", "cancelled"):
+                    yield f"event: done\ndata: {{\"status\": \"{reconciled.status.value}\"}}\n\n"
+                    break
             await asyncio.sleep(2)
 
     return StreamingResponse(event_stream(), media_type="text/event-stream")

agent/mcp_server.py

@@ -40,7 +40,7 @@
 if str(AGENT_DIR) not in sys.path:
     sys.path.insert(0, str(AGENT_DIR))
 
-from fastmcp import FastMCP
+from fastmcp import Context, FastMCP
 
 mcp = FastMCP("Vibe-Trading")
 
@@ -356,23 +356,40 @@ def list_swarm_presets() -> str:
 
 
 @mcp.tool
-def run_swarm(preset_name: str, variables: dict[str, str]) -> str:
-    """Run a swarm multi-agent team and return the final report.
+async def run_swarm(
+    preset_name: str,
+    variables: dict[str, str],
+    wait_seconds: int = 3600,
+    start_only: bool = False,
+    ctx: Context | None = None,
+) -> str:
+    """Run a swarm multi-agent team and stream progress back to the caller.
 
     Assembles a team of specialized agents that collaborate through a DAG workflow.
     For example, the 'investment_committee' preset runs bull analyst, bear analyst,
     risk officer, and portfolio manager in sequence.
 
     Use list_swarm_presets() to see available presets and their required variables.
 
+    The tool keeps the MCP call open via ``Context.report_progress`` while the
+    swarm runs, so the caller sees live "N/M tasks complete" updates instead
+    of timing out silently. Only if ``wait_seconds`` is exhausted does the
+    tool return early with the current ``run_id`` — call ``get_run_result``
+    afterwards to fetch the final report.
+
     Args:
         preset_name: Swarm preset name (e.g. 'investment_committee', 'quant_strategy_desk').
         variables: Required variables for the preset (e.g. {"target": "AAPL.US", "market": "US"}).
+        wait_seconds: Maximum seconds to keep the MCP call open. Default 3600
+            (1 hour); the progress-notification keepalive means the transport
+            stays connected for the full budget.
+        start_only: If True, kick off the run and return immediately with
+            ``run_id`` + current status. Ignores ``wait_seconds``.
     """
+    import asyncio
     import time
     from src.swarm.runtime import SwarmRuntime
     from src.swarm.store import SwarmStore, swarm_runs_root
-    from src.swarm.models import RunStatus
 
     swarm_dir = swarm_runs_root()
     store = SwarmStore(base_dir=swarm_dir)
@@ -387,32 +404,62 @@ def run_swarm(preset_name: str, variables: dict[str, str]) -> str:
     except ValueError as exc:
         return json.dumps({"status": "error", "error": f"DAG validation failed: {exc}"}, ensure_ascii=False)
 
-    # Poll until complete (max 30 minutes)
-    for _ in range(360):
-        time.sleep(5)
-        current = store.load_run(run.id)
-        if current is None:
-            return json.dumps({"status": "error", "error": "Run record lost"}, ensure_ascii=False)
-        if current.status in (RunStatus.completed, RunStatus.failed, RunStatus.cancelled):
-            from src.swarm.serialization import run_level_error, serialize_task
-
-            tasks = [serialize_task(t) for t in current.tasks]
-            return json.dumps(
-                {
-                    "status": current.status.value,
-                    "preset": preset_name,
-                    "run_id": current.id,
-                    "final_report": current.final_report,
-                    "error": run_level_error(current),
-                    "tasks": tasks,
-                    "total_input_tokens": current.total_input_tokens,
-                    "total_output_tokens": current.total_output_tokens,
-                },
-                ensure_ascii=False,
-                indent=2,
-            )
+    if start_only or wait_seconds <= 0:
+        return json.dumps(
+            _build_run_payload(store, run.id, preset_name, timed_out=False),
+            ensure_ascii=False,
+            indent=2,
+        )
 
-    return json.dumps({"status": "error", "error": "Swarm timed out after 30 minutes"}, ensure_ascii=False)
+    # Surface the run_id immediately in a fixed-format progress message so a
+    # caller whose transport drops mid-run (or whose MCP client enforces a
+    # hard tool-call timeout that ignores progress notifications) can still
+    # recover the run via ``get_run_result(run_id)``. Parsers should match
+    # ``swarm_started run_id=<id>`` literally; later frames are free-form.
+    if ctx is not None:
+        try:
+            await ctx.report_progress(
+                progress=0,
+                total=1,
+                message=f"swarm_started run_id={run.id} preset={preset_name}",
+            )
+        except Exception:
+            pass
+
+    terminal = {"completed", "failed", "cancelled"}
+    started_at = time.monotonic()
+    deadline = started_at + wait_seconds
+    while True:
+        payload = _build_run_payload(store, run.id, preset_name, timed_out=False)
+        if payload["status"] == "error":
+            return json.dumps(payload, ensure_ascii=False)
+        if payload["status"] in terminal:
+            return json.dumps(payload, ensure_ascii=False, indent=2)
+
+        # Emit a progress frame every loop, NOT only on state change — MCP
+        # clients use these as transport keepalive. A long task that doesn't
+        # transition for 30 minutes still needs ticks or the client times out.
+        # ``elapsed`` keeps the message content fresh so dedup-on-message
+        # clients still see updates.
+        if ctx is not None:
+            tasks = payload.get("tasks") or []
+            total = max(1, len(tasks))
+            done = sum(1 for t in tasks if t.get("status") in terminal)
+            elapsed = int(time.monotonic() - started_at)
+            try:
+                await ctx.report_progress(
+                    progress=done,
+                    total=total,
+                    message=f"{done}/{total} tasks complete · {elapsed}s elapsed (run {run.id})",
+                )
+            except Exception:
+                pass
+
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            payload = _build_run_payload(store, run.id, preset_name, timed_out=True)
+            return json.dumps(payload, ensure_ascii=False, indent=2)
+        await asyncio.sleep(min(5.0, remaining))
 
 
 # ---------------------------------------------------------------------------
@@ -561,28 +608,63 @@ def _get_swarm_store():
     return SwarmStore(base_dir=swarm_dir)
 
 
-def _run_to_dict(run) -> dict:
+def _run_to_dict(run, *, timed_out: bool = False, is_stale: bool = False) -> dict:
+    """Public projection of a (live-hydrated) :class:`SwarmRun`.
+
+    ``timed_out`` flips on only for the ``run_swarm`` wait-budget path. It does
+    not change the run's actual status — callers can still see ``running`` and
+    fetch the final report later via :func:`get_run_result`.
+
+    ``is_stale`` is a read-only signal: ``True`` means the run is still
+    ``running`` but its events.jsonl has been silent past the per-run
+    threshold. No disk state is changed by setting this — the explicit
+    :func:`reap_stale_runs` tool is what finalizes a stale run.
+    """
     from src.swarm.serialization import run_level_error, serialize_task
 
     return {
         "run_id": run.id,
         "status": run.status.value,
         "preset": run.preset_name,
         "created_at": run.created_at,
+        "completed_at": run.completed_at,
         "error": run_level_error(run),
         "tasks": [serialize_task(t) for t in run.tasks],
         "final_report": run.final_report,
         "total_input_tokens": run.total_input_tokens,
         "total_output_tokens": run.total_output_tokens,
+        "timed_out": timed_out,
+        "is_stale": is_stale,
     }
 
 
+def _build_run_payload(store, run_id: str, preset_name: str | None, *, timed_out: bool) -> dict:
+    """Reconcile + project a run for the MCP response.
+
+    Used by ``run_swarm`` (polling + start_only). Returns a normal payload on
+    success and a ``{"status": "error", ...}`` envelope when the run record
+    disappears (mid-run directory wipe / sandbox eviction).
+    """
+    run = store.load_run(run_id)
+    if run is None:
+        return {"status": "error", "error": "Run record lost", "run_id": run_id}
+    reconciled = store.reconcile_run(run, write=True)
+    payload = _run_to_dict(
+        reconciled,
+        timed_out=timed_out,
+        is_stale=store.is_run_stale(reconciled),
+    )
+    if preset_name:
+        payload["preset"] = preset_name
+    return payload
+
+
 @mcp.tool
 def get_swarm_status(run_id: str) -> str:
     """Get the current status of a swarm run.
 
-    Returns status, task progress, and token usage for the specified run.
-    Use this to poll a long-running swarm without blocking.
+    Returns status, task progress, token usage, and an ``is_stale`` flag for
+    the specified run. Use this to poll a long-running swarm without blocking.
 
     Args:
         run_id: The run ID returned by run_swarm.
@@ -591,15 +673,22 @@ def get_swarm_status(run_id: str) -> str:
     run = store.load_run(run_id)
     if run is None:
         return json.dumps({"status": "error", "error": f"Run {run_id} not found"}, ensure_ascii=False)
-    return json.dumps(_run_to_dict(run), ensure_ascii=False, indent=2)
+    reconciled = store.reconcile_run(run, write=True)
+    return json.dumps(
+        _run_to_dict(reconciled, is_stale=store.is_run_stale(reconciled)),
+        ensure_ascii=False,
+        indent=2,
+    )
 
 
 @mcp.tool
 def get_run_result(run_id: str) -> str:
-    """Get the final report and task summaries of a completed swarm run.
+    """Get the final report and task summaries of a swarm run.
 
-    Returns the final_report text and per-task summaries. If the run is
-    still in progress, returns current status instead.
+    Reconciles the run on read: an orphaned ``running`` run whose host
+    process exited will be transitioned to its real terminal status
+    (``completed`` / ``failed`` / ``cancelled`` derived from the task
+    statuses), so the caller never sees a permanent zombie.
 
     Args:
         run_id: The run ID returned by run_swarm.
@@ -608,15 +697,18 @@ def get_run_result(run_id: str) -> str:
     run = store.load_run(run_id)
     if run is None:
         return json.dumps({"status": "error", "error": f"Run {run_id} not found"}, ensure_ascii=False)
-    return json.dumps(_run_to_dict(run), ensure_ascii=False, indent=2)
+    reconciled = store.reconcile_run(run, write=True)
+    payload = _run_to_dict(reconciled, is_stale=store.is_run_stale(reconciled))
+    payload["ready"] = payload["status"] in {"completed", "failed", "cancelled"}
+    return json.dumps(payload, ensure_ascii=False, indent=2)
 
 
 @mcp.tool
 def list_runs(limit: int = 20) -> str:
     """List recent swarm runs sorted by creation time (newest first).
 
-    Returns run IDs, presets, statuses, and creation timestamps.
-    Use get_run_result(run_id) to fetch full details for a specific run.
+    Each row includes task counts and an ``is_stale`` flag so callers can
+    spot abandoned runs without a follow-up status call.
 
     Args:
         limit: Maximum number of runs to return (default 20).
@@ -625,19 +717,46 @@ def list_runs(limit: int = 20) -> str:
     runs = store.list_runs(limit=limit)
     items = []
     for run in runs:
+        # write=True so a zombie listed alongside live runs gets finalized;
+        # the cost is bounded by ``limit`` (default 20) and most rows are
+        # already terminal — reconcile is a no-op for those.
+        reconciled = store.reconcile_run(run, write=True)
+        counts = {"total": len(reconciled.tasks)}
+        for t in reconciled.tasks:
+            counts[t.status.value] = counts.get(t.status.value, 0) + 1
         items.append(
             {
-                "run_id": run.id,
-                "preset": run.preset_name,
-                "status": run.status.value,
-                "created_at": run.created_at,
-                "total_input_tokens": run.total_input_tokens,
-                "total_output_tokens": run.total_output_tokens,
+                "run_id": reconciled.id,
+                "preset": reconciled.preset_name,
+                "status": reconciled.status.value,
+                "is_stale": store.is_run_stale(reconciled),
+                "created_at": reconciled.created_at,
+                "completed_at": reconciled.completed_at,
+                "task_counts": counts,
+                "total_input_tokens": reconciled.total_input_tokens,
+                "total_output_tokens": reconciled.total_output_tokens,
             }
         )
     return json.dumps(items, ensure_ascii=False, indent=2)
 
 
+@mcp.tool
+def reap_stale_runs() -> str:
+    """Mark every ``running`` run whose host process died as ``failed``.
+
+    Walks the swarm store, applies the per-run stale threshold, and
+    finalizes any run that has gone silent past it (writes ``run.json`` +
+    ``tasks/*.json`` + appends a ``run_reaped`` event). Already-terminal
+    runs and still-alive runs are left untouched.
+
+    Returns:
+        JSON list of reaped run IDs (empty when nothing was stale).
+    """
+    store = _get_swarm_store()
+    reaped = store.reap_stale_running_runs()
+    return json.dumps({"reaped": reaped}, ensure_ascii=False, indent=2)
+
+
 # ---------------------------------------------------------------------------
 # Trade journal tool
 # ---------------------------------------------------------------------------