Fix three orchestrator bugs found during extended framework testing

src/hyperloom/agents/robustness/decision/action_ladder.py

@@ -9,10 +9,11 @@
 2. **diagnose** (medium) — ``alert(severity="medium")`` carrying evidence.
 3. **recommend** (high) — ``alert(severity="high")`` plus, for some symptoms, a
    symptom-specific remediation intent: ``kill_task`` (stale_lease),
-   ``delegate(recover)`` (gpu_memory_leaked), ``delegate(report)``
-   (``deadline_*`` wind-down and ``recover_unsuccessful`` finalization), or
-   ``prune_branch`` (stuck / no-lever families in ``_PRUNE_SYMPTOMS``). Every
-   other HIGH symptom is strategic: the alert alone, and Orchestration decides.
+   ``delegate(recover)`` (gpu_memory_leaked, local_server_unreachable),
+   ``delegate(report)`` (``deadline_*`` wind-down and ``recover_unsuccessful``
+   finalization), or ``prune_branch`` (stuck / no-lever families in
+   ``_PRUNE_SYMPTOMS``). Every other HIGH symptom is strategic: the alert
+   alone, and Orchestration decides.
 
 Strategic suggestions ride the alert ``detail.suggestion`` field. A per-key
 cooldown (``Symptom.dedup_key`` × ``cooldown_ticks``) prevents inbox flooding.
@@ -21,6 +22,7 @@
 
 from __future__ import annotations
 
+import hashlib
 import logging
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING, Any, Iterable
@@ -303,6 +305,38 @@ def _recommend(self, sym: Symptom) -> list[Intent]:
                 )
             )
             return intents
+        # Dead inference server -> ``delegate(recover, force_gpu_cleanup=True)``.
+        # ``recover``'s owner-pattern kill list already covers the atom/Magpie
+        # server process names (Magpie, EngineCore); this is the same remedy
+        # as gpu_memory_leaked, just triggered by an unreachable health check
+        # instead of a VRAM leak. ("server_lifecycle", named in this
+        # symptom's own suggestion text, is an internal warm-reuse config
+        # helper, not a real dispatchable action — PolicyGate would reject it
+        # as unknown_action.)
+        if sym.name == "local_server_unreachable":
+            evidence = dict(sym.evidence) if isinstance(sym.evidence, dict) else {}
+            # `_server_unreachable` emits one symptom per unreachable probe
+            # target and marks all of them HIGH together, so a tick-only key
+            # would collide across targets: the first delegate creates the
+            # task and the rest come back as duplicate-idempotency
+            # PolicyDenied, which incorrectly books a working recovery as a
+            # repeated policy denial. Disambiguate with the target itself.
+            target = str(sym.subject.get("url") or evidence.get("url") or "unknown")
+            target_key = hashlib.sha1(target.encode("utf-8")).hexdigest()[:8]
+            intents.append(
+                build_delegate(
+                    action_name="recover",
+                    params={
+                        "reason": "local_server_unreachable",
+                        "force_gpu_cleanup": True,
+                        "evidence": evidence,
+                    },
+                    idempotency_key=(
+                        f"recover-server-unreachable-tick-{self._last_tick_index}-{target_key}"
+                    ),
+                )
+            )
+            return intents
         # Wall-clock wind-down: ``delegate(report)`` lands a deterministic
         # report in the remaining budget before the deadline supervisor
         # SIGTERMs work; ``recover_unsuccessful`` is the finalization path.

src/hyperloom/agents/robustness/role/envelope.py

@@ -174,6 +174,8 @@ class IntentType(str, Enum):
         "pending_escalate_hint",
         "last_consumed_escalate_hint",
         "last_consumed_escalate_hint_ts",
+        "last_discarded_escalate_hint",
+        "last_discarded_escalate_hint_ts",
         "plateau_overrides",
         # CLOSE phase sequencer flag.
         "close_sequence_done",

src/hyperloom/agents/robustness/signals/local_health.py

@@ -202,7 +202,7 @@ def _server_unreachable(data: SourceData, cfg: LocalHealthConfig) -> list[Sympto
                 subject={"url": url},
                 source="local",
                 suggestion=(
-                    "delegate(server_lifecycle) to restart the inference server"
+                    "delegate(recover, force_gpu_cleanup=True) to restart the inference server"
                     if severity is SymptomSeverity.HIGH
                     else "monitor; alert orchestration if it persists"
                 ),
@@ -342,7 +342,7 @@ def _log_error_symptoms(data: SourceData) -> list[Symptom]:
                 subject={"pattern": pattern},
                 source="local",
                 suggestion=(
-                    "delegate(server_lifecycle) or escalate strategy"
+                    "delegate(recover, force_gpu_cleanup=True) or escalate strategy"
                     if severity is SymptomSeverity.HIGH
                     else "review log evidence with RCA before further action"
                 ),

src/hyperloom/agents/robustness/tests/test_decision_action_ladder.py

@@ -1216,3 +1216,82 @@ async def test_gpu_memory_leaked_idempotency_key_advances_with_tick():
     second_delegate = next(i for i in second.intents if i.type is IntentType.DELEGATE)
     assert first_delegate.payload["idempotency_key"] == "recover-gpu-leak-tick-0"
     assert second_delegate.payload["idempotency_key"] == "recover-gpu-leak-tick-5"
+
+
+# local_server_unreachable -> delegate(recover)
+
+
+def _server_unreachable_symptom(url: str, evidence: dict | None = None) -> Symptom:
+    return Symptom(
+        name="local_server_unreachable",
+        severity=SymptomSeverity.HIGH,
+        summary=f"local server probe {url} status=down",
+        evidence=evidence or {"url": url},
+        subject={"url": url},
+        source="local",
+        suggestion="delegate(recover, force_gpu_cleanup=True) to restart the inference server",
+    )
+
+
+async def test_local_server_unreachable_emits_alert_and_delegate_recover():
+    """``local_server_unreachable`` must route to the real ``recover`` action,
+    not the non-dispatchable ``server_lifecycle`` its own suggestion text names.
+    """
+    ladder = ActionLadder()
+    out = await ladder.decide(
+        [_server_unreachable_symptom("http://127.0.0.1:8000/health")],
+        tick_index=4,
+        now_unix=1.0,
+    )
+    types = [i.type for i in out.intents]
+    assert types == [IntentType.ALERT, IntentType.DELEGATE]
+
+    delegate = out.intents[1]
+    assert delegate.payload["action_name"] == "recover"
+    assert delegate.payload["params"]["force_gpu_cleanup"] is True
+    assert delegate.payload["params"]["reason"] == "local_server_unreachable"
+    assert delegate.payload["idempotency_key"].startswith("recover-server-unreachable-tick-4-")
+
+
+async def test_local_server_unreachable_idempotency_key_disambiguates_targets():
+    """Two unreachable targets in the same tick must not collide on idempotency_key,
+    or the second delegate comes back as a duplicate-idempotency PolicyDenied
+    that pollutes repeated_policy_denied tracking instead of just recovering.
+    """
+    ladder = ActionLadder()
+    out = await ladder.decide(
+        [
+            _server_unreachable_symptom("http://127.0.0.1:8000/health"),
+            _server_unreachable_symptom("http://127.0.0.1:8001/health"),
+        ],
+        tick_index=4,
+        now_unix=1.0,
+    )
+    delegate_keys = [i.payload["idempotency_key"] for i in out.intents if i.type is IntentType.DELEGATE]
+    assert len(delegate_keys) == 2
+    assert len(set(delegate_keys)) == 2
+    assert all(key.startswith("recover-server-unreachable-tick-4-") for key in delegate_keys)
+
+
+async def test_local_server_unreachable_idempotency_key_stable_for_same_target():
+    """The same target re-firing (e.g. a later tick after cooldown) must derive
+    the same per-target suffix, since that's what makes the disambiguator
+    deterministic rather than a source of new spurious duplicates.
+    """
+    ladder = ActionLadder()
+    first = await ladder.decide(
+        [_server_unreachable_symptom("http://127.0.0.1:8000/health")],
+        tick_index=4,
+        now_unix=1.0,
+    )
+    second = await ladder.decide(
+        [_server_unreachable_symptom("http://127.0.0.1:8000/health")],
+        tick_index=9,
+        now_unix=2.0,
+    )
+    first_key = next(i.payload["idempotency_key"] for i in first.intents if i.type is IntentType.DELEGATE)
+    second_key = next(i.payload["idempotency_key"] for i in second.intents if i.type is IntentType.DELEGATE)
+    first_suffix = first_key.rsplit("-", 1)[-1]
+    second_suffix = second_key.rsplit("-", 1)[-1]
+    assert first_suffix == second_suffix
+    assert first_key != second_key

src/hyperloom/agents/robustness/tests/test_signals_local_health.py

@@ -206,6 +206,29 @@ def test_a_seen_server_is_recorded_in_the_evidence():
     assert matched and matched[0].evidence["server_process_seen"] is True
 
 
+def test_all_targets_down_suggests_the_real_dispatchable_action():
+    """The HIGH suggestion must not point at ``server_lifecycle`` -- PolicyGate
+    rejects it as unknown_action; action_ladder routes this symptom to a real
+    ``delegate(recover, force_gpu_cleanup=True)``, and the suggestion text
+    reaching the orchestration prompt must say so instead.
+
+    A live server process is included so the idle-server-detection added
+    upstream (a refused port with no server and no benchmark client behind it
+    is treated as an expected idle stretch, not a fault) doesn't suppress the
+    symptom this test exists to check.
+    """
+    data = SourceData(
+        local_processes=_live_server(),
+        local_server_health=[
+            {"url": "http://localhost:30000", "reachable": False, "status": "error"},
+        ],
+    )
+    out = evaluate_local_health_signals(_ctx(), data)
+    matched = next(s for s in out if s.name == "local_server_unreachable")
+    assert "server_lifecycle" not in matched.suggestion
+    assert "delegate(recover" in matched.suggestion
+
+
 def test_no_unreachable_targets_is_silent():
     data = SourceData(
         local_server_health=[
@@ -230,6 +253,14 @@ def test_runtimeerror_pattern_is_medium_severity():
     assert matched and matched[0].severity is SymptomSeverity.MEDIUM
 
 
+def test_oom_pattern_suggests_the_real_dispatchable_action():
+    data = SourceData(local_log_errors=[{"pattern": "CUDA out of memory", "line": "torch ... CUDA out of memory ..."}])
+    out = evaluate_local_health_signals(_ctx(), data)
+    matched = next(s for s in out if s.name == "log_error_pattern")
+    assert "server_lifecycle" not in matched.suggestion
+    assert "delegate(recover" in matched.suggestion
+
+
 def test_log_error_groups_samples_by_pattern():
     data = SourceData(local_log_errors=[{"pattern": "RuntimeError", "line": f"err {i}"} for i in range(5)])
     out = evaluate_local_health_signals(_ctx(), data)

src/hyperloom/inference_optimizer/tests/test_coordinator_async_batch2_unit.py

@@ -2018,6 +2018,82 @@ async def _entered(*, from_phase, to_phase):
     assert coord.shared_state.stop_reason == "target_reached"
 
 
+@pytest.mark.asyncio
+async def test_advance_phase_hint_survives_transition_toward_explore(coord: Coordinator, monkeypatch) -> None:
+    """A skip_to_kernel hint set during FRAMEWORK_AGENT must survive an unrelated
+    FRAMEWORK_AGENT -> EXPLORE transition, since exit_normal_explore (the hint's
+    only consumer) only ever checks it once the phase is EXPLORE.
+    """
+    import hyperloom.orchestrator.phases.machine_state as ps
+
+    coord.shared_state.phase = "FRAMEWORK_AGENT"
+    coord.shared_state.pending_escalate_hint = "skip_to_kernel"
+    monkeypatch.setattr(ps, "compute_next_phase", lambda *a, **k: ("EXPLORE", "framework_phase_done", {}))
+
+    async def _entered(*, from_phase, to_phase):
+        return None
+
+    monkeypatch.setattr(coord.phase_machine, "_on_phase_entered", _entered)
+    await coord._advance_phase_if_needed()
+    assert (coord.shared_state.phase or "").upper() == "EXPLORE"
+    assert coord.shared_state.pending_escalate_hint == "skip_to_kernel"
+
+
+@pytest.mark.asyncio
+async def test_advance_phase_hint_discarded_when_not_headed_to_explore(coord: Coordinator, monkeypatch) -> None:
+    """A pending hint is genuinely stale once the transition target isn't
+    EXPLORE -- it can never reach exit_normal_explore's check again -- so this
+    is the one case the unrelated-transition cleanup should still clear it.
+
+    A discard is not a consumption: it must land in last_discarded_escalate_hint,
+    not last_consumed_escalate_hint, which specifically means "this hint drove
+    a transition" and this one never did.
+    """
+    import hyperloom.orchestrator.phases.machine_state as ps
+
+    coord.shared_state.phase = "FRAMEWORK_AGENT"
+    coord.shared_state.pending_escalate_hint = "skip_to_kernel"
+    monkeypatch.setattr(ps, "compute_next_phase", lambda *a, **k: ("SWEEP", "some_other_reason", {}))
+
+    async def _entered(*, from_phase, to_phase):
+        return None
+
+    monkeypatch.setattr(coord.phase_machine, "_on_phase_entered", _entered)
+    await coord._advance_phase_if_needed()
+    assert (coord.shared_state.phase or "").upper() == "SWEEP"
+    assert coord.shared_state.pending_escalate_hint == ""
+    assert coord.shared_state.last_discarded_escalate_hint == "skip_to_kernel"
+    assert coord.shared_state.last_discarded_escalate_hint_ts
+    assert coord.shared_state.last_consumed_escalate_hint == ""
+
+
+@pytest.mark.asyncio
+async def test_advance_phase_hint_consumed_when_it_drove_the_transition(coord: Coordinator, monkeypatch) -> None:
+    """The complementary case: a hint-driven transition must record consumption,
+    not a discard, so the two are distinguishable in the breakdown.
+    """
+    import hyperloom.orchestrator.phases.machine_state as ps
+
+    coord.shared_state.phase = "EXPLORE"
+    coord.shared_state.pending_escalate_hint = "skip_to_kernel"
+    monkeypatch.setattr(
+        ps,
+        "compute_next_phase",
+        lambda *a, **k: ("KERNEL_AGENT", "skip_to_kernel", {"hint": "skip_to_kernel"}),
+    )
+
+    async def _entered(*, from_phase, to_phase):
+        return None
+
+    monkeypatch.setattr(coord.phase_machine, "_on_phase_entered", _entered)
+    await coord._advance_phase_if_needed()
+    assert (coord.shared_state.phase or "").upper() == "KERNEL_AGENT"
+    assert coord.shared_state.pending_escalate_hint == ""
+    assert coord.shared_state.last_consumed_escalate_hint == "skip_to_kernel"
+    assert coord.shared_state.last_consumed_escalate_hint_ts
+    assert coord.shared_state.last_discarded_escalate_hint == ""
+
+
 # -- _materialize_approved_proposal -----------------------------------------
 def _pending(action_name: str, payload: dict, msg_id: str = "prop-1"):
     from hyperloom.orchestrator.loop.coordinator import PendingProposal

src/hyperloom/inference_optimizer/tests/test_coordinator_runtime.py

@@ -121,6 +121,28 @@ async def test_sub_agent_runner_no_executor_fails(tmp_path):
     res = await sub.run_task(task)
     assert res.state == "failed"
     assert "no runner" in res.error
+    assert res.error_class == "no_executor"
+    db.close()
+
+
+@pytest.mark.asyncio
+async def test_sub_agent_runner_executor_exception_sets_error_class(tmp_path):
+    """A raised executor exception must not collapse into the generic
+    unknown_error gap bucket -- error_class carries the exception's own class name.
+    """
+    db = SqliteConnection(tmp_path / "x.db")
+    locks = ResourceLockManager(SqliteLeaseBackend(db))
+    tr = TaskRegistry(db)
+    sub = SubAgentRunner(locks, tr)
+
+    async def exe(ctx):
+        raise TimeoutError("benchmark server never came up")
+
+    sub.register_executor("bench_runner", exe)
+    task = await tr.create(kind="bench_runner", params={}, idempotency_key="k-y")
+    res = await sub.run_task(task)
+    assert res.state == "failed"
+    assert res.error_class == "TimeoutError"
     db.close()
 
 

src/hyperloom/inference_optimizer/tests/test_phase_state_machine.py

@@ -257,6 +257,9 @@ def test_compute_next_phase_no_kernel_skips_kernel_phase():
         stop_reason="",
         pending_escalate_hint="skip_to_kernel",
         explore_search={},
+        # At least one specialist round this cycle, required for skip_to_kernel
+        # to fire at all (see test_exit_normal_explore_skip_to_kernel_*).
+        specialist_rounds=[{"proposals_total": 1, "proposals_kept": 0}],
         optimization_stack=[{"action": "explore"}],
     )
     out = phase_state.compute_next_phase(state, kernel_enabled=False)
@@ -267,6 +270,57 @@ def test_compute_next_phase_no_kernel_skips_kernel_phase():
     assert evidence.get("passed_through_reason") == "plateau_explore"
 
 
+def test_exit_normal_explore_skip_to_kernel_requires_a_tested_round():
+    """A skip_to_kernel hint must not end EXPLORE with zero validated work.
+
+    Reproduces the cumulative_gain_validated=0.00% session: the hint arrived
+    before EXPLORE ever dispatched a specialist round this cycle, and must not
+    be honored until one actually has.
+    """
+    state = SimpleNamespace(
+        phase="EXPLORE",
+        phase_started_unix=1_000_000.0,
+        max_minutes=0,
+        phase_budget_pct={},
+        pending_escalate_hint="skip_to_kernel",
+        explore_search={},
+        specialist_rounds=[],
+        macro_cycle=0,
+        optimization_stack=[{"action": "explore"}],
+        _now_unix=lambda: 1_000_000.0,
+    )
+    out = phase_state.exit_normal_explore(
+        state,
+        force_exit_hours_remaining=0.0,
+        force_exit_budget_pct=0.0,
+    )
+    assert out is None
+
+
+def test_exit_normal_explore_skip_to_kernel_fires_once_a_round_ran():
+    state = SimpleNamespace(
+        phase="EXPLORE",
+        phase_started_unix=1_000_000.0,
+        max_minutes=0,
+        phase_budget_pct={},
+        pending_escalate_hint="skip_to_kernel",
+        explore_search={},
+        specialist_rounds=[{"proposals_total": 1, "proposals_kept": 0}],
+        macro_cycle=0,
+        optimization_stack=[{"action": "explore"}],
+        _now_unix=lambda: 1_000_000.0,
+    )
+    out = phase_state.exit_normal_explore(
+        state,
+        force_exit_hours_remaining=0.0,
+        force_exit_budget_pct=0.0,
+    )
+    assert out is not None
+    reason, evidence = out
+    assert reason == "plateau_explore"
+    assert evidence.get("hint") == "skip_to_kernel"
+
+
 def test_compute_next_phase_terminal_overrides_phase():
     state = SimpleNamespace(
         phase="EXPLORE",