diff --git a/agentwatch/cli/main.py b/agentwatch/cli/main.py index fb594fc5..50fdf6f5 100644 --- a/agentwatch/cli/main.py +++ b/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()) diff --git a/tests/test_cli_export.py b/tests/test_cli_export.py index 26500bca..67a17298 100644 --- a/tests/test_cli_export.py +++ b/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 +