atlas-iac/testing/tests/test_hermes_hux_policy_receipts.py

"""HUX-05 run budgets and cancellation receipts.

Security obligations exercised: SO-40 (budget state per run against the
effective policy; crossing a limit emits ``budget.exhausted``), SO-41 (a
receipt says ``cancelled`` only when the process registry is empty and
carries every side effect the hook reports; a second stop returns the same
receipt).
"""

from __future__ import annotations

import json
import sys
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parents[2]
FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation"
if str(FOUNDATION) not in sys.path:
    sys.path.insert(0, str(FOUNDATION))

from hux import audit, contracts, identity, store  # noqa: E402
from hux import events as hux_events  # noqa: E402
from hux.server import build_router  # noqa: E402

SCHEMAS = contracts.load_all()
HEADERS = {"X-Hermes-Tenant-Identity": "slot-3", "X-Hux-Subject": "usr_0123456789abcdef", "X-Hux-Surface": "chat"}
OTHER = {**HEADERS, "X-Hux-Subject": "usr_fedcba9876543210"}
WORKER = {**HEADERS, "X-Hux-Surface": "worker", "X-Hux-Trust": "worker", "X-Hux-Relay-Key": "wk"}
ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"])
CONV = "conv_0001abcd"


@pytest.fixture
def router(tmp_path):
    return build_router(tmp_path, {"HUX_FLAGS": ALL_ON, "HUX_WORKER_KEY": "wk"})


@pytest.fixture
def events(monkeypatch):
    seen: list[dict] = []
    monkeypatch.setattr(hux_events, "emit", lambda store, identity, conversation_id, kind, summary, **extra: seen.append({"conversation_id": conversation_id, "kind": kind, "summary": summary, **extra}))
    return seen


def call(router, method, path, body=None, headers=WORKER):
    raw = b"" if body is None else json.dumps(body).encode()
    response = router.dispatch(method, path, headers, raw)
    return response.status, response.body


def valid(record):
    problems = contracts.validate_record(record, SCHEMAS)
    assert problems == [], problems
    return record


# --- budgets ------------------------------------------------------------------------

def test_budget_starts_empty_against_the_effective_policy(router):
    status, body = call(router, "GET", "/hux/v1/runs/run_9f/budget")
    assert status == 200 and valid(body)["run_id"] == "run_9f"
    assert body["spent"] == {k: 0 for k in ("tokens", "tool_calls", "wall_clock_seconds", "delegations", "spend_units", "subagents")}
    assert body["limits"]["tokens_per_run"] == 200000 and body["exhausted"] == []
    assert "revision" not in body and not any(k.startswith("_") for k in body)


def test_increments_accumulate_and_exhaust_each_limit(router, events):
    call(router, "PUT", "/hux/v1/policy", {"scope": {"level": "global"}, "autonomy": "safe",
         "budgets": {"tokens_per_run": 100, "tool_calls_per_run": 3, "subagents_per_run": 0, "scope": {"paths": ["/work"]}}}, HEADERS)
    status, body = call(router, "POST", "/hux/v1/runs/run_9f/budget", {"conversation_id": CONV, "tokens": 60, "tool_calls": 1})
    assert status == 200 and valid(body)["spent"]["tokens"] == 60
    assert body["exhausted"] == ["subagents_per_run"], "a zero limit is exhausted before any spend"
    assert body["limits"] == {"tokens_per_run": 100, "tool_calls_per_run": 3, "subagents_per_run": 0}, "scope is policy detail, not a limit"
    status, body = call(router, "POST", "/hux/v1/runs/run_9f/budget", {"tokens": 40, "tool_calls": 2, "delegations": 5})
    assert body["spent"] == {"tokens": 100, "tool_calls": 3, "wall_clock_seconds": 0, "delegations": 5, "spend_units": 0, "subagents": 0}
    assert body["exhausted"] == ["tokens_per_run", "tool_calls_per_run", "subagents_per_run"]
    assert [e["kind"] for e in events] == ["budget.exhausted", "budget.exhausted"]
    assert events[1]["summary"] == "Budget exhausted: tokens_per_run, tool_calls_per_run" and events[1]["run_id"] == "run_9f"
    assert events[1]["conversation_id"] == CONV, "the conversation learned on the first report sticks"
    status, body = call(router, "POST", "/hux/v1/runs/run_9f/budget", {"tokens": 1})
    assert status == 200 and len(events) == 2, "already exhausted limits do not re-emit"
    assert call(router, "GET", "/hux/v1/runs/run_9f/budget")[1] == body


def test_budget_follows_the_conversation_policy(router):
    _, conversation = call(router, "POST", "/hux/v1/conversations", {"title": "Budgeted"}, HEADERS)
    call(router, "PUT", "/hux/v1/policy", {"scope": {"level": "conversation", "scope_id": conversation["id"]}, "autonomy": "safe", "budgets": {"spend_units": 1}}, HEADERS)
    status, body = call(router, "POST", "/hux/v1/runs/run_c/budget", {"conversation_id": conversation["id"], "spend_units": 1})
    assert body["limits"] == {"spend_units": 1} and body["exhausted"] == ["spend_units"]
    assert call(router, "GET", "/hux/v1/runs/run_c/budget")[1]["limits"] == {"spend_units": 1}
    assert call(router, "GET", "/hux/v1/runs/run_c/budget", headers=OTHER)[1]["spent"]["spend_units"] == 0


@pytest.mark.parametrize("body", [[], {"tokens": -1}, {"tokens": True}, {"tokens": "5"}, {"conversation_id": "x"}])
def test_budget_rejects_bad_bodies(router, body):
    status, error = call(router, "POST", "/hux/v1/runs/run_9f/budget", body)
    assert status == 400 and valid(error)["code"] == "invalid"


# --- stop receipts ----------------------------------------------------------------

def test_stop_writes_a_receipt_and_a_second_stop_returns_it(router, events, tmp_path):
    side_effects = [{"description": "Partial file left in workspace", "reverted": True, "evidence": {"kind": "file", "id": "notes.md"}}]
    status, body = call(router, "POST", "/hux/v1/runs/run_9f/stop", {"process_registry_empty": True, "side_effects": side_effects, "conversation_id": CONV})
    assert status == 201 and valid(body)["outcome"] == "cancelled"
    assert body["requested_by"] == {"type": "system", "id": "worker"}
    assert body["requested_at"] == body["acknowledged_at"] == body["completed_at"]
    assert body["side_effects"] == side_effects and body["conversation_id"] == CONV and "revision" not in body
    status, again = call(router, "POST", "/hux/v1/runs/run_9f/stop", {"process_registry_empty": False})
    assert (status, again) == (200, body)
    assert [e["kind"] for e in events] == ["run.cancelled"]
    assert events[0]["evidence"] == [{"kind": "run", "id": body["id"]}] and events[0]["run_id"] == "run_9f"
    rows = [(r["action"], r.get("reason")) for r in audit.recent(store.TenantStore(tmp_path, identity.resolve(HEADERS, {}))) if r["action"] == "runs.stop"]
    assert rows == [("runs.stop", "cancelled"), ("runs.stop", "replayed")]


def test_only_the_gateway_may_vouch_for_an_empty_registry_and_a_failed_receipt_can_be_superseded(router, events, tmp_path):
    """F8 / SO-41: a human surface asserting ``process_registry_empty`` gets ``failed_to_cancel``; the worker's later real cancel supersedes it."""
    status, body = call(router, "POST", "/hux/v1/runs/run_9f/stop", {"process_registry_empty": True, "conversation_id": CONV}, HEADERS)
    assert status == 201 and valid(body)["outcome"] == "failed_to_cancel" and "completed_at" not in body
    first_requested = body["requested_at"]
    status, again = call(router, "POST", "/hux/v1/runs/run_9f/stop", {"process_registry_empty": True}, HEADERS)
    assert (status, again) == (200, body), "an identical failed stop replays"
    status, done = call(router, "POST", "/hux/v1/runs/run_9f/stop", {"process_registry_empty": True, "side_effects": [{"description": "x", "reverted": True}]})
    assert status == 201 and valid(done)["outcome"] == "cancelled" and done["requested_at"] == first_requested and "completed_at" in done
    assert done["conversation_id"] == CONV and done["requested_by"] == {"type": "system", "id": "worker"}
    tenant = store.TenantStore(tmp_path, identity.resolve(HEADERS, {}))
    stored = tenant.get("receipts", done["id"])
    assert stored["revision"] == 2 and contracts.validate_record(stored, SCHEMAS) == [], "stored receipts carry revision (F11)"
    status, third = call(router, "POST", "/hux/v1/runs/run_9f/stop", {"process_registry_empty": False})
    assert (status, third) == (200, done), "a terminal receipt never changes again"
    rows = [r.get("reason") for r in audit.recent(tenant) if r["action"] == "runs.stop"]
    assert rows == ["registry_state_not_from_gateway", "replayed", "superseded:cancelled", "replayed"]
    assert [e["summary"] for e in events] == ["Run stopped: failed_to_cancel (registry_state_not_from_gateway)", "Run stopped: cancelled (cancelled)"]
    status, body = call(router, "POST", "/hux/v1/runs/run_h/stop", {"already_complete": True}, HEADERS)
    assert valid(body)["outcome"] == "already_complete", "already_complete needs no registry claim"


def test_stop_outcomes_follow_the_process_registry(router):
    status, body = call(router, "POST", "/hux/v1/runs/run_a/stop", {"process_registry_empty": False, "side_effects": []})
    assert status == 201 and valid(body)["outcome"] == "failed_to_cancel" and "completed_at" not in body
    assert body["requested_by"] == {"type": "system", "id": "worker"}
    status, body = call(router, "POST", "/hux/v1/runs/run_b/stop", {"already_complete": True})
    assert valid(body)["outcome"] == "already_complete" and "completed_at" in body
    status, body = call(router, "POST", "/hux/v1/runs/run_c/stop", {})
    assert valid(body)["outcome"] == "failed_to_cancel", "no registry report means the stop is not proven"


@pytest.mark.parametrize("body", [[], {"side_effects": "none"}, {"side_effects": [{"description": ""}]}, {"conversation_id": "bad"}])
def test_stop_rejects_bad_bodies(router, body):
    status, error = call(router, "POST", "/hux/v1/runs/run_9f/stop", body)
    assert status == 400 and valid(error)["code"] == "invalid"


def test_receipts_are_per_tenant(router):
    call(router, "POST", "/hux/v1/runs/run_9f/stop", {"process_registry_empty": True})
    status, body = call(router, "POST", "/hux/v1/runs/run_9f/stop", {"process_registry_empty": False}, OTHER)
    assert status == 201 and body["outcome"] == "failed_to_cancel", "same run id, other subject, its own receipt"