UmbraTrans/docs/audio-system-bundle-validator.py

#!/usr/bin/env python3
# Copyright 2026 Umbra Tools (umbratools.com)
# SPDX-License-Identifier: Apache-2.0

"""Validate an audio-system bundle against the request in this repository.

This ships *with* the request, not after it. The point of asking for machine-readable
metadata rather than prose is that the sender can check their own work before it travels;
a round trip that comes back with forty precise errors beats one that comes back with
"mostly fine, I think".

Every problem is reported, not just the first. A validator that stops at the first failure
turns one exchange into three, because each run reveals one more problem and the sender has
to guess that there are others.

    python3 validate.py <bundle-directory>

Exits 0 when the bundle is usable, 1 when it is not, 2 when the bundle cannot be read at
all. Prints one line per problem with the record and field it applies to.

Only the standard library is used, deliberately: whoever runs this is running it on the
audio system, and adding a dependency to someone's working machine to check a file is not a
reasonable thing to ask. That means the JSON Schema checks here are the subset that can be
done without a library - enough to catch the mistakes that actually happen, and no more. A
full schema validator is the reader's job; this is a smoke alarm, not an inspector.
"""

from __future__ import annotations

import json
import re
import sys
from pathlib import Path

REQUIRED_FILES = (
    "README.md",
    "bundle.json",
    "VOCABULARY.md",
    "manifest.jsonl",
    "validate.py",
)

def fail(problems: list[str], message: str) -> None:
    problems.append(message)


def check_structure(root: Path, problems: list[str]) -> bool:
    """The bundle's shape. Missing pieces are reported and counted, not fatal per file."""

    missing = [name for name in REQUIRED_FILES if not (root / name).exists()]
    if missing:
        fail(problems, f"missing required file(s): {', '.join(missing)}")
    for directory in ("schemas", "examples"):
        if not (root / directory).is_dir():
            fail(problems, f"missing directory: {directory}/")
    return not missing


def load_manifest(root: Path, problems: list[str]) -> list[dict]:
    """Parse manifest.jsonl, reporting the line number of anything unreadable.

    Line numbers matter more than they look: the sender's next action is to open the file at
    the line the validator names, so an error saying "record 4 is broken" has cost them a
    search.
    """

    path = root / "manifest.jsonl"
    if not path.exists():
        return []
    records: list[dict] = []
    for number, line in enumerate(path.read_text().splitlines(), start=1):
        if not line.strip():
            continue
        try:
            record = json.loads(line)
        except json.JSONDecodeError as error:
            fail(problems, f"manifest.jsonl:{number}: not valid JSON ({error.msg})")
            continue
        if not isinstance(record, dict):
            fail(problems, f"manifest.jsonl:{number}: a record, not a {type(record).__name__}")
            continue
        records.append(record)
    if not records:
        fail(problems, "manifest.jsonl contains no records")
    return records


def check_required_fields(root: Path, records: list[dict], problems: list[str]) -> None:
    """Whatever the bundle index declares required, each record must actually have.

    Driven by bundle.json rather than a hardcoded list, because the sender knows their own
    records and I do not. A missing key that the sender said was required is the error this
    catches, and it is otherwise invisible until something downstream reads it.
    """

    index_path = root / "bundle.json"
    if not index_path.exists():
        return
    try:
        index = json.loads(index_path.read_text())
    except json.JSONDecodeError as error:
        fail(problems, f"bundle.json: not valid JSON ({error.msg})")
        return

    required = index.get("required_fields") or {}
    if not isinstance(required, dict):
        fail(problems, "bundle.json: required_fields must be an object of type -> [fields]")
        return

    for number, record in enumerate(records, start=1):
        kind = record.get("kind") or record.get("record_type") or record.get("type")
        for field in required.get(kind, []) or []:
            if field not in record:
                fail(problems, f"manifest.jsonl:{number}: ({kind}) missing required field '{field}'")


def check_identifiers_resolve(records: list[dict], problems: list[str]) -> None:
    """Every id that points at another record must point at one that exists.

    A dangling reference is the failure mode that makes a mapping look complete and be
    wrong: the record is present, the link is plausible, and nothing checks it.
    """

    known: set[str] = set()
    for record in records:
        for key in ("id", "uuid"):
            if isinstance(record.get(key), str):
                known.add(record[key])

    reference_fields = (
        "chapter_id", "book_id", "render_id", "parent_id", "source_id", "input_id",
        "voice_id", "takes", "inputs", "outputs", "chapters", "renders",
    )
    for number, record in enumerate(records, start=1):
        for field in reference_fields:
            if field not in record:
                continue
            value = record[field]
            candidates = value if isinstance(value, list) else [value]
            for one in candidates:
                if isinstance(one, dict):
                    one = one.get("id")
                if isinstance(one, str) and one not in known:
                    fail(
                        problems,
                        f"manifest.jsonl:{number}: '{field}' points at {one!r}, which is "
                        "not in the manifest",
                    )


#: Field names that name a location, which is useless without something to resolve it
#: against. A relative path with no declared base cannot be read at all, and an absolute
#: one leaks the machine layout into a document meant to travel.
PATH_FIELDS = frozenset({"path", "file", "filepath", "file_path", "location", "source_path"})

#: Fields that state a duration. Not flagged on their own - the request asks for units in
#: the name, and a sender who wrote `duration_ms` has done it. This exists to catch the
#: field that carries a time and says so nowhere, which is the one that is a factor of a
#: thousand wrong and still looks plausible.
DURATION_FIELDS = frozenset({"duration", "length", "elapsed", "runtime", "time"})


def check_conventions(records: list[dict], problems: list[str]) -> None:
    """Conventions the request asks for, checked mechanically.

    One diagnostic per record per problem. The first version of this iterated a pattern
    table inside the field loop and reported the same path twice, which is worth avoiding
    for its own sake: a tool whose job is precise errors loses credibility the first time
    it says the same thing twice.
    """

    for number, record in enumerate(records, start=1):
        for field in record:
            lowered = field.lower()
            if lowered in PATH_FIELDS and not (
                "relative_to" in record or "path_base" in record or "path_is_absolute" in record
            ):
                fail(
                    problems,
                    f"manifest.jsonl:{number}: '{field}' looks like a path and the record "
                    "does not say what it is relative to (add 'relative_to' or "
                    "'path_is_absolute')",
                )
            if lowered in DURATION_FIELDS:
                fail(
                    problems,
                    f"manifest.jsonl:{number}: '{field}' states a duration with no unit. "
                    "Rename it to say which - duration_ms, duration_s - rather than leaving "
                    "the reader to assume.",
                )


def check_credentials(records: list[dict], root: Path, problems: list[str]) -> None:
    """Refuse anything that looks like a secret.

    The bundle travels between machines and gets committed to repositories. A key or token
    in it would be a real incident, and this is cheap to check - a substring match catches
    the shapes people actually use.

    `manifest.jsonl` is scanned per record, with the line number, and then excluded from the
    file sweep. Scanning it twice reported every finding twice, which is noise in a tool
    whose only job is to be trusted about specifics.
    """

    suspicious = re.compile(
        r"(sk-[A-Za-z0-9]{16,}|ghp_[A-Za-z0-9]{20,}|AKIA[0-9A-Z]{12,}"
        r"|-----BEGIN[A-Z ]*PRIVATE KEY-----|eyJ[A-Za-z0-9_-]{20,}\.)"
    )
    for number, record in enumerate(records, start=1):
        if suspicious.search(json.dumps(record)):
            fail(
                problems,
                f"manifest.jsonl:{number}: looks like it contains a credential. Remove it "
                "before sending; this bundle is meant to be committable.",
            )
    for path in sorted(root.rglob("*")):
        if not path.is_file() or path.name == "manifest.jsonl":
            continue
        if path.suffix not in {".json", ".jsonl", ".md", ".env", ".yaml", ".yml"}:
            continue
        try:
            if suspicious.search(path.read_text(errors="ignore")):
                fail(
                    problems,
                    f"{path.relative_to(root)}: looks like it contains a credential",
                )
        except OSError:  # pragma: no cover - unreadable file is not a credential problem
            pass


def check_an_example_is_complete(root: Path, problems: list[str]) -> None:
    """At least one worked example, with something in it.

    The example is the part that actually teaches me the shape. A bundle with schemas and a
    manifest but no traced book leaves me inferring the relationships, which is the guessing
    this whole request exists to avoid.
    """

    examples = root / "examples"
    if not examples.is_dir():
        return
    books = [one for one in examples.iterdir() if one.is_dir()]
    if not books:
        fail(problems, "examples/ is empty: one real book traced end to end is the point")
        return
    for book in books:
        if not (book / "manifest.json").exists() or not (book / "lineage.json").exists():
            fail(
                problems,
                f"examples/{book.name}/: needs both manifest.json and lineage.json so the "
                "book can be traced from source to output",
            )


def main(argv: list[str]) -> int:
    if len(argv) != 2:
        print(__doc__)
        return 2
    root = Path(argv[1])
    if not root.is_dir():
        print(f"Not a directory: {root}")
        return 2

    problems: list[str] = []
    structure_ok = check_structure(root, problems)
    records = load_manifest(root, problems)
    check_required_fields(root, records, problems)
    check_identifiers_resolve(records, problems)
    check_conventions(records, problems)
    check_credentials(records, root, problems)
    check_an_example_is_complete(root, problems)

    if problems:
        print(f"{len(problems)} problem(s) in {root}:\n")
        for problem in problems:
            print(f"  - {problem}")
        print()
        if not structure_ok:
            print("The bundle is missing pieces listed above, so the checks below could not "
                  "say much. Fix those first and run this again.")
        return 1

    count = len(records)
    print(f"{root}: {count} record(s), structure complete, references resolve, "
          "no credentials found.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main(sys.argv))