"""Entscheidungsprotokoll: strukturiert, ohne Klartext, Ausführung nur mit Protokoll.""" import json import tempfile import unittest from pathlib import Path from aria_router.decision_log import DecisionLog, DecisionLogError from aria_router.gateway import Handling from tests.helpers import GatewayTestCase class DecisionLogTests(unittest.TestCase): def test_forbidden_keys_rejected(self): with tempfile.TemporaryDirectory() as tmp: log = DecisionLog(Path(tmp) / "d.jsonl") for record in ({"text": "x"}, {"a": {"prompt": "x"}}, {"a": [{"content": "x"}]}, {"clearance": {}}, {"reply_text": "x"}): with self.subTest(record=record), self.assertRaises(DecisionLogError): log.write(record) self.assertFalse((Path(tmp) / "d.jsonl").exists()) def test_records_are_json_lines(self): with tempfile.TemporaryDirectory() as tmp: log = DecisionLog(Path(tmp) / "d.jsonl") for index in range(3): log.write({"n": index, "prompt_sha256": "0" * 64}) lines = (Path(tmp) / "d.jsonl").read_text(encoding="utf-8").splitlines() self.assertEqual([json.loads(line)["n"] for line in lines], [0, 1, 2]) class DecisionLogRetentionTests(unittest.TestCase): """E17 (Kais, TG 12946): Entscheidungsprotokoll 30 Tage aufbewahren, dann löschen.""" DAY = 86400 NOW = 1_800_000_000.0 def _iso(self, seconds_ago): from datetime import datetime, timezone return datetime.fromtimestamp(self.NOW - seconds_ago, tz=timezone.utc).isoformat(timespec="milliseconds") def _seed(self, path, ages): path.write_text("".join(json.dumps({"ts": self._iso(a), "n": i}) + "\n" for i, a in enumerate(ages)), encoding="utf-8") def test_old_records_removed_new_and_unparseable_kept(self): with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "d.jsonl" self._seed(path, [40 * self.DAY, 31 * self.DAY, 29 * self.DAY, 1]) with open(path, "a", encoding="utf-8") as handle: handle.write(json.dumps({"n": "ohne-ts"}) + "\n") inode = path.stat().st_ino log = DecisionLog(path, 30 * self.DAY, clock=lambda: self.NOW) log.write({"ts": self._iso(0), "n": "neu"}) kept = [json.loads(line)["n"] for line in path.read_text(encoding="utf-8").splitlines()] self.assertEqual(kept, [2, 3, "ohne-ts", "neu"]) self.assertEqual(path.stat().st_ino, inode) # gleiche Datei, kein Ersetzen def test_prune_runs_at_most_once_per_interval(self): with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "d.jsonl" now = [self.NOW] log = DecisionLog(path, 30 * self.DAY, clock=lambda: now[0]) log.write({"ts": self._iso(0), "n": 0}) # erster Lauf kürzt, setzt Marker with open(path, "a", encoding="utf-8") as handle: # alter Datensatz nach dem Kürzen handle.write(json.dumps({"ts": self._iso(40 * self.DAY), "n": "alt"}) + "\n") now[0] += self.DAY - 1 log.write({"ts": self._iso(0), "n": 1}) self.assertIn("alt", [json.loads(l)["n"] for l in path.read_text(encoding="utf-8").splitlines()]) now[0] += 2 log.write({"ts": self._iso(0), "n": 2}) self.assertNotIn("alt", [json.loads(l)["n"] for l in path.read_text(encoding="utf-8").splitlines()]) def test_zero_retention_never_prunes(self): with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "d.jsonl" self._seed(path, [400 * self.DAY]) DecisionLog(path, clock=lambda: self.NOW).write({"ts": self._iso(0), "n": "neu"}) self.assertEqual(len(path.read_text(encoding="utf-8").splitlines()), 2) self.assertFalse((Path(tmp) / "d.jsonl.pruned").exists()) def test_config_key_bounds(self): from aria_router.config import ConfigurationError, parse_gateway_config from tests.helpers import config_dict, write_key for value, ok in ((0, True), (30, True), (3650, True), (-1, False), (3651, False), ("30", False)): with self.subTest(value=value), tempfile.TemporaryDirectory() as tmp: data = config_dict(Path(tmp), {"decision_log_retention_days": value}) write_key(Path(data["clearance"]["key_file"])) if ok: self.assertEqual(parse_gateway_config(data).decision_log_retention_days, value) else: with self.assertRaises(ConfigurationError): parse_gateway_config(data) class GatewayLogTests(GatewayTestCase): def test_execute_writes_decision_and_outcome(self): gateway = self.gateway() result = gateway.handle(self.request(1, "#public Frage")) records = [r for r in self.log_records(gateway) if r.get("job_id") == result.job_id] self.assertEqual([r["phase"] for r in records], ["decision", "outcome"]) self.assertEqual(records[0]["code"], "CLEARANCE_ISSUED") self.assertTrue(records[0]["authorization"]["clearance_issued"]) self.assertEqual(records[1]["adapter_status"], "success") self.assertEqual(records[1]["handling"], "reply") def test_routing_record_structured(self): gateway = self.gateway() gateway.handle(self.request(2, "#public Bitte debug mein Python")) record = self.decision_records(gateway)[-1] self.assertEqual(record["privacy"], "public") self.assertEqual(record["task_class"], "coding") self.assertEqual(record["routing"]["selected"], "codex") self.assertTrue(record["routing"]["tie_break"]) self.assertEqual({c["backend"] for c in record["routing"]["candidates"]}, {"codex", "claude"}) self.assertIn("code", record["capabilities"]) self.assertIn("codex", record["backend_state"]) self.assertEqual(record["authentication"]["principal"], "owner") def test_blocked_decision_logged_with_reason(self): gateway = self.gateway() gateway.handle(self.request(3, "/codex Angebot BMW")) record = self.decision_records(gateway)[-1] self.assertEqual(record["code"], "MANUAL_CHOICE_NOT_PERMITTED") codex = next(c for c in record["routing"]["candidates"] if c["backend"] == "codex") self.assertIn("PRIVACY_NOT_PERMITTED", codex["reasons"]) self.assertIn("customer.bmw", record["privacy_rules"]) def test_log_failure_blocks_execution(self): gateway = self.gateway() (self.root / "state").mkdir(exist_ok=True) (self.root / "state" / "decisions.jsonl").mkdir() result = gateway.handle(self.request(4, "#public Frage")) self.assertEqual((result.handling, result.code), (Handling.ERROR, "LOG_UNAVAILABLE")) self.assertEqual(gateway.adapters["codex"].runner.calls, []) self.assertEqual(gateway.status()["backends"]["codex"]["budget"]["jobs"], 0) def test_shadow_requires_log(self): gateway = self.gateway() (self.root / "state").mkdir(exist_ok=True) (self.root / "state" / "decisions.jsonl").mkdir() self.assertEqual(gateway.shadow(self.request(5, "hallo")).code, "LOG_UNAVAILABLE") if __name__ == "__main__": unittest.main()