fix(agent): route host pressure through bounded diagnosis

apps/api/tests/test_host_pressure_ansible_catalog.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import yaml
+
+from src.services.awooop_ansible_audit_service import (
+    _catalog_hints,
+    get_ansible_catalog_item,
+)
+
+ROOT = Path(__file__).resolve().parents[3]
+PLAYBOOK = ROOT / "infra" / "ansible" / "playbooks" / "110-host-pressure-readonly.yml"
+
+
+def test_host_pressure_alert_prefers_dedicated_no_write_catalog() -> None:
+    hints = _catalog_hints(
+        {
+            "alertname": "Host110SustainedModeratePressure",
+            "alert_category": "host_resource",
+            "affected_services": ["host-pressure-controller", "gitea_service"],
+            "signals": [
+                {
+                    "alert_name": "Host110SustainedModeratePressure",
+                    "labels": {
+                        "component": "host-pressure-controller",
+                        "host": "110",
+                    },
+                }
+            ],
+        },
+        None,
+    )
+
+    candidates = hints["candidates"]
+    assert candidates[0]["catalog_id"] == "ansible:110-host-pressure-readonly"
+    generic = next(row for row in candidates if row["catalog_id"] == "ansible:110-devops")
+    assert candidates[0]["match_score"] > generic["match_score"]
+    assert candidates[0]["auto_apply_enabled"] is False
+    assert candidates[0]["no_write_only"] is True
+
+
+def test_generic_110_fallback_remains_below_exact_host_pressure_route() -> None:
+    hints = _catalog_hints(
+        {
+            "alertname": "Host110SustainedModeratePressure",
+            "affected_services": ["host-pressure-controller"],
+        },
+        None,
+    )
+
+    candidates = hints["candidates"]
+    dedicated = next(
+        row
+        for row in candidates
+        if row["catalog_id"] == "ansible:110-host-pressure-readonly"
+    )
+    generic = next(
+        row for row in candidates if row["catalog_id"] == "ansible:110-devops"
+    )
+    assert dedicated["match_score"] > generic["match_score"]
+    assert generic["matched_keywords"] == ["110"]
+
+
+def test_host_pressure_playbook_is_bounded_and_read_only() -> None:
+    payload = yaml.safe_load(PLAYBOOK.read_text(encoding="utf-8"))
+    text = PLAYBOOK.read_text(encoding="utf-8")
+    catalog = get_ansible_catalog_item("ansible:110-host-pressure-readonly")
+
+    assert payload[0]["hosts"] == "host_110"
+    assert payload[0]["become"] is False
+    assert catalog is not None
+    assert catalog["auto_apply_enabled"] is False
+    assert catalog["no_write_only"] is True
+    assert "host-sustained-load-controller.py" in text
+    assert "gitea-queue-hook-backlog-playbook.py" in text
+    for forbidden in (
+        "systemctl restart",
+        "docker restart",
+        "kill -9",
+        "kubectl delete",
+        "reboot",
+    ):
+        assert forbidden not in text.lower()

apps/api/tests/test_repair_candidate_service.py

@@ -531,6 +531,34 @@ def test_promotion_summary_marks_controlled_runtime_when_apply_gate_passes() ->
     )
 
 
+def test_host_pressure_controller_is_never_drafted_as_systemd_restart() -> None:
+    service = RepairCandidateService()
+
+    template = service._build_playbook_draft_template(
+        coverage_gap={
+            "target_kind": "host_service",
+            "coverage_key": "host110sustainedmoderatepressure:host-pressure-controller",
+            "blocking_stage": "service_playbook_coverage",
+            "required_mcp_evidence_refs": ["metrics_or_logs_window"],
+        },
+        blockers=["playbook_observe_only"],
+        lane="promote_diagnostic_to_repair_playbook",
+        alertname="Host110SustainedModeratePressure",
+        target_resource="host-pressure-controller",
+        namespace="",
+        evidence=None,
+        playbook=None,
+    )
+
+    assert template["suggested_route"] == "host_pressure_readonly_after_owner_review"
+    assert template["repair_command_template"].startswith(
+        "/home/wooo/scripts/host-sustained-load-controller.py"
+    )
+    assert template["rollback_command_template"].startswith("no host mutation")
+    assert "systemctl restart" not in template["repair_command_template"]
+    assert "systemctl is-active" not in " ".join(template["verifier_plan_template"])
+
+
 @pytest.mark.asyncio
 async def test_postgres_slow_query_gap_prefills_database_owner_review_not_restart() -> None:
     incident = _incident()