Spaces:
Sleeping
Sleeping
File size: 1,752 Bytes
6bedc74 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | import json
from pathlib import Path
from typing import List
from src.schemas import AuditRecord, ToolCall, ToolResult
AUDIT_LOG_PATH = Path("logs/tool_audit.jsonl")
def write_audit_record(
call: ToolCall,
result: ToolResult,
) -> AuditRecord:
"""
Write one tool execution record to the JSONL audit log.
"""
AUDIT_LOG_PATH.parent.mkdir(
parents=True,
exist_ok=True,
)
record = AuditRecord(
call=call,
result=result,
)
with AUDIT_LOG_PATH.open(
"a",
encoding="utf-8",
) as file:
file.write(
record.model_dump_json()
+ "\n"
)
return record
def read_audit_records() -> List[AuditRecord]:
"""
Read all valid audit records from the JSONL log.
"""
if not AUDIT_LOG_PATH.exists():
return []
records: List[AuditRecord] = []
with AUDIT_LOG_PATH.open(
"r",
encoding="utf-8",
) as file:
for line in file:
line = line.strip()
if not line:
continue
try:
payload = json.loads(line)
records.append(
AuditRecord.model_validate(
payload
)
)
except (
json.JSONDecodeError,
ValueError,
):
# Ignore malformed lines rather than
# crashing the whole audit reader.
continue
return records
def clear_audit_log() -> None:
"""
Delete the audit log if it exists.
Primarily useful for tests and demo resets.
"""
if AUDIT_LOG_PATH.exists():
AUDIT_LOG_PATH.unlink() |