[FEAT] #513: Export Audit Reports directly to PDF via export-pdf CLI command

agentwatch/cli/main.py

@@ -2493,3 +2493,143 @@ async def _run() -> None:
             raise typer.Exit(1)
 
     asyncio.run(_run())
+
+
+# ─────────────────────────────────────────────
+# export-pdf command
+# ─────────────────────────────────────────────
+
+
+@session_app.command(name="export-pdf")
+def export_pdf(
+    session_id: str = typer.Argument(..., help="ID of the session to export"),
+    out: str = typer.Option("report.pdf", "--out", help="Custom output file path"),
+    api_url: str = typer.Option("http://localhost:8000", "--api"),
+    api_key: str | None = API_KEY_OPTION,
+) -> None:
+    """[bold]Export-pdf[/bold] a session replay to a PDF file."""
+
+    async def _run() -> None:
+        try:
+            import httpx
+        except ImportError:
+            console.print("[red]httpx not installed.[/red]")
+            raise typer.Exit(1)
+
+        async with httpx.AsyncClient() as client:
+            try:
+                resp = await client.get(
+                    f"{api_url}/api/v1/sessions/{session_id}/replay",
+                    headers=_api_headers(api_key),
+                    timeout=10.0,
+                )
+                if resp.status_code == 404:
+                    console.print(f"[red]Session {session_id} not found.[/red]")
+                    raise typer.Exit(1)
+                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)
+
+        try:
+            _data = resp.json()
+        except ValueError as exc:
+            console.print(f"[red]Unexpected API response: {exc}[/red]")
+            raise typer.Exit(1)
+
+        try:
+            # Build text lines for report
+            lines = [
+                "AgentWatch Session Replay Report",
+                "=================================",
+                f"Session ID: {session_id}",
+                f"Goal: {_data.get('goal', 'n/a')}",
+                f"Status: {_data.get('status', 'n/a')}",
+                f"Total Events: {_data.get('total_events', 0)}",
+                "",
+            ]
+
+            reasoning_audit = _data.get("reasoning_audit")
+            if reasoning_audit:
+                lines.extend(
+                    [
+                        "Reasoning Audit Summary",
+                        "-----------------------",
+                        f"Overall Score: {reasoning_audit.get('overall_score', 0.0)}",
+                        f"Hallucination Risk: {reasoning_audit.get('hallucination_risk', 0.0)}",
+                        f"Goal Alignment: {reasoning_audit.get('goal_alignment', 0.0)}",
+                        "",
+                    ]
+                )
+                findings = reasoning_audit.get("findings", [])
+                if findings:
+                    lines.append("Audit Findings:")
+                    for f in findings:
+                        lines.append(
+                            f"  - [{f.get('severity', 'medium')}] Step {f.get('step_index')}: {f.get('message')}"
+                        )
+                    lines.append("")
+
+            lines.extend(
+                [
+                    "Session Replay Steps",
+                    "--------------------",
+                ]
+            )
+            for step in _data.get("steps", []):
+                idx = step.get("index", 0)
+                annotations = ", ".join(step.get("annotations", [])) or "n/a"
+                lines.append(f"Step {idx}: Annotations: {annotations}")
+
+                event = step.get("event", {})
+                evt_type = event.get("event_type")
+                lines.append(f"  Event Type: {evt_type}")
+
+                tool_call = event.get("tool_call")
+                if tool_call:
+                    lines.append(f"  Tool: {tool_call.get('tool_name')}")
+                    if tool_call.get("raw_command"):
+                        lines.append(f"  Command: {tool_call.get('raw_command')}")
+
+                tool_result = event.get("tool_result")
+                if tool_result and tool_result.get("output"):
+                    out_snippet = str(tool_result.get("output"))[:100]
+                    lines.append(f"  Result: {out_snippet}...")
+
+                if event.get("planner_output_preview"):
+                    lines.append(
+                        f"  Planner Output: {event.get('planner_output_preview')[:100]}..."
+                    )
+                lines.append("")
+
+            # Render PDF / Text
+            try:
+                from reportlab.lib.pagesizes import letter  # noqa: PLC0415
+                from reportlab.pdfgen import canvas  # noqa: PLC0415
+
+                c = canvas.Canvas(out, pagesize=letter)
+                c.setTitle(f"AgentWatch Session Replay Report — {session_id}")
+                y = 750
+                for line in lines:
+                    wrapped = [line[i : i + 90] for i in range(0, len(line), 90)] or [""]
+                    for w in wrapped:
+                        c.drawString(50, y, w)
+                        y -= 14
+                        if y < 50:
+                            c.showPage()
+                            y = 750
+                c.save()
+            except ImportError:
+                with open(out, "w", encoding="utf-8") as f:
+                    f.write("\n".join(lines))
+
+            console.print(f"[green]{out} created successfully[/green]")
+        except PermissionError:
+            console.print(
+                f"[red]Cannot write to {out}. Please ensure the file is closed and try again.[/red]"
+            )
+            raise typer.Exit(1)
+
+    asyncio.run(_run())

tests/test_cli_export.py

@@ -115,3 +115,44 @@ def test_export_404(mock_httpx_client):
     result = runner.invoke(app, ["session", "export", "abc-123"])
     assert result.exit_code == 1
     assert "Session abc-123 not found" in result.stdout
+
+
+def test_export_pdf(mock_httpx_client, tmp_path):
+    mock_instance, mock_response = mock_httpx_client
+    out_file = tmp_path / "out.pdf"
+
+    result = runner.invoke(
+        app, ["session", "export-pdf", "abc-123", "--out", str(out_file)]
+    )
+    assert result.exit_code == 0
+    clean_stdout = result.stdout.replace("\n", "").replace("\r", "")
+    assert "out.pdf" in clean_stdout
+    assert "created successfully" in clean_stdout
+
+    # Verify that the PDF or file exists and is not empty
+    assert out_file.exists()
+    assert out_file.stat().st_size > 0
+
+
+def test_export_pdf_404(mock_httpx_client):
+    mock_instance, mock_response = mock_httpx_client
+    mock_response.status_code = 404
+
+    result = runner.invoke(app, ["session", "export-pdf", "abc-123"])
+    assert result.exit_code == 1
+    assert "Session abc-123 not found" in result.stdout
+
+
+def test_export_pdf_permission_error(mock_httpx_client, tmp_path):
+    mock_instance, mock_response = mock_httpx_client
+    out_file = tmp_path / "out.pdf"
+
+    # Mock open to raise PermissionError
+    with patch("builtins.open", side_effect=PermissionError("Permission denied")):
+        result = runner.invoke(
+            app, ["session", "export-pdf", "abc-123", "--out", str(out_file)]
+        )
+        assert result.exit_code == 1
+        assert "Cannot write to" in result.stdout
+        assert "Please ensure the file is closed" in result.stdout
+