Show & Tell: Phase 10.4 — ExecutionMonitor (async event queue, health scoring, and stall detection)

[web3guru888]

What is ExecutionMonitor?

ExecutionMonitor is the observability layer for Phase 10's goal execution pipeline. While PlanExecutor (#326) coordinates task dispatch, ExecutionMonitor watches every state transition in real time, computes a plan health score, detects stalled sub-tasks, and surfaces a MonitorView that the CognitiveCycle uses to decide whether to trigger the ReplanningEngine (Phase 10.5).

---

Architecture

PlanExecutor
  │
  │ emits ExecutionEvent (per state transition)
  ▼
asyncio.Queue[ExecutionEvent]
  │
  │ _consume() loop (background task)
  ▼
_apply(event)  ──►  _plans: dict[plan_id, dict[subtask_id, TaskProgress]]
                        │
                        ▼
                  _build_view()
                  ├── health_score  (SUCCESS / total_counted)
                  ├── stalled_ids   (RUNNING > stall_threshold_ms)
                  └── MonitorView   ◄── CognitiveCycle.view()
                        │
                        ▼
               CognitiveCycle._check_plan_health()
                  └── health < 0.5?  →  ReplanningEngine.trigger()

---

EventType Enum — 6 State Transitions

EventType	When emitted	State after
TASK_STARTED	Sub-task dispatched	RUNNING
TASK_COMPLETED	Dispatcher returns success	SUCCESS
TASK_FAILED	All retries exhausted	FAILED
TASK_RETRIED	Retry attempt begins	RUNNING
TASK_SKIPPED	Skipped due to upstream failure	SKIPPED
PLAN_COMPLETED	All waves finished	— (sets is_complete=True)

---

Health Score Formula

health = SUCCESS / max(1, SUCCESS + FAILED + SKIPPED + RUNNING)

RUNNING tasks count against health because they are uncertain — a running task might still fail. PENDING tasks (not yet started) are excluded.

Example: Diamond DAG

        A
       /       B   C
       \ /
        D

Execution events arrive in topological order:

Time	Event	State	health
t=0ms	A STARTED	RUNNING	0/1 = 0.00
t=50ms	A COMPLETED	SUCCESS	1/1 = 1.00
t=50ms	B,C STARTED	RUNNING	1/3 = 0.33
t=120ms	B COMPLETED	SUCCESS	2/3 = 0.67
t=130ms	C COMPLETED	SUCCESS	3/3 = 1.00
t=130ms	D STARTED	RUNNING	3/4 = 0.75
t=180ms	D COMPLETED	SUCCESS	4/4 = 1.00

---

Stall Detection Code

async def stalled_plans(self, stall_threshold_ms: float) -> list[MonitorView]:
    now_ms = time.time() * 1000
    stalled_views = []
    async with self._lock:
        for plan_id, tasks in self._plans.items():
            stalled_ids = frozenset(
                t.subtask_id for t in tasks.values()
                if t.state == ExecutionState.RUNNING
                and (now_ms - t.started_at_ms) > stall_threshold_ms
            )
            if stalled_ids:
                _STALLS_TOTAL.inc(len(stalled_ids))
                view = self._build_view(plan_id, tasks, stalled_ids)
                stalled_views.append(view)
    return stalled_views

A stall is: state == RUNNING AND elapsed_ms > stall_threshold_ms. Default threshold: 30 seconds.

---

5 Prometheus Metrics

Metric	Type	Labels
execmon_events_total	Counter	event_type
execmon_plans_tracked	Gauge	—
execmon_stalls_detected_total	Counter	—
execmon_health_score	Gauge	plan_id
execmon_view_latency_seconds	Histogram	—

PromQL

# Stall rate
rate(execmon_stalls_detected_total[5m]) * 60

# Degraded plans (health < 0.5)
count(execmon_health_score < 0.5)

# Event throughput
sum(rate(execmon_events_total[1m])) by (event_type)

---

Open Questions

1. Push vs Pull: Should PlanExecutor push events directly to ExecutionMonitor.ingest(), or should ExecutionMonitor poll PlanExecutor.snapshot() on a timer? Async queue push is lower latency but creates a dependency.

2. Multi-monitor fan-out: Can multiple ExecutionMonitor instances subscribe to the same PlanExecutor for different consumers (e.g., one for UI, one for replanning)?

3. Persistence: Should completed MonitorView snapshots be written to the Blackboard for post-mortem analysis, or kept in-memory with TTL eviction only?

Full spec: #330