homelab-codex-ws/services/control-plane/tests/test_observer_container_events.py

"""Tests for observer handling of node-agent container-state events.

Regression suite for the "dead/crash-looping container = silent" bug: the
observer's process_event event→status/incident map handled service_recovered,
service_healthy, service_unhealthy and healthcheck_failed but NOT
containers_not_running (emitted by node-agent for exited/dead/crash-loop
containers, and by stability-agent).  A dead container therefore left the
service at its last "healthy" status with no incident, so the supervisor saw no
drift and produced no action — the operator was never alerted.

Covers:
  1. containers_not_running → status=unhealthy + incident with the right trigger_type
  2. the incident's trigger_type is one the supervisor already routes to a
     container_restart (end-to-end reconcile, no supervisor change required)
  3. service_healthy after containers_not_running → incident auto-resolved
  4. container_restarting / container_state_unexpected → observational: no
     incident, status not flipped to unhealthy, but a trace is left behind
  5. idempotency: reprocessing containers_not_running does not multiply incidents
"""
from __future__ import annotations

import json
import sys
import time
from pathlib import Path

import pytest

# Observer lives outside the control-plane package; add scripts/ to path.
sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "scripts"))
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))

from observer.observer import Observer  # noqa: E402
import supervisor as supervisor_module  # noqa: E402
from supervisor import Supervisor, CONTAINER_RESTART_TRIGGERS  # noqa: E402


# ---------------------------------------------------------------------------
# Observer fixture — redirect every runtime path into a per-test tmp_path.
# Mirrors the fixture in test_incident_lifecycle.py (incl. OBSERVER_STATE_FILE,
# which is derived from STATE_DIR at import time).
# ---------------------------------------------------------------------------

@pytest.fixture
def observer(tmp_path, monkeypatch) -> Observer:
    import observer.observer as obs_mod

    world = tmp_path / "world"
    state = tmp_path / "state"
    events = tmp_path / "events"
    logs = tmp_path / "logs"
    repo = tmp_path / "repo"
    for d in (world, state, events, logs, repo / "inventory", repo / "hosts"):
        d.mkdir(parents=True, exist_ok=True)

    (repo / "inventory" / "topology.yaml").write_text(
        "nodes:\n  piha:\n    roles: [infra]\n    connectivity: {}\n"
    )

    monkeypatch.setattr(obs_mod, "WORLD_DIR", world)
    monkeypatch.setattr(obs_mod, "STATE_DIR", state)
    monkeypatch.setattr(obs_mod, "EVENTS_DIR", events)
    monkeypatch.setattr(obs_mod, "LOGS_DIR", logs)
    monkeypatch.setattr(obs_mod, "INVENTORY_TOPOLOGY", repo / "inventory" / "topology.yaml")
    monkeypatch.setattr(obs_mod, "REPO_ROOT", repo)
    monkeypatch.setattr(obs_mod, "FAILED_EVENTS_DIR", state / "observer_failed_events")
    monkeypatch.setattr(obs_mod, "OBSERVER_STATE_FILE", state / "observer_checkpoint.json")
    return Observer()


def _container_event(etype, *, node="piha", service="paperless",
                     severity="high", ts=None, payload=None):
    return {
        "id": f"evt-{node}-{ts or int(time.time())}-{etype}-{service}",
        "type": etype,
        "node": node,
        "service": service,
        "severity": severity,
        "timestamp": ts or int(time.time()),
        "message": f"Container '{service}' {etype}",
        "payload": payload or {"container": service, "status": "exited"},
    }


# ---------------------------------------------------------------------------
# 1. containers_not_running → unhealthy + incident
# ---------------------------------------------------------------------------

def test_containers_not_running_sets_unhealthy(observer):
    observer.process_event(_container_event("containers_not_running"))
    svc = observer.world_state["services"]["piha/paperless"]
    assert svc["status"] == "unhealthy"
    assert svc["incident_id"] is not None


def test_containers_not_running_creates_incident_with_trigger_type(observer):
    ev = _container_event("containers_not_running")
    observer.process_event(ev)

    svc = observer.world_state["services"]["piha/paperless"]
    inc = observer.world_state["incidents"][svc["incident_id"]]
    assert inc["status"] == "active"
    assert inc["node"] == "piha"
    assert inc["service"] == "paperless"
    # trigger_type is the event type, which the supervisor routes to container_restart.
    assert inc["trigger_type"] == "containers_not_running"


@pytest.mark.parametrize("etype", ["containers_not_running", "healthcheck_failed"])
def test_incident_trigger_type_is_recognized_by_supervisor(observer, etype):
    """The trigger_type the observer stamps MUST be one the supervisor routes to
    a container_restart — otherwise remediation silently never fires (or, for
    healthcheck_failed pre-etap-1, dead-ended in the broken redeploy path)."""
    observer.process_event(_container_event(etype))
    svc = observer.world_state["services"]["piha/paperless"]
    inc = observer.world_state["incidents"][svc["incident_id"]]
    assert inc["trigger_type"] == etype
    assert inc["trigger_type"] in CONTAINER_RESTART_TRIGGERS


def test_containers_not_running_after_healthy_transitions_to_unhealthy(observer):
    """A container that was healthy and then dies must flip to unhealthy — the
    exact real-world sequence that used to stay stuck at 'healthy'."""
    observer.process_event(_container_event("service_healthy", severity="info"))
    assert observer.world_state["services"]["piha/paperless"]["status"] == "healthy"

    observer.process_event(_container_event("containers_not_running"))
    svc = observer.world_state["services"]["piha/paperless"]
    assert svc["status"] == "unhealthy"
    assert svc["incident_id"] is not None


# ---------------------------------------------------------------------------
# 2. End-to-end: observer output → supervisor generates container_restart
# ---------------------------------------------------------------------------

@pytest.mark.parametrize("etype", ["containers_not_running", "healthcheck_failed"])
def test_supervisor_generates_container_restart_from_observer_output(
    observer, tmp_path, monkeypatch, etype
):
    """Integration: run the real observer to produce world state for a failing
    container, then run the real supervisor.reconcile() against it and assert a
    container_restart action lands in pending/.

    healthcheck_failed routes to container_restart since etap 1 (2026-07-29,
    recon D14/D15): it used to fall through to redeploy, which is broken as
    wired in the executor, so those events dead-ended. A restart plausibly
    heals a failing healthcheck; redeploy returns once the executor works
    (etap 2)."""
    # 1. Observer ingests the failure event and writes world state to disk.
    observer.process_event(_container_event(etype))
    observer._save_world()

    # 2. Point the supervisor at a fresh tmp tree; copy observer world output in.
    actions = tmp_path / "sup_actions"
    events = tmp_path / "sup_events"
    world = tmp_path / "sup_world"
    repo = tmp_path / "sup_repo"
    for d in (actions, events, world, repo / "hosts" / "piha"):
        d.mkdir(parents=True, exist_ok=True)

    import observer.observer as obs_mod
    for name in ("services.json", "incidents.json", "nodes.json"):
        src = obs_mod.WORLD_DIR / name
        if src.exists():
            (world / name).write_text(src.read_text())

    # Declare paperless as a desired service on piha so the drift loop considers it.
    (repo / "hosts" / "piha" / "services.yaml").write_text(
        "host: piha\nservices:\n  paperless:\n    role: kb\n    exposure: lan\n"
    )

    monkeypatch.setattr(supervisor_module, "ACTIONS_DIR", actions)
    monkeypatch.setattr(supervisor_module, "EVENTS_DIR", events)
    monkeypatch.setattr(supervisor_module, "WORLD_DIR", world)
    monkeypatch.setattr(supervisor_module, "REPO_ROOT", repo)

    sup = Supervisor()
    sup.reconcile()

    # action_id carries a per-incident timestamp suffix (2026-08-26 fix: two
    # different incidents for the same node/service must never collide in
    # cancelled/completed/failed — see supervisor._generate_recommendation),
    # so match by prefix rather than an exact id.
    matches = list((actions / "pending").glob("container-restart-piha-paperless-*.json"))
    assert len(matches) == 1, "supervisor did not generate exactly one container_restart"
    action_path = matches[0]
    action = json.loads(action_path.read_text())
    assert action["action_id"] == action_path.stem
    assert action["type"] == "container_restart"
    assert action["node"] == "piha"
    assert action["service"] == "paperless"
    assert action["payload"]["reason"] == etype
    # Must not ALSO fall through to the (broken until etap 2) redeploy path.
    assert not (actions / "pending" / "redeploy-piha-paperless.json").exists()


# ---------------------------------------------------------------------------
# 3. Recovery: service_healthy after containers_not_running auto-resolves
# ---------------------------------------------------------------------------

def test_service_healthy_resolves_containers_not_running_incident(observer):
    observer.process_event(_container_event("containers_not_running"))
    svc = observer.world_state["services"]["piha/paperless"]
    inc_id = svc["incident_id"]
    assert observer.world_state["incidents"][inc_id]["status"] == "active"

    observer.process_event(_container_event("service_healthy", severity="info"))
    svc = observer.world_state["services"]["piha/paperless"]
    assert svc["status"] == "healthy"
    assert svc["incident_id"] is None
    assert observer.world_state["incidents"][inc_id]["status"] == "resolved"


# ---------------------------------------------------------------------------
# 4. Observational events — no incident, no false remediation, but a trace
# ---------------------------------------------------------------------------

@pytest.mark.parametrize("etype", ["container_restarting", "container_state_unexpected"])
def test_observational_events_do_not_create_incident(observer, etype):
    observer.process_event(_container_event(etype, severity="low"))
    svc = observer.world_state["services"]["piha/paperless"]
    # No incident, and status is NOT flipped to unhealthy (would trigger a redeploy).
    assert svc.get("incident_id") is None
    assert svc["status"] != "unhealthy"
    assert not observer.world_state["incidents"]


@pytest.mark.parametrize("etype", ["container_restarting", "container_state_unexpected"])
def test_observational_events_leave_a_trace(observer, etype):
    """They must be visible in world state, not vanish into the void."""
    observer.process_event(_container_event(etype, severity="medium"))
    svc = observer.world_state["services"]["piha/paperless"]
    obs_note = svc.get("last_observation")
    assert obs_note is not None
    assert obs_note["type"] == etype


def test_container_restarting_does_not_clear_existing_unhealthy(observer):
    """A container that is already unhealthy (dead) and then reports a transient
    'restarting' must NOT be downgraded to healthy — the incident stands until a
    real service_healthy arrives."""
    observer.process_event(_container_event("containers_not_running"))
    svc = observer.world_state["services"]["piha/paperless"]
    assert svc["status"] == "unhealthy"
    inc_id = svc["incident_id"]

    observer.process_event(_container_event("container_restarting", severity="low"))
    svc = observer.world_state["services"]["piha/paperless"]
    assert svc["status"] == "unhealthy"
    assert svc["incident_id"] == inc_id
    assert observer.world_state["incidents"][inc_id]["status"] == "active"


# ---------------------------------------------------------------------------
# 5. Idempotency — reprocessing must not multiply incidents
# ---------------------------------------------------------------------------

def test_repeated_containers_not_running_does_not_multiply_incidents(observer):
    ev = _container_event("containers_not_running", ts=1784000000)
    observer.process_event(ev)
    observer.process_event(ev)
    observer.process_event(ev)

    active = [i for i in observer.world_state["incidents"].values()
              if i["status"] == "active"]
    assert len(active) == 1
    # Correlation bumps occurrence_count rather than opening new incidents.
    assert active[0]["occurrence_count"] == 3