UmbraAudio/scripts/audit_controller_bundle.py

#!/usr/bin/env python3
"""Audit a built bundle against the project it claims to describe.

The controller's validator checks that the bundle is internally consistent: a
required field is present, an id resolves, no credential is present. It cannot
check that the bundle is *true*. Every hash, byte size and duration in here is
something the audio system asserted about itself, and this is the file that goes
and checks those assertions against the filesystem.

That distinction is not pedantic. The failure this project has actually hit is a
record that says one thing and means another -- an approved chapter whose
published MP3 is four seconds longer than the approved WAV because padding was
added afterwards. A validator passes that bundle happily. Only re-measuring
finds it.

    python3 scripts/audit_controller_bundle.py --bundle /tmp/dh-bundle

Exits 0 when every checked claim holds, 1 when any does not. Prints one line per
problem and a summary of how many claims were checked.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import sqlite3
import wave
from collections import Counter, defaultdict
from pathlib import Path
from typing import Any


def load(path: Path) -> Any:
    return json.loads(path.read_text())


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(1 << 20), b""):
            digest.update(chunk)
    return digest.hexdigest()


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__,
                                     formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--bundle", required=True, help="bundle directory")
    parser.add_argument("--project", required=True, help="project directory name")
    parser.add_argument("--hash-renders", action="store_true", default=True,
                        help="re-hash every render file (slower, thorough)")
    parser.add_argument("--no-hash-renders", dest="hash_renders", action="store_false")
    args = parser.parse_args()

    bundle = Path(args.bundle)
    repo = Path(__file__).resolve().parent.parent
    book_dir = repo / args.project
    umbra = book_dir / ".umbra"

    index = load(bundle / "bundle.json")
    raw = (bundle / "manifest.jsonl").read_text().splitlines()
    records = [json.loads(line) for line in raw if line.strip()]
    by_id = {r["id"]: r for r in records}
    kinds = Counter(r["kind"] for r in records)

    problems: list[str] = []
    checked = 0

    def check(condition: bool, message: str) -> None:
        nonlocal checked
        checked += 1
        if not condition:
            problems.append(message)

    # 1. The index's own content hash must describe the file it ships with.
    manifest_bytes = (bundle / "manifest.jsonl").read_bytes()
    actual = hashlib.sha256(manifest_bytes).hexdigest()
    check(actual == index["content_hash"]["value"],
          f"bundle.json content_hash {index['content_hash']['value'][:16]}... "
          f"does not match manifest.jsonl ({actual[:16]}...)")
    check(index["counts"]["records_total"] == len(records),
          f"bundle.json claims {index['counts']['records_total']} records, file has {len(records)}")

    # 2. Every utterance's text hash must match the text it ships with.
    for record in records:
        if record["kind"] == "utterance":
            digest = hashlib.sha256(record["text"].encode("utf-8")).hexdigest()
            check(digest == record["text_sha256"],
                  f"utterance {record['id']}: text_sha256 does not match its own text")
            check(len(record["text"]) == record["text_char_count"],
                  f"utterance {record['id']}: text_char_count disagrees with text length")

    # 3. Every referenced id must resolve, including the ones the controller's
    #    validator does not check (speaker_id, render_id).
    for record in records:
        for field in ("book_id", "chapter_id", "speaker_id", "render_id", "voice_id"):
            value = record.get(field)
            if isinstance(value, str):
                check(value in by_id,
                      f"{record['kind']} {record['id']}: {field} -> {value!r} is not in the bundle")
        for field in ("chapters", "renders", "inputs", "published_artifact_ids"):
            for value in record.get(field, []) or []:
                check(value in by_id,
                      f"{record['kind']} {record['id']}: {field} contains "
                      f"{value!r}, not in the bundle")

    # 4. Render paths must exist, be the right size, and (optionally) hash right.
    renders = [r for r in records if r["kind"] == "render"]
    missing = unreadable = 0
    for record in renders:
        path = book_dir / ".umbra" / record["path"]
        if not path.exists():
            if record.get("file_present") is False:
                continue
            missing += 1
            problems.append(f"render {record['id']}: path {record['path']} does not exist")
            continue
        checked += 1
        size = path.stat().st_size
        check(size == record["byte_size"],
              f"render {record['id']}: byte_size {record['byte_size']} != actual {size}")
        if record.get("content_sha256") and args.hash_renders:
            check(sha256_file(path) == record["content_sha256"],
                  f"render {record['id']}: content_sha256 does not match the file on disk")
        if record["audio_format"] and record["audio_format"].get("codec") == "unreadable":
            unreadable += 1

    # 5. Durations must be re-derivable for WAV files, which is the one format
    #    the standard library can measure without trusting the recorder.
    sampled = 0
    for record in renders:
        if record["path"].endswith(".wav") and record.get("duration_ms"):
            path = book_dir / ".umbra" / record["path"]
            if not path.exists():
                continue
            try:
                with wave.open(str(path)) as handle:
                    rate = handle.getframerate()
                    if not rate:
                        continue
                    measured = round(handle.getnframes() / rate * 1000)
            except (OSError, wave.Error, EOFError):
                continue
            checked += 1
            sampled += 1
            check(measured == record["duration_ms"],
                  f"render {record['id']}: duration_ms "
                  f"{record['duration_ms']} != measured {measured}")

    # 6. The bundle's own counts must match the database it was built from.
    conn = sqlite3.connect(f"file:{umbra / 'project.db'}?mode=ro", uri=True)
    db_chapters = conn.execute("SELECT count(*) FROM chapters WHERE project_id=1").fetchone()[0]
    db_utterances = conn.execute(
        "SELECT count(*) FROM utterances u JOIN chapters c ON c.id=u.chapter_id "
        "WHERE c.project_id=1").fetchone()[0]
    db_names = conn.execute(
        "SELECT count(DISTINCT trim(speaker_name)) FROM utterances u JOIN chapters c "
        "ON c.id=u.chapter_id WHERE c.project_id=1 AND speaker_name IS NOT NULL").fetchone()[0]
    db_null_speakers = conn.execute(
        "SELECT count(*) FROM utterances u JOIN chapters c ON c.id=u.chapter_id "
        "WHERE c.project_id=1 AND (u.speaker_name IS NULL OR trim(u.speaker_name)='')"
    ).fetchone()[0]
    db_speaker_counts = dict(conn.execute(
        "SELECT trim(u.speaker_name), count(*) FROM utterances u JOIN chapters c "
        "ON c.id=u.chapter_id WHERE c.project_id=1 AND u.speaker_name IS NOT NULL "
        "GROUP BY trim(u.speaker_name)").fetchall())
    check(len([r for r in records if r["kind"] == "chapter"]) == db_chapters,
          f"chapter records != database chapters ({db_chapters})")
    check(len([r for r in records if r["kind"] == "utterance"]) == db_utterances,
          f"utterance records != database utterances ({db_utterances})")
    check(len([r for r in records if r["kind"] == "speaker"]) == db_names,
          f"speaker records != distinct speaker names in the database ({db_names})")
    conn.close()

    # 7. Speaker utterance counts must be true, because they are the number a
    #    mapping is most likely to trust. `utterance_count` is the exact count
    #    for that exact spelling; the implicit-narrator figure is checked
    #    separately because it is a deliberate inference, not a stored value.
    for record in records:
        if record["kind"] != "speaker":
            continue
        actual_count = db_speaker_counts.get(record["primary_name"])
        check(actual_count == record["utterance_count"],
              f"speaker {record['id']}: utterance_count {record['utterance_count']} "
              f"!= database {actual_count}")
        expected_implicit = record["utterance_count"] + (
            db_null_speakers if record["primary_name"].casefold() == "narrator" else 0
        )
        check(expected_implicit == record["utterance_count_including_implicit_narrator"],
              f"speaker {record['id']}: implicit-narrator count "
              f"{record['utterance_count_including_implicit_narrator']} "
              f"!= expected {expected_implicit}")

    # 7b. Every utterance's speaker must resolve, and every implicit-narrator
    #     row must actually have had a null speaker name. This is the assertion
    #     that would catch the bundle quietly inventing a narrator.
    null_rows = 0
    for record in records:
        if record["kind"] != "utterance":
            continue
        implicit = record["speaker_resolution"] == "implicit_narrator"
        if implicit:
            null_rows += 1
            check(record["speaker_name_raw"] in (None, ""),
                  f"utterance {record['id']}: marked implicit_narrator but "
                  f"speaker_name_raw is {record['speaker_name_raw']!r}")
        else:
            check(record["speaker_name_raw"] not in (None, ""),
                  f"utterance {record['id']}: marked explicit but speaker_name_raw is null")
        check(record["speaker_id"] is not None,
              f"utterance {record['id']}: speaker did not resolve")
    check(null_rows == db_null_speakers,
          f"{null_rows} utterances marked implicit_narrator != {db_null_speakers} "
          "rows with an empty speaker_name")

    # 8. Every chapter's utterance_count must be true.
    db_per_chapter = {}
    conn = sqlite3.connect(f"file:{umbra / 'project.db'}?mode=ro", uri=True)
    for row_id, count in conn.execute(
            "SELECT u.chapter_id, count(*) FROM utterances u JOIN chapters c "
            "ON c.id=u.chapter_id WHERE c.project_id=1 GROUP BY u.chapter_id").fetchall():
        db_per_chapter[f"ch-{row_id}"] = count
    conn.close()
    for record in records:
        if record["kind"] != "chapter":
            continue
        check(db_per_chapter.get(record["id"], 0) == record["utterance_count"],
              f"chapter {record['id']}: utterance_count {record['utterance_count']} "
              f"!= database {db_per_chapter.get(record['id'], 0)}")

    # 9. Casefold collisions must be recorded, not silently merged.
    collisions: dict[str, list[str]] = defaultdict(list)
    for record in records:
        if record["kind"] == "speaker":
            collisions[record["primary_name"].casefold()].append(record["primary_name"])
    real_collisions = {k: v for k, v in collisions.items() if len(v) > 1}
    for key, names in real_collisions.items():
        declared = [r for r in records if r["kind"] == "speaker"
                    and r.get("casefold_collision_group") == key]
        check(len(declared) == len(names),
              f"speaker name collision {key!r} ({names}) is not declared in the bundle")

    print(f"records by kind: {dict(kinds)}")
    print(f"claims checked:  {checked}")
    print(f"renders:         {len(renders)} "
          f"({missing} missing, {unreadable} unreadable, {sampled} durations re-measured)")
    print(f"speaker name collisions declared: "
          f"{ {k: sorted(v) for k, v in real_collisions.items()} }")
    print(f"implicit-narrator utterances:   {null_rows} (database agrees: "
          f"{null_rows == db_null_speakers})")

    if problems:
        print(f"\n{len(problems)} problem(s):\n")
        for problem in problems[:60]:
            print(f"  - {problem}")
        if len(problems) > 60:
            print(f"  ... and {len(problems) - 60} more")
        return 1

    print("\nEvery checked claim holds: hashes, byte sizes, durations, counts and "
          "references agree with the project on disk.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())