fix(eval): fail closed when the egress audit log is incomplete

packages/eval/harbor/egress-proxy/entrypoint.sh

@@ -5,7 +5,10 @@ STATE_DIR=/opt/maka-egress-state
 CERT_DIR=/opt/maka-egress
 
 mkdir -p "$STATE_DIR" "$CERT_DIR"
-: > "$STATE_DIR/hits.jsonl"
+# Create the audit log if this is the first start. Truncating here would erase
+# hits already written if the proxy restarts mid-trial, and the trial would
+# look like an honest zero-hit run.
+touch "$STATE_DIR/hits.jsonl"
 
 # confdir holds the CA private key and this cell's audit log, so it stays
 # container-private and only the certificate reaches the volume the subject

packages/eval/harbor/egress_filter.py

@@ -235,6 +235,7 @@ def blocked_response(rule_id: str):
 def append_audit(rule_id: str, host: str, normalized_path: str) -> None:
     AUDIT_PATH.parent.mkdir(parents=True, exist_ok=True)
     if AUDIT_PATH.exists() and AUDIT_PATH.stat().st_size >= MAX_AUDIT_BYTES:
+        write_truncation_marker()
         return
     record = {
         "ts": int(time.time() * 1000),
@@ -244,3 +245,40 @@ def append_audit(rule_id: str, host: str, normalized_path: str) -> None:
     }
     with AUDIT_PATH.open("a", encoding="utf-8") as stream:
         stream.write(json.dumps(record, ensure_ascii=True, separators=(",", ":")) + "\n")
+
+
+def write_truncation_marker() -> None:
+    if audit_already_truncated():
+        return
+    record = {
+        "ts": int(time.time() * 1000),
+        "ruleId": "audit_truncated",
+        "host": "",
+        "normalizedPath": "",
+    }
+    prefix = ""
+    if AUDIT_PATH.exists() and AUDIT_PATH.stat().st_size > 0:
+        with AUDIT_PATH.open("rb") as stream:
+            stream.seek(-1, os.SEEK_END)
+            if stream.read(1) != b"\n":
+                prefix = "\n"
+    with AUDIT_PATH.open("a", encoding="utf-8") as stream:
+        stream.write(prefix + json.dumps(record, ensure_ascii=True, separators=(",", ":")) + "\n")
+
+
+def audit_already_truncated() -> bool:
+    if not AUDIT_PATH.exists():
+        return False
+    with AUDIT_PATH.open("rb") as stream:
+        stream.seek(0, os.SEEK_END)
+        size = stream.tell()
+        stream.seek(max(0, size - 4096))
+        tail = stream.read().decode("utf-8", errors="ignore")
+    lines = [line for line in tail.splitlines() if line.strip()]
+    if not lines:
+        return False
+    try:
+        parsed = json.loads(lines[-1])
+    except json.JSONDecodeError:
+        return False
+    return isinstance(parsed, dict) and parsed.get("ruleId") == "audit_truncated"

packages/eval/harbor/test_egress_filter.py

@@ -57,14 +57,51 @@ def test_preserves_unrelated_network_and_rejects_malformed_urls(self) -> None:
             with self.assertRaises(ValueError, msg=url):
                 MODULE.contamination_rule(url)
 
-    def test_audit_is_bounded_and_policy_errors_fail_closed(self) -> None:
+    def _install_response_stub(self) -> None:
         class Response:
             @staticmethod
             def make(status, body, headers):
                 return {"status": status, "body": body, "headers": headers}
 
+        MODULE.http = SimpleNamespace(Response=Response)
+
+    def test_request_blocks_a_contamination_url_and_appends_one_audit_record(self) -> None:
         with tempfile.TemporaryDirectory() as directory:
-            MODULE.http = SimpleNamespace(Response=Response)
+            self._install_response_stub()
+            MODULE.AUDIT_PATH = Path(directory) / "hits.jsonl"
+            flow = type(
+                "Flow",
+                (),
+                {"request": type("Request", (), {"pretty_url": "https://tbench.ai/tasks"})()},
+            )()
+            MODULE.request(flow)
+            self.assertEqual(flow.response["status"], 451)
+            self.assertEqual(
+                flow.response["headers"]["X-Maka-Eval-Egress-Rule"], "tbench_domain"
+            )
+            lines = MODULE.AUDIT_PATH.read_text().splitlines()
+            self.assertEqual(len(lines), 1)
+            record = json.loads(lines[0])
+            self.assertEqual(record["ruleId"], "tbench_domain")
+            self.assertEqual(record["host"], "tbench.ai")
+            self.assertEqual(record["normalizedPath"], "/tasks")
+
+    def test_request_leaves_an_unrelated_url_unanswered(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            self._install_response_stub()
+            MODULE.AUDIT_PATH = Path(directory) / "hits.jsonl"
+            flow = type(
+                "Flow",
+                (),
+                {"request": type("Request", (), {"pretty_url": "https://example.com/"})()},
+            )()
+            MODULE.request(flow)
+            self.assertFalse(hasattr(flow, "response"))
+            self.assertFalse(MODULE.AUDIT_PATH.exists())
+
+    def test_audit_is_bounded_and_policy_errors_fail_closed(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            self._install_response_stub()
             MODULE.AUDIT_PATH = Path(directory) / "hits.jsonl"
             flow = type(
                 "Flow",
@@ -161,6 +198,52 @@ class TCPLayer:
             self.assertEqual(record["ruleId"], "raw_tunnel")
             self.assertEqual(record["host"], "ssh.github.com")
 
+    def test_audit_escapes_line_separators_so_python_and_typescript_agree(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            MODULE.AUDIT_PATH = Path(directory) / "hits.jsonl"
+            path = "/tasks/\u2028hidden"
+            MODULE.append_audit("tbench_domain", "tbench.ai", path)
+            raw = MODULE.AUDIT_PATH.read_text(encoding="utf-8")
+            self.assertNotIn("\u2028", raw)
+            self.assertIn("\\u2028", raw)
+            self.assertEqual(raw.count("\n"), 1)
+            self.assertEqual(len(raw.splitlines()), 1)
+            record = json.loads(raw)
+            self.assertEqual(record["normalizedPath"], path)
+            self.assertFalse(MODULE.audit_already_truncated())
+
+    def test_audit_writes_one_truncation_marker_when_the_byte_limit_is_reached(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            MODULE.AUDIT_PATH = Path(directory) / "hits.jsonl"
+            MODULE.AUDIT_PATH.write_bytes(b"x" * MODULE.MAX_AUDIT_BYTES)
+            MODULE.append_audit("tbench_domain", "tbench.ai", "/tasks")
+            records = [
+                json.loads(line)
+                for line in MODULE.AUDIT_PATH.read_text().splitlines()
+                if line.startswith("{")
+            ]
+            self.assertEqual(records[-1]["ruleId"], "audit_truncated")
+            size_after_marker = MODULE.AUDIT_PATH.stat().st_size
+            MODULE.append_audit("tbench_domain", "tbench.ai", "/other")
+            self.assertEqual(MODULE.AUDIT_PATH.stat().st_size, size_after_marker)
+            records_after = [
+                json.loads(line)
+                for line in MODULE.AUDIT_PATH.read_text().splitlines()
+                if line.startswith("{")
+            ]
+            self.assertEqual(
+                [record["ruleId"] for record in records_after if record["ruleId"] == "audit_truncated"],
+                ["audit_truncated"],
+            )
+
+    def test_truncation_probe_ignores_non_object_json_tails(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            MODULE.AUDIT_PATH = Path(directory) / "hits.jsonl"
+            MODULE.AUDIT_PATH.write_text('123\n"x"\n')
+            self.assertFalse(MODULE.audit_already_truncated())
+            MODULE.write_truncation_marker()
+            self.assertTrue(MODULE.audit_already_truncated())
+
 
 if __name__ == "__main__":
     unittest.main()