orchestrator/agent-team/tests/test_p3_failsafe_serve_default.py

"""P3 fail-safe serve default (design Decision 5; UNIT 0e).

The bound P3 build→verify + dispatch wiring is the new production ``serve``
default, but its factories are called EAGERLY at graph-build and the live
dispatch factory RAISES when ``AGENT_TEAM_REPO_OWNER`` / ``AGENT_TEAM_REPO_NAME``
are unset. These tests pin the fail-safe contract:

* :func:`agent_team.coordinator._p3_env_is_configured` truth table.
* :func:`agent_team.coordinator.failsafe_production_p3_wiring` —
  env-unset degrades to the INERT ``(None, None)`` pair with exactly ONE WARNING
  and ONE ``#agent-team`` inert notice via the lifecycle ``notify`` sink (never
  the ALARM path), and NEVER raises; env-set returns the live wiring pair.
* a Coordinator built with the env-unset (inert) result sets up cleanly and an
  approved task settles at the P2 BUILD terminus (no P3 nodes, no dispatch, no
  exception) — i.e. a task that would reach P3 parks short of build/dispatch.
* ``run-team.py serve`` binds the fail-safe pair; ``start`` / ``intake`` do not.
"""

from __future__ import annotations

import logging
from pathlib import Path
from typing import Any

import pytest

from agent_team import graph as graph_mod
from agent_team.coordinator import (
    Coordinator,
    _p3_env_is_configured,
    default_dispatch_node_factory,
    failsafe_production_p3_wiring,
)

try:  # InMemorySaver is the modern name; fall back on older langgraph.
    from langgraph.checkpoint.memory import InMemorySaver as _Saver
except ImportError:  # pragma: no cover - older langgraph
    from langgraph.checkpoint.memory import MemorySaver as _Saver


_P3_ENV = ("AGENT_TEAM_REPO_OWNER", "AGENT_TEAM_REPO_NAME")
_TOKEN_ENV = ("AGENT_TEAM_CI_READ_TOKEN", "GITHUB_TOKEN")


@pytest.fixture
def clean_p3_env(monkeypatch: pytest.MonkeyPatch) -> None:
    """Start each test from a fully unset P3 environment."""
    for name in (*_P3_ENV, *_TOKEN_ENV):
        monkeypatch.delenv(name, raising=False)


# --------------------------------------------------------------------------- #
# _p3_env_is_configured truth table
# --------------------------------------------------------------------------- #


def test_env_unset_is_not_configured(clean_p3_env: None) -> None:
    assert _p3_env_is_configured() is False


def test_owner_repo_without_token_is_not_configured(
    clean_p3_env: None, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setenv("AGENT_TEAM_REPO_OWNER", "org")
    monkeypatch.setenv("AGENT_TEAM_REPO_NAME", "repo")
    # No CI-read token -> the verifier could never read an authenticated pass.
    assert _p3_env_is_configured() is False


def test_token_without_owner_repo_is_not_configured(
    clean_p3_env: None, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setenv("AGENT_TEAM_CI_READ_TOKEN", "ghp_test")
    assert _p3_env_is_configured() is False


def test_owner_repo_and_ci_token_is_configured(
    clean_p3_env: None, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setenv("AGENT_TEAM_REPO_OWNER", "org")
    monkeypatch.setenv("AGENT_TEAM_REPO_NAME", "repo")
    monkeypatch.setenv("AGENT_TEAM_CI_READ_TOKEN", "ghp_test")
    assert _p3_env_is_configured() is True


def test_github_token_fallback_satisfies_ci_token(
    clean_p3_env: None, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setenv("AGENT_TEAM_REPO_OWNER", "org")
    monkeypatch.setenv("AGENT_TEAM_REPO_NAME", "repo")
    monkeypatch.setenv("GITHUB_TOKEN", "ghp_fallback")
    assert _p3_env_is_configured() is True


def test_blank_env_values_are_not_configured(
    clean_p3_env: None, monkeypatch: pytest.MonkeyPatch
) -> None:
    # Whitespace-only values must not count as configured (fail-closed).
    monkeypatch.setenv("AGENT_TEAM_REPO_OWNER", "   ")
    monkeypatch.setenv("AGENT_TEAM_REPO_NAME", "repo")
    monkeypatch.setenv("AGENT_TEAM_CI_READ_TOKEN", "ghp_test")
    assert _p3_env_is_configured() is False


# --------------------------------------------------------------------------- #
# failsafe_production_p3_wiring — env-unset degrade
# --------------------------------------------------------------------------- #


def test_env_unset_returns_inert_pair_warns_and_notifies(
    clean_p3_env: None, caplog: pytest.LogCaptureFixture
) -> None:
    notices: list[str] = []

    with caplog.at_level(logging.WARNING, logger="agent_team.coordinator"):
        build_verify, dispatch = failsafe_production_p3_wiring(
            notify=lambda msg, **_: notices.append(msg)
        )

    # Inert P3: no build->verify subgraph, no dispatch.
    assert build_verify is None
    assert dispatch is None

    # Exactly one WARNING about the inert P3 wiring.
    inert_warnings = [
        r
        for r in caplog.records
        if r.levelno == logging.WARNING and "INERT" in r.getMessage()
    ]
    assert len(inert_warnings) == 1

    # Exactly one #agent-team inert notice via the (non-ALARM) notify sink.
    assert len(notices) == 1
    assert "INERT" in notices[0]


def test_env_unset_without_notify_does_not_raise(clean_p3_env: None) -> None:
    # A token-less / channel-less serve still comes up inert with no notify sink.
    build_verify, dispatch = failsafe_production_p3_wiring(notify=None)
    assert build_verify is None
    assert dispatch is None


def test_inert_notify_failure_is_swallowed(clean_p3_env: None) -> None:
    def _boom(_msg: str, **_kw: Any) -> None:
        raise RuntimeError("slack down")

    # The notify sink raising must not propagate out of serve-start.
    build_verify, dispatch = failsafe_production_p3_wiring(notify=_boom)
    assert build_verify is None
    assert dispatch is None


# --------------------------------------------------------------------------- #
# failsafe_production_p3_wiring — env-set binds live wiring
# --------------------------------------------------------------------------- #


def test_env_set_returns_live_wiring_pair(
    clean_p3_env: None, monkeypatch: pytest.MonkeyPatch
) -> None:
    monkeypatch.setenv("AGENT_TEAM_REPO_OWNER", "test-org")
    monkeypatch.setenv("AGENT_TEAM_REPO_NAME", "test-repo")
    monkeypatch.setenv("AGENT_TEAM_CI_READ_TOKEN", "ghp_test")

    notices: list[str] = []
    build_verify, dispatch = failsafe_production_p3_wiring(
        notify=lambda msg, **_: notices.append(msg)
    )

    # Live pair: both factories present, no inert notice emitted.
    assert callable(build_verify)
    assert callable(dispatch)
    assert notices == []

    # The dispatch factory is the env-reading production factory.
    assert dispatch is default_dispatch_node_factory

    # Each live factory builds without raising and yields the expected shapes.
    bv_tuple = build_verify()
    assert isinstance(bv_tuple, tuple) and len(bv_tuple) == 3
    assert all(callable(part) for part in bv_tuple)
    assert callable(dispatch())


# --------------------------------------------------------------------------- #
# Coordinator built with the inert result sets up + a P3-bound task does not
# dispatch / crash (it settles at the P2 BUILD terminus).
# --------------------------------------------------------------------------- #


def test_coordinator_with_inert_wiring_builds_and_task_parks_short_of_p3(
    clean_p3_env: None, tmp_path: Path
) -> None:
    """env-unset -> graph builds, coordinator constructs, an approved task settles
    at the P2 BUILD terminus with no P3 nodes, no dispatch, and no exception."""
    from agent_team.graph import resume_task, start_task
    from agent_team.nodes import review_loop
    from agent_team.task_model import Phase, PipelineState, TaskStatus

    build_verify, dispatch = failsafe_production_p3_wiring(notify=None)
    assert (build_verify, dispatch) == (None, None)

    saver = _Saver()
    saved_invoker = review_loop._review_invoker
    try:
        review_loop.set_review_invoker(lambda prompt, **kw: "VERDICT: APPROVE\nok")

        def _plan_stub(state: PipelineState) -> PipelineState:
            return PipelineState(
                plan={"title": "t", "scope": ["src"], "phases": ["P1"]},
                current_phase=Phase.REVIEW.value,
                status=TaskStatus.ACTIVE.value,
            )

        coord = Coordinator(
            db_path=tmp_path / "inert.db",
            transport=_FakeTransport(),
            build_clarify_node=lambda: graph_mod.clarify_node,
            build_plan_node=lambda: _plan_stub,
            review_wiring=lambda: (
                review_loop.bind_review_node(),
                review_loop.route_after_review,
            ),
            build_verify_wiring=build_verify,
            dispatch_node_wiring=dispatch,
            build_checkpointer=lambda _path: saver,
        )
        coord.setup()

        # No P3 nodes were wired (inert): the graph stops at the P2 terminus.
        nodes = coord.graph.get_graph().nodes
        assert graph_mod.BUILD_NODE not in nodes
        assert graph_mod.VERIFY_NODE not in nodes

        # An approved task runs to the P2 BUILD terminus without dispatching or
        # raising (it never reaches a live build/dispatch).
        thread_id, _ = start_task(coord.graph, transport="slack")
        final = resume_task(coord.graph, thread_id=thread_id, answer="scope is X")
        assert final["current_phase"] == Phase.BUILD.value
    finally:
        review_loop._review_invoker = saved_invoker


class _FakeTransport:
    """Minimal non-posting transport for the inert-wiring coordinator test."""

    def post_question(self, **_kwargs: Any) -> str:
        return "ref"

    def parse_answer(self, raw: Any) -> tuple[str, Any, str]:  # pragma: no cover
        raise NotImplementedError