feat: support exporting compliance audit logs as CSV

agentwatch/api/server.py

@@ -12,6 +12,7 @@
 from collections import defaultdict
 from contextlib import asynccontextmanager
 from datetime import UTC, datetime, timedelta
+from pathlib import Path
 from typing import Any
 
 from fastapi import (
@@ -249,7 +250,8 @@ async def _pg_write_event(event: AgentEvent, tenant_id: str | None = None) -> No
 _confidence_scorer = ConfidenceScorer()
 _cost_tracker = CostTracker()
 _reasoning_auditor = ReasoningAuditor()
-_governance = GovernanceEngine()
+_audit_log_path = Path(os.getenv("AGENTWATCH_AUDIT_LOG_PATH", "data/audit-log.jsonl"))
+_governance = GovernanceEngine(audit_log_path=_audit_log_path)
 _compliance_reporter = ComplianceReporter(_governance, _collector)
 _alerting = AlertingEngine(
     AlertingConfig(
@@ -1265,7 +1267,16 @@ async def dashboard_top(_auth: None = Depends(_require_api_key)) -> dict[str, An
 
 
 @app.get("/api/v1/governance/compliance-report")
-async def compliance_report(_auth: None = Depends(_require_api_key)) -> dict[str, Any]:
+async def compliance_report(
+    _auth: None = Depends(_require_api_key),
+    format: str = Query("json", alias="format"),
+    include_allowed: bool = Query(False),
+):
+    if format == "csv":
+        from fastapi.responses import PlainTextResponse
+
+        csv_content = _compliance_reporter.generate_csv(include_allowed=include_allowed)
+        return PlainTextResponse(csv_content, media_type="text/csv")
     return _compliance_reporter.generate().to_dict()
 
 
@@ -1305,10 +1316,7 @@ async def eu_ai_act_report(_auth: None = Depends(_require_api_key)) -> dict[str,
         TechnicalDocumentation,
     )
 
-    # Derive the report from live telemetry and the active safety policy rather
-    # than static literals, so the conformity evidence reflects the running
-    # system (Article 15 record-keeping / accuracy-robustness requirements).
-    safety = _safety_engine.stats()  # {"checked", "blocked", "approved"}
+    safety = _safety_engine.stats()
     policy = _safety_engine.policy
     sessions = _collector.list_sessions(limit=200)
     checked = safety["checked"]
@@ -1350,7 +1358,6 @@ async def eu_ai_act_report(_auth: None = Depends(_require_api_key)) -> dict[str,
 
     pkg = EUAIActPackage()
     pkg.set_documentation(doc)
-    # Record-keeping evidence: derive decision-log entries from real sessions.
     sessions_used = sessions[:50]
     for session in sessions_used:
         pkg.log_decision(
@@ -1374,6 +1381,18 @@ async def eu_ai_act_report(_auth: None = Depends(_require_api_key)) -> dict[str,
     }
 
 
+@app.get("/api/v1/compliance/audit-log")
+async def compliance_audit_log_csv(
+    _auth: None = Depends(_require_api_key),
+    format: str = Query("csv", alias="format"),
+    include_allowed: bool = Query(False),
+):
+    from fastapi.responses import PlainTextResponse
+
+    csv_content = _compliance_reporter.generate_csv(include_allowed=include_allowed)
+    return PlainTextResponse(csv_content, media_type="text/csv")
+
+
 @app.post("/api/v1/demo/seed")
 async def seed_demo(_auth: None = Depends(_require_api_key)) -> dict[str, Any]:
     _seed_demo_data()

agentwatch/cli/main.py

@@ -7,12 +7,13 @@
 
 import asyncio
 import json
+import os
 import platform
 import sys
 import time
 from enum import Enum
 from pathlib import Path
-from typing import TYPE_CHECKING, NoReturn
+from typing import TYPE_CHECKING, Any, NoReturn
 
 import typer
 from rich import box
@@ -108,7 +109,6 @@ def main_callback(
 
 def _start_repl_session():
     """Run an interactive REPL shell for the CLI."""
-    import os
     import shlex
 
     from rich.panel import Panel
@@ -143,9 +143,7 @@ def _start_repl_session():
                 break
 
             if cmd_lower in ("clear", "cls"):
-                os.system(  # noqa: S605, S607
-                    "cls" if os.name == "nt" else "clear"
-                )  # nosec
+                console.clear()
                 continue
 
             args = shlex.split(cmd_line)
@@ -876,7 +874,6 @@ def top(
     async def _run() -> None:
         try:
             import asyncio
-            from typing import Any
 
             import httpx
             from rich.live import Live
@@ -1862,6 +1859,108 @@ async def _run() -> None:
     asyncio.run(_run())
 
 
+# ─────────────────────────────────────────────
+# compliance commands
+# ---------------------------------------------
+
+
+compliance_app = typer.Typer(
+    name="compliance",
+    help="Compliance audit log export and reporting.",
+    no_args_is_help=True,
+)
+app.add_typer(compliance_app)
+
+
+@compliance_app.command(name="export-csv")
+def compliance_export_csv(
+    output: Path = typer.Option(
+        Path("compliance-audit.csv"), "--output", "-o", help="Output CSV file path"
+    ),
+    include_allowed: bool = typer.Option(
+        False,
+        "--include-allowed",
+        help="Include allowed actions (denials only by default)",
+    ),
+    api_url: str = typer.Option("http://localhost:8000", "--api"),
+    api_key: str | None = API_KEY_OPTION,
+) -> None:
+    """Export compliance audit log as CSV from the running API server."""
+
+    async def _run() -> None:
+        try:
+            import httpx
+        except ImportError:
+            console.print("[red]httpx not installed. Run: pip install httpx[/red]")
+            raise typer.Exit(1)
+
+        async with httpx.AsyncClient() as client:
+            try:
+                params: dict[str, Any] = {"format": "csv"}
+                if include_allowed:
+                    params["include_allowed"] = "true"
+                resp = await client.get(
+                    f"{api_url}/api/v1/governance/compliance-report",
+                    headers=_api_headers(api_key),
+                    params=params,
+                    timeout=10.0,
+                )
+                resp.raise_for_status()
+            except httpx.HTTPStatusError as exc:
+                _handle_http_status_error(exc, api_url)
+            except httpx.HTTPError as exc:
+                console.print(f"[red]Failed to connect to API at {api_url}: {exc}[/red]")
+                raise typer.Exit(1)
+
+        with open(output, "w", encoding="utf-8", newline="") as f:
+            f.write(resp.text)
+        console.print(f"[green]{output} created successfully[/green]")
+
+    asyncio.run(_run())
+
+
+_DEFAULT_AUDIT_LOG_PATH = Path(os.getenv("AGENTWATCH_AUDIT_LOG_PATH", "data/audit-log.jsonl"))
+
+
+@compliance_app.command(name="export-local")
+def compliance_export_local(
+    output: Path = typer.Option(
+        Path("compliance-audit.csv"), "--output", "-o", help="Output CSV file path"
+    ),
+    include_allowed: bool = typer.Option(
+        False,
+        "--include-allowed",
+        help="Include allowed actions (denials only by default)",
+    ),
+    audit_log: Path = typer.Option(
+        _DEFAULT_AUDIT_LOG_PATH,
+        "--audit-log",
+        help="Path to the persisted audit log JSONL file",
+    ),
+) -> None:
+    """Export compliance audit log from a local JSONL file as CSV."""
+    from agentwatch.governance.compliance_reporter import ComplianceReporter
+    from agentwatch.governance.engine import GovernanceEngine
+
+    engine = GovernanceEngine(audit_log_path=audit_log)
+    loaded = len(engine._audit_log)
+    if loaded == 0:
+        console.print(
+            f"[yellow]No audit entries found in {audit_log}. "
+            "The server persists audit data when AGENTWATCH_AUDIT_LOG_PATH is set.[/yellow]"
+        )
+        return
+
+    reporter = ComplianceReporter(engine)
+    csv_content = reporter.generate_csv(include_allowed=include_allowed)
+
+    with open(output, "w", encoding="utf-8", newline="") as f:
+        f.write(csv_content)
+    # Count actual exported rows (subtract 1 for header)
+    exported = max(0, csv_content.count("\n") - 1)
+    console.print(f"[green]{output} created successfully ({exported} entries exported)[/green]")
+
+
 # ─────────────────────────────────────────────
 # Version
 # ---------------------------------------------

agentwatch/governance/compliance_reporter.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import csv
+import io
 from dataclasses import dataclass
 from datetime import UTC, datetime
 from typing import TYPE_CHECKING, Any
@@ -28,6 +30,38 @@ def to_dict(self) -> dict[str, Any]:
             "findings": self.findings,
         }
 
+    def to_csv(self) -> str:
+        buf = io.StringIO()
+        writer = csv.writer(buf)
+        writer.writerow(
+            [
+                "audit_id",
+                "timestamp",
+                "principal_id",
+                "event_type",
+                "resource",
+                "action",
+                "allowed",
+                "session_id",
+                "details",
+            ]
+        )
+        for entry in self.findings.get("sample_denials", []):
+            writer.writerow(
+                [
+                    entry.get("audit_id", ""),
+                    entry.get("timestamp", ""),
+                    entry.get("principal_id", ""),
+                    entry.get("event_type", ""),
+                    entry.get("resource", ""),
+                    entry.get("action", ""),
+                    entry.get("allowed", ""),
+                    entry.get("session_id", ""),
+                    entry.get("details", ""),
+                ]
+            )
+        return buf.getvalue()
+
 
 class ComplianceReporter:
     def __init__(self, governance: GovernanceEngine, collector: TraceCollector | None = None):
@@ -56,3 +90,40 @@ def generate(self) -> ComplianceReport:
             summary=summary,
             findings=findings,
         )
+
+    def generate_csv(self, *, include_allowed: bool = False) -> str:
+        total = len(self._governance._audit_log)
+        audit_log = self._governance.get_audit_log(limit=total)
+        if not include_allowed:
+            audit_log = [entry for entry in audit_log if not entry.allowed]
+
+        buf = io.StringIO()
+        writer = csv.writer(buf)
+        writer.writerow(
+            [
+                "audit_id",
+                "timestamp",
+                "principal_id",
+                "event_type",
+                "resource",
+                "action",
+                "allowed",
+                "session_id",
+                "details",
+            ]
+        )
+        for entry in audit_log:
+            writer.writerow(
+                [
+                    entry.audit_id,
+                    entry.timestamp.isoformat(),
+                    entry.principal_id or "",
+                    entry.event_type.value,
+                    entry.resource,
+                    entry.action,
+                    entry.allowed,
+                    entry.session_id or "",
+                    entry.details,
+                ]
+            )
+        return buf.getvalue()

agentwatch/governance/engine.py

@@ -178,6 +178,47 @@ def __init__(
         self._audit_path = audit_log_path
         self._entry_counter = 0
 
+        if self._audit_path:
+            self.load_from_disk()
+
+    def load_from_disk(self) -> int:
+        """Load persisted audit entries from the JSONL file into memory.
+
+        Returns the number of entries loaded.
+        """
+        if not self._audit_path or not self._audit_path.exists():
+            return 0
+
+        count = 0
+        try:
+            with open(self._audit_path, encoding="utf-8") as f:
+                for line in f:
+                    line = line.strip()
+                    if not line:
+                        continue
+                    try:
+                        data = json.loads(line)
+                        entry = AuditEntry(
+                            audit_id=data["audit_id"],
+                            timestamp=datetime.fromisoformat(data["timestamp"]),
+                            principal_id=data.get("principal_id"),
+                            event_type=AuditEventType(data["event_type"]),
+                            resource=data["resource"],
+                            action=data["action"],
+                            allowed=data["allowed"],
+                            session_id=data.get("session_id"),
+                            details=data.get("details", {}),
+                        )
+                        self._audit_log.append(entry)
+                        count += 1
+                    except (json.JSONDecodeError, KeyError, ValueError, TypeError) as exc:
+                        logger.warning("Skipping corrupt audit entry: %s", exc)
+        except OSError as exc:
+            logger.warning("Failed to read audit log from %s: %s", self._audit_path, exc)
+
+        logger.info("Loaded %d audit entries from %s", count, self._audit_path)
+        return count
+
     def register_principal(self, principal: Principal) -> None:
         self._principals[principal.principal_id] = principal
         logger.debug("Registered principal: %s (%s)", principal.name, principal.roles)