From b1229b3f37e609e3a320f6bc75b73ab8e34097c1 Mon Sep 17 00:00:00 2001 From: Robert Date: Mon, 27 Jul 2026 19:01:47 +0200 Subject: [PATCH 1/3] feat: operator console for phase-1 review The dashboard was a form per endpoint -- fine for poking the API, useless for what phase 1 actually requires: looking at what the engine decided and reviewing every hard_deny by hand before enforcement goes on. That work was CLI-only. The console shows a live decision feed with per-state filtering and a distribution bar, the phase-readiness criteria with their verdict, and the hard-deny events laid out for review. Two read-only endpoints back it: /v1/decisions and /v1/rollout/report. It lives inside AETHERYA rather than in a separate frontend because this page exposes every decision and every action an agent proposed -- the most sensitive data in the system. A separate app would need its own deployment, its own auth and its own path to the audit trail: three new things to protect to show the same data. Server-rendered with no build step, it ships in the container that is already running and keeps the trail behind one boundary. The API stays open, so a richer frontend can still be built on top without touching the engine. Gated behind AETHERYA_CONSOLE_API_KEY when set, open otherwise like the rest of the decision profile; the page and the route comment both say the port must not be public. Removed _dashboard_html (~11KB of unreachable HTML) rather than leaving it to rot next to its replacement, and dropped the test that asserted its form controls -- test_console.py covers the new page. Verified against the running stack: rebuilt, healthy, and the console renders real traffic across log_only, deny and hard_deny. 1194 tests, 100% coverage under CI's dependency set, mypy strict, ruff and black clean. Co-Authored-By: Claude Opus 5 --- CHANGELOG.md | 6 + src/aetherya/api.py | 86 ++++++++ src/aetherya/api_server.py | 378 ++++------------------------------- src/aetherya/console.py | 277 +++++++++++++++++++++++++ tests/test_api_server.py | 13 +- tests/test_console.py | 341 +++++++++++++++++++++++++++++++ tests/test_rollout_report.py | 40 ++++ 7 files changed, 788 insertions(+), 353 deletions(-) create mode 100644 src/aetherya/console.py create mode 100644 tests/test_console.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 20cf5e6..d00f771 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,12 @@ All notable changes to this project are documented in this file. ## Unreleased +### Added (operator console) + +- `console.py` — an operator console replacing the API dashboard at `/`. The old page was a form per endpoint: useful for poking the API, useless for the job phase 1 actually requires, which is looking at what the engine decided and reviewing every `hard_deny` by hand before enforcement is switched on. The new page shows a live decision feed with per-state filtering and a distribution bar, the phase-readiness criteria with their verdict, and a hard-deny review list. Server-rendered, no build step, no external assets, so it ships in the same container as the engine and the audit trail stays behind one network boundary instead of two. +- `GET /v1/decisions?limit=&state=` and `GET /v1/rollout/report?phase=` — read-only views over the audit trail backing the console. Gated behind `AETHERYA_CONSOLE_API_KEY` when that is set; open otherwise, like the rest of the decision profile. **This view exposes every recorded action, so the port must not be public** — stated in the page footer and in the route comment. +- Removed the previous `_dashboard_html` (~11KB of now-unreachable HTML) rather than leaving it to rot alongside its replacement. + ### Added (phase-1 sidecar) - `enforcement.py` + `enforcement.phase` in policy — the rollout posture is now configuration. The engine always computes a full decision; the phase decides how much is enforced, so advancing is a config change and never a code change, and all three phases exist from day one. `apply_enforcement` never mutates the decision: the audit records what the engine ruled, not what a partially-enforcing deployment did about it, and conflating those would make phase-1 data worthless. diff --git a/src/aetherya/api.py b/src/aetherya/api.py index e77a2ad..eeed4a2 100644 --- a/src/aetherya/api.py +++ b/src/aetherya/api.py @@ -1,6 +1,7 @@ from __future__ import annotations import hmac +import json import os from dataclasses import dataclass from pathlib import Path @@ -27,6 +28,7 @@ from aetherya.parser import parse_user_input from aetherya.pipeline import run_pipeline from aetherya.rate_limiter import RateLimitConfig, RateLimiter, build_rate_limiter +from aetherya.rollout_report import DEFAULT_MIN_DAYS, DEFAULT_MIN_DECISIONS, build_report @dataclass(frozen=True) @@ -355,6 +357,90 @@ def decide(self, payload: Any) -> tuple[int, dict[str, Any]]: {"ok": False, "error_type": type(exc).__name__, "error": str(exc)}, ) + def decisions(self, params: dict[str, Any] | None = None) -> tuple[int, dict[str, Any]]: + """ + Recent decisions from the audit trail, newest first. + + Read-only view for the operator console. Reads the tail of the file + rather than holding an index: at console volumes that is simpler and + cannot drift from the authoritative record, which is the file itself. + """ + try: + query = params or {} + if self.settings.audit_path is None: + raise ValueError("audit_path is disabled in API settings") + + limit = max(1, min(int(query.get("limit", 100) or 100), 1000)) + state_filter = _as_optional_str(query.get("state"), field_name="state") + + events: list[dict[str, Any]] = [] + path = self.settings.audit_path + lines = path.read_text(encoding="utf-8").splitlines() if path.exists() else [] + for raw in reversed(lines): + if len(events) >= limit: + break + if not raw.strip(): + continue + try: + event = json.loads(raw) + except json.JSONDecodeError: + continue + if not isinstance(event, dict): + continue + decision = event.get("decision") or {} + state = str(decision.get("state", "unknown")) + if state_filter and state != state_filter: + continue + context = event.get("context") or {} + action_ctx = context.get("action") if isinstance(context, dict) else {} + events.append( + { + "ts": event.get("ts"), + "actor": event.get("actor"), + "action": str(event.get("action", ""))[:400], + "state": state, + "allowed": decision.get("allowed"), + "risk_score": decision.get("risk_score"), + "reason": decision.get("reason"), + "violated_principle": decision.get("violated_principle"), + "tool": (action_ctx.get("tool") if isinstance(action_ctx, dict) else None), + "escalated": bool( + isinstance(context, dict) and context.get("intent_escalation") + ), + "decision_id": event.get("decision_id"), + } + ) + + return (200, {"ok": True, "count": len(events), "decisions": events}) + except Exception as exc: + return ( + 400, + {"ok": False, "error_type": type(exc).__name__, "error": str(exc)}, + ) + + def rollout(self, params: dict[str, Any] | None = None) -> tuple[int, dict[str, Any]]: + """Phase measurement for the console — same data as `aetherya rollout report`.""" + try: + query = params or {} + if self.settings.audit_path is None: + raise ValueError("audit_path is disabled in API settings") + + cfg = load_policy_config(self.settings.policy_path) + phase = int(query.get("phase") or cfg.enforcement.phase) + report = build_report( + self.settings.audit_path, + current_phase=phase, + min_decisions=int(query.get("min_decisions", DEFAULT_MIN_DECISIONS) or 0) + or DEFAULT_MIN_DECISIONS, + min_days=float(query.get("min_days", DEFAULT_MIN_DAYS) or 0) or DEFAULT_MIN_DAYS, + ) + return (200, {"ok": True, "report": report.to_dict()}) + except Exception as exc: + return ( + 400, + {"ok": False, "error_type": type(exc).__name__, "error": str(exc)}, + ) + def audit_verify(self, payload: Any) -> tuple[int, dict[str, Any]]: try: body = _as_mapping(payload, field_name="audit verify payload") diff --git a/src/aetherya/api_server.py b/src/aetherya/api_server.py index bc1789c..fe20ba1 100644 --- a/src/aetherya/api_server.py +++ b/src/aetherya/api_server.py @@ -1,15 +1,18 @@ from __future__ import annotations import argparse +import hmac import json +import os import sys from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any -from urllib.parse import urlsplit +from urllib.parse import parse_qs, urlsplit from aetherya.api import AetheryaAPI, APISettings from aetherya.config import load_policy_config +from aetherya.console import console_html from aetherya.constitution import warmup_semantic_model @@ -17,344 +20,6 @@ class RequestTooLargeError(ValueError): pass -def _dashboard_html() -> str: - return """ - - - - - AETHERYA API Dashboard - - - -
-
-

AETHERYA API Dashboard

-

Interactive panel for health checks, policy decisions, signed approvals, and audit-chain verification.

-
- -
-
-

Health

-
- - idle -
-

Quick runtime check: policy, fingerprint and defaults.

-
- -
-

Decide

- - - - - - -
- - - idle -
-
- -
-

Confirmation Sign (Admin)

- - - - - - - - -
- - idle -
-
- -
-

Confirmation Verify (Admin)

- - - - - - - - -
- - idle -
-
- -
-

Audit Verify

-
- - -
- - -
- - idle -
-
-
- -
-

Response

- 
{\n  "ok": true,\n  "message": "Ready"\n}
-
-
- - - - -""" - - class AetheryaHTTPRequestHandler(BaseHTTPRequestHandler): api: AetheryaAPI | None = None max_body_bytes: int = 1_048_576 @@ -408,11 +73,42 @@ def _parse_json_body(self) -> dict[str, Any]: raise ValueError("request body must be a JSON object") return payload + def _query(self) -> dict[str, Any]: + raw = parse_qs(urlsplit(self.path).query) + return {key: values[0] for key, values in raw.items() if values} + def _handle_request(self) -> None: method = self.command.upper() path = urlsplit(self.path).path - if method == "GET" and path in {"/", "/dashboard"}: - self._send_html(200, _dashboard_html()) + if method == "GET" and path in {"/", "/console", "/dashboard"}: + self._send_html(200, console_html()) + return + + # Read-only console data. Gated behind AETHERYA_CONSOLE_API_KEY when it + # is set; open otherwise, like the rest of the decision profile. This + # view exposes every recorded action, so the port must not be public. + if method == "GET" and path in {"/v1/decisions", "/v1/rollout/report"}: + if self.api is None: + self._send_json(500, {"ok": False, "error": "api not configured"}) + return + expected = os.getenv("AETHERYA_CONSOLE_API_KEY", "").strip() + if expected: + provided = self.headers.get("X-AETHERYA-Console-Key", "").strip() + if not provided or not hmac.compare_digest(provided, expected): + self._send_json( + 401, + { + "ok": False, + "error_type": "Unauthorized", + "error": "missing or invalid console key", + }, + ) + return + query = self._query() + status, body = ( + self.api.decisions(query) if path == "/v1/decisions" else self.api.rollout(query) + ) + self._send_json(status, body) return payload: dict[str, Any] = {} diff --git a/src/aetherya/console.py b/src/aetherya/console.py new file mode 100644 index 0000000..4a9bc4f --- /dev/null +++ b/src/aetherya/console.py @@ -0,0 +1,277 @@ +from __future__ import annotations + +# --------------------------------------------------------------------------- +# Operator console. +# +# The previous dashboard was a form per endpoint — useful for poking the API, +# useless for the job phase 1 actually requires: looking at what the engine +# decided and reviewing every hard_deny by hand before enforcement is switched +# on. This is that tool. +# +# Server-rendered, no build step, no external assets. It ships in the same +# container as the engine, which keeps the audit trail behind exactly one +# network boundary instead of two. +# +# SECURITY: this page exposes every decision and every action an agent proposed. +# Do not publish port 8080. Set AETHERYA_CONSOLE_API_KEY to require a key. +# --------------------------------------------------------------------------- + + +def console_html() -> str: + return """ + + + + +ÆTHERYA — Operator Console + + + +
+
+

ÆTHERYA operator console

+
loading…
+
+ +
+
+

Decision feed

+

Newest first, straight from the audit trail.

+
+
+
+
+ + + + +
timeactorstateriskactionreason
loading…
+
+
+ +
+

Phase readiness

+

Exit criteria for the configured phase.

+
loading…
+
+
+
+ +
+

Hard-deny review

+

+ Every one of these needs human eyes before enforcement is switched on. + A false positive here is a detection bug — add the case to the corpus and + fix the rule rather than loosening the policy. +

+
loading…
+
+ +
+ Read-only view of the audit trail. Do not expose this port publicly. +
+
+ + + + +""" diff --git a/tests/test_api_server.py b/tests/test_api_server.py index 2f02c16..217c9aa 100644 --- a/tests/test_api_server.py +++ b/tests/test_api_server.py @@ -13,7 +13,6 @@ AetheryaHTTPRequestHandler, RequestTooLargeError, _build_handler, - _dashboard_html, build_server, main, serve_api, @@ -72,16 +71,6 @@ def end_headers(self) -> None: assert fake_html.wfile.getvalue().decode("utf-8") == "

x

" -def test_dashboard_template_contains_controls() -> None: - html = _dashboard_html() - assert "AETHERYA API Dashboard" in html - assert "/v1/decide" in html - assert "/v1/audit/verify" in html - assert "/v1/confirmation/sign" in html - assert "/v1/confirmation/verify" in html - assert "candidate_response" in html - - def test_handler_parse_json_body_branches() -> None: class FakeHandler: def __init__(self, content_length: str, raw: bytes, max_body_bytes: int = 64): @@ -163,7 +152,7 @@ def _send_html(self, status: int, html: str) -> None: dashboard = FakeHandler(method="GET", path="/") AetheryaHTTPRequestHandler._handle_request(dashboard) # noqa: SLF001 assert dashboard.sent_html[0][0] == 200 - assert "AETHERYA API Dashboard" in dashboard.sent_html[0][1] + assert "Operator Console" in dashboard.sent_html[0][1] get_ok = FakeHandler(method="GET", path="/health") AetheryaHTTPRequestHandler._handle_request(get_ok) # noqa: SLF001 diff --git a/tests/test_console.py b/tests/test_console.py new file mode 100644 index 0000000..4100268 --- /dev/null +++ b/tests/test_console.py @@ -0,0 +1,341 @@ +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +import pytest + +from aetherya.api import AetheryaAPI, APISettings +from aetherya.audit import AuditLogger +from aetherya.console import console_html + + +def _api(tmp_path: Path, audit: Path | None = None) -> AetheryaAPI: + return AetheryaAPI( + APISettings( + policy_path=Path("config/policy.yaml"), + audit_path=audit if audit is not None else tmp_path / "decisions.jsonl", + ) + ) + + +def _seed(path: Path, rows: list[tuple[str, str]]) -> None: + """rows: (actor, state).""" + logger = AuditLogger(str(path), policy_fingerprint="sha256:test") + for actor, state in rows: + logger.log( + actor=actor, + action=f"{state} action by {actor}", + decision={ + "allowed": state in {"allow", "log_only"}, + "risk_score": 171 if state == "hard_deny" else 0, + "reason": f"{state}: because", + "state": state, + }, + context={"mode": "operative", "action": {"tool": "shell"}}, + ) + + +# --------------------------------------------------------------------------- +# Decision feed +# --------------------------------------------------------------------------- + + +def test_decisions_are_returned_newest_first(tmp_path: Path) -> None: + audit = tmp_path / "decisions.jsonl" + _seed(audit, [("first", "allow"), ("second", "allow"), ("third", "hard_deny")]) + + code, body = _api(tmp_path, audit).decisions() + assert code == 200 + assert [d["actor"] for d in body["decisions"]] == ["third", "second", "first"] + + +def test_decisions_can_be_filtered_by_state(tmp_path: Path) -> None: + audit = tmp_path / "decisions.jsonl" + _seed(audit, [("a", "allow"), ("b", "hard_deny"), ("c", "allow"), ("d", "hard_deny")]) + + _, body = _api(tmp_path, audit).decisions({"state": "hard_deny"}) + assert body["count"] == 2 + assert {d["actor"] for d in body["decisions"]} == {"b", "d"} + + +def test_decisions_respect_the_limit(tmp_path: Path) -> None: + audit = tmp_path / "decisions.jsonl" + _seed(audit, [(f"a{i}", "allow") for i in range(30)]) + + _, body = _api(tmp_path, audit).decisions({"limit": 5}) + assert body["count"] == 5 + + +def test_the_limit_is_clamped(tmp_path: Path) -> None: + audit = tmp_path / "decisions.jsonl" + _seed(audit, [("a", "allow")]) + api = _api(tmp_path, audit) + + assert api.decisions({"limit": 0})[1]["ok"] is True + assert api.decisions({"limit": 99_999})[1]["ok"] is True + + +def test_decisions_expose_the_fields_the_console_renders(tmp_path: Path) -> None: + audit = tmp_path / "decisions.jsonl" + _seed(audit, [("bot", "hard_deny")]) + + row = _api(tmp_path, audit).decisions()[1]["decisions"][0] + for field in ("ts", "actor", "action", "state", "risk_score", "reason", "tool"): + assert field in row + + +def test_a_missing_audit_file_yields_an_empty_feed(tmp_path: Path) -> None: + """A fresh deployment has no decisions yet; that is not an error.""" + code, body = _api(tmp_path, tmp_path / "nothing-here.jsonl").decisions() + assert code == 200 + assert body["count"] == 0 + + +def test_malformed_lines_are_skipped_in_the_feed(tmp_path: Path) -> None: + audit = tmp_path / "decisions.jsonl" + _seed(audit, [("a", "allow")]) + with audit.open("a", encoding="utf-8") as handle: + handle.write("not json\n") + handle.write("\n") + handle.write('"bare string"\n') + + assert _api(tmp_path, audit).decisions()[1]["count"] == 1 + + +def test_actions_are_truncated_in_the_feed(tmp_path: Path) -> None: + audit = tmp_path / "decisions.jsonl" + AuditLogger(str(audit)).log( + actor="a", + action="x" * 900, + decision={"allowed": True, "risk_score": 0, "reason": "ok", "state": "allow"}, + context={}, + ) + assert len(_api(tmp_path, audit).decisions()[1]["decisions"][0]["action"]) <= 400 + + +def test_escalated_requests_are_flagged(tmp_path: Path) -> None: + audit = tmp_path / "decisions.jsonl" + AuditLogger(str(audit)).log( + actor="a", + action="dd if=/dev/zero of=/dev/sda", + decision={"allowed": False, "risk_score": 100, "reason": "bad", "state": "hard_deny"}, + context={"intent_escalation": {"escalated": True}}, + ) + assert _api(tmp_path, audit).decisions()[1]["decisions"][0]["escalated"] is True + + +def test_decisions_fail_when_audit_is_disabled(tmp_path: Path) -> None: + api = AetheryaAPI(APISettings(audit_path=None)) + code, body = api.decisions() + assert code == 400 + assert "audit_path is disabled" in body["error"] + + +# --------------------------------------------------------------------------- +# Rollout endpoint +# --------------------------------------------------------------------------- + + +def test_rollout_report_is_served(tmp_path: Path) -> None: + audit = tmp_path / "decisions.jsonl" + _seed(audit, [("a", "allow"), ("b", "hard_deny")]) + + code, body = _api(tmp_path, audit).rollout() + assert code == 200 + report = body["report"] + assert report["states"] == {"allow": 1, "hard_deny": 1} + assert report["ready_to_advance"] is False + assert len(report["hard_deny_events"]) == 1 + + +def test_rollout_defaults_to_the_configured_phase(tmp_path: Path) -> None: + audit = tmp_path / "decisions.jsonl" + _seed(audit, [("a", "allow")]) + + body = _api(tmp_path, audit).rollout()[1] + assert body["report"]["current_phase"] == 1 + + +def test_rollout_phase_can_be_overridden(tmp_path: Path) -> None: + audit = tmp_path / "decisions.jsonl" + _seed(audit, [("a", "allow")]) + + body = _api(tmp_path, audit).rollout({"phase": 2})[1] + assert body["report"]["current_phase"] == 2 + + +def test_rollout_reports_a_missing_audit_file(tmp_path: Path) -> None: + code, body = _api(tmp_path, tmp_path / "nope.jsonl").rollout() + assert code == 400 + assert "audit file not found" in body["error"] + + +def test_rollout_fails_when_audit_is_disabled() -> None: + code, body = AetheryaAPI(APISettings(audit_path=None)).rollout() + assert code == 400 + assert "audit_path is disabled" in body["error"] + + +def test_rollout_accepts_custom_thresholds(tmp_path: Path) -> None: + audit = tmp_path / "decisions.jsonl" + _seed(audit, [("a", "allow")] * 3) + + body = _api(tmp_path, audit).rollout({"min_decisions": 1, "min_days": 0.0})[1] + window = next(c for c in body["report"]["criteria"] if c["name"] == "sufficient_window") + assert window["passed"] is True + + +# --------------------------------------------------------------------------- +# The page itself +# --------------------------------------------------------------------------- + + +def test_console_page_is_self_contained() -> None: + """ + No build step and no external assets: the page ships with the engine, so a + deployment is one container and the audit trail sits behind one boundary. + """ + html = console_html() + assert "" in html.lower() + assert 'src="http' not in html + assert 'href="http' not in html + assert "cdn" not in html.lower() + + +def test_console_page_calls_the_endpoints_it_needs() -> None: + html = console_html() + for endpoint in ("/health", "/v1/decisions", "/v1/rollout/report"): + assert endpoint in html + + +def test_console_page_escapes_rendered_values() -> None: + """The feed renders actor-supplied strings; they must not be injected raw.""" + html = console_html() + assert "const esc" in html + assert "&lt;" in html or "<" in html + + +def test_console_warns_about_exposure() -> None: + assert "Do not expose this port publicly" in console_html() + + +# --------------------------------------------------------------------------- +# HTTP surface +# --------------------------------------------------------------------------- + + +def _server(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Any: + import threading + + from aetherya.api_server import build_server + + audit = tmp_path / "decisions.jsonl" + _seed(audit, [("a", "allow"), ("b", "hard_deny")]) + api = _api(tmp_path, audit) + server = build_server(host="127.0.0.1", port=0, api=api, max_body_bytes=65536) + threading.Thread(target=server.serve_forever, daemon=True).start() + return server + + +def _get(server: Any, path: str, headers: dict[str, str] | None = None) -> Any: + import urllib.request + + request = urllib.request.Request( # noqa: S310 + f"http://127.0.0.1:{server.server_port}{path}", headers=headers or {} + ) + try: + with urllib.request.urlopen(request, timeout=5) as response: # noqa: S310 + return response.status, response.read().decode("utf-8") + except urllib.error.HTTPError as exc: # type: ignore[name-defined] + return exc.code, exc.read().decode("utf-8") + + +def test_console_is_served_at_the_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + server = _server(tmp_path, monkeypatch) + try: + for path in ("/", "/console", "/dashboard"): + status, body = _get(server, path) + assert status == 200 + assert "Operator Console" in body + finally: + server.shutdown() + + +def test_decisions_endpoint_is_reachable(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + server = _server(tmp_path, monkeypatch) + try: + status, body = _get(server, "/v1/decisions?limit=5&state=hard_deny") + assert status == 200 + payload = json.loads(body) + assert payload["count"] == 1 + finally: + server.shutdown() + + +def test_rollout_endpoint_is_reachable(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + server = _server(tmp_path, monkeypatch) + try: + status, body = _get(server, "/v1/rollout/report") + assert status == 200 + assert json.loads(body)["report"]["current_phase"] == 1 + finally: + server.shutdown() + + +def test_console_key_is_enforced_when_configured( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """ + This view exposes every recorded action. When a key is set it is required; + without one the routes are open like the rest of the decision profile, which + is why the port must not be public. + """ + import urllib.error # noqa: F401 - imported for _get's except clause + + monkeypatch.setenv("AETHERYA_CONSOLE_API_KEY", "s3cret") + server = _server(tmp_path, monkeypatch) + try: + assert _get(server, "/v1/decisions")[0] == 401 + assert _get(server, "/v1/rollout/report")[0] == 401 + + status, _ = _get(server, "/v1/decisions", {"X-AETHERYA-Console-Key": "s3cret"}) + assert status == 200 + + assert _get(server, "/v1/decisions", {"X-AETHERYA-Console-Key": "wrong"})[0] == 401 + finally: + server.shutdown() + + +def test_console_page_is_not_key_gated(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """The shell is inert HTML; the data behind it is what the key protects.""" + monkeypatch.setenv("AETHERYA_CONSOLE_API_KEY", "s3cret") + server = _server(tmp_path, monkeypatch) + try: + assert _get(server, "/")[0] == 200 + finally: + server.shutdown() + + +def test_console_routes_fail_when_the_api_is_not_configured() -> None: + """Defensive: the handler must answer rather than raise if wiring is missing.""" + import io + + from aetherya.api_server import AetheryaHTTPRequestHandler + + class _Bare(AetheryaHTTPRequestHandler): + def __init__(self) -> None: # noqa: D107 + self.api = None + self.command = "GET" + self.path = "/v1/decisions" + self.headers = {} # type: ignore[assignment] + self.sent: list[tuple[int, dict[str, Any]]] = [] + self.wfile = io.BytesIO() + + def _send_json(self, status: int, body: dict[str, Any]) -> None: # type: ignore[override] + self.sent.append((status, body)) + + handler = _Bare() + handler._handle_request() # noqa: SLF001 + assert handler.sent[0][0] == 500 diff --git a/tests/test_rollout_report.py b/tests/test_rollout_report.py index f714d8e..163c853 100644 --- a/tests/test_rollout_report.py +++ b/tests/test_rollout_report.py @@ -331,3 +331,43 @@ def test_json_output_is_machine_readable( payload: dict[str, Any] = json.loads(capsys.readouterr().out) assert payload["window"]["total_decisions"] == 2 assert isinstance(payload["criteria"], list) + + +def test_timestamps_that_are_not_strings_are_ignored(tmp_path: Path) -> None: + path = tmp_path / "decisions.jsonl" + path.write_text( + json.dumps( + { + "actor": "a", + "action": "x", + "ts": 12345, + "decision": {"allowed": True, "risk_score": 0, "reason": "ok", "state": "allow"}, + "context": {}, + } + ) + + "\n", + encoding="utf-8", + ) + assert build_report(path).first_event is None + + +def test_text_output_without_timestamps_omits_the_range( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + path = tmp_path / "decisions.jsonl" + path.write_text( + json.dumps( + { + "actor": "a", + "action": "x", + "decision": {"allowed": True, "risk_score": 0, "reason": "ok", "state": "allow"}, + "context": {}, + } + ) + + "\n", + encoding="utf-8", + ) + main(["--audit-path", str(path)]) + out = capsys.readouterr().out + assert "window" in out + assert " -> " not in out From d0a5ae5ea37ac69a0ae12957e8593c3f8b5ef6ba Mon Sep 17 00:00:00 2001 From: Robert Date: Mon, 27 Jul 2026 22:43:54 +0200 Subject: [PATCH 2/3] feat: OpenAI-compatible gateway for agents you cannot modify MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The sidecar client requires editing the agent loop. Most agent runtimes — OpenClaw, LiteLLM, anything that accepts a custom OpenAI-compatible provider — cannot be edited that way. This gates them by changing what the agent talks to instead of what it does: the gateway speaks chat-completions, forwards upstream to OpenAI or Anthropic, and rules on every proposed tool call before the agent sees it. Integration is a config change; OpenClaw is not touched. A refused call is removed from the response and replaced with a line of text explaining why, so the model can replan. A silently dropped call just gets retried. Notable decisions: - Gating requires buffering. A verdict cannot be formed from half a tool call, so `stream: true` still returns SSE frames but emits them after the verdict. The protocol survives; token-by-token latency does not. Documented as a cost, not hidden. - Streamed tool calls carry `index`. Clients reassemble by it; two calls emitted without one merge into a single malformed call. - An actor absent from the capability matrix is denied on every call, and that is invisible until phase 3 because `deny` is not enforced before it. The gateway refuses to start at phase 3 with an unknown actor and warns otherwise. - SDKs are imported by name, following llm_provider: both are optional extras, and a module-level import would make them mandatory for type-checking and for deployments using the other provider. - Malformed tool arguments are gated as raw text rather than dropped. Dropping them would skip the check on exactly the input most likely to be evasive. - Escalations are refused at phase 3, not held: a single chat-completions call has no approval round trip. Asserted in tests so the doc cannot drift. Verified: the real OpenAI SDK parses the gateway's output in both streaming and non-streaming modes, and a refused call does not reach the client. The live upstream path needs an API key and is exercised by scripts/gateway_smoke.py, which is not part of CI. 100% coverage, mypy strict + ruff + black clean, and green under a CI simulation with the optional SDKs absent. Co-Authored-By: Claude Opus 5 --- Makefile | 8 +- README.md | 1 + docs/gateway-openclaw.md | 169 ++++++++ docs/index.md | 1 + pyproject.toml | 1 + scripts/gateway_smoke.py | 151 +++++++ src/aetherya/gateway.py | 466 ++++++++++++++++++++ src/aetherya/gateway_server.py | 237 +++++++++++ tests/test_gateway.py | 757 +++++++++++++++++++++++++++++++++ tests/test_gateway_server.py | 466 ++++++++++++++++++++ 10 files changed, 2256 insertions(+), 1 deletion(-) create mode 100644 docs/gateway-openclaw.md create mode 100644 scripts/gateway_smoke.py create mode 100644 src/aetherya/gateway.py create mode 100644 src/aetherya/gateway_server.py create mode 100644 tests/test_gateway.py create mode 100644 tests/test_gateway_server.py diff --git a/Makefile b/Makefile index 7cdd380..b1f46a0 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: fmt lint type test cov check security_baseline chaos_benchmark pipeline_benchmark pipeline_memory_soak property_tests audit_fuzz openai_shadow_smoke pre_api_gate api_serve api_decision_serve api_approvals_serve +.PHONY: fmt lint type test cov check security_baseline chaos_benchmark pipeline_benchmark pipeline_memory_soak property_tests audit_fuzz openai_shadow_smoke pre_api_gate api_serve api_decision_serve api_approvals_serve gateway_serve gateway_smoke fmt: ruff format src tests @@ -52,6 +52,12 @@ api_decision_serve: api_approvals_serve: python -m aetherya.api_server --service-mode approvals --host 127.0.0.1 --port 8081 +gateway_serve: + python -m aetherya.gateway_server --host 127.0.0.1 --port 8090 + +gateway_smoke: + python scripts/gateway_smoke.py --provider anthropic --phase 2 + api_stop: @echo "Liberando puerto 8080..." -fuser -k 8080/tcp || true diff --git a/README.md b/README.md index e670295..c5692a6 100644 --- a/README.md +++ b/README.md @@ -144,6 +144,7 @@ python -m aetherya.audit_verify --audit-path "$AUDIT" --require-chain --json \ | HTTP API | [docs/api.md](docs/api.md) | | Integrations (Python, CLI, agents) | [docs/integrations.md](docs/integrations.md) | | Production rollout phases | [docs/rollout-phases.md](docs/rollout-phases.md) | +| Gateway for OpenClaw & other agents | [docs/gateway-openclaw.md](docs/gateway-openclaw.md) | | Testing & benchmarks | [docs/testing-and-benchmarks.md](docs/testing-and-benchmarks.md) | | Release & verification | [docs/release-and-verification.md](docs/release-and-verification.md) | diff --git a/docs/gateway-openclaw.md b/docs/gateway-openclaw.md new file mode 100644 index 0000000..996d792 --- /dev/null +++ b/docs/gateway-openclaw.md @@ -0,0 +1,169 @@ +# Putting ÆTHERYA in front of an agent you did not write + +The sidecar in [rollout-phases.md](rollout-phases.md) requires editing the agent +loop. Most agent runtimes — OpenClaw, LiteLLM, LangChain, anything that accepts +a custom OpenAI-compatible provider — cannot be edited that way, or you do not +want to fork them. + +The gateway solves it by changing what the agent talks to instead of what it +does. It speaks the OpenAI chat-completions protocol, so the runtime treats it +as just another provider: + +``` +OpenClaw ──► ÆTHERYA gateway ──► OpenAI | Anthropic + │ + every tool call the model proposes is ruled on + before the agent ever sees it +``` + +The agent never learns there is a boundary. A tool call the policy refuses is +removed from the response and replaced with a line of text explaining why, so +the model can replan — and the audit trail records the decision either way. + +--- + +## Run it + +```bash +export ANTHROPIC_API_KEY=sk-ant-... +aetherya-gateway --provider anthropic --model claude-opus-5 --actor openclaw +``` + +or + +```bash +export OPENAI_API_KEY=sk-... +aetherya-gateway --provider openai --model gpt-4o-mini --actor openclaw +``` + +It binds `127.0.0.1:8090` and prints its posture: + +``` +[gateway] anthropic/claude-opus-5 · phase 1 (shadow) · http://127.0.0.1:8090/v1 +``` + +| Flag | Default | Notes | +|---|---|---| +| `--provider` | `anthropic` | `anthropic` or `openai` | +| `--model` | `claude-opus-5` | Passed upstream; a request may override it | +| `--actor` | `openclaw` | **Must exist in `capability_matrix.actors`** — see below | +| `--phase` | from `policy.yaml` | `1` shadow · `2` hard-deny · `3` full | +| `--audit-path` | `audit/decisions.jsonl` | Feeds the operator console | +| `--effort` | `high` | Anthropic only | + +**The gateway holds your upstream API key and authenticates nobody.** Keep it on +loopback. Binding it elsewhere prints a warning; it is not a suggestion. + +--- + +## Point OpenClaw at it + +In `openclaw.json`: + +```json5 +{ + agents: { + defaults: { + model: { primary: "aetherya/claude-opus-5" } + } + }, + models: { + providers: { + aetherya: { + baseUrl: "http://localhost:8090/v1", + apiKey: "not-used", + api: "openai-completions", + timeoutSeconds: 300, + models: [ + { + id: "claude-opus-5", + name: "Claude Opus 5 via ÆTHERYA", + contextWindow: 200000, + maxTokens: 8192 + } + ] + } + } + } +} +``` + +`apiKey` is deliberately junk: OpenClaw talks to the gateway, and the gateway +holds the real credential. That is a property worth keeping — the agent process +never sees your provider key. + +Then `openclaw configure` (existing install) or `openclaw onboard` (fresh). + +--- + +## Add your actor to the capability matrix + +This is the one step that is easy to skip and expensive to skip: + +```yaml +# config/policy.yaml +capability_matrix: + actors: + openclaw: + roles: [operator] + tools: [] + operations: [] +``` + +An actor the matrix has never heard of is denied on **every** call. In phases 1 +and 2 that is invisible, because `deny` is not enforced — so the +misconfiguration sits there silently until phase 3 turns your agent into a +brick. The gateway therefore refuses to start at phase 3 with an unknown actor, +and warns at phases 1 and 2. + +Note that the tools your agent uses must also be in `execution_gate.allowed_tools`. +A tool that is not on the allowlist is a `hard_deny` — the tier phase 2 enforces. +Run in phase 1 first and read the audit trail; that is exactly what it is for. + +--- + +## Check it before trusting it + +```bash +export ANTHROPIC_API_KEY=sk-ant-... +python scripts/gateway_smoke.py --provider anthropic --phase 2 +``` + +This starts the gateway, points a real OpenAI SDK client at it, and asks a real +model to run `rm -rf /` with a `shell` tool available. It reports what the model +proposed, what the policy ruled, and what survived to the client. Exits non-zero +if a destructive call got through. Costs a few cents. + +--- + +## What this costs you + +**Streaming latency.** A verdict needs the whole tool call — you cannot rule on +half of one. So the upstream response is completed before anything reaches the +client. `stream: true` still returns SSE frames and clients keep working, but +the tokens arrive in a burst at the end rather than as they are generated. For a +tool-using agent this is usually invisible; for a chat UI it is not. + +**A hop.** One extra process between the agent and the provider, holding your +key. Gate the port accordingly. + +**Tool vocabulary mapping.** The gateway does not know your agent's tools, so it +maps them generically: tool name → `tool`, arguments → `parameters`, and a +readable rendering of the arguments → `raw_input` (which is what ProceduralGuard +and JailbreakGuard read, so a destructive command hidden inside an argument is +still seen). If your tools carry structure the policy should reason about +specifically, encode it in `execution_gate.allowed_parameters` rather than +relying on the generic mapping. + +--- + +## What it does not do + +- **It does not sandbox.** ÆTHERYA decides; execution isolation is a separate + concern and a complementary one. +- **It does not gate model output text**, only proposed tool calls. Use + `output_gate` for response content. +- **It does not handle phase-3 confirmations.** An `escalate` at phase 3 is + refused like a `deny`, because there is no approval round trip in a single + chat-completions call. Agents that need held-for-approval semantics want the + sidecar in [rollout-phases.md](rollout-phases.md#phase-3--full-enforcement). diff --git a/docs/index.md b/docs/index.md index adf52b6..afe28d2 100644 --- a/docs/index.md +++ b/docs/index.md @@ -53,6 +53,7 @@ See a runnable simulation: [`examples/agent_integration.py`](../examples/agent_i |------|----------------| | [integrations.md](./integrations.md) | How to integrate — Python, CLI, agent loop, wrap-a-tool pattern | | [rollout-phases.md](./rollout-phases.md) | Shadow → hard-deny → full enforcement, with exit criteria per phase | +| [gateway-openclaw.md](./gateway-openclaw.md) | OpenAI-compatible gateway: gate an agent you cannot modify (OpenClaw, LiteLLM, …) | | [architecture.md](./architecture.md) | Full pipeline, component roles, fail-closed guarantee | | [security-model.md](./security-model.md) | Threat model, JailbreakGuard, confirmation + replay, audit chain | | [policy-model.md](./policy-model.md) | Modes, Constitution, semantic thresholds, policy.yaml structure | diff --git a/pyproject.toml b/pyproject.toml index 017cd48..502cc22 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -83,6 +83,7 @@ aetherya = "aetherya.cli:main" aetherya-api = "aetherya.api_server:main" aetherya-decision-server = "aetherya.decision_server:main" aetherya-approvals-server = "aetherya.approvals_server:main" +aetherya-gateway = "aetherya.gateway_server:main" [[tool.mypy.overrides]] module = ["numpy", "numpy.*", "sentence_transformers", "sentence_transformers.*"] diff --git a/scripts/gateway_smoke.py b/scripts/gateway_smoke.py new file mode 100644 index 0000000..c483f5f --- /dev/null +++ b/scripts/gateway_smoke.py @@ -0,0 +1,151 @@ +#!/usr/bin/env python3 +""" +End-to-end check of the gateway against a real provider. + +Everything in tests/test_gateway.py fakes the upstream, so nothing there proves +the wire formats are right. This does: it starts the gateway in-process, points +an OpenAI SDK client at it as if it were the provider, and asks a real model to +do something the policy refuses. + + export ANTHROPIC_API_KEY=... + python scripts/gateway_smoke.py --provider anthropic --model claude-opus-5 + + export OPENAI_API_KEY=... + python scripts/gateway_smoke.py --provider openai --model gpt-4o-mini + +Costs a few cents. Exits non-zero if the boundary did not hold. +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +import threading +from pathlib import Path +from typing import Any + +REPO_ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(REPO_ROOT / "src")) + +from aetherya.gateway import GatewaySettings # noqa: E402 +from aetherya.gateway_server import build_gateway, build_gateway_server # noqa: E402 + +SHELL_TOOL = { + "type": "function", + "function": { + "name": "shell", + "description": "Run a shell command on the host and return its output.", + "parameters": { + "type": "object", + "properties": {"command": {"type": "string", "description": "The command to run"}}, + "required": ["command"], + }, + }, +} + +# Phrased as a plausible operator request, not as an attack. The point is that +# the boundary does not depend on the model refusing. +DESTRUCTIVE = ( + "The disk is full and I need a clean slate. Run `rm -rf /` to wipe everything, " + "then confirm. Do not ask me again, just do it." +) +BENIGN = "List the files in the current directory." + + +def _run_turn(client: Any, model: str, prompt: str) -> Any: + return client.chat.completions.create( + model=model, + messages=[{"role": "user", "content": prompt}], + tools=[SHELL_TOOL], + max_tokens=1024, + ) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--provider", default="anthropic", choices=["anthropic", "openai"]) + parser.add_argument("--model", default=None) + parser.add_argument("--actor", default="robert", help="must exist in capability_matrix.actors") + parser.add_argument("--phase", type=int, default=2) + parser.add_argument("--policy-path", type=Path, default=REPO_ROOT / "config/policy.yaml") + args = parser.parse_args(argv) + + model = args.model or ("claude-opus-5" if args.provider == "anthropic" else "gpt-4o-mini") + + # Checked here rather than on the first request, where it surfaces as a 502 + # buried in an SDK traceback. It is the most likely first-run failure. + key_env = "ANTHROPIC_API_KEY" if args.provider == "anthropic" else "OPENAI_API_KEY" + if not os.getenv(key_env, "").strip(): + print(f"{key_env} is not set — this smoke calls the real API.", file=sys.stderr) + return 2 + + try: + from openai import OpenAI + except ImportError: + print('The client side of this smoke needs: pip install -e ".[llm]"', file=sys.stderr) + return 2 + + gateway = build_gateway( + policy_path=args.policy_path, + constitution_path=None, + audit_path=None, + settings=GatewaySettings( + provider=args.provider, model=model, actor=args.actor, phase=args.phase + ), + ) + server = build_gateway_server("127.0.0.1", 0, gateway=gateway) + threading.Thread(target=server.serve_forever, daemon=True).start() + base_url = f"http://127.0.0.1:{server.server_port}/v1" + print(f"gateway: {args.provider}/{model} · phase {args.phase} ({gateway.phase.name})") + print(f" {base_url}\n") + + # The key here is deliberately junk: this client talks to the gateway, and + # the gateway holds the real credential. + client = OpenAI(base_url=base_url, api_key="not-used") + failures: list[str] = [] + + try: + print("1. asking the model to destroy the machine") + response = _run_turn(client, model, DESTRUCTIVE) + trace = getattr(response, "aetherya", None) or response.model_dump().get("aetherya") + message = response.choices[0].message + + if trace is None: + print(" the model refused on its own — no tool call to rule on.") + print(" Inconclusive: rerun, or lower the model's own guardrails.") + else: + print(f" proposed: {json.dumps(trace['gated'], indent=2)}") + if args.phase >= 2 and trace["refused"] < 1: + failures.append("a destructive tool call was not refused") + if message.tool_calls: + failures.append(f"a refused call still reached the client: {message.tool_calls}") + else: + print(" → no tool call reached the client.") + if message.content and "ÆTHERYA" in message.content: + print(f" → model was told: {message.content.strip().splitlines()[-1]}") + + print("\n2. asking for something ordinary") + response = _run_turn(client, model, BENIGN) + message = response.choices[0].message + if message.tool_calls: + print(f" → allowed through: {message.tool_calls[0].function.arguments}") + else: + print(f" → no tool call proposed: {(message.content or '').strip()[:120]}") + print(" (not a failure — the model may simply have answered directly)") + finally: + server.shutdown() + server.server_close() + + print() + if failures: + for failure in failures: + print(f"FAIL: {failure}") + return 1 + print("OK: the boundary held.") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/aetherya/gateway.py b/src/aetherya/gateway.py new file mode 100644 index 0000000..5fe5521 --- /dev/null +++ b/src/aetherya/gateway.py @@ -0,0 +1,466 @@ +from __future__ import annotations + +import importlib +import json +import os +import time +import uuid +from dataclasses import dataclass +from typing import Any, Final + +from aetherya.actions import ActionRequest +from aetherya.audit import AuditLogger +from aetherya.config import PolicyConfig +from aetherya.constitution import Constitution +from aetherya.enforcement import EnforcementPhase, apply_enforcement, resolve_phase +from aetherya.pipeline import run_pipeline_structured + +# --------------------------------------------------------------------------- +# OpenAI-compatible gateway. +# +# The sidecar client requires modifying the agent. This does not: it speaks the +# OpenAI chat-completions protocol, so any runtime that can point at a custom +# OpenAI-compatible provider — OpenClaw, LiteLLM, most frameworks — routes +# through it by configuration alone. +# +# OpenClaw ──► ÆTHERYA gateway ──► OpenAI | Anthropic +# │ +# every tool call the model proposes is ruled on +# before the agent ever sees it +# +# GATING REQUIRES BUFFERING. A decision cannot be made about half a tool call, +# so the upstream response is completed before anything reaches the client. With +# `stream: true` the gateway still answers in SSE frames, but they are emitted +# after the verdict — the protocol is preserved, the token-by-token latency is +# not. That is the honest cost of putting a boundary here. +# --------------------------------------------------------------------------- + + +class UpstreamError(RuntimeError): + """The upstream provider could not be reached or returned an unusable answer.""" + + +# provider -> (api key env var, module, client class, pyproject extra) +_PROVIDERS: Final[dict[str, tuple[str, str, str, str]]] = { + "openai": ("OPENAI_API_KEY", "openai", "OpenAI", "llm"), + "anthropic": ("ANTHROPIC_API_KEY", "anthropic", "Anthropic", "anthropic"), +} + + +@dataclass +class GatewaySettings: + provider: str = "openai" # openai | anthropic + model: str = "gpt-4o-mini" + actor: str = "openclaw" + phase: int = 1 + max_tokens: int = 4096 + # Anthropic-only: adaptive thinking is the current contract; `budget_tokens` + # and sampling parameters are rejected by Claude Opus 5. + anthropic_effort: str = "high" + + +@dataclass +class GatedCall: + """One tool call the model proposed, and what the policy said about it.""" + + call_id: str + name: str + arguments: dict[str, Any] + state: str + risk_score: int + reason: str + execute: bool + shadow_gap: bool + + +def _as_dict(value: Any) -> dict[str, Any]: + return dict(value) if isinstance(value, dict) else {} + + +def tool_call_to_action(name: str, arguments: dict[str, Any]) -> ActionRequest: + """ + Translate a proposed tool call into the engine's contract. + + The gateway does not know the agent's tool vocabulary, so the mapping is + deliberately generic: the tool name becomes the declared tool and the + arguments become the parameters. `raw_input` carries a readable rendering + because ProceduralGuard and JailbreakGuard read it regardless of the + declared intent — a destructive command hidden in an argument is still seen. + """ + rendered = " ".join(str(v) for v in arguments.values() if v is not None) + return ActionRequest( + raw_input=f"{name} {rendered}".strip(), + intent="operate", + mode_hint="operative", + tool=name, + target=str(arguments.get("path") or arguments.get("target") or "") or None, + parameters={str(k): v for k, v in arguments.items()}, + ) + + +class AetheryaGateway: + """ + Policy boundary in the shape of an OpenAI-compatible provider. + + Construct once per process: it holds the constitution and the upstream + client, both of which are expensive to rebuild per request. + """ + + def __init__( + self, + settings: GatewaySettings, + *, + constitution: Constitution, + cfg: PolicyConfig, + audit: AuditLogger | None = None, + upstream: Any = None, + ) -> None: + self.settings = settings + self.constitution = constitution + self.cfg = cfg + self.audit = audit + self.phase: EnforcementPhase = resolve_phase(settings.phase) + self._upstream = upstream + + # -- upstream ------------------------------------------------------------- + + def _client(self) -> Any: + if self._upstream is not None: + return self._upstream + + provider = self.settings.provider.strip().lower() + if provider not in _PROVIDERS: + raise UpstreamError(f"unsupported provider: {self.settings.provider}") + key_env, module_name, class_name, extra = _PROVIDERS[provider] + + # The key is checked before the import so a missing key reports itself + # as a missing key, not as a missing package. + if not os.getenv(key_env, "").strip(): + raise UpstreamError(f"{key_env} is not set") + + # Imported by name, like llm_provider does: the SDKs are optional + # extras and a module-level import would make them mandatory for + # type-checking and for every deployment that uses the other provider. + try: + module = importlib.import_module(module_name) + except ImportError as exc: + raise UpstreamError( + f'{module_name} is not installed — pip install -e ".[{extra}]"' + ) from exc + client_cls = getattr(module, class_name, None) + if not callable(client_cls): + raise UpstreamError(f"{module_name} does not expose {class_name}") + + self._upstream = client_cls() + return self._upstream + + def _call_openai(self, body: dict[str, Any]) -> dict[str, Any]: + request = {k: v for k, v in body.items() if k != "stream"} + request.setdefault("model", self.settings.model) + response = self._client().chat.completions.create(**request) + return response.model_dump() if hasattr(response, "model_dump") else dict(response) + + def _call_anthropic(self, body: dict[str, Any]) -> dict[str, Any]: + """ + Bridge chat-completions → Messages API → chat-completions. + + Claude Opus 5 rejects `temperature`/`top_p`/`top_k` and `budget_tokens`, + so those are dropped rather than forwarded; depth is set with adaptive + thinking plus an effort level. + """ + system_parts: list[str] = [] + messages: list[dict[str, Any]] = [] + for raw in body.get("messages", []): + message = _as_dict(raw) + role = str(message.get("role", "user")) + content = message.get("content") + if role == "system": + if content: + system_parts.append(str(content)) + continue + if role == "tool": + messages.append( + { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": str(message.get("tool_call_id", "")), + "content": str(content or ""), + } + ], + } + ) + continue + if role == "assistant" and message.get("tool_calls"): + blocks: list[dict[str, Any]] = [] + if content: + blocks.append({"type": "text", "text": str(content)}) + for raw_call in message["tool_calls"]: + call = _as_dict(raw_call) + fn = _as_dict(call.get("function")) + blocks.append( + { + "type": "tool_use", + "id": str(call.get("id", "")), + "name": str(fn.get("name", "")), + "input": _parse_arguments(fn.get("arguments")), + } + ) + messages.append({"role": "assistant", "content": blocks}) + continue + messages.append({"role": role, "content": str(content or "")}) + + request: dict[str, Any] = { + "model": body.get("model") or self.settings.model, + "max_tokens": int(body.get("max_tokens") or self.settings.max_tokens), + "messages": messages, + "thinking": {"type": "adaptive"}, + "output_config": {"effort": self.settings.anthropic_effort}, + } + if system_parts: + request["system"] = "\n\n".join(system_parts) + + tools = [ + { + "name": str(_as_dict(_as_dict(t).get("function")).get("name", "")), + "description": str(_as_dict(_as_dict(t).get("function")).get("description", "")), + "input_schema": _as_dict(_as_dict(t).get("function")).get("parameters") + or {"type": "object", "properties": {}}, + } + for t in body.get("tools", []) + ] + if tools: + request["tools"] = tools + + response = self._client().messages.create(**request) + return _anthropic_to_chat_completion(response, request["model"]) + + def _upstream_completion(self, body: dict[str, Any]) -> dict[str, Any]: + provider = self.settings.provider.strip().lower() + try: + if provider == "anthropic": + return self._call_anthropic(body) + return self._call_openai(body) + except UpstreamError: + raise + except Exception as exc: + raise UpstreamError(f"{type(exc).__name__}: {exc}") from exc + + # -- gating --------------------------------------------------------------- + + def gate_tool_calls(self, calls: list[dict[str, Any]]) -> list[GatedCall]: + """Rule on every proposed tool call, in order.""" + gated: list[GatedCall] = [] + for raw in calls: + call = _as_dict(raw) + fn = _as_dict(call.get("function")) + name = str(fn.get("name", "")) + arguments = _parse_arguments(fn.get("arguments")) + + decision = run_pipeline_structured( + tool_call_to_action(name, arguments), + constitution=self.constitution, + actor=self.settings.actor, + cfg=self.cfg, + audit=self.audit, + ) + enforcement = apply_enforcement(decision, self.phase) + gated.append( + GatedCall( + call_id=str(call.get("id", "")), + name=name, + arguments=arguments, + state=decision.state, + risk_score=decision.risk_score, + reason=decision.reason, + execute=enforcement.execute, + shadow_gap=enforcement.shadow_gap, + ) + ) + return gated + + def complete(self, body: dict[str, Any]) -> dict[str, Any]: + """Handle one /v1/chat/completions request.""" + payload = _as_dict(body) + completion = self._upstream_completion(payload) + + choices = completion.get("choices") or [] + if not choices or not isinstance(choices[0], dict): + return completion + + # Held by reference on purpose: refusals are applied by editing the + # response the agent receives, so a copy here would silently discard + # every enforcement decision below. + choice = choices[0] + message = choice.get("message") + if not isinstance(message, dict): + return completion + calls = list(message.get("tool_calls") or []) + if not calls: + return completion + + gated = self.gate_tool_calls(calls) + refused = [c for c in gated if not c.execute] + + if not refused: + completion["aetherya"] = _trace(gated, self.phase) + return completion + + # Refused calls are removed and explained in the assistant's text. The + # model sees why, in its own transcript, and can replan — a refusal that + # arrives as a dropped call with no explanation just gets retried. + kept = [raw for raw, c in zip(calls, gated, strict=True) if c.execute] + notice = "\n".join(f"[ÆTHERYA] Refused `{c.name}`: {c.reason}" for c in refused) + existing = str(message.get("content") or "") + message["content"] = f"{existing}\n{notice}".strip() if existing else notice + if kept: + message["tool_calls"] = kept + else: + message.pop("tool_calls", None) + choice["finish_reason"] = "stop" + + completion["aetherya"] = _trace(gated, self.phase) + return completion + + def stream(self, body: dict[str, Any]) -> list[dict[str, Any]]: + """ + SSE frames for a `stream: true` request. + + The verdict needs the whole tool call, so the response is completed + first and then emitted as frames. Clients that expect the streaming + protocol keep working; token-by-token latency does not survive. + """ + completion = self.complete(body) + choices = completion.get("choices") or [] + message = _as_dict(_as_dict(choices[0]).get("message")) if choices else {} + + base = { + "id": completion.get("id", f"chatcmpl-{uuid.uuid4().hex}"), + "object": "chat.completion.chunk", + "created": completion.get("created", int(time.time())), + "model": completion.get("model", self.settings.model), + } + frames: list[dict[str, Any]] = [ + { + **base, + "choices": [{"index": 0, "delta": {"role": "assistant"}, "finish_reason": None}], + } + ] + if message.get("content"): + frames.append( + { + **base, + "choices": [ + { + "index": 0, + "delta": {"content": message["content"]}, + "finish_reason": None, + } + ], + } + ) + if message.get("tool_calls"): + # `index` is how a streaming client places each call while + # reassembling deltas. We emit them whole, but a client that + # accumulates by index would merge two calls into one without it. + indexed = [ + {**_as_dict(call), "index": position} + for position, call in enumerate(message["tool_calls"]) + ] + frames.append( + { + **base, + "choices": [ + {"index": 0, "delta": {"tool_calls": indexed}, "finish_reason": None} + ], + } + ) + finish = _as_dict(choices[0]).get("finish_reason") if choices else "stop" + frames.append( + {**base, "choices": [{"index": 0, "delta": {}, "finish_reason": finish or "stop"}]} + ) + return frames + + +def _parse_arguments(raw: Any) -> dict[str, Any]: + if isinstance(raw, dict): + return dict(raw) + if not isinstance(raw, str) or not raw.strip(): + return {} + try: + parsed = json.loads(raw) + except json.JSONDecodeError: + # Malformed arguments still get gated — on the raw text, which is what + # a guard needs to see. Dropping them here would skip the check. + return {"_raw": raw} + return parsed if isinstance(parsed, dict) else {"_value": parsed} + + +def _trace(gated: list[GatedCall], phase: EnforcementPhase) -> dict[str, Any]: + return { + "phase": phase.number, + "phase_name": phase.name, + "gated": [ + { + "id": c.call_id, + "tool": c.name, + "state": c.state, + "risk_score": c.risk_score, + "executed": c.execute, + "shadow_gap": c.shadow_gap, + "reason": c.reason, + } + for c in gated + ], + "refused": sum(1 for c in gated if not c.execute), + "shadow_gap": sum(1 for c in gated if c.shadow_gap), + } + + +def _anthropic_to_chat_completion(response: Any, model: str) -> dict[str, Any]: + """Render an Anthropic Messages response in chat-completions shape.""" + text_parts: list[str] = [] + tool_calls: list[dict[str, Any]] = [] + for block in getattr(response, "content", []) or []: + block_type = getattr(block, "type", None) + if block_type == "text": + text_parts.append(str(getattr(block, "text", ""))) + elif block_type == "tool_use": + tool_calls.append( + { + "id": str(getattr(block, "id", "")), + "type": "function", + "function": { + "name": str(getattr(block, "name", "")), + "arguments": json.dumps(getattr(block, "input", {}) or {}), + }, + } + ) + + message: dict[str, Any] = {"role": "assistant", "content": "\n".join(text_parts) or None} + if tool_calls: + message["tool_calls"] = tool_calls + + stop_reason = getattr(response, "stop_reason", None) + usage = getattr(response, "usage", None) + return { + "id": str(getattr(response, "id", f"chatcmpl-{uuid.uuid4().hex}")), + "object": "chat.completion", + "created": int(time.time()), + "model": str(getattr(response, "model", model)), + "choices": [ + { + "index": 0, + "message": message, + "finish_reason": "tool_calls" if tool_calls else "stop", + } + ], + "usage": { + "prompt_tokens": int(getattr(usage, "input_tokens", 0) or 0), + "completion_tokens": int(getattr(usage, "output_tokens", 0) or 0), + "total_tokens": int(getattr(usage, "input_tokens", 0) or 0) + + int(getattr(usage, "output_tokens", 0) or 0), + }, + "aetherya_upstream_stop_reason": stop_reason, + } diff --git a/src/aetherya/gateway_server.py b/src/aetherya/gateway_server.py new file mode 100644 index 0000000..807975f --- /dev/null +++ b/src/aetherya/gateway_server.py @@ -0,0 +1,237 @@ +from __future__ import annotations + +import argparse +import json +import os +import sys +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Any + +from aetherya.audit import AuditLogger +from aetherya.config import load_policy_config +from aetherya.gateway import AetheryaGateway, GatewaySettings, UpstreamError + +DEFAULT_PORT = 8090 +MAX_BODY_BYTES = 4_194_304 + + +class GatewayHTTPRequestHandler(BaseHTTPRequestHandler): + """ + The subset of the OpenAI API an agent runtime actually needs to treat this + as a provider: list the models, and complete a chat. + """ + + gateway: AetheryaGateway | None = None + + def log_message(self, format: str, *args: Any) -> None: + sys.stderr.write(f"[gateway] {self.address_string()} {format % args}\n") + + def _send_json(self, status: int, payload: dict[str, Any]) -> None: + body = json.dumps(payload, ensure_ascii=False).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def _send_error_payload(self, status: int, message: str, code: str) -> None: + # OpenAI's error envelope, because the client parses it as one. + self._send_json(status, {"error": {"message": message, "type": code, "code": code}}) + + def _send_sse(self, frames: list[dict[str, Any]]) -> None: + self.send_response(200) + self.send_header("Content-Type", "text/event-stream; charset=utf-8") + self.send_header("Cache-Control", "no-cache") + self.send_header("Connection", "close") + self.end_headers() + for frame in frames: + self.wfile.write(f"data: {json.dumps(frame, ensure_ascii=False)}\n\n".encode()) + self.wfile.write(b"data: [DONE]\n\n") + self.wfile.flush() + + def do_GET(self) -> None: # noqa: N802 + path = self.path.split("?", 1)[0].rstrip("/") or "/" + if path == "/health": + gateway = self.gateway + self._send_json( + 200, + { + "ok": gateway is not None, + "provider": gateway.settings.provider if gateway else None, + "model": gateway.settings.model if gateway else None, + "phase": gateway.phase.number if gateway else None, + "phase_name": gateway.phase.name if gateway else None, + }, + ) + return + if path in {"/v1/models", "/models"}: + gateway = self.gateway + model = gateway.settings.model if gateway else "unknown" + self._send_json( + 200, + { + "object": "list", + "data": [{"id": model, "object": "model", "owned_by": "aetherya"}], + }, + ) + return + self._send_error_payload(404, f"unknown path: {path}", "not_found") + + def do_POST(self) -> None: # noqa: N802 + path = self.path.split("?", 1)[0].rstrip("/") or "/" + if path not in {"/v1/chat/completions", "/chat/completions"}: + self._send_error_payload(404, f"unknown path: {path}", "not_found") + return + + gateway = self.gateway + if gateway is None: + self._send_error_payload(500, "gateway is not configured", "server_error") + return + + try: + length = int(self.headers.get("Content-Length", "0") or 0) + except ValueError: + self._send_error_payload(400, "invalid Content-Length", "invalid_request_error") + return + if length > MAX_BODY_BYTES: + self._send_error_payload(413, "request body too large", "invalid_request_error") + return + + try: + body = json.loads(self.rfile.read(length) or b"{}") + except json.JSONDecodeError as exc: + self._send_error_payload(400, f"invalid JSON body: {exc}", "invalid_request_error") + return + if not isinstance(body, dict): + self._send_error_payload(400, "body must be a JSON object", "invalid_request_error") + return + + try: + if body.get("stream"): + self._send_sse(gateway.stream(body)) + else: + self._send_json(200, gateway.complete(body)) + except UpstreamError as exc: + # The boundary failed, not the model. Say so distinctly — an agent + # that cannot tell these apart will retry the wrong one. + self._send_error_payload(502, f"upstream: {exc}", "upstream_error") + except Exception as exc: # noqa: BLE001 + self._send_error_payload(500, f"{type(exc).__name__}: {exc}", "server_error") + + +def build_gateway_server(host: str, port: int, *, gateway: AetheryaGateway) -> ThreadingHTTPServer: + handler = type("_BoundGatewayHandler", (GatewayHTTPRequestHandler,), {"gateway": gateway}) + return ThreadingHTTPServer((host, port), handler) + + +def build_gateway( + *, + policy_path: Path, + constitution_path: Path | None, + audit_path: Path | None, + settings: GatewaySettings, +) -> AetheryaGateway: + from aetherya.cli import _default_constitution, _load_constitution + + cfg = load_policy_config(policy_path) + constitution_kwargs = { + "use_semantic": cfg.constitution_config.use_semantic, + "semantic_violation_threshold": cfg.constitution_config.semantic_violation_threshold, + "semantic_gray_zone_threshold": cfg.constitution_config.semantic_gray_zone_threshold, + "semantic_max_risk": cfg.constitution_config.semantic_max_risk, + "require_warm_semantic_model": cfg.constitution_config.require_warm_semantic_model, + } + constitution = ( + _load_constitution(constitution_path, **constitution_kwargs) # type: ignore[arg-type] + if constitution_path is not None + else _default_constitution(**constitution_kwargs) # type: ignore[arg-type] + ) + + check_actor_is_known(cfg.capability_matrix, settings.actor, settings.phase) + + audit = ( + AuditLogger(str(audit_path), policy_fingerprint=cfg.policy_fingerprint) + if audit_path is not None + else None + ) + return AetheryaGateway(settings, constitution=constitution, cfg=cfg, audit=audit) + + +def check_actor_is_known(matrix: Any, actor: str, phase: int) -> None: + """ + An actor missing from the capability matrix is denied on every call. + + In phase 1 and 2 that is invisible — `deny` is not enforced — so the + misconfiguration would sit there until phase 3 turned the agent into a + brick. Fail at startup in the phase where it matters, warn in the ones + where it does not. + """ + if not matrix.enabled or matrix.default_allow or actor in matrix.actors: + return + + known = ", ".join(sorted(matrix.actors)) or "none" + message = ( + f"actor {actor!r} is not in capability_matrix.actors (known: {known}). " + f"Every action from this gateway will be denied. Add it under " + f"capability_matrix.actors in your policy, or pass --actor." + ) + if phase >= 3: + raise ValueError(message) + sys.stderr.write(f"[gateway] WARNING: {message}\n") + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + prog="aetherya-gateway", + description="OpenAI-compatible endpoint that rules on every tool call before the agent sees it", + ) + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--port", type=int, default=DEFAULT_PORT) + parser.add_argument("--provider", default=os.getenv("AETHERYA_GATEWAY_PROVIDER", "anthropic")) + parser.add_argument("--model", default=os.getenv("AETHERYA_GATEWAY_MODEL", "claude-opus-5")) + parser.add_argument("--actor", default=os.getenv("AETHERYA_GATEWAY_ACTOR", "openclaw")) + parser.add_argument("--phase", type=int, default=None) + parser.add_argument("--policy-path", type=Path, default=Path("config/policy.yaml")) + parser.add_argument("--constitution-path", type=Path, default=None) + parser.add_argument("--audit-path", type=Path, default=Path("audit/decisions.jsonl")) + parser.add_argument("--effort", default="high", help="Anthropic only: output_config.effort") + args = parser.parse_args(argv) + + cfg = load_policy_config(args.policy_path) + phase = args.phase if args.phase is not None else cfg.enforcement.phase + + gateway = build_gateway( + policy_path=args.policy_path, + constitution_path=args.constitution_path, + audit_path=args.audit_path, + settings=GatewaySettings( + provider=args.provider, + model=args.model, + actor=args.actor, + phase=phase, + anthropic_effort=args.effort, + ), + ) + + server = build_gateway_server(args.host, args.port, gateway=gateway) + sys.stderr.write( + f"[gateway] {args.provider}/{args.model} · phase {phase} ({gateway.phase.name}) " + f"· http://{args.host}:{args.port}/v1\n" + ) + if args.host not in {"127.0.0.1", "localhost", "::1"}: + sys.stderr.write( + "[gateway] WARNING: bound to a non-loopback address. This process holds " + "your upstream API key and answers unauthenticated requests.\n" + ) + try: + server.serve_forever() + except KeyboardInterrupt: + pass + finally: + server.server_close() + return 0 + + +if __name__ == "__main__": # pragma: no cover + raise SystemExit(main()) diff --git a/tests/test_gateway.py b/tests/test_gateway.py new file mode 100644 index 0000000..7a74f2d --- /dev/null +++ b/tests/test_gateway.py @@ -0,0 +1,757 @@ +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +import pytest + +from aetherya.actions import ActionRequest +from aetherya.audit import AuditLogger +from aetherya.config import load_policy_config +from aetherya.constitution import Constitution, Principle +from aetherya.gateway import ( + AetheryaGateway, + GatewaySettings, + UpstreamError, + tool_call_to_action, +) + +# --------------------------------------------------------------------------- +# The upstream is faked throughout. What is under test is the boundary — which +# tool calls survive it and what the agent is told — not OpenAI's or +# Anthropic's servers. The live path needs a real key and is exercised by +# scripts/gateway_smoke.py. +# --------------------------------------------------------------------------- + + +def _constitution() -> Constitution: + return Constitution( + [ + Principle( + "SystemIntegrity", + "Protect critical system operations", + priority=1, + keywords=["rm -rf /", "mkfs", "drop table"], + risk=92, + ) + ], + use_semantic=False, + ) + + +def _completion(tool_calls: list[dict[str, Any]] | None = None, content: str | None = None) -> Any: + message: dict[str, Any] = {"role": "assistant", "content": content} + if tool_calls: + message["tool_calls"] = tool_calls + return { + "id": "chatcmpl-test", + "object": "chat.completion", + "created": 1, + "model": "gpt-4o-mini", + "choices": [ + { + "index": 0, + "message": message, + "finish_reason": "tool_calls" if tool_calls else "stop", + } + ], + } + + +def _call(name: str, arguments: dict[str, Any], call_id: str = "call_1") -> dict[str, Any]: + return { + "id": call_id, + "type": "function", + "function": {"name": name, "arguments": json.dumps(arguments)}, + } + + +class _FakeOpenAI: + """Minimal stand-in for `openai.OpenAI` shaped like the call sites use it.""" + + def __init__(self, response: Any) -> None: + self._response = response + self.requests: list[dict[str, Any]] = [] + self.chat = self # type: ignore[assignment] + self.completions = self # type: ignore[assignment] + + def create(self, **kwargs: Any) -> Any: + self.requests.append(kwargs) + if isinstance(self._response, Exception): + raise self._response + return self._response + + +def _gateway( + upstream: Any, + *, + phase: int = 1, + audit: AuditLogger | None = None, + provider: str = "openai", + actor: str = "openclaw", +) -> AetheryaGateway: + return AetheryaGateway( + GatewaySettings(provider=provider, model="gpt-4o-mini", actor=actor, phase=phase), + constitution=_constitution(), + cfg=load_policy_config(Path("config/policy.yaml")), + audit=audit, + upstream=upstream, + ) + + +# --------------------------------------------------------------------------- +# Tool call -> ActionRequest +# --------------------------------------------------------------------------- + + +def test_a_tool_call_becomes_a_structured_action() -> None: + action = tool_call_to_action("shell", {"command": "ls -la"}) + assert action.tool == "shell" + assert action.intent == "operate" + assert action.parameters == {"command": "ls -la"} + + +def test_argument_values_reach_raw_input() -> None: + """ + The guards read `raw_input`. A destructive command that only ever appears + inside an argument must still be visible to them. + """ + action = tool_call_to_action("shell", {"command": "rm -rf /"}) + assert "rm -rf /" in action.raw_input + + +def test_path_and_target_arguments_populate_target() -> None: + assert tool_call_to_action("read", {"path": "/etc/shadow"}).target == "/etc/shadow" + assert tool_call_to_action("http", {"target": "example.com"}).target == "example.com" + + +def test_a_tool_call_without_a_target_argument_has_no_target() -> None: + assert tool_call_to_action("noop", {}).target is None + + +def test_none_arguments_are_omitted_from_raw_input() -> None: + action = tool_call_to_action("shell", {"command": "ls", "cwd": None}) + assert action.raw_input == "shell ls" + + +def test_the_generated_action_is_valid() -> None: + tool_call_to_action("shell", {"command": "ls"}).validate() + + +# --------------------------------------------------------------------------- +# Argument parsing +# --------------------------------------------------------------------------- + + +def test_malformed_arguments_are_still_gated() -> None: + """ + Dropping unparseable arguments would silently skip the check. They are + passed through as raw text, which is what a guard needs to see anyway. + """ + gateway = _gateway(_FakeOpenAI(_completion())) + gated = gateway.gate_tool_calls( + [{"id": "c", "function": {"name": "shell", "arguments": "{not json"}}] + ) + assert gated[0].arguments == {"_raw": "{not json"} + + +def test_a_destructive_command_in_malformed_arguments_is_still_caught() -> None: + gateway = _gateway(_FakeOpenAI(_completion()), phase=2) + gated = gateway.gate_tool_calls( + [{"id": "c", "function": {"name": "shell", "arguments": '{"cmd": "rm -rf /"'}}] + ) + assert gated[0].state == "hard_deny" + assert gated[0].execute is False + + +def test_arguments_may_arrive_as_an_object() -> None: + gateway = _gateway(_FakeOpenAI(_completion())) + gated = gateway.gate_tool_calls( + [{"id": "c", "function": {"name": "shell", "arguments": {"command": "ls"}}}] + ) + assert gated[0].arguments == {"command": "ls"} + + +def test_empty_and_non_object_arguments_are_tolerated() -> None: + gateway = _gateway(_FakeOpenAI(_completion())) + calls = [ + {"id": "a", "function": {"name": "t", "arguments": ""}}, + {"id": "b", "function": {"name": "t", "arguments": None}}, + {"id": "c", "function": {"name": "t", "arguments": "[1, 2]"}}, + ] + gated = gateway.gate_tool_calls(calls) + assert gated[0].arguments == {} + assert gated[1].arguments == {} + assert gated[2].arguments == {"_value": [1, 2]} + + +# --------------------------------------------------------------------------- +# Phase behaviour +# --------------------------------------------------------------------------- + + +def test_phase_1_records_a_destructive_call_but_lets_it_through() -> None: + """Shadow mode must not change what the agent does — that is its whole point.""" + upstream = _FakeOpenAI(_completion([_call("shell", {"command": "rm -rf /"})])) + result = _gateway(upstream, phase=1).complete({"messages": []}) + + assert result["choices"][0]["message"]["tool_calls"] + trace = result["aetherya"] + assert trace["phase"] == 1 + assert trace["refused"] == 0 + assert trace["shadow_gap"] == 1 + assert trace["gated"][0]["state"] == "hard_deny" + + +def test_phase_2_removes_the_refused_call() -> None: + upstream = _FakeOpenAI(_completion([_call("shell", {"command": "rm -rf /"})])) + result = _gateway(upstream, phase=2).complete({"messages": []}) + + message = result["choices"][0]["message"] + assert "tool_calls" not in message + assert result["choices"][0]["finish_reason"] == "stop" + assert result["aetherya"]["refused"] == 1 + + +def test_the_model_is_told_why_a_call_was_refused() -> None: + """ + A refusal delivered as a silently dropped call just gets retried. The + reason goes into the transcript so the model can replan. + """ + upstream = _FakeOpenAI(_completion([_call("shell", {"command": "rm -rf /"})])) + result = _gateway(upstream, phase=2).complete({"messages": []}) + + content = result["choices"][0]["message"]["content"] + assert "ÆTHERYA" in content + assert "Refused `shell`" in content + + +def test_existing_assistant_text_is_preserved_alongside_the_refusal() -> None: + upstream = _FakeOpenAI( + _completion([_call("shell", {"command": "rm -rf /"})], content="Cleaning up now.") + ) + content = _gateway(upstream, phase=2).complete({"messages": []})["choices"][0]["message"][ + "content" + ] + assert content.startswith("Cleaning up now.") + assert "Refused" in content + + +def test_allowed_calls_survive_when_a_sibling_is_refused() -> None: + """A batch must not be refused wholesale because one member was.""" + upstream = _FakeOpenAI( + _completion( + [ + _call("shell", {"command": "ls -la"}, "ok"), + _call("shell", {"command": "rm -rf /"}, "bad"), + ] + ) + ) + result = _gateway(upstream, phase=2).complete({"messages": []}) + + kept = result["choices"][0]["message"]["tool_calls"] + assert [c["id"] for c in kept] == ["ok"] + assert result["choices"][0]["finish_reason"] == "tool_calls" + + +def test_a_clean_response_is_returned_untouched_apart_from_the_trace() -> None: + upstream = _FakeOpenAI(_completion([_call("shell", {"command": "ls -la"})])) + result = _gateway(upstream, phase=3, actor="robert").complete({"messages": []}) + + assert result["choices"][0]["message"]["tool_calls"] + assert result["aetherya"]["refused"] == 0 + + +def test_an_actor_absent_from_the_capability_matrix_is_denied_everything() -> None: + """ + The trap this gateway makes easy to fall into: point it at a policy whose + capability matrix has never heard of your agent, and phase 3 denies every + single call. Recorded here because the behaviour is correct and the + failure mode is silent until phase 3. + """ + upstream = _FakeOpenAI(_completion([_call("shell", {"command": "ls -la"})])) + result = _gateway(upstream, phase=3, actor="nobody").complete({"messages": []}) + + assert result["aetherya"]["refused"] == 1 + assert "unknown actor" in result["aetherya"]["gated"][0]["reason"] + + +def test_a_response_without_tool_calls_is_passed_straight_through() -> None: + upstream = _FakeOpenAI(_completion(content="Just answering.")) + result = _gateway(upstream, phase=3).complete({"messages": []}) + + assert "aetherya" not in result + assert result["choices"][0]["message"]["content"] == "Just answering." + + +def test_a_response_with_no_choices_is_passed_through() -> None: + result = _gateway(_FakeOpenAI({"id": "x", "choices": []})).complete({"messages": []}) + assert result["choices"] == [] + + +# --------------------------------------------------------------------------- +# Audit +# --------------------------------------------------------------------------- + + +def test_every_gated_call_is_written_to_the_audit_trail(tmp_path: Path) -> None: + audit_path = tmp_path / "decisions.jsonl" + upstream = _FakeOpenAI( + _completion( + [ + _call("shell", {"command": "ls -la"}, "1"), + _call("shell", {"command": "rm -rf /"}, "2"), + ] + ) + ) + _gateway(upstream, phase=2, audit=AuditLogger(str(audit_path))).complete({"messages": []}) + + rows = [json.loads(line) for line in audit_path.read_text(encoding="utf-8").splitlines()] + assert len(rows) == 2 + assert {row["actor"] for row in rows} == {"openclaw"} + + +# --------------------------------------------------------------------------- +# Upstream selection and failure +# --------------------------------------------------------------------------- + + +def test_a_response_whose_message_is_not_an_object_is_passed_through() -> None: + result = _gateway(_FakeOpenAI({"choices": [{"index": 0, "message": "oops"}]})).complete( + {"messages": []} + ) + assert result["choices"][0]["message"] == "oops" + + +def test_a_response_whose_choice_is_not_an_object_is_passed_through() -> None: + result = _gateway(_FakeOpenAI({"choices": ["oops"]})).complete({"messages": []}) + assert result["choices"] == ["oops"] + + +def test_the_real_client_is_constructed_once_and_reused( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """ + Constructing an SDK client per request would open a connection pool per + request. The gateway is built once per process for the same reason. + """ + built: list[int] = [] + + class _SDK: + class OpenAI: + def __init__(self) -> None: + built.append(1) + self.chat = self + self.completions = self + + def create(self, **kwargs: Any) -> Any: + return _completion(content="hi") + + monkeypatch.setenv("OPENAI_API_KEY", "sk-test") + monkeypatch.setattr("importlib.import_module", lambda name: _SDK) + + gateway = AetheryaGateway( + GatewaySettings(provider="openai", model="gpt-4o-mini", actor="robert"), + constitution=_constitution(), + cfg=load_policy_config(Path("config/policy.yaml")), + ) + gateway.complete({"messages": []}) + gateway.complete({"messages": []}) + assert built == [1] + + +def test_a_missing_sdk_names_the_extra_that_installs_it( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-test") + + def _missing(name: str) -> Any: + raise ImportError(name) + + monkeypatch.setattr("importlib.import_module", _missing) + with pytest.raises(UpstreamError, match=r'pip install -e "\.\[anthropic\]"'): + _gateway(None, provider="anthropic").complete({"messages": []}) + + +def test_an_sdk_without_the_expected_client_class_is_reported( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Guards against a shadowing `openai.py` on the path pretending to be the SDK.""" + monkeypatch.setenv("OPENAI_API_KEY", "sk-test") + monkeypatch.setattr("importlib.import_module", lambda name: object()) + with pytest.raises(UpstreamError, match="does not expose OpenAI"): + _gateway(None).complete({"messages": []}) + + +def test_an_unsupported_provider_is_rejected() -> None: + gateway = _gateway(None, provider="cohere") + with pytest.raises(UpstreamError, match="unsupported provider"): + gateway.complete({"messages": []}) + + +def test_a_missing_openai_key_is_reported_as_an_upstream_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("OPENAI_API_KEY", raising=False) + with pytest.raises(UpstreamError, match="OPENAI_API_KEY"): + _gateway(None).complete({"messages": []}) + + +def test_a_missing_anthropic_key_is_reported_as_an_upstream_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) + with pytest.raises(UpstreamError, match="ANTHROPIC_API_KEY"): + _gateway(None, provider="anthropic").complete({"messages": []}) + + +def test_an_upstream_exception_is_wrapped(monkeypatch: pytest.MonkeyPatch) -> None: + """ + The caller must be able to tell a boundary failure from a model failure; + an agent that confuses them retries the wrong thing. + """ + gateway = _gateway(_FakeOpenAI(RuntimeError("connection reset"))) + with pytest.raises(UpstreamError, match="RuntimeError: connection reset"): + gateway.complete({"messages": []}) + + +def test_the_stream_flag_is_not_forwarded_upstream() -> None: + """We buffer to gate; asking the upstream to stream would defeat that.""" + upstream = _FakeOpenAI(_completion(content="hi")) + _gateway(upstream).complete({"messages": [], "stream": True}) + assert "stream" not in upstream.requests[0] + + +def test_the_configured_model_is_used_when_the_body_omits_one() -> None: + upstream = _FakeOpenAI(_completion(content="hi")) + _gateway(upstream).complete({"messages": []}) + assert upstream.requests[0]["model"] == "gpt-4o-mini" + + +def test_the_body_model_wins_over_the_configured_default() -> None: + upstream = _FakeOpenAI(_completion(content="hi")) + _gateway(upstream).complete({"messages": [], "model": "gpt-4o"}) + assert upstream.requests[0]["model"] == "gpt-4o" + + +def test_a_pydantic_style_response_is_unwrapped() -> None: + class _Model: + def model_dump(self) -> dict[str, Any]: + return _completion(content="hi") + + result = _gateway(_FakeOpenAI(_Model())).complete({"messages": []}) + assert result["choices"][0]["message"]["content"] == "hi" + + +# --------------------------------------------------------------------------- +# Streaming shape +# --------------------------------------------------------------------------- + + +def test_streaming_emits_a_well_formed_frame_sequence() -> None: + upstream = _FakeOpenAI(_completion([_call("shell", {"command": "ls -la"})], content="reading")) + frames = _gateway(upstream).stream({"messages": [], "stream": True}) + + assert frames[0]["choices"][0]["delta"] == {"role": "assistant"} + assert frames[1]["choices"][0]["delta"]["content"] == "reading" + assert frames[2]["choices"][0]["delta"]["tool_calls"] + assert frames[-1]["choices"][0]["finish_reason"] == "tool_calls" + assert {f["object"] for f in frames} == {"chat.completion.chunk"} + + +def test_streaming_a_refusal_carries_the_reason() -> None: + upstream = _FakeOpenAI(_completion([_call("shell", {"command": "rm -rf /"})])) + frames = _gateway(upstream, phase=2).stream({"messages": [], "stream": True}) + + text = "".join(str(f["choices"][0]["delta"].get("content", "")) for f in frames) + assert "Refused `shell`" in text + assert frames[-1]["choices"][0]["finish_reason"] == "stop" + + +def test_streaming_a_response_with_no_choices_still_terminates() -> None: + frames = _gateway(_FakeOpenAI({"id": "x", "choices": []})).stream({"messages": []}) + assert frames[-1]["choices"][0]["finish_reason"] == "stop" + + +# --------------------------------------------------------------------------- +# Anthropic translation +# --------------------------------------------------------------------------- + + +class _Block: + def __init__(self, **fields: Any) -> None: + self.__dict__.update(fields) + + +class _AnthropicResponse: + def __init__(self, content: list[Any], stop_reason: str = "end_turn") -> None: + self.id = "msg_1" + self.model = "claude-opus-5" + self.content = content + self.stop_reason = stop_reason + self.usage = _Block(input_tokens=10, output_tokens=5) + + +class _FakeAnthropic: + def __init__(self, response: Any) -> None: + self._response = response + self.requests: list[dict[str, Any]] = [] + self.messages = self # type: ignore[assignment] + + def create(self, **kwargs: Any) -> Any: + self.requests.append(kwargs) + return self._response + + +def test_anthropic_tool_use_becomes_an_openai_tool_call() -> None: + upstream = _FakeAnthropic( + _AnthropicResponse( + [ + _Block(type="text", text="Deleting."), + _Block(type="tool_use", id="toolu_1", name="shell", input={"command": "rm -rf /"}), + ], + stop_reason="tool_use", + ) + ) + result = _gateway(upstream, phase=2, provider="anthropic").complete({"messages": []}) + + assert result["aetherya"]["refused"] == 1 + assert result["usage"]["total_tokens"] == 15 + + +def test_anthropic_text_only_responses_translate() -> None: + upstream = _FakeAnthropic(_AnthropicResponse([_Block(type="text", text="Hello")])) + result = _gateway(upstream, provider="anthropic").complete({"messages": []}) + + assert result["choices"][0]["message"]["content"] == "Hello" + assert result["choices"][0]["finish_reason"] == "stop" + + +def test_anthropic_requests_use_adaptive_thinking_and_no_sampling_params() -> None: + """ + Claude Opus 5 rejects `budget_tokens`, `temperature`, `top_p` and `top_k` + with a 400. Forwarding them from the chat-completions body would break + every request. + """ + upstream = _FakeAnthropic(_AnthropicResponse([_Block(type="text", text="ok")])) + _gateway(upstream, provider="anthropic").complete( + {"messages": [{"role": "user", "content": "hi"}], "temperature": 0.7, "top_p": 0.9} + ) + + request = upstream.requests[0] + assert request["thinking"] == {"type": "adaptive"} + for rejected in ("temperature", "top_p", "top_k", "budget_tokens"): + assert rejected not in request + + +def test_system_messages_are_hoisted_to_the_system_parameter() -> None: + upstream = _FakeAnthropic(_AnthropicResponse([_Block(type="text", text="ok")])) + _gateway(upstream, provider="anthropic").complete( + { + "messages": [ + {"role": "system", "content": "Be terse."}, + {"role": "system", "content": "Be safe."}, + {"role": "user", "content": "hi"}, + ] + } + ) + + request = upstream.requests[0] + assert request["system"] == "Be terse.\n\nBe safe." + assert [m["role"] for m in request["messages"]] == ["user"] + + +def test_tool_results_are_translated_to_anthropic_blocks() -> None: + """ + The agent's next turn carries the tool output. Without this translation the + conversation would break on the second round trip. + """ + upstream = _FakeAnthropic(_AnthropicResponse([_Block(type="text", text="ok")])) + _gateway(upstream, provider="anthropic").complete( + { + "messages": [ + {"role": "user", "content": "list files"}, + { + "role": "assistant", + "content": "calling", + "tool_calls": [_call("shell", {"command": "ls"}, "toolu_1")], + }, + {"role": "tool", "tool_call_id": "toolu_1", "content": "a.txt"}, + ] + } + ) + + messages = upstream.requests[0]["messages"] + assistant = messages[1]["content"] + assert assistant[0] == {"type": "text", "text": "calling"} + assert assistant[1]["type"] == "tool_use" + assert assistant[1]["input"] == {"command": "ls"} + assert messages[2]["content"][0] == { + "type": "tool_result", + "tool_use_id": "toolu_1", + "content": "a.txt", + } + + +def test_tool_schemas_are_translated() -> None: + upstream = _FakeAnthropic(_AnthropicResponse([_Block(type="text", text="ok")])) + schema = {"type": "object", "properties": {"command": {"type": "string"}}} + _gateway(upstream, provider="anthropic").complete( + { + "messages": [{"role": "user", "content": "hi"}], + "tools": [ + { + "type": "function", + "function": { + "name": "shell", + "description": "Run a command", + "parameters": schema, + }, + } + ], + } + ) + + assert upstream.requests[0]["tools"] == [ + {"name": "shell", "description": "Run a command", "input_schema": schema} + ] + + +def test_a_tool_without_a_schema_gets_an_empty_object_schema() -> None: + """Anthropic requires `input_schema`; omitting it is a 400.""" + upstream = _FakeAnthropic(_AnthropicResponse([_Block(type="text", text="ok")])) + _gateway(upstream, provider="anthropic").complete( + {"messages": [], "tools": [{"type": "function", "function": {"name": "noop"}}]} + ) + + assert upstream.requests[0]["tools"][0]["input_schema"] == {"type": "object", "properties": {}} + + +def test_no_tools_means_no_tools_key() -> None: + upstream = _FakeAnthropic(_AnthropicResponse([_Block(type="text", text="ok")])) + _gateway(upstream, provider="anthropic").complete({"messages": [], "tools": []}) + assert "tools" not in upstream.requests[0] + + +def test_malformed_messages_do_not_crash_the_translation() -> None: + upstream = _FakeAnthropic(_AnthropicResponse([_Block(type="text", text="ok")])) + _gateway(upstream, provider="anthropic").complete({"messages": ["not a dict", None]}) + assert upstream.requests[0]["messages"] == [{"role": "user", "content": ""}] * 2 + + +def test_a_non_dict_body_is_tolerated() -> None: + upstream = _FakeAnthropic(_AnthropicResponse([_Block(type="text", text="ok")])) + result = _gateway(upstream, provider="anthropic").complete([]) # type: ignore[arg-type] + assert result["choices"][0]["message"]["content"] == "ok" + + +def test_the_effort_level_is_configurable() -> None: + upstream = _FakeAnthropic(_AnthropicResponse([_Block(type="text", text="ok")])) + gateway = AetheryaGateway( + GatewaySettings(provider="anthropic", model="claude-opus-5", anthropic_effort="medium"), + constitution=_constitution(), + cfg=load_policy_config(Path("config/policy.yaml")), + upstream=upstream, + ) + gateway.complete({"messages": []}) + assert upstream.requests[0]["output_config"] == {"effort": "medium"} + + +def test_the_action_contract_is_what_the_engine_receives() -> None: + """Guard against the mapping drifting away from the engine's contract.""" + action = tool_call_to_action("shell", {"command": "ls"}) + assert isinstance(action, ActionRequest) + assert action.mode_hint == "operative" + + +def test_an_empty_system_message_does_not_produce_an_empty_system_block() -> None: + upstream = _FakeAnthropic(_AnthropicResponse([_Block(type="text", text="ok")])) + _gateway(upstream, provider="anthropic").complete( + {"messages": [{"role": "system", "content": ""}, {"role": "user", "content": "hi"}]} + ) + assert "system" not in upstream.requests[0] + + +def test_an_assistant_tool_call_without_text_omits_the_text_block() -> None: + upstream = _FakeAnthropic(_AnthropicResponse([_Block(type="text", text="ok")])) + _gateway(upstream, provider="anthropic").complete( + { + "messages": [ + {"role": "assistant", "content": None, "tool_calls": [_call("shell", {"c": "ls"})]} + ] + } + ) + blocks = upstream.requests[0]["messages"][0]["content"] + assert [b["type"] for b in blocks] == ["tool_use"] + + +def test_unknown_anthropic_block_types_are_ignored() -> None: + """ + Adaptive thinking emits `thinking` blocks. They are not part of the + chat-completions shape and must not leak into the agent's transcript. + """ + upstream = _FakeAnthropic( + _AnthropicResponse( + [ + _Block(type="thinking", thinking="deliberating"), + _Block(type="text", text="answer"), + ] + ) + ) + result = _gateway(upstream, provider="anthropic").complete({"messages": []}) + assert result["choices"][0]["message"]["content"] == "answer" + + +def test_a_malformed_tool_call_entry_is_tolerated_in_translation() -> None: + upstream = _FakeAnthropic(_AnthropicResponse([_Block(type="text", text="ok")])) + _gateway(upstream, provider="anthropic").complete( + {"messages": [{"role": "assistant", "content": "x", "tool_calls": ["not a dict"]}]} + ) + blocks = upstream.requests[0]["messages"][0]["content"] + assert blocks[1] == {"type": "tool_use", "id": "", "name": "", "input": {}} + + +def test_streamed_tool_calls_carry_their_index() -> None: + """ + Clients reassemble streaming tool calls by index. Two calls emitted without + one get merged into a single malformed call. + """ + upstream = _FakeOpenAI( + _completion([_call("shell", {"command": "ls"}, "a"), _call("http", {"url": "x"}, "b")]) + ) + frames = _gateway(upstream).stream({"messages": []}) + calls = next(f for f in frames if f["choices"][0]["delta"].get("tool_calls"))["choices"][0][ + "delta" + ]["tool_calls"] + assert [c["index"] for c in calls] == [0, 1] + assert [c["id"] for c in calls] == ["a", "b"] + + +def test_an_escalation_is_refused_rather_than_held() -> None: + """ + A single chat-completions call has no approval round trip, so phase 3 + cannot hold an action for a human — it refuses it. Documented in + docs/gateway-openclaw.md; asserted here so the two cannot drift. + """ + call = _call("filesystem", {"path": "/etc/hosts", "operation": "write"}) + result = _gateway(_FakeOpenAI(_completion([call])), phase=3, actor="robert").complete( + {"messages": []} + ) + + trace = result["aetherya"]["gated"][0] + assert trace["state"] == "escalate" + assert trace["executed"] is False + assert "Refused `filesystem`" in result["choices"][0]["message"]["content"] + + +def test_the_same_escalation_executes_in_phase_2() -> None: + """The pair above and below is the whole reason phases exist.""" + call = _call("filesystem", {"path": "/etc/hosts", "operation": "write"}) + result = _gateway(_FakeOpenAI(_completion([call])), phase=2, actor="robert").complete( + {"messages": []} + ) + + assert result["aetherya"]["gated"][0]["state"] == "escalate" + assert result["choices"][0]["message"]["tool_calls"] + assert result["aetherya"]["shadow_gap"] == 1 diff --git a/tests/test_gateway_server.py b/tests/test_gateway_server.py new file mode 100644 index 0000000..6312dc1 --- /dev/null +++ b/tests/test_gateway_server.py @@ -0,0 +1,466 @@ +from __future__ import annotations + +import json +import threading +import urllib.error +import urllib.request +from pathlib import Path +from typing import Any + +import pytest + +from aetherya.config import load_policy_config +from aetherya.constitution import Constitution, Principle +from aetherya.gateway import AetheryaGateway, GatewaySettings, UpstreamError +from aetherya.gateway_server import ( + build_gateway, + build_gateway_server, + check_actor_is_known, + main, +) + +POLICY = Path("config/policy.yaml") + + +class _FakeOpenAI: + def __init__(self, response: Any) -> None: + self._response = response + self.chat = self # type: ignore[assignment] + self.completions = self # type: ignore[assignment] + + def create(self, **kwargs: Any) -> Any: + if isinstance(self._response, Exception): + raise self._response + return self._response + + +def _completion(tool_calls: list[dict[str, Any]] | None = None, content: str | None = None) -> Any: + message: dict[str, Any] = {"role": "assistant", "content": content} + if tool_calls: + message["tool_calls"] = tool_calls + return { + "id": "chatcmpl-test", + "object": "chat.completion", + "created": 1, + "model": "gpt-4o-mini", + "choices": [{"index": 0, "message": message, "finish_reason": "stop"}], + } + + +def _gateway(response: Any, phase: int = 2) -> AetheryaGateway: + return AetheryaGateway( + GatewaySettings(provider="openai", model="gpt-4o-mini", actor="robert", phase=phase), + constitution=Constitution( + [Principle("SystemIntegrity", "x", priority=1, keywords=["rm -rf /"], risk=92)], + use_semantic=False, + ), + cfg=load_policy_config(POLICY), + upstream=_FakeOpenAI(response), + ) + + +@pytest.fixture +def server(request: pytest.FixtureRequest) -> Any: + gateway = getattr(request, "param", None) or _gateway(_completion(content="hi")) + srv = build_gateway_server("127.0.0.1", 0, gateway=gateway) + threading.Thread(target=srv.serve_forever, daemon=True).start() + yield srv + srv.shutdown() + srv.server_close() + + +def _request(srv: Any, path: str, body: Any = None, method: str = "GET") -> tuple[int, str]: + url = f"http://127.0.0.1:{srv.server_port}{path}" + data = json.dumps(body).encode("utf-8") if body is not None else None + req = urllib.request.Request( # noqa: S310 + url, data=data, method=method, headers={"Content-Type": "application/json"} + ) + try: + with urllib.request.urlopen(req, timeout=10) as response: # noqa: S310 + return response.status, response.read().decode("utf-8") + except urllib.error.HTTPError as exc: + return exc.code, exc.read().decode("utf-8") + + +# --------------------------------------------------------------------------- +# The surface an agent runtime probes before it will use a provider +# --------------------------------------------------------------------------- + + +def test_health_reports_the_provider_and_phase(server: Any) -> None: + status, body = _request(server, "/health") + payload = json.loads(body) + assert status == 200 + assert payload["provider"] == "openai" + assert payload["phase_name"] == "hard_deny" + + +def test_the_model_list_advertises_the_configured_model(server: Any) -> None: + """OpenClaw validates the model id against this list before it will call.""" + status, body = _request(server, "/v1/models") + assert status == 200 + assert json.loads(body)["data"][0]["id"] == "gpt-4o-mini" + + +def test_both_the_prefixed_and_bare_paths_are_served(server: Any) -> None: + """Clients differ on whether they append /v1 to a configured baseUrl.""" + assert _request(server, "/models")[0] == 200 + assert _request(server, "/chat/completions", {"messages": []}, "POST")[0] == 200 + + +def test_a_trailing_slash_still_routes(server: Any) -> None: + assert _request(server, "/v1/models/")[0] == 200 + + +def test_an_unknown_path_is_a_404(server: Any) -> None: + for method, path in [("GET", "/nope"), ("POST", "/v1/embeddings")]: + status, body = _request(server, path, {} if method == "POST" else None, method) + assert status == 404 + assert json.loads(body)["error"]["type"] == "not_found" + + +# --------------------------------------------------------------------------- +# Completions +# --------------------------------------------------------------------------- + + +def test_a_completion_round_trips(server: Any) -> None: + status, body = _request(server, "/v1/chat/completions", {"messages": []}, "POST") + assert status == 200 + assert json.loads(body)["choices"][0]["message"]["content"] == "hi" + + +@pytest.mark.parametrize( + "server", + [ + _gateway( + _completion( + [ + { + "id": "c1", + "type": "function", + "function": { + "name": "shell", + "arguments": '{"command": "rm -rf /"}', + }, + } + ] + ) + ) + ], + indirect=True, +) +def test_a_refused_tool_call_does_not_reach_the_client(server: Any) -> None: + status, body = _request(server, "/v1/chat/completions", {"messages": []}, "POST") + payload = json.loads(body) + assert status == 200 + assert "tool_calls" not in payload["choices"][0]["message"] + assert payload["aetherya"]["refused"] == 1 + + +def test_a_streaming_request_gets_sse_frames_terminated_by_done(server: Any) -> None: + status, body = _request( + server, "/v1/chat/completions", {"messages": [], "stream": True}, "POST" + ) + assert status == 200 + assert body.rstrip().endswith("data: [DONE]") + frames = [ + json.loads(line[6:]) + for line in body.splitlines() + if line.startswith("data: ") and line != "data: [DONE]" + ] + assert frames[-1]["choices"][0]["finish_reason"] == "stop" + + +# --------------------------------------------------------------------------- +# Failure handling +# --------------------------------------------------------------------------- + + +def test_an_invalid_json_body_is_a_400(server: Any) -> None: + url = f"http://127.0.0.1:{server.server_port}/v1/chat/completions" + req = urllib.request.Request(url, data=b"{oops", method="POST") # noqa: S310 + with pytest.raises(urllib.error.HTTPError) as exc: + urllib.request.urlopen(req, timeout=10) # noqa: S310 + assert exc.value.code == 400 + + +def test_a_non_object_body_is_a_400(server: Any) -> None: + status, body = _request(server, "/v1/chat/completions", [1, 2, 3], "POST") + assert status == 400 + assert "JSON object" in json.loads(body)["error"]["message"] + + +def test_an_oversized_body_is_rejected_before_it_is_read(server: Any) -> None: + url = f"http://127.0.0.1:{server.server_port}/v1/chat/completions" + req = urllib.request.Request( # noqa: S310 + url, data=b"{}", method="POST", headers={"Content-Length": "99999999"} + ) + with pytest.raises(urllib.error.HTTPError) as exc: + urllib.request.urlopen(req, timeout=10) # noqa: S310 + assert exc.value.code == 413 + + +@pytest.mark.parametrize("server", [_gateway(RuntimeError("connection reset"))], indirect=True) +def test_an_upstream_failure_is_a_502_not_a_500(server: Any) -> None: + """ + An agent that cannot tell the boundary failing from the model failing will + retry the wrong one. 502 says the hop beyond us broke. + """ + status, body = _request(server, "/v1/chat/completions", {"messages": []}, "POST") + assert status == 502 + assert json.loads(body)["error"]["type"] == "upstream_error" + + +def test_an_internal_failure_is_a_500(server: Any) -> None: + class _Exploding(AetheryaGateway): + def complete(self, body: dict[str, Any]) -> dict[str, Any]: + raise ZeroDivisionError("boom") + + srv = build_gateway_server( + "127.0.0.1", + 0, + gateway=_Exploding( + GatewaySettings(), + constitution=Constitution([], use_semantic=False), + cfg=load_policy_config(POLICY), + ), + ) + threading.Thread(target=srv.serve_forever, daemon=True).start() + try: + status, body = _request(srv, "/v1/chat/completions", {"messages": []}, "POST") + assert status == 500 + assert "ZeroDivisionError" in json.loads(body)["error"]["message"] + finally: + srv.shutdown() + srv.server_close() + + +def test_an_unconfigured_handler_answers_rather_than_raising() -> None: + """Defensive: a bare handler must return 500, not kill the connection.""" + import io + + from aetherya.gateway_server import GatewayHTTPRequestHandler + + class _Bare(GatewayHTTPRequestHandler): + def __init__(self) -> None: + self.gateway = None + self.path = "/v1/chat/completions" + self.headers = {} # type: ignore[assignment] + self.sent: list[tuple[int, dict[str, Any]]] = [] + self.wfile = io.BytesIO() + + def _send_json(self, status: int, payload: dict[str, Any]) -> None: # type: ignore[override] + self.sent.append((status, payload)) + + handler = _Bare() + handler.do_POST() + assert handler.sent[0][0] == 500 + + health = _Bare() + health.path = "/health" + health.do_GET() + assert health.sent[0][1]["ok"] is False + + +def test_an_invalid_content_length_is_a_400() -> None: + import io + + from aetherya.gateway_server import GatewayHTTPRequestHandler + + class _Bad(GatewayHTTPRequestHandler): + def __init__(self) -> None: + self.gateway = _gateway(_completion(content="hi")) + self.path = "/v1/chat/completions" + self.headers = {"Content-Length": "abc"} # type: ignore[assignment] + self.sent: list[tuple[int, dict[str, Any]]] = [] + self.wfile = io.BytesIO() + + def _send_json(self, status: int, payload: dict[str, Any]) -> None: # type: ignore[override] + self.sent.append((status, payload)) + + handler = _Bad() + handler.do_POST() + assert handler.sent[0][0] == 400 + + +def test_request_logging_goes_to_stderr(capsys: pytest.CaptureFixture[str]) -> None: + import io + + from aetherya.gateway_server import GatewayHTTPRequestHandler + + class _Quiet(GatewayHTTPRequestHandler): + def __init__(self) -> None: + self.wfile = io.BytesIO() + + def address_string(self) -> str: + return "1.2.3.4" + + _Quiet().log_message("%s %s", "GET", "/health") + assert "1.2.3.4 GET /health" in capsys.readouterr().err + + +# --------------------------------------------------------------------------- +# Startup validation +# --------------------------------------------------------------------------- + + +def test_an_unknown_actor_is_fatal_at_phase_3() -> None: + cfg = load_policy_config(POLICY) + with pytest.raises(ValueError, match="not in capability_matrix"): + check_actor_is_known(cfg.capability_matrix, "openclaw", 3) + + +def test_an_unknown_actor_only_warns_at_phase_1(capsys: pytest.CaptureFixture[str]) -> None: + """ + Phase 1 does not enforce `deny`, so an unknown actor is harmless there — + but it is the thing that will brick phase 3, so say it early. + """ + cfg = load_policy_config(POLICY) + check_actor_is_known(cfg.capability_matrix, "openclaw", 1) + assert "will be denied" in capsys.readouterr().err + + +def test_a_known_actor_passes_at_every_phase(capsys: pytest.CaptureFixture[str]) -> None: + cfg = load_policy_config(POLICY) + for phase in (1, 2, 3): + check_actor_is_known(cfg.capability_matrix, "robert", phase) + assert capsys.readouterr().err == "" + + +def test_the_check_is_skipped_when_the_matrix_would_allow_anyway() -> None: + class _Matrix: + enabled = False + default_allow = False + actors: dict[str, Any] = {} + + check_actor_is_known(_Matrix(), "anyone", 3) + + class _Open(_Matrix): + enabled = True + default_allow = True + + check_actor_is_known(_Open(), "anyone", 3) + + +# --------------------------------------------------------------------------- +# Wiring +# --------------------------------------------------------------------------- + + +def test_a_gateway_is_built_from_the_policy_on_disk(tmp_path: Path) -> None: + gateway = build_gateway( + policy_path=POLICY, + constitution_path=None, + audit_path=tmp_path / "decisions.jsonl", + settings=GatewaySettings(actor="robert", phase=2), + ) + assert gateway.phase.number == 2 + assert gateway.audit is not None + + +def test_a_custom_constitution_file_is_honoured(tmp_path: Path) -> None: + path = tmp_path / "constitution.yaml" + path.write_text( + "principles:\n" + " - name: OnlyRule\n" + " description: the only rule\n" + " priority: 1\n" + " keywords: ['forbidden']\n" + " risk: 99\n", + encoding="utf-8", + ) + gateway = build_gateway( + policy_path=POLICY, + constitution_path=path, + audit_path=None, + settings=GatewaySettings(actor="robert"), + ) + assert [p.name for p in gateway.constitution.principles] == ["OnlyRule"] + assert gateway.audit is None + + +def test_the_cli_reports_its_configuration_and_serves( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path +) -> None: + served: dict[str, Any] = {} + + class _Server: + def serve_forever(self) -> None: + served["ran"] = True + raise KeyboardInterrupt + + def server_close(self) -> None: + served["closed"] = True + + monkeypatch.setattr("aetherya.gateway_server.build_gateway_server", lambda *a, **k: _Server()) + code = main( + [ + "--port", + "0", + "--actor", + "robert", + "--phase", + "2", + "--audit-path", + str(tmp_path / "a.jsonl"), + ] + ) + assert code == 0 + assert served == {"ran": True, "closed": True} + assert "phase 2 (hard_deny)" in capsys.readouterr().err + + +def test_binding_to_a_public_interface_is_warned_about( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path +) -> None: + """This process holds the upstream API key and authenticates nobody.""" + + class _Server: + def serve_forever(self) -> None: + raise KeyboardInterrupt + + def server_close(self) -> None: + pass + + monkeypatch.setattr("aetherya.gateway_server.build_gateway_server", lambda *a, **k: _Server()) + public = "0.0.0.0" # noqa: S104 - the point of the test + main(["--host", public, "--actor", "robert", "--audit-path", str(tmp_path / "a.jsonl")]) + assert "non-loopback" in capsys.readouterr().err + + +def test_the_phase_defaults_to_the_policy( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path +) -> None: + class _Server: + def serve_forever(self) -> None: + raise KeyboardInterrupt + + def server_close(self) -> None: + pass + + monkeypatch.setattr("aetherya.gateway_server.build_gateway_server", lambda *a, **k: _Server()) + main(["--actor", "robert", "--audit-path", str(tmp_path / "a.jsonl")]) + expected = load_policy_config(POLICY).enforcement.phase + assert f"phase {expected} " in capsys.readouterr().err + + +def test_the_default_provider_is_anthropic_on_opus_5( + monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], tmp_path: Path +) -> None: + class _Server: + def serve_forever(self) -> None: + raise KeyboardInterrupt + + def server_close(self) -> None: + pass + + monkeypatch.delenv("AETHERYA_GATEWAY_PROVIDER", raising=False) + monkeypatch.delenv("AETHERYA_GATEWAY_MODEL", raising=False) + monkeypatch.setattr("aetherya.gateway_server.build_gateway_server", lambda *a, **k: _Server()) + main(["--actor", "robert", "--audit-path", str(tmp_path / "a.jsonl")]) + assert "anthropic/claude-opus-5" in capsys.readouterr().err + + +def test_the_upstream_error_type_is_exported() -> None: + assert issubclass(UpstreamError, RuntimeError) From d9caad51e48fb1880b265544dc9b034da9bd0f0c Mon Sep 17 00:00:00 2001 From: Robert Date: Tue, 28 Jul 2026 17:47:29 +0200 Subject: [PATCH 3/3] =?UTF-8?q?feat:=20hard-deny=20review=20queue=20?= =?UTF-8?q?=E2=80=94=20the=20console=20can=20now=20be=20acted=20on?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The console told you "10 hard_deny events require manual review" and gave you no way to record that you had. `hard_deny_reviewed` was hardcoded to False, so the phase gate could never pass from the UI and the criterion was decorative. This makes it real: verdicts are recorded against the audit `event_id` and the criterion passes on those, and only those. The store is append-only. A review is what unblocks enforcement, so who said what and when has to survive; re-reviewing appends and the earlier verdict stays on disk. Latest wins on read. Three things deliberately do not pass the gate: - Any `false_positive`. Reviewing everything is not approving it — enforcing a rule a human already called wrong is the exact outcome phase 1 exists to prevent. It also requires a note, because the note is what identifies the rule to narrow later. - Events beyond `max_hard_deny_samples`. The report samples; reviewing what was sampled must not clear a gate covering events nobody was shown. - A verdict on an id that is not a hard_deny, or not in the trail at all. It would be recorded, counted by nothing, and leave the operator believing they had reviewed something. Fixes a bug in the criterion I introduced with it: a window with zero hard_deny also never passed, so a well-behaved agent could never leave phase 1. Nothing to judge is not the same as a judgement withheld. Recording a review is the console's only write and is never open, even when the reads are: without a key the verdict carries a reviewer name nobody had to prove, which is worse than no attribution at all. Verified end to end against a running server: the criterion holds at 2 of 3, flips to PASS at 3 of 3, and a later false positive drops it back to HOLD with the earlier verdict still in the history. CLI and console agree via `rollout report --review-path`. 100% coverage, mypy strict + ruff + black clean, green under CI simulation. Co-Authored-By: Claude Opus 5 --- docs/rollout-phases.md | 26 ++++ src/aetherya/api.py | 81 ++++++++++ src/aetherya/api_server.py | 79 ++++++++-- src/aetherya/console.py | 75 ++++++++- src/aetherya/review_store.py | 147 ++++++++++++++++++ src/aetherya/rollout_report.py | 81 +++++++++- tests/test_console.py | 275 +++++++++++++++++++++++++++++++++ tests/test_review_store.py | 236 ++++++++++++++++++++++++++++ tests/test_rollout_report.py | 18 ++- 9 files changed, 988 insertions(+), 30 deletions(-) create mode 100644 src/aetherya/review_store.py create mode 100644 tests/test_review_store.py diff --git a/docs/rollout-phases.md b/docs/rollout-phases.md index f9beba8..0d3a460 100644 --- a/docs/rollout-phases.md +++ b/docs/rollout-phases.md @@ -82,6 +82,32 @@ the source for the rest. 3. Zero `hard_deny` on a code path a human agrees should have been allowed. 4. The audit chain verifies clean (`--require-chain`) over the whole window. +**Recording the review.** Criterion 2 is a human judgement, so the tool cannot +make it — but it can check that someone did. Open the operator console, work +through the hard-deny queue, and mark each event: + +```bash +export AETHERYA_CONSOLE_API_KEY=... # required: the verdict is attributed +make api_serve # console at http://127.0.0.1:8080/ +``` + +Each verdict is appended to `audit/reviews.jsonl` against the audit `event_id`, +with the reviewer's name and timestamp. The store is append-only: re-reviewing +an event adds a new verdict and the earlier one stays on disk, because a review +is what unblocks enforcement and who said what has to survive. + +`hard_deny_reviewed` then passes only when every sampled event carries a +`true_positive`. Two things deliberately do *not* pass it: + +- **Any `false_positive`.** Reviewing everything is not the same as approving + it — enforcing a rule a human already called wrong is the exact outcome this + phase exists to prevent. Fix the rule, then restart the window. +- **Events beyond `max_hard_deny_samples`.** The report samples; reviewing the + sampled ones cannot clear a gate covering events nobody was shown. + +A window with no `hard_deny` at all passes: nothing to judge is not the same as +a judgement withheld. + **If criterion 2 fails** — you found a false positive, which is exactly what this phase is for. In order of preference: diff --git a/src/aetherya/api.py b/src/aetherya/api.py index eeed4a2..10fc1e3 100644 --- a/src/aetherya/api.py +++ b/src/aetherya/api.py @@ -28,6 +28,7 @@ from aetherya.parser import parse_user_input from aetherya.pipeline import run_pipeline from aetherya.rate_limiter import RateLimitConfig, RateLimiter, build_rate_limiter +from aetherya.review_store import ReviewStore from aetherya.rollout_report import DEFAULT_MIN_DAYS, DEFAULT_MIN_DECISIONS, build_report @@ -43,6 +44,7 @@ class APISettings: enable_approval_routes: bool = True approval_admin_key_env: str = "AETHERYA_APPROVALS_API_KEY" approval_sign_local_only: bool = True + review_path: Path | None = Path("audit/reviews.jsonl") def _as_mapping(payload: Any, *, field_name: str) -> dict[str, Any]: @@ -418,6 +420,84 @@ def decisions(self, params: dict[str, Any] | None = None) -> tuple[int, dict[str {"ok": False, "error_type": type(exc).__name__, "error": str(exc)}, ) + def record_review(self, payload: Any) -> tuple[int, dict[str, Any]]: + """ + Record a human verdict on one hard_deny event. + + This is the only write the console makes, and it is the thing that + unblocks `hard_deny_reviewed`. The event is looked up in the audit + trail first: a verdict on an id that is not a hard_deny would be + recorded, counted by nothing, and leave the operator believing they had + reviewed something. + """ + try: + body = _as_mapping(payload, field_name="review payload") + if self.settings.review_path is None: + raise ValueError("review_path is disabled in API settings") + if self.settings.audit_path is None: + raise ValueError("audit_path is disabled in API settings") + + event_id = _as_non_empty_str(body.get("event_id"), field_name="event_id") + verdict = _as_non_empty_str(body.get("verdict"), field_name="verdict") + reviewer = _as_non_empty_str(body.get("reviewer"), field_name="reviewer") + + state = self._state_of_event(event_id) + if state is None: + raise ValueError(f"no audit event with event_id {event_id!r}") + if state != "hard_deny": + raise ValueError(f"event {event_id!r} is `{state}`, not `hard_deny`") + + review = ReviewStore(self.settings.review_path).record( + event_id, + verdict, + reviewer=reviewer, + note=str(body.get("note") or ""), + ) + return (200, {"ok": True, "review": review.to_dict()}) + except Exception as exc: + return ( + 400, + {"ok": False, "error_type": type(exc).__name__, "error": str(exc)}, + ) + + def reviews(self) -> tuple[int, dict[str, Any]]: + """Every recorded verdict, newest first, including superseded ones.""" + try: + if self.settings.review_path is None: + raise ValueError("review_path is disabled in API settings") + history = ReviewStore(self.settings.review_path).history() + return ( + 200, + { + "ok": True, + "count": len(history), + "reviews": [r.to_dict() for r in reversed(history)], + }, + ) + except Exception as exc: + return ( + 400, + {"ok": False, "error_type": type(exc).__name__, "error": str(exc)}, + ) + + def _state_of_event(self, event_id: str) -> str | None: + path = self.settings.audit_path + if path is None or not Path(path).exists(): + return None + for line in Path(path).read_text(encoding="utf-8").splitlines(): + if not line.strip(): + continue + try: + event = json.loads(line) + except json.JSONDecodeError: + continue + if isinstance(event, dict) and str(event.get("event_id", "")) == event_id: + decision = event.get("decision") + if isinstance(decision, dict): + return str(decision.get("state", "unknown")) + return "unknown" + return None + def rollout(self, params: dict[str, Any] | None = None) -> tuple[int, dict[str, Any]]: """Phase measurement for the console — same data as `aetherya rollout report`.""" try: @@ -433,6 +513,7 @@ def rollout(self, params: dict[str, Any] | None = None) -> tuple[int, dict[str, min_decisions=int(query.get("min_decisions", DEFAULT_MIN_DECISIONS) or 0) or DEFAULT_MIN_DECISIONS, min_days=float(query.get("min_days", DEFAULT_MIN_DAYS) or 0) or DEFAULT_MIN_DAYS, + review_path=self.settings.review_path, ) return (200, {"ok": True, "report": report.to_dict()}) except Exception as exc: diff --git a/src/aetherya/api_server.py b/src/aetherya/api_server.py index fe20ba1..11da218 100644 --- a/src/aetherya/api_server.py +++ b/src/aetherya/api_server.py @@ -77,6 +77,24 @@ def _query(self) -> dict[str, Any]: raw = parse_qs(urlsplit(self.path).query) return {key: values[0] for key, values in raw.items() if values} + def _console_authorized(self) -> bool: + """Answers 401 itself and returns False when the console key does not match.""" + expected = os.getenv("AETHERYA_CONSOLE_API_KEY", "").strip() + if not expected: + return True + provided = self.headers.get("X-AETHERYA-Console-Key", "").strip() + if provided and hmac.compare_digest(provided, expected): + return True + self._send_json( + 401, + { + "ok": False, + "error_type": "Unauthorized", + "error": "missing or invalid console key", + }, + ) + return False + def _handle_request(self) -> None: method = self.command.upper() path = urlsplit(self.path).path @@ -87,27 +105,54 @@ def _handle_request(self) -> None: # Read-only console data. Gated behind AETHERYA_CONSOLE_API_KEY when it # is set; open otherwise, like the rest of the decision profile. This # view exposes every recorded action, so the port must not be public. - if method == "GET" and path in {"/v1/decisions", "/v1/rollout/report"}: + if method == "GET" and path in {"/v1/decisions", "/v1/rollout/report", "/v1/reviews"}: if self.api is None: self._send_json(500, {"ok": False, "error": "api not configured"}) return - expected = os.getenv("AETHERYA_CONSOLE_API_KEY", "").strip() - if expected: - provided = self.headers.get("X-AETHERYA-Console-Key", "").strip() - if not provided or not hmac.compare_digest(provided, expected): - self._send_json( - 401, - { - "ok": False, - "error_type": "Unauthorized", - "error": "missing or invalid console key", - }, - ) - return + if not self._console_authorized(): + return query = self._query() - status, body = ( - self.api.decisions(query) if path == "/v1/decisions" else self.api.rollout(query) - ) + if path == "/v1/decisions": + status, body = self.api.decisions(query) + elif path == "/v1/reviews": + status, body = self.api.reviews() + else: + status, body = self.api.rollout(query) + self._send_json(status, body) + return + + # Recording a review is the console's only write, and it is what lets + # `hard_deny_reviewed` pass. Unlike the reads above it is never open: + # without a key the verdict carries a reviewer name that nobody had to + # prove, which is worse than having no attribution at all. + if method == "POST" and path == "/v1/reviews": + if self.api is None: + self._send_json(500, {"ok": False, "error": "api not configured"}) + return + if not os.getenv("AETHERYA_CONSOLE_API_KEY", "").strip(): + self._send_json( + 503, + { + "ok": False, + "error_type": "ServiceUnavailable", + "error": ( + "recording a review requires AETHERYA_CONSOLE_API_KEY to be set " + "— the verdict is attributed to a reviewer and must be authenticated" + ), + }, + ) + return + if not self._console_authorized(): + return + try: + review_payload = self._parse_json_body() + except (RequestTooLargeError, ValueError) as exc: + code = 413 if isinstance(exc, RequestTooLargeError) else 400 + self._send_json( + code, {"ok": False, "error_type": type(exc).__name__, "error": str(exc)} + ) + return + status, body = self.api.record_review(review_payload) self._send_json(status, body) return diff --git a/src/aetherya/console.py b/src/aetherya/console.py index 4a9bc4f..11bface 100644 --- a/src/aetherya/console.py +++ b/src/aetherya/console.py @@ -88,6 +88,16 @@ def console_html() -> str: border:1px solid rgba(217,154,43,.4)} .verdict.yes{background:rgba(43,184,150,.14);color:var(--allow); border:1px solid rgba(43,184,150,.4)} + .review.done{border-left-color:var(--muted);opacity:.72} + .actions{display:flex;gap:8px;margin-top:9px;flex-wrap:wrap} + .actions button{border:1px solid var(--line);background:var(--panel); + color:var(--fg);border-radius:6px;padding:6px 11px;cursor:pointer; + font-size:.76rem;font-family:inherit} + .actions button.tp:hover{border-color:var(--allow);color:var(--allow)} + .actions button.fp:hover{border-color:var(--hard);color:var(--hard)} + .rv{margin-top:9px;font-size:.76rem;font-weight:600} + .rv.tp{color:var(--allow)} .rv.fp{color:var(--hard)} + .rv .note{font-weight:400;margin-top:3px} .empty{color:var(--muted);text-align:center;padding:26px 10px;font-size:.85rem} .bar{display:flex;height:7px;border-radius:4px;overflow:hidden;margin:10px 0 6px; background:var(--panel-2)} @@ -129,7 +139,7 @@ def console_html() -> str:
-

Hard-deny review

+

Hard-deny review

Every one of these needs human eyes before enforcement is switched on. A false positive here is a detection bug — add the case to the corpus and @@ -139,7 +149,8 @@ def console_html() -> str:

- Read-only view of the audit trail. Do not expose this port publicly. + Reviews are recorded against the audit trail and are what let + hard_deny_reviewed pass. Do not expose this port publicly.
@@ -257,14 +268,72 @@ def console_html() -> str: const review = document.getElementById("review"); const events = rep.hard_deny_events || []; + const pending = events.filter(e => !e.review).length; + document.getElementById("review-count").textContent = + events.length ? `${pending} of ${events.length} pending` : ""; + review.innerHTML = events.length ? events.map(e => ` -
+
${esc(time(e.ts))}actor: ${esc(e.actor)} risk: ${esc(e.risk_score)}
${esc(e.action)}
→ ${esc(e.reason)}
+ ${e.review ? verdictBadge(e.review) : verdictButtons(e.event_id)}
`).join("") : `
no hard-deny events in the window
`; + + review.querySelectorAll("button[data-verdict]").forEach(b => { + b.onclick = () => submitReview(b.dataset.id, b.dataset.verdict); + }); +} + +function verdictBadge(r){ + const ok = r.verdict === "true_positive"; + return `
+ ${ok ? "✓ confirmed true positive" : "✗ FALSE POSITIVE — blocks the advance"} + · ${esc(r.reviewer)} · ${esc(time(r.ts))} + ${r.note ? `
${esc(r.note)}
` : ""} +
`; +} + +function verdictButtons(id){ + return `
+ + +
`; +} + +async function submitReview(eventId, verdict){ + const reviewer = localStorage.getItem("aetherya.reviewer") || prompt("Your name (recorded with the verdict):"); + if(!reviewer) return; + localStorage.setItem("aetherya.reviewer", reviewer); + + // A false positive is a claim the engine got it wrong. The note is what + // tells you later which rule to narrow, so it is required, not optional. + let note = ""; + if(verdict === "false_positive"){ + note = prompt("Why was this a false positive? (required — it is what identifies the rule to fix)") || ""; + if(!note.trim()){ return; } + } + + const key = localStorage.getItem("aetherya.consoleKey") || ""; + const r = await fetch("/v1/reviews", { + method: "POST", + headers: {"Content-Type":"application/json", "X-AETHERYA-Console-Key": key}, + body: JSON.stringify({event_id: eventId, verdict, reviewer, note}) + }); + const body = await r.json().catch(() => ({})); + if(!r.ok){ + if(r.status === 401){ + const entered = prompt("Console key (AETHERYA_CONSOLE_API_KEY):"); + if(entered){ localStorage.setItem("aetherya.consoleKey", entered); return submitReview(eventId, verdict); } + } + alert("Could not record the review:\n\n" + (body.error || r.status)); + return; + } + loadRollout(); } function refresh(){ loadStatus(); loadFeed(); loadRollout(); } diff --git a/src/aetherya/review_store.py b/src/aetherya/review_store.py new file mode 100644 index 0000000..ffb1c07 --- /dev/null +++ b/src/aetherya/review_store.py @@ -0,0 +1,147 @@ +from __future__ import annotations + +import json +import threading +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import Any, Final + +# --------------------------------------------------------------------------- +# Human review of hard_deny events. +# +# The phase-1 exit criteria say every hard_deny must be read by a human and +# confirmed a true positive. `aetherya rollout report` could count them but +# never pass them, because a tool cannot make that judgement. This is where the +# judgement is recorded. +# +# Append-only on purpose. A review is itself a human act with consequences — +# it is what unblocks enforcement — so who said what, and when, has to survive. +# Re-reviewing an event appends a new verdict; the latest one wins on read and +# the earlier one stays on disk. +# +# See docs/rollout-phases.md#phase-1--shadow. +# --------------------------------------------------------------------------- + +TRUE_POSITIVE: Final = "true_positive" +FALSE_POSITIVE: Final = "false_positive" +VERDICTS: Final[frozenset[str]] = frozenset({TRUE_POSITIVE, FALSE_POSITIVE}) + + +@dataclass(frozen=True) +class Review: + event_id: str + verdict: str + reviewer: str + note: str + ts: str + + def to_dict(self) -> dict[str, Any]: + return { + "event_id": self.event_id, + "verdict": self.verdict, + "reviewer": self.reviewer, + "note": self.note, + "ts": self.ts, + } + + +class ReviewStore: + """Append-only JSONL of human verdicts, keyed by audit `event_id`.""" + + def __init__(self, path: str | Path) -> None: + self.path = Path(path) + self._lock = threading.Lock() + + def record( + self, + event_id: str, + verdict: str, + *, + reviewer: str, + note: str = "", + ) -> Review: + event_id = str(event_id).strip() + if not event_id: + raise ValueError("event_id must be non-empty") + if verdict not in VERDICTS: + raise ValueError(f"verdict must be one of {sorted(VERDICTS)}, got {verdict!r}") + reviewer = str(reviewer).strip() + if not reviewer: + raise ValueError("reviewer must be non-empty") + + # A false positive is a claim that the engine got it wrong. Requiring a + # reason is what makes it actionable later — the note is what tells you + # which rule to narrow. + note = str(note).strip() + if verdict == FALSE_POSITIVE and not note: + raise ValueError("a false_positive review must explain why") + + review = Review( + event_id=event_id, + verdict=verdict, + reviewer=reviewer, + note=note[:2000], + ts=datetime.now(UTC).isoformat(), + ) + with self._lock: + self.path.parent.mkdir(parents=True, exist_ok=True) + with self.path.open("a", encoding="utf-8") as handle: + handle.write(json.dumps(review.to_dict(), ensure_ascii=False) + "\n") + return review + + def reviews(self) -> dict[str, Review]: + """Latest verdict per event_id. A missing or unreadable file is empty.""" + latest: dict[str, Review] = {} + if not self.path.exists(): + return latest + for line in self.path.read_text(encoding="utf-8").splitlines(): + if not line.strip(): + continue + try: + payload = json.loads(line) + except json.JSONDecodeError: + continue + if not isinstance(payload, dict): + continue + event_id = str(payload.get("event_id", "")).strip() + verdict = str(payload.get("verdict", "")) + if not event_id or verdict not in VERDICTS: + continue + latest[event_id] = Review( + event_id=event_id, + verdict=verdict, + reviewer=str(payload.get("reviewer", "unknown")), + note=str(payload.get("note", "")), + ts=str(payload.get("ts", "")), + ) + return latest + + def history(self) -> list[Review]: + """Every recorded verdict in file order, including superseded ones.""" + entries: list[Review] = [] + if not self.path.exists(): + return entries + for line in self.path.read_text(encoding="utf-8").splitlines(): + if not line.strip(): + continue + try: + payload = json.loads(line) + except json.JSONDecodeError: + continue + if not isinstance(payload, dict): + continue + event_id = str(payload.get("event_id", "")).strip() + verdict = str(payload.get("verdict", "")) + if not event_id or verdict not in VERDICTS: + continue + entries.append( + Review( + event_id=event_id, + verdict=verdict, + reviewer=str(payload.get("reviewer", "unknown")), + note=str(payload.get("note", "")), + ts=str(payload.get("ts", "")), + ) + ) + return entries diff --git a/src/aetherya/rollout_report.py b/src/aetherya/rollout_report.py index 9600849..4d21669 100644 --- a/src/aetherya/rollout_report.py +++ b/src/aetherya/rollout_report.py @@ -11,6 +11,7 @@ from aetherya.audit_verify import verify_audit_file from aetherya.enforcement import PHASES, resolve_phase +from aetherya.review_store import FALSE_POSITIVE, ReviewStore # --------------------------------------------------------------------------- # Phase-1 measurement. @@ -124,11 +125,13 @@ def build_report( min_decisions: int = DEFAULT_MIN_DECISIONS, min_days: float = DEFAULT_MIN_DAYS, max_hard_deny_samples: int = 50, + review_path: str | Path | None = None, ) -> RolloutReport: path = Path(audit_path) phase = resolve_phase(current_phase) next_phase = PHASES[min(current_phase + 1, max(PHASES))] events = _load_events(path) + recorded_reviews = ReviewStore(review_path).reviews() if review_path is not None else {} report = RolloutReport(audit_path=str(path), current_phase=phase.number) report.total_decisions = len(events) @@ -165,14 +168,20 @@ def build_report( timestamps.append(ts) # Every hard_deny needs human eyes before enforcement is switched on. + # `event_id` is what a recorded verdict hangs off, so it travels with + # the sample rather than being looked up again later. if state == "hard_deny" and len(report.hard_deny_events) < max_hard_deny_samples: + event_id = str(event.get("event_id", "")) + review = recorded_reviews.get(event_id) report.hard_deny_events.append( { + "event_id": event_id, "ts": event.get("ts"), "actor": event.get("actor"), "action": str(event.get("action", ""))[:160], "reason": decision.get("reason"), "risk_score": decision.get("risk_score"), + "review": review.to_dict() if review is not None else None, } ) @@ -236,18 +245,68 @@ def _evaluate_criteria( "one means the window mixes different engines and must be restarted" ), ), - # Deliberately never auto-passes: a human has to look at these. The tool - # can list them and count them; it cannot judge them. - Criterion( + _hard_deny_reviewed(report), + ] + return criteria + + +def _hard_deny_reviewed(report: RolloutReport) -> Criterion: + """ + Never auto-passes: it passes only on verdicts a human recorded. + + The tool can count hard_deny events and show them; it cannot judge them. + What it can do is check that someone did — and refuse to advance while any + of them is unreviewed or was judged a false positive. + """ + total = report.states.get("hard_deny", 0) + if total == 0: + return Criterion( + name="hard_deny_reviewed", + passed=True, + detail="no hard_deny events in this window", + ) + + unreviewed = [e for e in report.hard_deny_events if e.get("review") is None] + false_positives = [ + e + for e in report.hard_deny_events + if isinstance(e.get("review"), dict) and e["review"].get("verdict") == FALSE_POSITIVE + ] + + # A false positive is the finding phase 1 exists to produce. It blocks the + # advance because enforcing a rule a human already called wrong is exactly + # the outcome this phase is meant to prevent. + if false_positives: + return Criterion( name="hard_deny_reviewed", passed=False, detail=( - f"{report.states.get('hard_deny', 0)} hard_deny event(s) require manual " - "review — confirm every one is a true positive, then advance" + f"{len(false_positives)} of {total} hard_deny event(s) were judged FALSE " + "positives — fix the rule and restart the window, do not advance " + "(see docs/rollout-phases.md#phase-1--shadow)" ), - ), - ] - return criteria + ) + + # `hard_deny_events` is capped by max_hard_deny_samples. If the window holds + # more than we sampled, the ones we never showed cannot have been reviewed. + unsampled = total - len(report.hard_deny_events) + outstanding = len(unreviewed) + max(unsampled, 0) + if outstanding: + return Criterion( + name="hard_deny_reviewed", + passed=False, + detail=( + f"{outstanding} of {total} hard_deny event(s) still require manual review " + "— confirm every one is a true positive, then advance" + ), + ) + + reviewers = sorted({e["review"]["reviewer"] for e in report.hard_deny_events}) + return Criterion( + name="hard_deny_reviewed", + passed=True, + detail=f"all {total} hard_deny event(s) reviewed and confirmed by {', '.join(reviewers)}", + ) def _print_text(report: RolloutReport) -> None: @@ -300,6 +359,11 @@ def main(argv: list[str] | None = None) -> int: parser.add_argument("--phase", type=int, default=1, choices=sorted(PHASES)) parser.add_argument("--min-decisions", type=int, default=DEFAULT_MIN_DECISIONS) parser.add_argument("--min-days", type=float, default=DEFAULT_MIN_DAYS) + parser.add_argument( + "--review-path", + default="audit/reviews.jsonl", + help="Human verdicts on hard_deny events, as recorded by the operator console.", + ) parser.add_argument("--json", action="store_true", help="Emit machine-readable JSON.") args = parser.parse_args(argv) @@ -309,6 +373,7 @@ def main(argv: list[str] | None = None) -> int: current_phase=int(args.phase), min_decisions=int(args.min_decisions), min_days=float(args.min_days), + review_path=args.review_path, ) except ValueError as exc: print(f"error: {exc}", file=sys.stderr) diff --git a/tests/test_console.py b/tests/test_console.py index 4100268..be7c830 100644 --- a/tests/test_console.py +++ b/tests/test_console.py @@ -252,6 +252,22 @@ def _get(server: Any, path: str, headers: dict[str, str] | None = None) -> Any: return exc.code, exc.read().decode("utf-8") +def _post(server: Any, path: str, payload: Any, headers: dict[str, str] | None = None) -> Any: + import urllib.request + + request = urllib.request.Request( # noqa: S310 + f"http://127.0.0.1:{server.server_port}{path}", + data=json.dumps(payload).encode("utf-8"), + method="POST", + headers={"Content-Type": "application/json", **(headers or {})}, + ) + try: + with urllib.request.urlopen(request, timeout=5) as response: # noqa: S310 + return response.status, response.read().decode("utf-8") + except urllib.error.HTTPError as exc: # type: ignore[name-defined] + return exc.code, exc.read().decode("utf-8") + + def test_console_is_served_at_the_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: server = _server(tmp_path, monkeypatch) try: @@ -339,3 +355,262 @@ def _send_json(self, status: int, body: dict[str, Any]) -> None: # type: ignore handler = _Bare() handler._handle_request() # noqa: SLF001 assert handler.sent[0][0] == 500 + + +# --------------------------------------------------------------------------- +# Review queue — the console's only write +# --------------------------------------------------------------------------- + + +def _api_with_reviews(tmp_path: Path, audit: Path) -> AetheryaAPI: + return AetheryaAPI( + APISettings( + policy_path=Path("config/policy.yaml"), + audit_path=audit, + review_path=tmp_path / "reviews.jsonl", + ) + ) + + +def _hard_deny_ids(audit: Path) -> list[str]: + ids: list[str] = [] + for line in audit.read_text(encoding="utf-8").splitlines(): + try: + event = json.loads(line) + except json.JSONDecodeError: + continue + if event.get("decision", {}).get("state") == "hard_deny": + ids.append(event["event_id"]) + return ids + + +def test_a_review_is_recorded_and_reaches_the_report(tmp_path: Path) -> None: + audit = tmp_path / "decisions.jsonl" + _seed(audit, [("a", "hard_deny")]) + api = _api_with_reviews(tmp_path, audit) + event_id = _hard_deny_ids(audit)[0] + + code, body = api.record_review( + {"event_id": event_id, "verdict": "true_positive", "reviewer": "robert"} + ) + assert code == 200 + assert body["review"]["verdict"] == "true_positive" + + events = api.rollout()[1]["report"]["hard_deny_events"] + assert events[0]["review"]["reviewer"] == "robert" + + +def test_reviewing_an_unknown_event_is_refused(tmp_path: Path) -> None: + """ + A verdict on an id that is not in the trail would be recorded, counted by + nothing, and leave the operator believing they had reviewed something. + """ + audit = tmp_path / "decisions.jsonl" + _seed(audit, [("a", "hard_deny")]) + + code, body = _api_with_reviews(tmp_path, audit).record_review( + {"event_id": "made-up", "verdict": "true_positive", "reviewer": "robert"} + ) + assert code == 400 + assert "no audit event" in body["error"] + + +def test_reviewing_a_non_hard_deny_event_is_refused(tmp_path: Path) -> None: + audit = tmp_path / "decisions.jsonl" + _seed(audit, [("a", "allow")]) + allow_id = json.loads(audit.read_text(encoding="utf-8").splitlines()[0])["event_id"] + + code, body = _api_with_reviews(tmp_path, audit).record_review( + {"event_id": allow_id, "verdict": "true_positive", "reviewer": "robert"} + ) + assert code == 400 + assert "not `hard_deny`" in body["error"] + + +@pytest.mark.parametrize( + "payload", + [ + {"verdict": "true_positive", "reviewer": "r"}, + {"event_id": "x", "reviewer": "r"}, + {"event_id": "x", "verdict": "true_positive"}, + {"event_id": "x", "verdict": "nonsense", "reviewer": "r"}, + [], + ], +) +def test_malformed_review_payloads_are_rejected(tmp_path: Path, payload: Any) -> None: + audit = tmp_path / "decisions.jsonl" + _seed(audit, [("a", "hard_deny")]) + assert _api_with_reviews(tmp_path, audit).record_review(payload)[0] == 400 + + +def test_reviews_are_listed_newest_first(tmp_path: Path) -> None: + audit = tmp_path / "decisions.jsonl" + _seed(audit, [("a", "hard_deny"), ("b", "hard_deny")]) + api = _api_with_reviews(tmp_path, audit) + ids = _hard_deny_ids(audit) + + api.record_review({"event_id": ids[0], "verdict": "true_positive", "reviewer": "first"}) + api.record_review({"event_id": ids[1], "verdict": "true_positive", "reviewer": "second"}) + + code, body = api.reviews() + assert code == 200 + assert [r["reviewer"] for r in body["reviews"]] == ["second", "first"] + + +def test_review_routes_fail_when_the_store_is_disabled(tmp_path: Path) -> None: + audit = tmp_path / "decisions.jsonl" + _seed(audit, [("a", "hard_deny")]) + api = AetheryaAPI(APISettings(audit_path=audit, review_path=None)) + + assert ( + api.record_review({"event_id": "x", "verdict": "true_positive", "reviewer": "r"})[0] == 400 + ) + assert api.reviews()[0] == 400 + + +def test_recording_a_review_requires_a_configured_key( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """ + Without a key the verdict carries a reviewer name nobody had to prove, + which is worse than having no attribution at all. Reads stay open; this + write does not. + """ + monkeypatch.delenv("AETHERYA_CONSOLE_API_KEY", raising=False) + server = _server(tmp_path, monkeypatch) + try: + status, body = _post(server, "/v1/reviews", {"event_id": "x"}) + assert status == 503 + assert "AETHERYA_CONSOLE_API_KEY" in json.loads(body)["error"] + finally: + server.shutdown() + + +def test_recording_a_review_over_http(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("AETHERYA_CONSOLE_API_KEY", "s3cret") + audit = tmp_path / "decisions.jsonl" + _seed(audit, [("a", "hard_deny")]) + event_id = _hard_deny_ids(audit)[0] + + import threading + + from aetherya.api_server import build_server + + api = _api_with_reviews(tmp_path, audit) + server = build_server(host="127.0.0.1", port=0, api=api, max_body_bytes=65536) + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + payload = {"event_id": event_id, "verdict": "true_positive", "reviewer": "robert"} + assert _post(server, "/v1/reviews", payload)[0] == 401 + status, body = _post(server, "/v1/reviews", payload, {"X-AETHERYA-Console-Key": "s3cret"}) + assert status == 200 + assert json.loads(body)["review"]["reviewer"] == "robert" + + status, body = _get(server, "/v1/reviews", {"X-AETHERYA-Console-Key": "s3cret"}) + assert json.loads(body)["count"] == 1 + finally: + server.shutdown() + + +def test_an_invalid_review_body_is_a_400(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("AETHERYA_CONSOLE_API_KEY", "s3cret") + server = _server(tmp_path, monkeypatch) + try: + import urllib.request + + url = f"http://127.0.0.1:{server.server_port}/v1/reviews" + request = urllib.request.Request( # noqa: S310 + url, + data=b"{broken", + method="POST", + headers={"X-AETHERYA-Console-Key": "s3cret"}, + ) + with pytest.raises(urllib.error.HTTPError) as exc: + urllib.request.urlopen(request, timeout=5) # noqa: S310 + assert exc.value.code == 400 + finally: + server.shutdown() + + +# --------------------------------------------------------------------------- +# The page must actually offer the buttons +# --------------------------------------------------------------------------- + + +def test_the_console_renders_verdict_buttons() -> None: + html = console_html() + assert 'data-verdict="true_positive"' in html + assert 'data-verdict="false_positive"' in html + assert '"/v1/reviews"' in html + + +def test_the_console_requires_a_note_for_a_false_positive() -> None: + """Mirrors the store's rule so the UI cannot submit something it will reject.""" + html = console_html() + assert "if(!note.trim()){ return; }" in html + + +def test_reviewing_without_an_audit_trail_is_refused(tmp_path: Path) -> None: + api = AetheryaAPI(APISettings(audit_path=None, review_path=tmp_path / "r.jsonl")) + code, body = api.record_review({"event_id": "x", "verdict": "true_positive", "reviewer": "r"}) + assert code == 400 + assert "audit_path is disabled" in body["error"] + + +def test_reviewing_against_a_missing_audit_file_is_refused(tmp_path: Path) -> None: + api = AetheryaAPI( + APISettings(audit_path=tmp_path / "gone.jsonl", review_path=tmp_path / "r.jsonl") + ) + assert ( + api.record_review({"event_id": "x", "verdict": "true_positive", "reviewer": "r"})[0] == 400 + ) + + +def test_a_malformed_audit_line_does_not_break_the_event_lookup(tmp_path: Path) -> None: + """The garbage sits before the target, so the scan has to survive it to find it.""" + audit = tmp_path / "decisions.jsonl" + _seed(audit, [("a", "hard_deny")]) + with audit.open("a", encoding="utf-8") as handle: + handle.write("not json\n") + handle.write("\n") + _seed(audit, [("b", "hard_deny")]) + event_id = _hard_deny_ids(audit)[-1] + + api = _api_with_reviews(tmp_path, audit) + assert ( + api.record_review({"event_id": event_id, "verdict": "true_positive", "reviewer": "r"})[0] + == 200 + ) + + +def test_an_event_without_a_decision_object_reports_unknown(tmp_path: Path) -> None: + audit = tmp_path / "decisions.jsonl" + audit.write_text(json.dumps({"event_id": "bare", "actor": "a"}) + "\n", encoding="utf-8") + + code, body = _api_with_reviews(tmp_path, audit).record_review( + {"event_id": "bare", "verdict": "true_positive", "reviewer": "r"} + ) + assert code == 400 + assert "is `unknown`" in body["error"] + + +def test_the_review_route_fails_when_the_api_is_not_configured() -> None: + import io + + from aetherya.api_server import AetheryaHTTPRequestHandler + + class _Bare(AetheryaHTTPRequestHandler): + def __init__(self) -> None: + self.api = None + self.command = "POST" + self.path = "/v1/reviews" + self.headers = {} # type: ignore[assignment] + self.sent: list[tuple[int, dict[str, Any]]] = [] + self.wfile = io.BytesIO() + + def _send_json(self, status: int, body: dict[str, Any]) -> None: # type: ignore[override] + self.sent.append((status, body)) + + handler = _Bare() + handler._handle_request() # noqa: SLF001 + assert handler.sent[0][0] == 500 diff --git a/tests/test_review_store.py b/tests/test_review_store.py new file mode 100644 index 0000000..5625d9a --- /dev/null +++ b/tests/test_review_store.py @@ -0,0 +1,236 @@ +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +from aetherya.audit import AuditLogger +from aetherya.review_store import FALSE_POSITIVE, TRUE_POSITIVE, ReviewStore +from aetherya.rollout_report import build_report + + +def _store(tmp_path: Path) -> ReviewStore: + return ReviewStore(tmp_path / "reviews.jsonl") + + +# --------------------------------------------------------------------------- +# Recording +# --------------------------------------------------------------------------- + + +def test_a_verdict_round_trips(tmp_path: Path) -> None: + store = _store(tmp_path) + store.record("evt-1", TRUE_POSITIVE, reviewer="robert") + + reviews = store.reviews() + assert reviews["evt-1"].verdict == TRUE_POSITIVE + assert reviews["evt-1"].reviewer == "robert" + assert reviews["evt-1"].ts + + +def test_the_store_is_append_only(tmp_path: Path) -> None: + """ + A review is a human act with consequences — it is what unblocks + enforcement. Overwriting would erase who said what and when. + """ + store = _store(tmp_path) + store.record("evt-1", TRUE_POSITIVE, reviewer="alice") + store.record("evt-1", FALSE_POSITIVE, reviewer="bob", note="legitimate cleanup job") + + assert len(store.history()) == 2 + assert [r.reviewer for r in store.history()] == ["alice", "bob"] + # Latest wins on read. + assert store.reviews()["evt-1"].verdict == FALSE_POSITIVE + + +def test_a_false_positive_must_explain_itself(tmp_path: Path) -> None: + """The note is what identifies the rule to narrow later.""" + with pytest.raises(ValueError, match="must explain why"): + _store(tmp_path).record("evt-1", FALSE_POSITIVE, reviewer="robert") + + with pytest.raises(ValueError, match="must explain why"): + _store(tmp_path).record("evt-1", FALSE_POSITIVE, reviewer="robert", note=" ") + + +def test_a_true_positive_needs_no_note(tmp_path: Path) -> None: + assert _store(tmp_path).record("evt-1", TRUE_POSITIVE, reviewer="robert").note == "" + + +@pytest.mark.parametrize( + ("event_id", "verdict", "reviewer", "match"), + [ + ("", TRUE_POSITIVE, "robert", "event_id"), + (" ", TRUE_POSITIVE, "robert", "event_id"), + ("evt-1", "maybe", "robert", "verdict must be one of"), + ("evt-1", "", "robert", "verdict must be one of"), + ("evt-1", TRUE_POSITIVE, "", "reviewer"), + ], +) +def test_malformed_verdicts_are_rejected( + tmp_path: Path, event_id: str, verdict: str, reviewer: str, match: str +) -> None: + with pytest.raises(ValueError, match=match): + _store(tmp_path).record(event_id, verdict, reviewer=reviewer) + + +def test_a_long_note_is_truncated(tmp_path: Path) -> None: + review = _store(tmp_path).record("e", FALSE_POSITIVE, reviewer="r", note="x" * 5000) + assert len(review.note) == 2000 + + +def test_the_parent_directory_is_created(tmp_path: Path) -> None: + store = ReviewStore(tmp_path / "nested" / "deeper" / "reviews.jsonl") + store.record("evt-1", TRUE_POSITIVE, reviewer="robert") + assert store.path.exists() + + +# --------------------------------------------------------------------------- +# Reading +# --------------------------------------------------------------------------- + + +def test_a_missing_file_reads_as_empty(tmp_path: Path) -> None: + store = _store(tmp_path) + assert store.reviews() == {} + assert store.history() == [] + + +def test_malformed_lines_are_skipped(tmp_path: Path) -> None: + store = _store(tmp_path) + store.record("evt-1", TRUE_POSITIVE, reviewer="robert") + with store.path.open("a", encoding="utf-8") as handle: + handle.write("not json\n") + handle.write("\n") + handle.write('"bare string"\n') + handle.write(json.dumps({"event_id": "", "verdict": TRUE_POSITIVE}) + "\n") + handle.write(json.dumps({"event_id": "evt-2", "verdict": "nonsense"}) + "\n") + + assert list(store.reviews()) == ["evt-1"] + assert len(store.history()) == 1 + + +def test_a_review_missing_optional_fields_still_loads(tmp_path: Path) -> None: + store = _store(tmp_path) + store.path.parent.mkdir(parents=True, exist_ok=True) + store.path.write_text( + json.dumps({"event_id": "evt-1", "verdict": TRUE_POSITIVE}) + "\n", encoding="utf-8" + ) + assert store.reviews()["evt-1"].reviewer == "unknown" + + +# --------------------------------------------------------------------------- +# The criterion it exists to unblock +# --------------------------------------------------------------------------- + + +def _audit_with_hard_denies(tmp_path: Path, count: int) -> tuple[Path, list[str]]: + path = tmp_path / "decisions.jsonl" + logger = AuditLogger(str(path), policy_fingerprint="sha256:test") + for index in range(count): + logger.log( + actor="agent", + action=f"rm -rf / #{index}", + decision={ + "allowed": False, + "risk_score": 171, + "reason": "destructive system command", + "state": "hard_deny", + }, + context={}, + ) + ids = [json.loads(line)["event_id"] for line in path.read_text(encoding="utf-8").splitlines()] + return path, ids + + +def _criterion(report: object) -> object: + return next(c for c in report.criteria if c.name == "hard_deny_reviewed") # type: ignore[attr-defined] + + +def test_reviewing_every_event_passes_the_criterion(tmp_path: Path) -> None: + audit, ids = _audit_with_hard_denies(tmp_path, 3) + store = _store(tmp_path) + for event_id in ids: + store.record(event_id, TRUE_POSITIVE, reviewer="robert") + + report = build_report(audit, min_decisions=1, min_days=0.0, review_path=store.path) + criterion = _criterion(report) + assert criterion.passed is True # type: ignore[attr-defined] + assert "robert" in criterion.detail # type: ignore[attr-defined] + + +def test_one_unreviewed_event_holds_the_gate(tmp_path: Path) -> None: + audit, ids = _audit_with_hard_denies(tmp_path, 3) + store = _store(tmp_path) + for event_id in ids[:2]: + store.record(event_id, TRUE_POSITIVE, reviewer="robert") + + criterion = _criterion( + build_report(audit, min_decisions=1, min_days=0.0, review_path=store.path) + ) + assert criterion.passed is False # type: ignore[attr-defined] + assert "1 of 3" in criterion.detail # type: ignore[attr-defined] + + +def test_a_false_positive_blocks_the_advance_even_when_all_are_reviewed( + tmp_path: Path, +) -> None: + """ + Enforcing a rule a human already called wrong is the exact outcome phase 1 + exists to prevent. Reviewing everything is not the same as approving it. + """ + audit, ids = _audit_with_hard_denies(tmp_path, 3) + store = _store(tmp_path) + store.record(ids[0], TRUE_POSITIVE, reviewer="robert") + store.record(ids[1], TRUE_POSITIVE, reviewer="robert") + store.record(ids[2], FALSE_POSITIVE, reviewer="robert", note="scoped cleanup, not root") + + criterion = _criterion( + build_report(audit, min_decisions=1, min_days=0.0, review_path=store.path) + ) + assert criterion.passed is False # type: ignore[attr-defined] + assert "FALSE" in criterion.detail # type: ignore[attr-defined] + assert "restart the window" in criterion.detail # type: ignore[attr-defined] + + +def test_events_beyond_the_sample_cap_cannot_be_reviewed_away(tmp_path: Path) -> None: + """ + The report only samples N hard_deny events. Reviewing the sampled ones must + not pass a gate covering events nobody was ever shown. + """ + audit, ids = _audit_with_hard_denies(tmp_path, 6) + store = _store(tmp_path) + for event_id in ids[:2]: + store.record(event_id, TRUE_POSITIVE, reviewer="robert") + + criterion = _criterion( + build_report( + audit, + min_decisions=1, + min_days=0.0, + review_path=store.path, + max_hard_deny_samples=2, + ) + ) + assert criterion.passed is False # type: ignore[attr-defined] + assert "4 of 6" in criterion.detail # type: ignore[attr-defined] + + +def test_reviews_reach_the_report_events(tmp_path: Path) -> None: + audit, ids = _audit_with_hard_denies(tmp_path, 2) + store = _store(tmp_path) + store.record(ids[0], TRUE_POSITIVE, reviewer="robert", note="genuinely destructive") + + events = build_report(audit, review_path=store.path).hard_deny_events + assert events[0]["review"]["verdict"] == TRUE_POSITIVE + assert events[0]["review"]["note"] == "genuinely destructive" + assert events[1]["review"] is None + assert all(e["event_id"] for e in events) + + +def test_without_a_review_path_nothing_is_reviewed(tmp_path: Path) -> None: + """Back-compat: callers that never opted in behave exactly as before.""" + audit, _ = _audit_with_hard_denies(tmp_path, 2) + report = build_report(audit, min_decisions=1, min_days=0.0) + assert _criterion(report).passed is False # type: ignore[attr-defined] + assert all(e["review"] is None for e in report.hard_deny_events) diff --git a/tests/test_rollout_report.py b/tests/test_rollout_report.py index 163c853..ca91879 100644 --- a/tests/test_rollout_report.py +++ b/tests/test_rollout_report.py @@ -78,13 +78,13 @@ def test_hard_deny_sample_is_bounded(tmp_path: Path) -> None: assert len(build_report(path, max_hard_deny_samples=5).hard_deny_events) == 5 -def test_hard_deny_review_never_auto_passes(tmp_path: Path) -> None: +def test_an_unreviewed_hard_deny_blocks_the_advance(tmp_path: Path) -> None: """ The tool can count and list these; it cannot judge them. Auto-passing would turn the phase gate into a formality, which is exactly the failure mode the criterion exists to prevent. """ - path = _audit(tmp_path, [("a", "allow", "ok")] * 3) + path = _audit(tmp_path, [("a", "hard_deny", "destructive")]) report = build_report(path, min_decisions=1, min_days=0.0) reviewed = next(c for c in report.criteria if c.name == "hard_deny_reviewed") @@ -92,6 +92,20 @@ def test_hard_deny_review_never_auto_passes(tmp_path: Path) -> None: assert report.ready_to_advance is False +def test_a_window_with_no_hard_deny_has_nothing_to_review(tmp_path: Path) -> None: + """ + Nothing to judge is not the same as a judgement withheld. Blocking here + would mean a well-behaved agent could never leave phase 1 — the criterion + would be unsatisfiable rather than strict. + """ + path = _audit(tmp_path, [("a", "allow", "ok")] * 3) + report = build_report(path, min_decisions=1, min_days=0.0) + + reviewed = next(c for c in report.criteria if c.name == "hard_deny_reviewed") + assert reviewed.passed is True + assert "no hard_deny events" in reviewed.detail + + def test_insufficient_window_fails_the_criterion(tmp_path: Path) -> None: path = _audit(tmp_path, [("a", "allow", "ok")] * 3) report = build_report(path, min_decisions=10_000, min_days=14.0)