feat(api): §10 JSON-lines call logging (full prompt + seed, injectable sink)
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
55
api/app/call_log.py
Normal file
55
api/app/call_log.py
Normal file
@@ -0,0 +1,55 @@
|
||||
"""Structured JSON-lines logging of every model call (charter §10 — the seed and
|
||||
the full prompt logged with every call; §4 — free eval/replay infrastructure).
|
||||
|
||||
One JSON object per line: everything needed to replay the call (canon_log +
|
||||
messages + model + seed) against a candidate model. No secrets to redact — the
|
||||
client never sends keys, the canon log carries no PII, and the prompt is exactly
|
||||
what we want on record.
|
||||
"""
|
||||
|
||||
import json
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
from typing import Callable
|
||||
|
||||
from . import config
|
||||
|
||||
|
||||
def _default_write(line: str) -> None:
|
||||
path = config.call_log_path()
|
||||
if path:
|
||||
with open(path, "a", encoding="utf-8") as handle:
|
||||
handle.write(line + "\n")
|
||||
else:
|
||||
sys.stdout.write(line + "\n")
|
||||
|
||||
|
||||
def record(
|
||||
*,
|
||||
role: str,
|
||||
model: str,
|
||||
options: dict,
|
||||
messages: list[dict],
|
||||
canon_log: dict,
|
||||
ok: bool,
|
||||
latency_ms: int,
|
||||
response: str | None = None,
|
||||
error: str | None = None,
|
||||
write: Callable[[str], None] = _default_write,
|
||||
) -> dict:
|
||||
rec: dict = {
|
||||
"ts": datetime.now(timezone.utc).isoformat(),
|
||||
"role": role,
|
||||
"model": model,
|
||||
"options": options,
|
||||
"messages": messages,
|
||||
"canon_log": canon_log,
|
||||
"ok": ok,
|
||||
"latency_ms": latency_ms,
|
||||
}
|
||||
if ok:
|
||||
rec["response"] = response
|
||||
else:
|
||||
rec["error"] = error
|
||||
write(json.dumps(rec, ensure_ascii=False))
|
||||
return rec
|
||||
34
api/tests/test_call_log.py
Normal file
34
api/tests/test_call_log.py
Normal file
@@ -0,0 +1,34 @@
|
||||
import json
|
||||
|
||||
from app.call_log import record
|
||||
|
||||
|
||||
def test_success_record_shape():
|
||||
captured = []
|
||||
rec = record(
|
||||
role="narrator", model="qwen3.5:latest", options={"seed": 7},
|
||||
messages=[{"role": "user", "content": "x"}], canon_log={"a": 1},
|
||||
ok=True, latency_ms=120, response="prose", write=captured.append,
|
||||
)
|
||||
assert len(captured) == 1
|
||||
parsed = json.loads(captured[0])
|
||||
assert parsed["ok"] is True
|
||||
assert parsed["response"] == "prose"
|
||||
assert parsed["options"]["seed"] == 7
|
||||
assert parsed["role"] == "narrator"
|
||||
assert parsed["canon_log"] == {"a": 1}
|
||||
assert "error" not in parsed
|
||||
assert "ts" in parsed
|
||||
assert rec["response"] == "prose"
|
||||
|
||||
|
||||
def test_failure_record_shape():
|
||||
captured = []
|
||||
record(
|
||||
role="narrator", model="m", options={}, messages=[], canon_log={},
|
||||
ok=False, latency_ms=5, error="ModelError: down", write=captured.append,
|
||||
)
|
||||
parsed = json.loads(captured[0])
|
||||
assert parsed["ok"] is False
|
||||
assert parsed["error"] == "ModelError: down"
|
||||
assert "response" not in parsed
|
||||
Reference in New Issue
Block a user