Anton
052591281d
- Complete agent system with planning and execution - Bridge API with FastAPI server - WhatsApp-style mobile dashboard - Enhanced web search with query improvement - Security guards and policy checks - Comprehensive documentation
44 lines
1.4 KiB
Python
44 lines
1.4 KiB
Python
"""Append-only JSONL audit logger."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Optional
|
|
|
|
|
|
class AuditLogger:
|
|
"""Writes structured JSONL audit records to workspace/runs/<run_id>/audit.jsonl.
|
|
|
|
Required events: run_started, plan_created, step_started,
|
|
step_finished, state_saved, run_finished.
|
|
"""
|
|
|
|
def __init__(self, run_id: str, runs_dir: Path, trace_id: Optional[str] = None) -> None:
|
|
self._run_id = run_id
|
|
self._trace_id = trace_id
|
|
self._run_dir = runs_dir / run_id
|
|
self._run_dir.mkdir(parents=True, exist_ok=True)
|
|
self._log_path = self._run_dir / "audit.jsonl"
|
|
|
|
def log(
|
|
self,
|
|
event_type: str,
|
|
step_id: Optional[str] = None,
|
|
status: str = "ok",
|
|
details: Optional[dict[str, Any]] = None,
|
|
) -> None:
|
|
"""Append a single audit record to audit.jsonl."""
|
|
record: dict[str, Any] = {
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
"trace_id": self._trace_id,
|
|
"run_id": self._run_id,
|
|
"layer": "agent",
|
|
"event_type": event_type,
|
|
"step_id": step_id,
|
|
"status": status,
|
|
"details": details or {},
|
|
}
|
|
with self._log_path.open("a", encoding="utf-8") as fh:
|
|
fh.write(json.dumps(record, ensure_ascii=False) + "\n")
|