Merge 8fae48e4ba into 6e33d7e84b

litellm/__init__.py

@@ -156,6 +156,7 @@ _custom_logger_compatible_callbacks_literal = Literal[
     "compression_interception",
     "newrelic",
     "signoz",
+    "asqav",
 ]
 cold_storage_custom_logger: Optional[_custom_logger_compatible_callbacks_literal] = None
 logged_real_time_event_types: Optional[Union[List[str], Literal["*"]]] = None

litellm/integrations/asqav/__init__.py

@@ -0,0 +1,5 @@
+"""Asqav local-first, tamper-evident audit-log callback for LiteLLM."""
+
+from litellm.integrations.asqav.asqav import AsqavLogger
+
+__all__ = ("AsqavLogger",)

litellm/integrations/asqav/asqav.py

@@ -0,0 +1,376 @@
+"""Asqav local-first audit-log callback for LiteLLM.
+
+Each LLM call appends one record to a local JSONL file. Every record carries a
+SHA-256 hash over its own canonical fields plus the previous record's hash, so
+the file forms a tamper-evident chain that can be checked offline with the
+standard library alone.
+
+- No dependencies beyond litellm itself, and no network calls.
+- Fail-soft: an error while building or writing a record is logged and never
+  reaches the LLM call.
+- Message and response content are stored as SHA-256 digests by default, so an
+  auditor can prove a payload existed without the log holding it.
+"""
+
+import asyncio
+import hashlib
+import itertools
+import json
+import os
+import threading
+import time
+from collections.abc import Iterable, Mapping, Sequence
+from datetime import datetime, timezone
+from typing import BinaryIO, Final
+
+from pydantic import ConfigDict, TypeAdapter, ValidationError
+
+from litellm._logging import verbose_logger
+from litellm.integrations.custom_logger import CustomLogger
+
+__all__ = ("AsqavLogger",)
+
+# prev_hash of the first record in a chain (it has no predecessor).
+GENESIS_HASH: Final = "0" * 64
+
+# Overridable with ASQAV_LOG_PATH or the log_path argument.
+DEFAULT_LOG_PATH: Final = os.path.join(os.path.expanduser("~"), ".litellm_asqav_audit.jsonl")
+
+_TAIL_CHUNK_BYTES: Final = 4096
+
+_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
+_OBJECT_MAPPING: Final = TypeAdapter(Mapping[object, object], config=ConfigDict(hide_input_in_errors=True))
+_OBJECT_SEQUENCE: Final = TypeAdapter(Sequence[object], config=ConfigDict(hide_input_in_errors=True))
+
+# Header and key values that must never reach the log.
+_SENSITIVE_KEYS: Final = frozenset({"user_api_key", "Authorization", "authorization", "token", "api_key"})
+
+# Proxy identity fields copied from litellm_params.metadata so each record says who made the call.
+_PROXY_IDENTITY_KEYS: Final = frozenset(
+    {
+        "user_api_key_user_id",
+        "user_api_key_team_id",
+        "user_api_key_org_id",
+        "user_api_key_alias",
+        "user_id",
+        "team_id",
+        "org_id",
+    }
+)
+
+
+def _sha256_hex(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+def _canonical_bytes(record: Mapping[str, object]) -> bytes:
+    """Deterministic serialisation for hashing: sorted keys, no whitespace.
+
+    Values JSON cannot encode are stored as their ``str()`` form, so the bytes
+    written to disk and the bytes hashed are the same after a JSON round trip.
+    """
+    return json.dumps(record, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")
+
+
+def _content_digest(value: object) -> str | None:
+    """SHA-256 hex digest of a content value, or None when there is no content."""
+    if value is None:
+        return None
+    return _sha256_hex(json.dumps(value, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8"))
+
+
+def _read_tail(fh: BinaryIO, size: int) -> bytes:
+    """Read back from the end of fh, doubling the window until it holds the whole last line."""
+    for shift in itertools.count():
+        read_size = min(_TAIL_CHUNK_BYTES << shift, size)
+        fh.seek(size - read_size)
+        tail = fh.read(read_size)
+        if read_size == size or b"\n" in tail.rstrip(b"\n"):
+            return tail
+    return b""
+
+
+def _str_keyed(value: object) -> Mapping[str, object]:
+    """The string-keyed entries of value when it is a mapping, else an empty mapping."""
+    if not isinstance(value, Mapping):
+        return {}
+    entries: Final = _OBJECT_MAPPING.validate_python(value)
+    return {key: item for key, item in entries.items() if isinstance(key, str)}
+
+
+def _int_or_none(value: object) -> int | None:
+    return value if isinstance(value, int) and not isinstance(value, bool) else None
+
+
+def _str_or_none(value: object) -> str | None:
+    return value if isinstance(value, str) else None
+
+
+def _first_choice(response_obj: object) -> object:
+    choices: Final[object] = getattr(response_obj, "choices", None)
+    if isinstance(choices, list | tuple) and choices:
+        return _OBJECT_SEQUENCE.validate_python(choices)[0]
+    return None
+
+
+def _response_content(response_obj: object) -> object:
+    message: Final[object] = getattr(_first_choice(response_obj), "message", None)
+    content: Final[object] = getattr(message, "content", None)
+    return content
+
+
+def _latency_ms(start_time: object, end_time: object) -> int | None:
+    if isinstance(start_time, datetime) and isinstance(end_time, datetime):
+        return int((end_time - start_time).total_seconds() * 1000)
+    return None
+
+
+def _proxy_identity(kwargs: Mapping[str, object]) -> Mapping[str, object]:
+    """Identity fields from litellm_params.metadata; raw keys and auth headers are never copied."""
+    litellm_params: Final = _str_keyed(kwargs.get("litellm_params"))
+    proxy_metadata: Final = _str_keyed(litellm_params.get("metadata"))
+    return {
+        key: value
+        for key, value in proxy_metadata.items()
+        if key in _PROXY_IDENTITY_KEYS and key not in _SENSITIVE_KEYS
+    }
+
+
+def _call_id(kwargs: Mapping[str, object]) -> str:
+    standard_logging_object: Final = _str_keyed(kwargs.get("standard_logging_object"))
+    candidates: Final = (
+        standard_logging_object.get("id"),
+        standard_logging_object.get("litellm_call_id"),
+        kwargs.get("litellm_call_id"),
+        kwargs.get("id"),
+    )
+    found: Final = next((str(candidate) for candidate in candidates if candidate), None)
+    return found if found is not None else str(int(time.time() * 1e6))
+
+
+def _extract_loggable(
+    kwargs: Mapping[str, object],
+    response_obj: object,
+    start_time: object,
+    end_time: object,
+    status: str,
+) -> Mapping[str, object]:
+    """Metadata and content digests for one callback invocation.
+
+    Message content and response text are never stored in the clear here; only
+    their SHA-256 digests appear, so a caller can prove a payload existed
+    without the log reconstructing it.
+    """
+    root_metadata: Final = _str_keyed(kwargs.get("metadata")) or _str_keyed(kwargs.get("litellm_metadata"))
+    usage: Final[object] = getattr(response_obj, "usage", None)
+    hidden_params: Final = _str_keyed(getattr(response_obj, "_hidden_params", None))
+    first_choice: Final = _first_choice(response_obj)
+    model: Final = kwargs.get("model")
+
+    return {
+        "call_id": _call_id(kwargs),
+        "model": model if isinstance(model, str) else "",
+        "status": status,
+        "latency_ms": _latency_ms(start_time, end_time),
+        "prompt_tokens": _int_or_none(getattr(usage, "prompt_tokens", None)),
+        "completion_tokens": _int_or_none(getattr(usage, "completion_tokens", None)),
+        "total_tokens": _int_or_none(getattr(usage, "total_tokens", None)),
+        "finish_reason": _str_or_none(getattr(first_choice, "finish_reason", None)),
+        "provider_request_id": _str_or_none(hidden_params.get("x-request-id"))
+        or _str_or_none(hidden_params.get("cf-ray")),
+        "messages_digest": _content_digest(kwargs.get("messages")),
+        "response_content_digest": _content_digest(_response_content(response_obj)),
+        # Root metadata wins over the proxy identity copy, matching the user's own tags.
+        "metadata": {**_proxy_identity(kwargs), **root_metadata},
+    }
+
+
+def _private_opener(path: str, flags: int) -> int:
+    """Create the log owner-only (0o600) whatever the process umask."""
+    return os.open(path, flags, 0o600)
+
+
+def _redact_from_env() -> bool:
+    return os.environ.get("ASQAV_REDACT_CONTENT", "true").lower() != "false"
+
+
+class AsqavLogger(CustomLogger):
+    """Tamper-evident local-first audit-log callback for LiteLLM.
+
+    Enable with ``litellm.callbacks = ["asqav"]`` or ``callbacks: ["asqav"]``
+    under ``litellm_settings`` in the proxy config.
+
+    Configuration (environment variables, overridden by constructor arguments):
+
+    ASQAV_LOG_PATH
+        Path to the JSONL audit log. Defaults to ~/.litellm_asqav_audit.jsonl.
+
+    ASQAV_REDACT_CONTENT
+        Set to "false" to store message and response content in the clear
+        instead of as SHA-256 digests. Defaults to "true" (digests only).
+
+    Multi-worker limitation: this logger expects a single audit writer per log
+    file. Its lock serialises threads within one process, not separate OS
+    processes. In a multi-worker proxy, give each worker its own
+    ASQAV_LOG_PATH or run one dedicated writer; otherwise records from several
+    workers share seq and prev_hash values and verify_chain reports a break.
+    """
+
+    def __init__(
+        self,
+        log_path: str | None = None,
+        redact_content: bool | None = None,
+    ) -> None:
+        super().__init__()  # pyright: ignore[reportUnknownMemberType]  # CustomLogger.__init__ declares untyped **kwargs
+        self._log_path: Final[str] = log_path or os.environ.get("ASQAV_LOG_PATH") or DEFAULT_LOG_PATH
+        self._redact_content: Final[bool] = _redact_from_env() if redact_content is None else redact_content
+        self._lock: Final = threading.Lock()
+        self._call_count: int = 0
+        self._prev_hash: str = GENESIS_HASH
+        # Resume from an existing log so the chain continues across restarts.
+        self._load_chain_tail()
+
+    def __repr__(self) -> str:
+        return f"AsqavLogger(log_path={self._log_path!r}, redact_content={self._redact_content})"
+
+    @property
+    def log_path(self) -> str:
+        return self._log_path
+
+    def _load_chain_tail(self) -> None:
+        """Read the last record of an existing log to resume seq and prev_hash."""
+        try:
+            if not os.path.exists(self._log_path):
+                return
+            with open(self._log_path, "rb") as fh:
+                size: Final = fh.seek(0, os.SEEK_END)
+                if size == 0:
+                    return
+                tail: Final = _read_tail(fh, size)
+            lines: Final = [line for line in tail.split(b"\n") if line.strip()]
+            if not lines:
+                return
+            last_record: Final = _JSON_OBJECT.validate_python(json.loads(lines[-1].decode("utf-8")))
+            record_hash: Final = last_record.get("record_hash")
+            seq: Final = last_record.get("seq")
+            self._prev_hash = record_hash if isinstance(record_hash, str) else GENESIS_HASH
+            self._call_count = seq + 1 if isinstance(seq, int) else 0
+        except (OSError, ValueError, ValidationError):
+            verbose_logger.debug("[AsqavLogger] Could not load chain tail", exc_info=True)
+
+    def _build_and_append(
+        self,
+        kwargs: Mapping[str, object],
+        response_obj: object,
+        start_time: object,
+        end_time: object,
+        status: str,
+    ) -> None:
+        """Build one audit record and append it to the JSONL log.
+
+        seq and prev_hash are assigned and the file is written under one lock,
+        so the on-disk order always matches the chain order. Chain state only
+        advances after a successful write: a failed write drops the record and
+        the chain continues from the last record on disk.
+        """
+        try:
+            loggable: Final = _extract_loggable(kwargs, response_obj, start_time, end_time, status)
+            clear_content: Final[Mapping[str, object]] = (
+                {}
+                if self._redact_content
+                else {"messages": kwargs.get("messages"), "response_content": _response_content(response_obj)}
+            )
+            with self._lock:
+                # Every field enters the hash, so an auditor can reproduce it from the log alone.
+                hashable: Final[Mapping[str, object]] = {
+                    "seq": self._call_count,
+                    "ts": datetime.now(tz=timezone.utc).isoformat(),
+                    "prev_hash": self._prev_hash,
+                    **loggable,
+                    **clear_content,
+                }
+                record_hash: Final = _sha256_hex(_canonical_bytes(hashable))
+                if not self._write_record({**hashable, "record_hash": record_hash}):
+                    return
+                self._prev_hash = record_hash
+                self._call_count += 1
+        except Exception:  # noqa: BLE001  # an audit-log failure must never break the LLM call
+            verbose_logger.debug("[AsqavLogger] Unhandled error while building a record", exc_info=True)
+
+    def _write_record(self, record: Mapping[str, object]) -> bool:
+        """Append one record to the log file. Returns False if the write failed."""
+        try:
+            parent: Final = os.path.dirname(self._log_path)
+            if parent:
+                os.makedirs(parent, exist_ok=True)
+            # Lock down a file an earlier run created under a permissive umask; new
+            # files are created 0o600 by _private_opener.
+            if os.path.exists(self._log_path):
+                os.chmod(self._log_path, 0o600)
+            with open(self._log_path, "a", encoding="utf-8", opener=_private_opener) as fh:
+                fh.write(json.dumps(record, separators=(",", ":"), default=str) + "\n")
+        except (OSError, TypeError, ValueError):
+            verbose_logger.warning("[AsqavLogger] Failed to write audit record", exc_info=True)
+            return False
+        else:
+            return True
+
+    def log_success_event(
+        self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object
+    ) -> None:
+        self._build_and_append(kwargs, response_obj, start_time, end_time, "success")
+
+    def log_failure_event(
+        self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object
+    ) -> None:
+        self._build_and_append(kwargs, response_obj, start_time, end_time, "failure")
+
+    async def async_log_success_event(
+        self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object
+    ) -> None:
+        await asyncio.to_thread(self._build_and_append, kwargs, response_obj, start_time, end_time, "success")
+
+    async def async_log_failure_event(
+        self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object
+    ) -> None:
+        await asyncio.to_thread(self._build_and_append, kwargs, response_obj, start_time, end_time, "failure")
+
+    def verify_chain(self, log_path: str | None = None) -> tuple[bool, str]:
+        """Check the audit log at log_path (default: this logger's log).
+
+        Returns (True, "ok") when every record's hash matches its content and its
+        prev_hash matches the previous record's hash, else (False, reason) for
+        the first violation found. Not called on the request path.
+        """
+        path: Final = log_path or self._log_path
+        try:
+            with open(path, encoding="utf-8") as fh:
+                return _verify_lines(fh)
+        except FileNotFoundError:
+            return False, f"log file not found: {path}"
+        except (OSError, ValueError, ValidationError) as exc:
+            return False, f"verification error: {exc}"
+
+
+def _verify_lines(lines: Iterable[str]) -> tuple[bool, str]:
+    prev_hash = GENESIS_HASH  # rebind-ok: walks the chain one record at a time
+    for lineno, raw_line in enumerate(lines, start=1):
+        line = raw_line.strip()
+        if not line:
+            continue
+        record = _JSON_OBJECT.validate_python(json.loads(line))
+        stored_hash = record.get("record_hash")
+        computed_hash = _sha256_hex(_canonical_bytes({k: v for k, v in record.items() if k != "record_hash"}))
+        if computed_hash != stored_hash:
+            return (
+                False,
+                f"line {lineno}: hash mismatch (stored={str(stored_hash)[:12]}, computed={computed_hash[:12]})",
+            )
+        record_prev = record.get("prev_hash", GENESIS_HASH)
+        if record_prev != prev_hash:
+            return (
+                False,
+                f"line {lineno}: prev_hash chain break (expected={prev_hash[:12]}, got={str(record_prev)[:12]})",
+            )
+        prev_hash = computed_hash
+    return True, "ok"

litellm/litellm_core_utils/custom_logger_registry.py

@@ -14,6 +14,7 @@ from litellm import _custom_logger_compatible_callbacks_literal
 from litellm.integrations.agentops import AgentOps
 from litellm.integrations.anthropic_cache_control_hook import AnthropicCacheControlHook
 from litellm.integrations.argilla import ArgillaLogger
+from litellm.integrations.asqav import AsqavLogger
 from litellm.integrations.azure_sentinel.azure_sentinel import AzureSentinelLogger
 from litellm.integrations.azure_storage.azure_storage import AzureBlobStorageLogger
 from litellm.integrations.bitbucket import BitBucketPromptManager
@@ -121,6 +122,7 @@ class CustomLoggerRegistry:
         "ternary": TernaryLogger,
         "posthog": PostHogLogger,
         "newrelic": NewRelicLogger,
+        "asqav": AsqavLogger,
     }
 
     try:

litellm/litellm_core_utils/litellm_logging.py

@@ -205,6 +205,7 @@ from litellm.utils import get_base_model_from_metadata, print_verbose
 
 from ..integrations.argilla import ArgillaLogger
 from ..integrations.arize.arize_phoenix import ArizePhoenixLogger
+from ..integrations.asqav import AsqavLogger
 from ..integrations.athina import AthinaLogger
 from ..integrations.azure_sentinel.azure_sentinel import AzureSentinelLogger
 from ..integrations.azure_storage.azure_storage import AzureBlobStorageLogger
@@ -5922,6 +5923,13 @@ def build_custom_logger_compatible_class(
         newrelic_logger: Final = NewRelicLogger()
         _in_memory_loggers.append(newrelic_logger)
         return newrelic_logger
+    elif logging_integration == "asqav":
+        for callback in _in_memory_loggers:
+            if isinstance(callback, AsqavLogger):
+                return callback
+        asqav_logger: Final = AsqavLogger()
+        _in_memory_loggers.append(asqav_logger)
+        return asqav_logger
     return None
 
 
@@ -6334,6 +6342,10 @@ def get_custom_logger_compatible_class(
                     return callback
                 if isinstance(callback, NewRelicLogger):
                     return callback
+        elif logging_integration == "asqav":
+            for callback in _in_memory_loggers:
+                if isinstance(callback, AsqavLogger):
+                    return callback
         return None
 
     except Exception as e:

tests/unit/integrations/asqav/test_asqav.py

@@ -0,0 +1,462 @@
+"""Tests for the Asqav local-first audit-log callback.
+
+No network and no provider calls: every test writes to a tmp_path log.
+
+The chain checks re-derive record hashes with hashlib and json directly, so they
+also prove an auditor can verify the log without litellm installed.
+"""
+
+import asyncio
+import hashlib
+import json
+import os
+import stat
+import threading
+from collections.abc import Mapping, Sequence
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Final
+
+import pytest
+
+import litellm
+from litellm.integrations.asqav import AsqavLogger
+from litellm.integrations.asqav.asqav import GENESIS_HASH
+from litellm.litellm_core_utils import litellm_logging as logging_module
+from litellm.litellm_core_utils.custom_logger_registry import CustomLoggerRegistry
+from litellm.types.utils import Choices, Message, ModelResponse, Usage
+from tests._wait_helpers import wait_until
+
+START: Final = datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
+END: Final = datetime(2026, 1, 1, 0, 0, 1, tzinfo=timezone.utc)
+
+
+def _kwargs(content: str = "hello", call_id: str = "call-1") -> dict[str, object]:
+    return {
+        "model": "gpt-4o",
+        "messages": [{"role": "user", "content": content}],
+        "litellm_call_id": call_id,
+    }
+
+
+def _response(content: str = "world") -> ModelResponse:
+    return ModelResponse(
+        id="chatcmpl-test",
+        model="gpt-4o",
+        choices=[Choices(finish_reason="stop", index=0, message=Message(content=content, role="assistant"))],
+        usage=Usage(prompt_tokens=10, completion_tokens=5, total_tokens=15),
+    )
+
+
+def _logger(path: Path, redact_content: bool = True) -> AsqavLogger:
+    return AsqavLogger(log_path=str(path), redact_content=redact_content)
+
+
+def _append(logger: AsqavLogger, count: int, prefix: str = "msg") -> None:
+    for i in range(count):
+        logger.log_success_event(_kwargs(content=f"{prefix}-{i}"), _response(content=f"resp-{i}"), START, END)
+
+
+def _records(path: Path) -> list[dict[str, object]]:
+    return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]
+
+
+def _rewrite(path: Path, records: Sequence[Mapping[str, object]]) -> None:
+    path.write_text("".join(json.dumps(r, separators=(",", ":")) + "\n" for r in records), encoding="utf-8")
+
+
+def _stdlib_hash(record: Mapping[str, object]) -> str:
+    """The record hash, recomputed the way an auditor would with only the standard library."""
+    body = {k: v for k, v in record.items() if k != "record_hash"}
+    return hashlib.sha256(json.dumps(body, sort_keys=True, separators=(",", ":")).encode("utf-8")).hexdigest()
+
+
+def test_first_record_starts_the_chain(tmp_path: Path) -> None:
+    path = tmp_path / "audit.jsonl"
+    _logger(path).log_success_event(_kwargs(), _response(), START, END)
+
+    [record] = _records(path)
+    assert record["seq"] == 0
+    assert record["prev_hash"] == GENESIS_HASH
+    assert record["status"] == "success"
+    assert record["record_hash"] == _stdlib_hash(record)
+
+
+def test_records_link_and_hash_reproduces_with_stdlib(tmp_path: Path) -> None:
+    path = tmp_path / "audit.jsonl"
+    logger = _logger(path)
+    _append(logger, 10)
+
+    records = _records(path)
+    assert [r["seq"] for r in records] == list(range(10))
+    assert records[0]["prev_hash"] == GENESIS_HASH
+    for previous, current in zip(records, records[1:]):
+        assert current["prev_hash"] == previous["record_hash"]
+    assert all(r["record_hash"] == _stdlib_hash(r) for r in records)
+    assert logger.verify_chain() == (True, "ok")
+
+
+def test_record_carries_usage_latency_and_call_metadata(tmp_path: Path) -> None:
+    path = tmp_path / "audit.jsonl"
+    _logger(path).log_success_event(_kwargs(call_id="call-abc"), _response(), START, END)
+
+    [record] = _records(path)
+    assert record["call_id"] == "call-abc"
+    assert record["model"] == "gpt-4o"
+    assert record["latency_ms"] == 1000
+    assert (record["prompt_tokens"], record["completion_tokens"], record["total_tokens"]) == (10, 5, 15)
+    assert record["finish_reason"] == "stop"
+
+
+def test_latency_is_null_when_times_are_not_datetimes(tmp_path: Path) -> None:
+    path = tmp_path / "audit.jsonl"
+    _logger(path).log_success_event(_kwargs(), _response(), None, None)
+
+    [record] = _records(path)
+    assert record["latency_ms"] is None
+
+
+def test_verify_detects_an_edited_field(tmp_path: Path) -> None:
+    path = tmp_path / "audit.jsonl"
+    logger = _logger(path)
+    _append(logger, 5)
+    records = _records(path)
+    _rewrite(path, [*records[:2], {**records[2], "model": "tampered-model"}, *records[3:]])
+
+    ok, reason = logger.verify_chain()
+
+    assert ok is False
+    assert reason.startswith("line 3: hash mismatch")
+
+
+def test_verify_detects_a_rehashed_prev_hash_edit(tmp_path: Path) -> None:
+    path = tmp_path / "audit.jsonl"
+    logger = _logger(path)
+    _append(logger, 5)
+    records = _records(path)
+    forged = {**records[3], "prev_hash": "a" * 64}
+    _rewrite(path, [*records[:3], {**forged, "record_hash": _stdlib_hash(forged)}, *records[4:]])
+
+    ok, reason = logger.verify_chain()
+
+    assert ok is False
+    assert reason.startswith("line 4: prev_hash chain break")
+
+
+def test_verify_detects_a_deleted_record(tmp_path: Path) -> None:
+    path = tmp_path / "audit.jsonl"
+    logger = _logger(path)
+    _append(logger, 5)
+    records = _records(path)
+    _rewrite(path, [*records[:2], *records[3:]])
+
+    ok, reason = logger.verify_chain()
+
+    assert ok is False
+    assert reason.startswith("line 3: prev_hash chain break")
+
+
+def test_verify_reports_a_missing_file(tmp_path: Path) -> None:
+    missing = tmp_path / "missing.jsonl"
+
+    assert _logger(tmp_path / "audit.jsonl").verify_chain(str(missing)) == (False, f"log file not found: {missing}")
+
+
+def test_verify_skips_blank_lines(tmp_path: Path) -> None:
+    path = tmp_path / "audit.jsonl"
+    logger = _logger(path)
+    _append(logger, 2)
+    first, second = path.read_text(encoding="utf-8").splitlines()
+    path.write_text(f"{first}\n\n{second}\n", encoding="utf-8")
+
+    assert logger.verify_chain() == (True, "ok")
+
+
+def test_verify_reports_a_line_that_is_not_json(tmp_path: Path) -> None:
+    path = tmp_path / "audit.jsonl"
+    logger = _logger(path)
+    _append(logger, 1)
+    with path.open("a", encoding="utf-8") as fh:
+        fh.write("not json\n")
+
+    ok, reason = logger.verify_chain()
+
+    assert ok is False
+    assert reason.startswith("verification error:")
+
+
+def test_chain_and_seq_resume_after_restart(tmp_path: Path) -> None:
+    path = tmp_path / "audit.jsonl"
+    _append(_logger(path), 5, prefix="first")
+    last_before_restart = _records(path)[-1]
+
+    restarted = _logger(path)
+    _append(restarted, 1, prefix="second")
+
+    records = _records(path)
+    assert len(records) == 6
+    assert records[5]["seq"] == 5
+    assert records[5]["prev_hash"] == last_before_restart["record_hash"]
+    assert restarted.verify_chain() == (True, "ok")
+
+
+def test_restart_resumes_when_last_record_exceeds_tail_window(tmp_path: Path) -> None:
+    """A last record bigger than the 4 KB first read window still resumes the chain."""
+    path = tmp_path / "audit.jsonl"
+    big_kwargs = {**_kwargs(content="big"), "metadata": {"blob": "x" * 10_000}}
+    _logger(path).log_success_event(big_kwargs, _response(), START, END)
+    assert os.path.getsize(path) > 4096
+
+    restarted = _logger(path)
+    _append(restarted, 2)
+
+    records = _records(path)
+    assert [r["seq"] for r in records] == [0, 1, 2]
+    assert restarted.verify_chain() == (True, "ok")
+
+
+def test_restart_on_a_blank_file_starts_a_new_chain(tmp_path: Path) -> None:
+    path = tmp_path / "audit.jsonl"
+    path.write_text("\n\n", encoding="utf-8")
+
+    restarted = _logger(path)
+    _append(restarted, 1)
+
+    [record] = _records(path)
+    assert (record["seq"], record["prev_hash"]) == (0, GENESIS_HASH)
+    assert restarted.verify_chain() == (True, "ok")
+
+
+def test_restart_after_a_corrupt_last_line_keeps_logging_and_verify_flags_it(tmp_path: Path) -> None:
+    path = tmp_path / "audit.jsonl"
+    _append(_logger(path), 1)
+    with path.open("a", encoding="utf-8") as fh:
+        fh.write("{truncated\n")
+
+    restarted = _logger(path)
+    _append(restarted, 1, prefix="after")
+
+    lines = path.read_text(encoding="utf-8").splitlines()
+    assert len(lines) == 3
+    assert json.loads(lines[2])["messages_digest"] is not None
+    ok, reason = restarted.verify_chain()
+    assert ok is False
+    assert reason.startswith("verification error:")
+
+
+def test_failure_event_is_logged(tmp_path: Path) -> None:
+    path = tmp_path / "audit.jsonl"
+    _logger(path).log_failure_event(_kwargs(), None, START, END)
+
+    [record] = _records(path)
+    assert record["status"] == "failure"
+    assert record["prompt_tokens"] is None
+
+
+def test_malformed_response_still_writes_a_record(tmp_path: Path) -> None:
+    path = tmp_path / "audit.jsonl"
+    _logger(path).log_success_event(_kwargs(), object(), START, END)
+
+    [record] = _records(path)
+    assert record["status"] == "success"
+    assert record["response_content_digest"] is None
+
+
+def test_failed_write_does_not_raise_or_advance_the_chain(tmp_path: Path) -> None:
+    blocker = tmp_path / "logs"
+    blocker.write_text("", encoding="utf-8")
+    path = blocker / "audit.jsonl"
+    logger = _logger(path)
+
+    logger.log_success_event(_kwargs(call_id="call-dropped"), _response(), START, END)
+    blocker.unlink()
+    logger.log_success_event(_kwargs(call_id="call-kept"), _response(), START, END)
+
+    [record] = _records(path)
+    assert record["call_id"] == "call-kept"
+    assert record["seq"] == 0
+    assert record["prev_hash"] == GENESIS_HASH
+
+
+def test_unexpected_error_while_building_does_not_raise_or_advance_the_chain(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    path = tmp_path / "audit.jsonl"
+    logger = _logger(path)
+
+    def _explode(_record: Mapping[str, object]) -> bool:
+        raise RuntimeError("boom")
+
+    monkeypatch.setattr(logger, "_write_record", _explode)
+    logger.log_success_event(_kwargs(call_id="call-dropped"), _response(), START, END)
+    monkeypatch.undo()
+    logger.log_success_event(_kwargs(call_id="call-kept"), _response(), START, END)
+
+    [record] = _records(path)
+    assert (record["call_id"], record["seq"], record["prev_hash"]) == ("call-kept", 0, GENESIS_HASH)
+
+
+def test_content_is_digested_by_default(tmp_path: Path) -> None:
+    path = tmp_path / "audit.jsonl"
+    _logger(path).log_success_event(_kwargs(content="my private prompt"), _response("my private answer"), START, END)
+
+    raw = path.read_text(encoding="utf-8")
+    [record] = _records(path)
+    assert "my private prompt" not in raw
+    assert "my private answer" not in raw
+    expected_digest = hashlib.sha256(
+        json.dumps([{"role": "user", "content": "my private prompt"}], sort_keys=True, separators=(",", ":")).encode()
+    ).hexdigest()
+    assert record["messages_digest"] == expected_digest
+    assert isinstance(record["response_content_digest"], str)
+
+
+def test_redact_content_false_stores_plaintext(tmp_path: Path) -> None:
+    path = tmp_path / "audit.jsonl"
+    logger = _logger(path, redact_content=False)
+    logger.log_success_event(_kwargs(content="visible prompt"), _response("visible answer"), START, END)
+
+    [record] = _records(path)
+    assert record["messages"] == [{"role": "user", "content": "visible prompt"}]
+    assert record["response_content"] == "visible answer"
+    assert logger.verify_chain() == (True, "ok")
+
+
+def test_env_vars_configure_path_and_redaction(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    path = tmp_path / "from-env.jsonl"
+    monkeypatch.setenv("ASQAV_LOG_PATH", str(path))
+    monkeypatch.setenv("ASQAV_REDACT_CONTENT", "false")
+
+    logger = AsqavLogger()
+    logger.log_success_event(_kwargs(content="env prompt"), _response(), START, END)
+
+    assert logger.log_path == str(path)
+    assert _records(path)[0]["messages"] == [{"role": "user", "content": "env prompt"}]
+
+
+def test_proxy_identity_is_kept_and_secrets_are_dropped(tmp_path: Path) -> None:
+    path = tmp_path / "audit.jsonl"
+    kwargs = {
+        **_kwargs(),
+        "litellm_params": {
+            "metadata": {
+                "user_api_key_user_id": "user-abc",
+                "user_api_key_team_id": "team-xyz",
+                "user_api_key_org_id": "org-123",
+                "user_api_key_alias": "my-key",
+                "user_api_key": "hashed-key-value",
+                "Authorization": "placeholder-auth-header",
+                "requester_ip_address": "10.0.0.1",
+            }
+        },
+    }
+    _logger(path).log_success_event(kwargs, _response(), START, END)
+
+    metadata = _records(path)[0]["metadata"]
+    assert metadata == {
+        "user_api_key_user_id": "user-abc",
+        "user_api_key_team_id": "team-xyz",
+        "user_api_key_org_id": "org-123",
+        "user_api_key_alias": "my-key",
+    }
+
+
+def test_non_json_metadata_is_stored_as_text_and_verifies(tmp_path: Path) -> None:
+    path = tmp_path / "audit.jsonl"
+    logger = _logger(path)
+    kwargs = {**_kwargs(), "metadata": {"requested_at": START}}
+    logger.log_success_event(kwargs, _response(), START, END)
+
+    assert _records(path)[0]["metadata"] == {"requested_at": str(START)}
+    assert logger.verify_chain() == (True, "ok")
+
+
+@pytest.mark.skipif(os.name == "nt", reason="POSIX file modes")
+def test_log_file_is_private(tmp_path: Path) -> None:
+    path = tmp_path / "audit.jsonl"
+    path.write_text("", encoding="utf-8")
+    path.chmod(0o644)
+
+    _logger(path).log_success_event(_kwargs(), _response(), START, END)
+
+    assert stat.S_IMODE(path.stat().st_mode) == 0o600
+
+
+def test_concurrent_callbacks_keep_disk_order_equal_to_chain_order(tmp_path: Path) -> None:
+    path = tmp_path / "audit.jsonl"
+    logger = _logger(path)
+    threads_count, per_thread = 8, 25
+    barrier = threading.Barrier(threads_count)
+
+    def worker(thread_id: int) -> None:
+        barrier.wait()
+        _append(logger, per_thread, prefix=f"t{thread_id}")
+
+    threads = [threading.Thread(target=worker, args=(i,)) for i in range(threads_count)]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+
+    assert [r["seq"] for r in _records(path)] == list(range(threads_count * per_thread))
+    assert logger.verify_chain() == (True, "ok")
+
+
+def test_async_hooks_write_records(tmp_path: Path) -> None:
+    path = tmp_path / "audit.jsonl"
+    logger = _logger(path)
+
+    async def run() -> None:
+        await logger.async_log_success_event(_kwargs(content="async-ok"), _response(), START, END)
+        await logger.async_log_failure_event(_kwargs(content="async-fail"), None, START, END)
+
+    asyncio.run(run())
+
+    assert [r["status"] for r in _records(path)] == ["success", "failure"]
+    assert logger.verify_chain() == (True, "ok")
+
+
+def test_asqav_is_a_registered_callback_name() -> None:
+    assert "asqav" in litellm._known_custom_logger_compatible_callbacks
+    assert CustomLoggerRegistry.CALLBACK_CLASS_STR_TO_CLASS_TYPE["asqav"] is AsqavLogger
+
+
+def test_callbacks_asqav_builds_one_logger_and_reuses_it(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    """`callbacks=["asqav"]` goes through litellm_logging, which must hand back one instance."""
+    monkeypatch.setenv("ASQAV_LOG_PATH", str(tmp_path / "audit.jsonl"))
+    monkeypatch.setattr(logging_module, "_in_memory_loggers", [])
+
+    assert logging_module.get_custom_logger_compatible_class("asqav") is None
+
+    first = logging_module._init_custom_logger_compatible_class(
+        logging_integration="asqav", internal_usage_cache=None, llm_router=None, custom_logger_init_args={}
+    )
+    second = logging_module._init_custom_logger_compatible_class(
+        logging_integration="asqav", internal_usage_cache=None, llm_router=None, custom_logger_init_args={}
+    )
+
+    assert isinstance(first, AsqavLogger)
+    assert second is first
+    assert first.log_path == str(tmp_path / "audit.jsonl")
+    assert logging_module.get_custom_logger_compatible_class("asqav") is first
+
+
+def test_completion_with_callbacks_asqav_appends_a_verifiable_record(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A real litellm.completion call (mock_response, no network) lands one chained record."""
+    path = tmp_path / "audit.jsonl"
+    monkeypatch.setenv("ASQAV_LOG_PATH", str(path))
+    monkeypatch.setattr(logging_module, "_in_memory_loggers", [])
+    monkeypatch.setattr(litellm, "callbacks", ["asqav"])
+
+    litellm.completion(
+        model="gpt-4o", messages=[{"role": "user", "content": "hi"}], mock_response="hello from the mock"
+    )
+    wait_until(lambda: path.exists() and bool(_records(path)), message="asqav record written")
+
+    [record] = _records(path)
+    assert record["status"] == "success"
+    assert record["model"] == "gpt-4o"
+    assert record["record_hash"] == _stdlib_hash(record)
+    assert "hello from the mock" not in path.read_text(encoding="utf-8")