judgevet.adapters.outbound.audit_jsonl

Append each audit record to a local file as one JSON line.

JsonlAuditSink is the reference AuditSink. It writes one compact UTF-8 JSON object per JudgmentRecord, terminated by a newline, with no byte order mark. Source: https://jsonlines.org/. Source: https://github.com/ndjson/ndjson-spec. Keys follow the record's field order, and every field appears on every line. A None field is written as JSON null. The timestamp is RFC 3339 text from datetime.isoformat(), with a +00:00 offset. Source: https://github.com/cloudevents/spec/blob/main/cloudevents/spec.md. Answers use the API wire shape. JSON object keys are strings, so a Score answer's legend and probabilities read back with string keys. The encoder refuses NaN and infinity, because neither is JSON. Source: https://docs.python.org/3/library/json.html.

The sink opens the file at construction with O_APPEND and mode 0o600. Each line reaches the file in one os.write call under a lock. On a local POSIX file, O_APPEND makes the seek and the write one atomic step. That guarantee does not hold on NFS with several appenders. Source: https://man7.org/linux/man-pages/man2/open.2.html. Rotation and retention belong to the caller.

Examples:

import tempfile
from datetime import UTC, datetime
from pathlib import Path

from judgevet.adapters.outbound.audit_jsonl import JsonlAuditSink
from judgevet.domain.audit import JudgmentRecord

record = JudgmentRecord(
    timestamp=datetime(2026, 9, 30, tzinfo=UTC),
    outcome="error",
    requested_model="jev-latest",
    questions={"q1": "noul"},
)
with tempfile.TemporaryDirectory() as folder:
    path = Path(folder) / "audit.jsonl"
    with JsonlAuditSink(path) as sink:
        sink.record(record)
    assert len(path.read_text(encoding="utf-8").splitlines()) == 1

import tempfile
from pathlib import Path

with tempfile.TemporaryDirectory() as folder:
    sink = JsonlAuditSink(Path(folder) / "audit.jsonl", fsync=True)
    sink.close()
    assert sink.fsync