atlas-iac/testing/tests/test_hermes_handoff_json_rules.py

"""Contracts for the JSON-shaped handoff acceptance evaluators."""

from __future__ import annotations

import datetime as dt
import json

import pytest

from testing.tests.test_hermes_handoff_support import (
    load_handoff_module,
    outcome,
    spec,
    step,
)

evaluators = load_handoff_module("hermes_handoff_evaluators")
model = load_handoff_module("hermes_handoff_model")

NOW = dt.datetime(2026, 8, 17, 12, 0, 0, tzinfo=dt.timezone.utc)


def evaluate(built, payload):
    body = payload if isinstance(payload, str) else json.dumps(payload)
    return evaluators.evaluate(built, {"s": outcome(stdout=body)})


def one(rule: str, expect: dict):
    return spec(rule, {"step": "s", **expect}, steps=(step("s"),))


def test_json_field_compares_paths_against_expected_values() -> None:
    built = one(
        "json_field", {"fields": {"authenticated": True, "transport": ["a", "b"]}}
    )

    assert (
        evaluate(built, {"authenticated": True, "transport": "b"}).status == model.PASS
    )

    wrong = evaluate(built, {"authenticated": False, "transport": "c"})
    assert wrong.status == model.FAIL
    assert "authenticated=False" in wrong.reason and "transport='c'" in wrong.reason

    assert evaluate(built, {"authenticated": True}).status == model.NOT_RUN


def test_json_field_skips_rather_than_guesses_on_unusable_output() -> None:
    built = one("json_field", {"fields": {"a": 1}})

    assert evaluate(built, "not json").status == model.NOT_RUN
    failed = evaluators.evaluate(built, {"s": outcome(returncode=1)})
    assert failed.status == model.NOT_RUN
    assert evaluate(built, '{"a": 1, "a": 1}').status == model.NOT_RUN
    assert evaluate(built, '{"a": NaN}').status == model.NOT_RUN


def test_json_equality_does_not_confuse_booleans_and_integers() -> None:
    built = one("json_field", {"fields": {"enabled": True}})
    assert evaluate(built, {"enabled": 1}).status == model.FAIL
    array = one("all_of_field", {"fields": {"enabled": False}})
    assert evaluate(array, [{"enabled": 0}]).status == model.FAIL


def test_json_numeric_enforces_inclusive_bounds_and_rejects_non_numbers() -> None:
    built = one("json_numeric", {"fields": {"latency_ms": {"min": 1, "max": 1000}}})

    assert evaluate(built, {"latency_ms": 500}).status == model.PASS
    assert evaluate(built, {"latency_ms": 1}).status == model.PASS

    low = evaluate(built, {"latency_ms": 0})
    assert low.status == model.FAIL and "below minimum" in low.reason

    high = evaluate(built, {"latency_ms": 5000})
    assert high.status == model.FAIL and "above maximum" in high.reason

    assert evaluate(built, {"latency_ms": True}).status == model.NOT_RUN
    assert evaluate(built, {"latency_ms": "fast"}).status == model.NOT_RUN
    assert evaluate(built, {}).status == model.NOT_RUN


def test_json_recent_treats_stale_evidence_as_a_failure() -> None:
    built = one("json_recent", {"now": NOW, "fields": {"checked_at": 3600}})

    fresh = evaluate(built, {"checked_at": "2026-08-17T11:30:00Z"})
    assert (
        fresh.status == model.PASS
        and fresh.evidence["checked_at"]["age_seconds"] == 1800
    )

    stale = evaluate(built, {"checked_at": "2026-08-16T00:00:00Z"})
    assert stale.status == model.FAIL and "limit 3600s" in stale.reason

    naive = evaluate(built, {"checked_at": "2026-08-17T11:30:00"})
    assert naive.status == model.NOT_RUN
    assert evaluate(built, {"checked_at": "2026-08-17T12:00:01Z"}).status == model.FAIL

    assert evaluate(built, {"checked_at": "yesterday"}).status == model.NOT_RUN
    assert evaluate(built, {}).status == model.NOT_RUN
    invalid_clock = one("json_recent", {"now": "now", "fields": {"checked_at": 1}})
    assert (
        evaluate(invalid_clock, {"checked_at": "2026-08-17T11:30:00Z"}).status
        == model.NOT_RUN
    )


def test_all_of_field_selects_keys_and_reports_every_mismatch() -> None:
    built = one(
        "all_of_field",
        {"key_field": "number", "keys": [14, 15], "fields": {"merged": True}},
    )
    payload = [
        {"number": 14, "merged": True},
        {"number": 15, "merged": False},
        {"number": 16, "merged": False},
    ]

    evaluation = evaluate(built, payload)

    assert evaluation.status == model.FAIL
    assert "15: merged=False" in evaluation.reason
    assert "16" not in evaluation.reason
    assert evaluation.evidence["examined"] == [14, 15]


def test_all_of_field_skips_when_an_expected_key_is_absent_or_nothing_matched() -> None:
    keyed = one(
        "all_of_field",
        {"key_field": "number", "keys": [14], "fields": {"merged": True}},
    )
    absent = evaluate(keyed, [{"number": 99, "merged": True}])
    assert absent.status == model.NOT_RUN and "expected keys absent" in absent.reason

    unkeyed = one("all_of_field", {"fields": {"status": ["done"]}})
    assert evaluate(unkeyed, []).status == model.NOT_RUN
    assert evaluate(unkeyed, {"not": "an array"}).status == model.NOT_RUN


def test_all_of_field_skips_when_the_named_array_path_is_absent() -> None:
    built = spec(
        "all_of_field",
        {"step": "s", "array": "items", "fields": {"a": 1}},
        steps=(step("s"),),
    )

    evaluation = evaluate(built, {"other": []})

    assert evaluation.status == model.NOT_RUN
    assert "array path is absent" in evaluation.reason


def test_all_of_field_treats_an_absent_field_as_a_mismatch_not_a_crash() -> None:
    built = one("all_of_field", {"key_field": "name", "fields": {"suspended": False}})

    evaluation = evaluate(built, [{"name": "a"}])

    assert evaluation.status == model.FAIL
    assert "suspended=None" in evaluation.reason


def test_all_of_field_honours_exemptions_and_non_empty_paths() -> None:
    exempted = one(
        "all_of_field",
        {
            "key_field": "name",
            "exempt_keys": ["parked"],
            "fields": {"suspended": False},
        },
    )
    payload = [
        {"name": "parked", "suspended": True},
        {"name": "live", "suspended": False},
    ]
    assert evaluate(exempted, payload).status == model.PASS

    non_empty = one("all_of_field", {"key_field": "id", "non_empty": ["title"]})
    assert evaluate(non_empty, [{"id": "a", "title": "work"}]).status == model.PASS
    blank = evaluate(non_empty, [{"id": "a", "title": ""}, {"id": "b"}])
    assert blank.status == model.FAIL and "is empty" in blank.reason


def test_json_record_captures_named_paths_or_skips() -> None:
    built = one(
        "json_record", {"record": {"numbers": "[].number", "refs": "[].head.ref"}}
    )
    payload = [
        {"number": 14, "head": {"ref": "a"}},
        {"number": 15, "head": {"ref": "b"}},
    ]

    evaluation = evaluate(built, payload)
    assert evaluation.status == model.PASS
    assert evaluation.evidence == {"numbers": [14, 15], "refs": ["a", "b"]}

    missing = evaluate(built, [{"number": 14}])
    assert missing.status == model.NOT_RUN and "absent" in missing.reason


def routing_log(*records: dict) -> str:
    return "\n".join(json.dumps(record) for record in records)


def routing_spec(**extra):
    return spec(
        "routing_evidence",
        {
            "step": "s",
            "providers": ("codex", "claude"),
            "efforts": ("high", "xhigh"),
            "lanes": ("route", "worker"),
            "require_fallback_evidence": True,
            **extra,
        },
        steps=(step("s"),),
    )


FULL_LOG = routing_log(
    {"ts": "2026-08-17T11:00:00Z", "model": "route/codex/sol/high"},
    {"ts": "2026-08-17T11:30:00Z", "model": "route/claude/opus/xhigh"},
    {
        "ts": "2026-08-17T11:45:00Z",
        "model": "worker/codex/sol/xhigh",
        "fallback_reason": "unavailable",
    },
    {"ts": "2026-08-17T11:50:00Z", "model": "qwen2.5:14b", "tier": "classifier"},
)


def test_routing_evidence_summarises_lanes_providers_efforts_and_fallbacks() -> None:
    evaluation = evaluate(routing_spec(), FULL_LOG)

    assert evaluation.status == model.PASS
    assert evaluation.evidence["providers"] == ["claude", "codex"]
    assert evaluation.evidence["efforts"] == ["high", "xhigh"]
    assert evaluation.evidence["lanes"] == ["route", "worker"]
    assert evaluation.evidence["fallback_reasons"] == ["unavailable"]
    assert evaluation.evidence["records"] == 4 and evaluation.evidence["routed"] == 3


def test_routing_evidence_names_exactly_what_the_window_did_not_show() -> None:
    partial = routing_log(
        {"ts": "2026-08-17T11:00:00Z", "model": "route/codex/sol/high"}
    )

    evaluation = evaluate(routing_spec(), partial)

    assert evaluation.status == model.FAIL
    assert "provider:claude" in evaluation.reason
    assert "effort:xhigh" in evaluation.reason
    assert "lane:worker" in evaluation.reason
    assert "fallback:none-observed" in evaluation.reason


def test_routing_evidence_rejects_a_clipped_tail_and_empty_input() -> None:
    clipped = '{"ts": "2026-08-17T11:0\n' + FULL_LOG

    assert evaluate(routing_spec(), clipped).status == model.NOT_RUN
    assert evaluate(routing_spec(), "").status == model.NOT_RUN
    assert evaluate(routing_spec(), "garbage\n").status == model.NOT_RUN


def test_routing_evidence_can_require_freshness() -> None:
    fresh = evaluate(routing_spec(max_age_seconds=3600, now=NOW), FULL_LOG)
    assert fresh.status == model.PASS
    assert max(fresh.evidence["age_seconds"].values()) <= 3600

    stale = evaluate(
        routing_spec(max_age_seconds=60, now=NOW + dt.timedelta(days=2)), FULL_LOG
    )
    assert stale.status == model.FAIL and "freshness" in stale.reason

    undated = routing_log(
        {"model": "route/codex/sol/high"}, {"model": "route/claude/opus/xhigh"}
    )
    unknown = evaluate(routing_spec(max_age_seconds=60, now=NOW), undated)
    assert unknown.status == model.NOT_RUN


def test_routing_freshness_is_required_per_provider_not_only_globally() -> None:
    rows = routing_log(
        {"ts": "2026-08-15T11:00:00Z", "model": "route/claude/opus/high"},
        {"ts": "2026-08-17T11:59:00Z", "model": "route/codex/sol/xhigh"},
        {
            "ts": "2026-08-17T11:59:00Z",
            "model": "worker/codex/sol/high",
            "fallback_reason": "down",
        },
    )
    evaluation = evaluate(routing_spec(max_age_seconds=3600, now=NOW), rows)
    assert evaluation.status == model.FAIL
    assert "freshness:provider:claude" in evaluation.reason
    missing_dimension = evaluate(
        routing_spec(max_age_seconds=3600, now=NOW),
        routing_log(
            {"ts": "2026-08-17T11:59:00Z", "model": "route/codex/sol/high"},
            {
                "ts": "2026-08-17T11:59:00Z",
                "model": "worker/codex/sol/xhigh",
                "fallback_reason": "down",
            },
        ),
    )
    assert missing_dimension.status == model.FAIL


@pytest.mark.parametrize(
    "body",
    [
        '["not-an-object"]',
        '{"ts":"2026-08-17T12:00:01Z","model":"route/codex/sol/high"}',
        '{"ts":"2026-08-17T11:00:00Z"}',
        '{"ts":"2026-08-17T11:00:00Z","model":"bad"}',
        '{"ts":"2026-08-17T11:00:00Z","model":"route/codex/sol/high","fallback_reason":0}',
    ],
)
def test_routing_rows_fail_closed_on_structural_defects(body: str) -> None:
    assert evaluate(routing_spec(now=NOW), body).status in {model.FAIL, model.NOT_RUN}
    invalid_clock = routing_spec(now="now")
    assert evaluate(invalid_clock, FULL_LOG).status == model.NOT_RUN


def flux_spec(*expected: str):
    return spec(
        "flux_health",
        {"step": "s", "expected_suspensions": expected},
        steps=(step("s"),),
    )


def test_flux_health_separates_parked_objects_from_broken_ones() -> None:
    rows = (
        "flux-system/core\t1\t1\tfalse\tReady=True,Healthy=True,\n"
        "flux-system/parked\t1\t1\ttrue\tReady=False,\n"
        "flux-system/broken\t1\t1\tfalse\tReady=False,Healthy=False,\n"
    )

    evaluation = evaluators.evaluate(
        flux_spec("flux-system/parked"), {"s": outcome(stdout=rows)}
    )

    assert evaluation.status == model.FAIL
    assert evaluation.evidence["unhealthy"] == ["flux-system/broken"]
    assert evaluation.evidence["unexpected_suspensions"] == []
    assert "unhealthy: flux-system/broken" in evaluation.reason


def test_flux_health_flags_a_suspension_nobody_declared() -> None:
    rows = "flux-system/core\t1\t1\tfalse\tReady=True,\nflux-system/parked\t1\t1\ttrue\tReady=False,\n"

    evaluation = evaluators.evaluate(flux_spec(), {"s": outcome(stdout=rows)})

    assert evaluation.status == model.FAIL
    assert evaluation.evidence["unexpected_suspensions"] == ["flux-system/parked"]


def test_flux_health_fails_closed_on_in_flight_or_unknown_conditions() -> None:
    rows = (
        "flux-system/reconciling\t1\t1\tfalse\tReconciling=True,Ready=False,\n"
        "flux-system/degraded\t1\t1\tfalse\tReady=False,Healthy=True,\n"
        "flux-system/waiting\t1\t1\tfalse\tReady=Unknown,\n"
    )

    evaluation = evaluators.evaluate(flux_spec(), {"s": outcome(stdout=rows)})

    assert evaluation.status == model.FAIL
    assert evaluation.evidence["total"] == 3


def test_blank_rows_in_a_projection_are_ignored_by_both_log_parsers() -> None:
    padded = "\n\n" + FULL_LOG + "\n\n"
    assert evaluate(routing_spec(), padded).status == model.PASS

    rows = "\n\nflux-system/core\t1\t1\tfalse\tReady=True,\n\n"
    evaluation = evaluators.evaluate(flux_spec(), {"s": outcome(stdout=rows)})
    assert evaluation.status == model.PASS and evaluation.evidence["total"] == 1


def test_flux_health_skips_on_an_empty_or_failed_projection() -> None:
    assert (
        evaluators.evaluate(flux_spec(), {"s": outcome(stdout="")}).status
        == model.NOT_RUN
    )
    assert (
        evaluators.evaluate(flux_spec(), {"s": outcome(returncode=1)}).status
        == model.NOT_RUN
    )
    malformed = "flux-system/core\t1\t0\tfalse\tReady=True,\n"
    assert (
        evaluators.evaluate(flux_spec(), {"s": outcome(stdout=malformed)}).status
        == model.FAIL
    )


@pytest.mark.parametrize(
    "row",
    [
        "too\tfew",
        "BAD NAME\t1\t1\tfalse\tReady=True,",
        "flux-system/core\tx\t1\tfalse\tReady=True,",
        "flux-system/core\t1\t1\tmaybe\tReady=True,",
        "flux-system/core\t1\t1\tfalse\tbroken,",
        "flux-system/core\t1\t1\tfalse\tReady=Yes,",
        "flux-system/core\t1\t1\tfalse\tReady=True,Ready=True,",
    ],
)
def test_flux_rows_reject_malformed_columns_names_generations_and_conditions(
    row: str,
) -> None:
    assert (
        evaluators.evaluate(flux_spec(), {"s": outcome(stdout=row)}).status
        == model.NOT_RUN
    )