"""Agent-side HUX hook driven end-to-end against the real service over HTTP. Security obligations exercised: SO-35 (only the human surface decides; the hook's worker identity cannot), SO-36 (a ``once`` approval releases the gate exactly once), SO-37 (the gate compares the canonical argument hash the hook recorded at request time), SO-39 (an external side effect under ``autonomous`` still waits for a human), SO-40 (an exhausted budget blocks new approvals), SO-41 (a stop is done only when the receipt exists), SO-11 and SO-07 (raw arguments and tool output never reach the tenant ledger). """ from __future__ import annotations import json import sys import threading from pathlib import Path import pytest ROOT = Path(__file__).resolve().parents[2] FOUNDATION = ROOT / "dockerfiles" / "hermes-hux-foundation" HOOK_ROOT = ROOT / "dockerfiles" / "hermes-worker-hux" for entry in (FOUNDATION, HOOK_ROOT): if str(entry) not in sys.path: sys.path.insert(0, str(entry)) from hux import contracts # noqa: E402 from hux.http import serve # noqa: E402 from hux.server import build_router # noqa: E402 from hux_hook import HuxClient, HuxServiceError, after_tool, before_tool, canonical_argument_hash, on_stop, record_spend # noqa: E402 SCHEMAS = contracts.load_all() ALL_ON = ",".join(card["flag"] for card in contracts.load_flags()["cards"]) SUBJECT = "usr_0123456789abcdef" IDENTITY = {"tenant_slot": "slot-3", "subject": SUBJECT, "surface": "worker", "trust": "worker"} HUMAN = {"tenant_slot": "slot-3", "subject": SUBJECT, "surface": "chat", "trust": "router"} CANARY = "CANARY-7f3a-SECRET-VALUE" ARGS = {"path": "notes.md", "content": CANARY} RUN = "run_hook_1" def start(tmp_path: Path, flags: str = ALL_ON) -> tuple[str, object]: """Run the real service on an ephemeral loopback port for one test.""" router = build_router(tmp_path, {"HUX_FLAGS": flags, "HUX_ROUTER_KEY": "rk", "HUX_WORKER_KEY": "wk"}) server = serve(router, "127.0.0.1", 0) threading.Thread(target=server.serve_forever, daemon=True).start() return f"http://127.0.0.1:{server.server_address[1]}", server @pytest.fixture def service(tmp_path): base, server = start(tmp_path) human = HuxClient(base, HUMAN, key="rk") conversation = human.post("/hux/v1/conversations", {"title": "hook test"}).body["id"] yield {"base": base, "root": tmp_path, "agent": HuxClient(base, IDENTITY, key="wk"), "human": human, "conv": conversation} server.shutdown() def set_policy(human: HuxClient, autonomy: str, budgets: dict | None = None) -> None: body = {"scope": {"level": "global"}, "autonomy": autonomy} if budgets: body["budgets"] = budgets human.put("/hux/v1/policy", body) def ledger_text(root: Path) -> str: return "\n".join(p.read_text(errors="ignore") for p in root.rglob("*") if p.is_file()) def events_of(human: HuxClient, conversation: str) -> list[dict]: items = human.get(f"/hux/v1/conversations/{conversation}/events").body["items"] for item in items: assert contracts.validate_record(item, SCHEMAS) == [] return items # --- approval -> pending -> human decision -> gate once ------------------------------ def test_once_approval_releases_exactly_once(service): """SO-35, SO-36, SO-37: pending until a human decides, then one release; the second gate is blocked.""" agent, human, conv = service["agent"], service["human"], service["conv"] first = before_tool(agent, RUN, conv, "write_file", ARGS, "write_files", risk="low", turn=1) assert (first.proceed, first.reason) == (False, "approval_required") assert first.approval_id and first.approval_id.startswith("apr_") record = human.get(f"/hux/v1/approvals/{first.approval_id}").body assert contracts.validate_record(record, SCHEMAS) == [] assert record["request"]["evidence"][0]["hash"] == canonical_argument_hash("write_file", ARGS) assert CANARY not in json.dumps(record) with pytest.raises(HuxServiceError) as denied: agent.post(f"/hux/v1/approvals/{first.approval_id}", {"choice": "once"}) assert denied.value.status == 403 human.post(f"/hux/v1/approvals/{first.approval_id}", {"choice": "once"}) second = before_tool(agent, RUN, conv, "write_file", {"content": CANARY, "path": "notes.md"}, "write_files", risk="low") assert second == second.__class__(True, first.approval_id, "released") third = before_tool(agent, RUN, conv, "write_file", ARGS, "write_files", risk="low") assert third.proceed is False and "consumed" in third.reason kinds = [e["kind"] for e in events_of(human, conv)] assert kinds.count("side_effect.released") == 1 and "side_effect.blocked" in kinds and "approval.requested" in kinds assert CANARY not in ledger_text(service["root"]) def test_different_arguments_after_approval_are_blocked(service): """SO-37: an approval for one argument hash never releases another.""" agent, human, conv = service["agent"], service["human"], service["conv"] pending = before_tool(agent, RUN, conv, "shell", {"cmd": "ls"}, "shell") human.post(f"/hux/v1/approvals/{pending.approval_id}", {"choice": "once"}) other = before_tool(agent, RUN, conv, "shell", {"cmd": "rm -rf /"}, "shell") assert other.proceed is False and other.reason == "approval_required" def test_denied_and_session_choices(service): """A denial fails closed; a session grant lets a later call in the same conversation auto-approve.""" agent, human, conv = service["agent"], service["human"], service["conv"] pending = before_tool(agent, RUN, conv, "shell", {"cmd": "ls"}, "shell") human.post(f"/hux/v1/approvals/{pending.approval_id}", {"choice": "deny"}) assert before_tool(agent, RUN, conv, "shell", {"cmd": "ls"}, "shell") == pending.__class__(False, pending.approval_id, "approval_denied") pending2 = before_tool(agent, RUN, conv, "shell", {"cmd": "pwd"}, "shell") human.post(f"/hux/v1/approvals/{pending2.approval_id}", {"choice": "session"}) assert before_tool(agent, "run_hook_2", conv, "shell", {"cmd": "whoami"}, "shell").proceed is True def test_external_side_effect_under_autonomous_still_asks(service): """SO-39: autonomy never auto-allows an external side effect.""" agent, human, conv = service["agent"], service["human"], service["conv"] set_policy(human, "autonomous") decision = before_tool(agent, RUN, conv, "send_email", {"to": "x@example.com"}, "send_message", external=True, risk="high") assert (decision.proceed, decision.reason) == (False, "approval_required") local = before_tool(agent, RUN, conv, "write_file", ARGS, "write_files") assert local.proceed is True and local.reason == "released" assert before_tool(agent, RUN, conv, "write_file", ARGS, "write_files").proceed is False def test_budget_exhaustion_blocks_new_approvals(service): """SO-40: once the run budget is spent the hook reports budget_exhausted and never proceeds.""" agent, human, conv = service["agent"], service["human"], service["conv"] set_policy(human, "autonomous", {"tool_calls_per_run": 2}) state = record_spend(agent, RUN, conv, tool_calls=2, tokens=10, bogus=5) assert state is not None and contracts.validate_record(state, SCHEMAS) == [] assert state["exhausted"] == ["tool_calls_per_run"] decision = before_tool(agent, RUN, conv, "write_file", ARGS, "write_files") assert decision == decision.__class__(False, None, "budget_exhausted") assert record_spend(agent, RUN, conv, tokens=-3)["spent"]["tokens"] == 10 assert record_spend(agent, RUN, tool_calls=1)["spent"]["tool_calls"] == 3 def test_stop_receipt_is_written_once(service): """SO-41: the receipt reports what really happened and a repeat returns the same record.""" agent, human, conv = service["agent"], service["human"], service["conv"] effects = [{"description": "partial notes.md", "reverted": True}] receipt = on_stop(agent, RUN, conv, process_registry_empty=True, side_effects=effects) assert contracts.validate_record(receipt, SCHEMAS) == [] assert receipt["outcome"] == "cancelled" and receipt["side_effects"] == effects assert on_stop(agent, RUN, conv, process_registry_empty=False)["id"] == receipt["id"] failed = on_stop(agent, "run_hook_3", None, process_registry_empty=False) assert failed["outcome"] == "failed_to_cancel" and "conversation_id" not in failed assert on_stop(agent, "run_hook_4", conv, True, already_complete=True)["outcome"] == "already_complete" assert [e["kind"] for e in events_of(human, conv)].count("run.cancelled") == 2 def test_after_tool_records_status_only(service): """SO-11: tool.result carries sizes and status; the output and arguments never land.""" agent, conv = service["agent"], service["conv"] digest = canonical_argument_hash("shell", {"cmd": CANARY}) event = after_tool(agent, RUN, conv, "shell", True, 512, turn=3, argument_hash=digest, duration_ms=40, exit_code=0) assert event["kind"] == "tool.result" and event["detail"] == {"tool": "shell", "ok": True, "bytes": 512, "duration_ms": 40, "exit_code": 0} assert event["evidence"][0]["hash"] == digest and event["turn"] == 3 replay = after_tool(agent, RUN, conv, "shell", True, 512, turn=3, argument_hash=digest) assert replay["id"] == event["id"] plain = after_tool(agent, RUN, conv, "shell", False, -1) assert plain["detail"] == {"tool": "shell", "ok": False, "bytes": 0} and "evidence" not in plain assert CANARY not in ledger_text(service["root"]) def test_bad_ids_and_service_errors_fail_closed(service): """Malformed run or conversation ids never reach a tool execution.""" agent, conv = service["agent"], service["conv"] assert before_tool(agent, "run/../x", conv, "shell", {}, "shell").reason == "invalid_run_id" assert before_tool(agent, RUN, "not an id", "shell", {}, "shell").reason == "service_error:invalid" assert before_tool(agent, RUN, conv, "shell", {}, "teleport").reason == "service_error:invalid" def test_gate_failures_fail_closed(service, monkeypatch): """A gate that errors, answers nonsense or says no keeps the tool from running.""" agent, human, conv = service["agent"], service["human"], service["conv"] set_policy(human, "autonomous") real_post = agent.post def broken_gate(path, body, idempotency_key=None): if path.endswith("/gate"): raise HuxServiceError(500, "invalid", "boom") return real_post(path, body, idempotency_key) monkeypatch.setattr(agent, "post", broken_gate) decision = before_tool(agent, RUN, conv, "write_file", ARGS, "write_files") assert decision.proceed is False and decision.reason == "service_error:invalid" and decision.approval_id class Odd: body = "not a dict" monkeypatch.setattr(agent, "post", lambda path, body, idempotency_key=None: Odd() if path.endswith("/gate") else real_post(path, body, idempotency_key)) assert before_tool(agent, RUN, conv, "write_file", ARGS, "write_files").reason == "gate_invalid" class NoProceed: body = {"proceed": False} monkeypatch.setattr(agent, "post", lambda path, body, idempotency_key=None: NoProceed() if path.endswith("/gate") else real_post(path, body, idempotency_key)) assert before_tool(agent, RUN, conv, "write_file", ARGS, "write_files").reason == "gate_blocked" monkeypatch.setattr(agent, "post", lambda path, body, idempotency_key=None: Odd()) assert before_tool(agent, RUN, conv, "write_file", ARGS, "write_files").reason == "approval_invalid" def test_unreachable_service_fails_closed_for_gate_and_open_for_telemetry(tmp_path): """No service: the gate says no, telemetry returns None, nothing raises into the loop.""" base, server = start(tmp_path) server.shutdown() server.server_close() agent = HuxClient(base, IDENTITY, key="wk", timeout=1) decision = before_tool(agent, RUN, "conv_0001abcd", "shell", {"cmd": "ls"}, "shell") assert decision == decision.__class__(False, None, "hux_unavailable") assert after_tool(agent, RUN, "conv_0001abcd", "shell", True, 1) is None assert record_spend(agent, RUN, "conv_0001abcd", tool_calls=1) is None assert on_stop(agent, RUN, "conv_0001abcd", True) is None def test_autonomy_flag_off_is_reported_and_fails_closed(tmp_path): """SO-50: with hux.autonomy off, capabilities says so and the hook refuses every side effect.""" base, server = start(tmp_path, "hux.foundation,hux.activity_timeline,hux.projects") try: agent = HuxClient(base, IDENTITY, key="wk") caps = agent.capabilities() assert caps["reachable"] and caps["cards"]["HUX-11"] and caps["cards"]["HUX-01"] and not caps["cards"]["HUX-05"] decision = before_tool(agent, RUN, "conv_0001abcd", "shell", {"cmd": "ls"}, "shell") assert decision == decision.__class__(False, None, "autonomy_off") finally: server.shutdown()