From af97dcc34f6656d4bf74bd3b099b4b973fa421d1 Mon Sep 17 00:00:00 2001 From: arcgod-design Date: Fri, 19 Jun 2026 01:53:58 +0530 Subject: [PATCH 1/5] feat: support exporting compliance audit logs as CSV (closes #148) --- agentwatch/cli/main.py | 83 ++++++++++++++++++- agentwatch/governance/compliance_reporter.py | 62 +++++++++++++++ tests/test_compliance.py | 84 ++++++++++++++++++++ 3 files changed, 227 insertions(+), 2 deletions(-) diff --git a/agentwatch/cli/main.py b/agentwatch/cli/main.py index 4bb5582b..f73ddd43 100644 --- a/agentwatch/cli/main.py +++ b/agentwatch/cli/main.py @@ -10,7 +10,7 @@ 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 @@ -859,7 +859,6 @@ def top( async def _run() -> None: try: import asyncio - from typing import Any import httpx from rich.live import Live @@ -1840,6 +1839,86 @@ 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.""" + + 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/compliance/audit-log", + 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()) + + +@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)" + ), +) -> None: + """Export compliance audit log from a local GovernanceEngine as CSV.""" + from agentwatch.governance.compliance_reporter import ComplianceReporter + from agentwatch.governance.engine import GovernanceEngine + + engine = GovernanceEngine() + 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) + console.print(f"[green]{output} created successfully[/green]") + + # ───────────────────────────────────────────── # Entrypoint # --------------------------------------------- diff --git a/agentwatch/governance/compliance_reporter.py b/agentwatch/governance/compliance_reporter.py index 111aeda0..d6351fa2 100644 --- a/agentwatch/governance/compliance_reporter.py +++ b/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,34 @@ 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 +86,35 @@ def generate(self) -> ComplianceReport: summary=summary, findings=findings, ) + + def generate_csv(self, *, include_allowed: bool = False) -> str: + audit_log = self._governance.get_audit_log(limit=10_000) + 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() diff --git a/tests/test_compliance.py b/tests/test_compliance.py index 5c142821..5341a11e 100644 --- a/tests/test_compliance.py +++ b/tests/test_compliance.py @@ -2,9 +2,18 @@ from __future__ import annotations +import csv +import io from datetime import UTC, datetime from agentwatch.governance.causal import AdverseOutcome, attribute +from agentwatch.governance.compliance_reporter import ComplianceReporter +from agentwatch.governance.engine import ( + AuditEventType, + GovernanceEngine, + Permission, + Principal, +) from agentwatch.governance.eu_ai_act import ( EUAIActPackage, TechnicalDocumentation, @@ -229,3 +238,78 @@ def test_iso42001_report_counts(): rep = ams.report() assert rep.metrics["open_incidents"] == 1 assert rep.metrics["high_risks"] == 1 + + +# ───────────────────────────────────────────── +# CMP-010 — CSV Export (Issue #148) +# ───────────────────────────────────────────── + + +def _make_engine_with_entries() -> GovernanceEngine: + engine = GovernanceEngine() + viewer = Principal(principal_id="u1", name="Viewer", roles=["viewer"]) + admin = Principal(principal_id="u2", name="Admin", roles=["admin"]) + engine.register_principal(viewer) + engine.register_principal(admin) + engine.check_permission("u1", Permission.SAFETY_OVERRIDE, "policy") + engine.check_permission("u1", Permission.ADMIN_ALL, "config") + engine.record_action( + "u2", AuditEventType.CONFIG_CHANGE, "config", "updated", allowed=True + ) + return engine + + +def test_compliance_report_to_csv_has_header(): + engine = _make_engine_with_entries() + reporter = ComplianceReporter(engine) + report = reporter.generate() + csv_output = report.to_csv() + reader = csv.reader(io.StringIO(csv_output)) + header = next(reader) + assert "audit_id" in header + assert "event_type" in header + assert "allowed" in header + + +def test_compliance_report_to_csv_contains_denials(): + engine = _make_engine_with_entries() + reporter = ComplianceReporter(engine) + report = reporter.generate() + csv_output = report.to_csv() + assert "permission_denied" in csv_output + + +def test_compliance_report_to_csv_empty_when_no_denials(): + engine = GovernanceEngine() + engine.register_principal(Principal(principal_id="a", name="A", roles=["admin"])) + engine.record_action( + "a", AuditEventType.CONFIG_CHANGE, "c", "u", allowed=True + ) + reporter = ComplianceReporter(engine) + report = reporter.generate() + csv_output = report.to_csv() + reader = csv.reader(io.StringIO(csv_output)) + rows = list(reader) + assert len(rows) == 1 + + +def test_compliance_reporter_generate_csv_denials_only(): + engine = _make_engine_with_entries() + reporter = ComplianceReporter(engine) + csv_output = reporter.generate_csv(include_allowed=False) + reader = csv.reader(io.StringIO(csv_output)) + rows = list(reader) + assert len(rows) >= 2 + for row in rows[1:]: + assert "permission_denied" in row[3] or "safety_override" in row[3] + + +def test_compliance_reporter_generate_csv_includes_allowed(): + engine = _make_engine_with_entries() + reporter = ComplianceReporter(engine) + csv_output = reporter.generate_csv(include_allowed=True) + reader = csv.reader(io.StringIO(csv_output)) + rows = list(reader) + assert len(rows) >= 3 + actions = [row[5] for row in rows[1:]] + assert "updated" in actions From 64b162eba5daea081fc80e731336b9ed9632649d Mon Sep 17 00:00:00 2001 From: arcgod-design Date: Fri, 19 Jun 2026 01:57:34 +0530 Subject: [PATCH 2/5] fix: apply Black formatting to cli/main.py --- agentwatch/cli/main.py | 19 +++- agentwatch/governance/compliance_reporter.py | 96 +++++++++++--------- 2 files changed, 66 insertions(+), 49 deletions(-) diff --git a/agentwatch/cli/main.py b/agentwatch/cli/main.py index f73ddd43..60eb81fa 100644 --- a/agentwatch/cli/main.py +++ b/agentwatch/cli/main.py @@ -883,7 +883,9 @@ async def _run() -> None: def generate_dashboard(data, error_msg=None): if error_msg: return Panel( - f"[red]{error_msg}[/red]", title="AgentWatch Top Error", border_style="red" + f"[red]{error_msg}[/red]", + title="AgentWatch Top Error", + border_style="red", ) table = Table(show_header=True, header_style="bold magenta", expand=True) @@ -911,7 +913,9 @@ def generate_dashboard(data, error_msg=None): ) return Panel( - table, title="[cyan]AgentWatch Top - Active Agent Loops[/cyan]", border_style="cyan" + table, + title="[cyan]AgentWatch Top - Active Agent Loops[/cyan]", + border_style="cyan", ) async def poll_loop(live_display: Any) -> None: @@ -932,7 +936,8 @@ async def poll_loop(live_display: Any) -> None: await asyncio.sleep(refresh_rate) with Live( - generate_dashboard({"top_sessions": []}), refresh_per_second=1.0 / refresh_rate + generate_dashboard({"top_sessions": []}), + refresh_per_second=1.0 / refresh_rate, ) as live: await poll_loop(live) @@ -1858,7 +1863,9 @@ def compliance_export_csv( 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)" + 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, @@ -1903,7 +1910,9 @@ def compliance_export_local( 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)" + False, + "--include-allowed", + help="Include allowed actions (denials only by default)", ), ) -> None: """Export compliance audit log from a local GovernanceEngine as CSV.""" diff --git a/agentwatch/governance/compliance_reporter.py b/agentwatch/governance/compliance_reporter.py index d6351fa2..7136ae30 100644 --- a/agentwatch/governance/compliance_reporter.py +++ b/agentwatch/governance/compliance_reporter.py @@ -33,29 +33,33 @@ def to_dict(self) -> dict[str, Any]: 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", - ]) + 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", ""), - ]) + 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() @@ -94,27 +98,31 @@ def generate_csv(self, *, include_allowed: bool = False) -> str: buf = io.StringIO() writer = csv.writer(buf) - writer.writerow([ - "audit_id", - "timestamp", - "principal_id", - "event_type", - "resource", - "action", - "allowed", - "session_id", - "details", - ]) + 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, - ]) + 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() From f15e7d428cd3c39257ab2788226f1d7327f5dc14 Mon Sep 17 00:00:00 2001 From: arcgod-design Date: Tue, 23 Jun 2026 22:41:11 +0530 Subject: [PATCH 3/5] fix: persist audit log to JSONL, fix export-local to load from disk, add CSV format to API endpoint - GovernanceEngine: add load_from_disk() to reload persisted JSONL entries - Server: enable audit_log_path with AGENTWATCH_AUDIT_LOG_PATH env var - Server: add ?format=csv to compliance-report endpoint + dedicated /api/v1/compliance/audit-log - CLI export-local: load from JSONL file instead of empty engine, add --audit-log option - CLI export-csv: hit correct endpoint (/api/v1/governance/compliance-report) Closes #148 --- agentwatch/api/server.py | 33 ++++++++++++++++++++------ agentwatch/cli/main.py | 24 +++++++++++++++---- agentwatch/governance/engine.py | 41 +++++++++++++++++++++++++++++++++ 3 files changed, 86 insertions(+), 12 deletions(-) diff --git a/agentwatch/api/server.py b/agentwatch/api/server.py index 297e03ee..49219982 100644 --- a/agentwatch/api/server.py +++ b/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 ( @@ -236,7 +237,8 @@ async def _pg_write_event(event: AgentEvent) -> None: _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( @@ -1174,7 +1176,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() @@ -1214,10 +1225,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"] @@ -1259,7 +1267,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( @@ -1283,6 +1290,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() diff --git a/agentwatch/cli/main.py b/agentwatch/cli/main.py index 60eb81fa..e314b68e 100644 --- a/agentwatch/cli/main.py +++ b/agentwatch/cli/main.py @@ -1870,7 +1870,7 @@ def compliance_export_csv( api_url: str = typer.Option("http://localhost:8000", "--api"), api_key: str | None = API_KEY_OPTION, ) -> None: - """Export compliance audit log as CSV.""" + """Export compliance audit log as CSV from the running API server.""" async def _run() -> None: try: @@ -1885,7 +1885,7 @@ async def _run() -> None: if include_allowed: params["include_allowed"] = "true" resp = await client.get( - f"{api_url}/api/v1/compliance/audit-log", + f"{api_url}/api/v1/governance/compliance-report", headers=_api_headers(api_key), params=params, timeout=10.0, @@ -1904,6 +1904,9 @@ async def _run() -> None: asyncio.run(_run()) +_DEFAULT_AUDIT_LOG_PATH = Path("data/audit-log.jsonl") + + @compliance_app.command(name="export-local") def compliance_export_local( output: Path = typer.Option( @@ -1914,18 +1917,29 @@ def compliance_export_local( "--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 GovernanceEngine as CSV.""" + """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() + 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]" + ) 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) - console.print(f"[green]{output} created successfully[/green]") + console.print(f"[green]{output} created successfully ({loaded} entries exported)[/green]") # ───────────────────────────────────────────── diff --git a/agentwatch/governance/engine.py b/agentwatch/governance/engine.py index 90ad8911..1feeb8ec 100644 --- a/agentwatch/governance/engine.py +++ b/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) 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) From 0fc441432af8417c71a911d50c1e2cef0e5d9a8c Mon Sep 17 00:00:00 2001 From: arcgod-design Date: Fri, 26 Jun 2026 02:33:49 +0530 Subject: [PATCH 4/5] fix: remove unused format param, fix CSV entry count, remove 10k limit, catch TypeError in audit log loader (closes #148) --- agentwatch/cli/main.py | 4 +++- agentwatch/governance/compliance_reporter.py | 3 ++- agentwatch/governance/engine.py | 2 +- 3 files changed, 6 insertions(+), 3 deletions(-) diff --git a/agentwatch/cli/main.py b/agentwatch/cli/main.py index e314b68e..fc80ec8b 100644 --- a/agentwatch/cli/main.py +++ b/agentwatch/cli/main.py @@ -1939,7 +1939,9 @@ def compliance_export_local( with open(output, "w", encoding="utf-8", newline="") as f: f.write(csv_content) - console.print(f"[green]{output} created successfully ({loaded} entries exported)[/green]") + # 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]") # ───────────────────────────────────────────── diff --git a/agentwatch/governance/compliance_reporter.py b/agentwatch/governance/compliance_reporter.py index 7136ae30..67b0d9ee 100644 --- a/agentwatch/governance/compliance_reporter.py +++ b/agentwatch/governance/compliance_reporter.py @@ -92,7 +92,8 @@ def generate(self) -> ComplianceReport: ) def generate_csv(self, *, include_allowed: bool = False) -> str: - audit_log = self._governance.get_audit_log(limit=10_000) + 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] diff --git a/agentwatch/governance/engine.py b/agentwatch/governance/engine.py index 1feeb8ec..5fd71e82 100644 --- a/agentwatch/governance/engine.py +++ b/agentwatch/governance/engine.py @@ -211,7 +211,7 @@ def load_from_disk(self) -> int: ) self._audit_log.append(entry) count += 1 - except (json.JSONDecodeError, KeyError, ValueError) as exc: + 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) From 3c53f273de629b8b2d2ac1e67b5a101a3b3e5b0d Mon Sep 17 00:00:00 2001 From: arcgod-design Date: Sun, 5 Jul 2026 15:04:09 +0530 Subject: [PATCH 5/5] fix: replace os.system with console.clear, use env var for audit log path --- agentwatch/cli/main.py | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) diff --git a/agentwatch/cli/main.py b/agentwatch/cli/main.py index d05d47ad..88dcca4f 100644 --- a/agentwatch/cli/main.py +++ b/agentwatch/cli/main.py @@ -7,6 +7,7 @@ import asyncio import json +import os import platform import sys import time @@ -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) @@ -1921,7 +1919,7 @@ async def _run() -> None: asyncio.run(_run()) -_DEFAULT_AUDIT_LOG_PATH = Path("data/audit-log.jsonl") +_DEFAULT_AUDIT_LOG_PATH = Path(os.getenv("AGENTWATCH_AUDIT_LOG_PATH", "data/audit-log.jsonl")) @compliance_app.command(name="export-local")