feat: tool call audit log

Every orchestrator tool invocation is recorded to home/{user}/tool_audit/YYYY-MM-DD.jsonl.
Each entry captures: timestamp, user, tool, args (truncated), status (ok/error/denied),
result length, and a 300-char result snippet.

- tool_audit.py: JSONL writer with per-file asyncio locks; read_recent / read_recent_all_users helpers
- tools/__init__.py: hook in call_tool() — fire-and-forget record on every dispatch
- routers/audit.py: GET /api/audit/recent and /api/audit/stats (admin-only)
- tools/files.py: add home_root() to file_read allowed roots so agents can read audit JSONL

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Scott Idem
2026-05-05 19:55:59 -04:00
parent ddf44a2aee
commit 584ae679a6
5 changed files with 244 additions and 10 deletions

View File

@@ -207,12 +207,28 @@ async def call_tool(name: str, args: dict, callables: dict | None = None) -> str
Pass `callables` (from get_tools_for_role) to enforce role restrictions.
Falls back to the full _CALLABLES dict if omitted.
Every call is recorded to the tool audit log (tool_audit.py).
"""
import asyncio
import tool_audit
from persona import get_user
user = get_user() or "unknown"
dispatch = callables if callables is not None else _CALLABLES
fn = dispatch.get(name)
if fn is None:
asyncio.create_task(tool_audit.record(user, name, args, "denied"))
return f"Tool not available or access denied: {name}"
return await fn(**args)
try:
result = await fn(**args)
asyncio.create_task(tool_audit.record(user, name, args, "ok", result))
return result
except Exception as e:
asyncio.create_task(tool_audit.record(user, name, args, "error", str(e)))
raise
# ── OpenAI JSON Schema conversion ────────────────────────────────────────────

View File

@@ -16,12 +16,21 @@ logger = logging.getLogger(__name__)
# Directories the orchestrator is allowed to read from.
# Paths are resolved (symlinks followed, ~ expanded) at import time.
_ALLOWED_ROOTS: list[Path] = [
Path.home() / "agents_sync",
Path.home() / "OSIT_dev",
Path.home() / "DgrZone_Nextcloud",
Path.home() / "OSIT_Nextcloud",
]
def _build_allowed_roots() -> list[Path]:
roots = [
Path.home() / "agents_sync",
Path.home() / "OSIT_dev",
Path.home() / "DgrZone_Nextcloud",
Path.home() / "OSIT_Nextcloud",
]
try:
from config import settings
roots.append(settings.home_root())
except Exception:
pass
return roots
_ALLOWED_ROOTS: list[Path] = _build_allowed_roots()
# Hard cap on file size to prevent accidental context blowout
_MAX_BYTES = 50_000 # ~50 KB
@@ -221,8 +230,10 @@ DECLARATIONS = [
name="file_read",
description=(
"Read a local file and return its contents. "
"Allowed directories: ~/agents_sync/, ~/OSIT_dev/, ~/DgrZone_Nextcloud/, ~/OSIT_Nextcloud/. "
"Use this to read documentation, notes, CLAUDE.md files, or config references. "
"Allowed directories: ~/agents_sync/, ~/OSIT_dev/, ~/DgrZone_Nextcloud/, ~/OSIT_Nextcloud/, "
"and the Cortex home/ directory (persona memory, tool audit logs, etc.). "
"Use this to read documentation, notes, CLAUDE.md files, config references, "
"or tool audit logs at home/{user}/tool_audit/YYYY-MM-DD.jsonl. "
"If given a directory path, returns a directory listing instead."
),
parameters=types.Schema(