511 lines
20 KiB
Python
511 lines
20 KiB
Python
#!/usr/bin/env python3
|
|
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Controller alignment reports (spec 21 §6.6) - read-only, never guesses.
|
|
|
|
Emits the envelope the controller diffs mechanically:
|
|
|
|
{"report": "alignment", "schema_version": 1, "phase": <n>, "from": "umbraaudio-m4",
|
|
"node_contract_version": 1, "generated_at": ..., "verdict": pass|fail|blocked,
|
|
"evidence": {...per-phase...}, "findings": [...]}
|
|
|
|
Two rules the controller stated and this tool enforces on itself:
|
|
|
|
* A phase with no evidence is ``blocked``, never ``pass``.
|
|
* A field that cannot be produced is ``null`` **plus a finding**, never omitted.
|
|
|
|
Phases that need the other side are emitted as ``blocked`` with the reason, so the absence
|
|
is visible instead of silent.
|
|
|
|
Usage:
|
|
python scripts/alignment_report.py --phase 1
|
|
python scripts/alignment_report.py --phase 2
|
|
python scripts/alignment_report.py --phase 3 --project "Last Dwarf"
|
|
python scripts/alignment_report.py --phase 3-set --project "Last Dwarf"
|
|
python scripts/alignment_report.py --phase all --out-dir /tmp/alignment
|
|
python scripts/alignment_report.py --phase all --stdout-only
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import sqlite3
|
|
import sys
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
from umbraaudio.config import list_projects
|
|
from umbraaudio.definitions import DEFINITION_SET_VERSION, declaration, vocabularies
|
|
from umbraaudio.source_units import key_report
|
|
|
|
SCHEMA_VERSION = 1
|
|
NODE_ID = "umbraaudio-m4"
|
|
NODE_CONTRACT_VERSION = 1
|
|
|
|
# Controller §6.6.1: we choose the probe set deterministically, they send us those units.
|
|
PROBE_STRIDE = 17
|
|
PROBE_CAP = 200
|
|
|
|
|
|
def _now() -> str:
|
|
return datetime.now(UTC).isoformat(timespec="seconds")
|
|
|
|
|
|
def _finding(fid: str, severity: str, description: str, affects: list[str]) -> dict[str, Any]:
|
|
return {"id": fid, "severity": severity, "description": description, "affects": affects}
|
|
|
|
|
|
def _envelope(phase: Any, verdict: str, evidence: dict[str, Any], findings: list) -> dict[str, Any]:
|
|
return {
|
|
"report": "alignment",
|
|
"schema_version": SCHEMA_VERSION,
|
|
"phase": phase,
|
|
"from": NODE_ID,
|
|
"node_contract_version": NODE_CONTRACT_VERSION,
|
|
"generated_at": _now(),
|
|
"verdict": verdict,
|
|
"evidence": evidence,
|
|
"findings": findings,
|
|
}
|
|
|
|
|
|
def _ro(project_dir: str) -> sqlite3.Connection:
|
|
db = Path(project_dir) / ".umbra" / "project.db"
|
|
return sqlite3.connect(f"file:{db.resolve()}?mode=ro", uri=True)
|
|
|
|
|
|
def _alias_map(project_dir: str) -> tuple[dict[str, str], list[dict[str, Any]]]:
|
|
path = Path(project_dir) / ".umbra" / "speakers.json"
|
|
if not path.exists():
|
|
return {}, []
|
|
profiles = json.loads(path.read_text(encoding="utf-8")).get("speakers", [])
|
|
alias: dict[str, str] = {}
|
|
for s in profiles:
|
|
primary = s.get("primary_name", "")
|
|
if primary:
|
|
alias[primary.casefold()] = primary
|
|
for a in s.get("aliases") or []:
|
|
if a:
|
|
alias[a.casefold()] = primary
|
|
return alias, profiles
|
|
|
|
|
|
# --------------------------------------------------------------------------- phase 1
|
|
def phase_1() -> dict[str, Any]:
|
|
"""Vocabulary diff. Both sides emit the same shape and are compared mechanically."""
|
|
evidence = {"vocabularies": vocabularies(), "definition_set_version": DEFINITION_SET_VERSION}
|
|
findings: list[dict[str, Any]] = []
|
|
findings.append(
|
|
_finding(
|
|
"V-01",
|
|
"minor",
|
|
"Values marked 'extracted' in provenance() are read out of executable code and "
|
|
"can change without a deployment; values marked 'declared' are fixed by a spec "
|
|
"or by the controller. Compare provenance before treating a diff as a conflict.",
|
|
["definition_set_version"],
|
|
)
|
|
)
|
|
findings.append(
|
|
_finding(
|
|
"V-02",
|
|
"minor",
|
|
"'speaker' is deliberately NOT published as a character kind. Controller §1.2: a "
|
|
"speaker is a character that has lines in a work - a derived fact, not a kind.",
|
|
["character_kind"],
|
|
)
|
|
)
|
|
verdict = "pass" if evidence["vocabularies"] else "blocked"
|
|
return _envelope(1, verdict, evidence, findings)
|
|
|
|
|
|
# --------------------------------------------------------------------------- phase 2
|
|
def _project_phase_2(project_dir: str) -> dict[str, Any]:
|
|
con = _ro(project_dir)
|
|
try:
|
|
alias, profiles = _alias_map(project_dir)
|
|
chapters = con.execute("SELECT COUNT(*) FROM chapters").fetchone()[0]
|
|
utterances = con.execute("SELECT COUNT(*) FROM utterances").fetchone()[0]
|
|
dialogue = con.execute(
|
|
"SELECT COUNT(*) FROM utterances WHERE lower(type)='dialogue'"
|
|
).fetchone()[0]
|
|
names = [
|
|
r[0]
|
|
for r in con.execute(
|
|
"SELECT DISTINCT speaker_name FROM utterances "
|
|
"WHERE lower(type)='dialogue' AND speaker_name IS NOT NULL"
|
|
)
|
|
]
|
|
canonical = {alias.get(n.casefold(), n) for n in names}
|
|
unresolved = sorted({n for n in names if n.casefold() not in alias})
|
|
narrator = [n for n in names if n.casefold() == "narrator"]
|
|
# Controller §1.2: a speaker is a CHARACTER that has lines. Every canonical name
|
|
# here has lines by construction, so the interesting number is the reverse:
|
|
# characters in the profile file that never speak.
|
|
narrator_chars = {alias.get("narrator", "narrator")}
|
|
characters_with_lines = canonical - narrator_chars
|
|
characters_without_lines = len(profiles) - len(characters_with_lines)
|
|
|
|
data_dir = Path(project_dir) / ".umbra"
|
|
vp = data_dir / "voice_profiles.json"
|
|
voice_profiles = 0
|
|
if vp.exists():
|
|
raw = json.loads(vp.read_text(encoding="utf-8"))
|
|
voice_profiles = len(raw.get("speakers", raw) if isinstance(raw, dict) else raw)
|
|
refs = sum(1 for _ in (data_dir / "voices").rglob("*_ref.wav")) if (
|
|
data_dir / "voices"
|
|
).is_dir() else 0
|
|
vmatrix = sum(1 for _ in (data_dir / "voice_prompts").rglob("*.pt")) if (
|
|
data_dir / "voice_prompts"
|
|
).is_dir() else 0
|
|
|
|
return {
|
|
"project": Path(project_dir).name,
|
|
"speaker_names": len(names),
|
|
"characters": len(canonical),
|
|
"speakers": len(characters_with_lines),
|
|
"characters_without_lines": characters_without_lines,
|
|
"utterances": utterances,
|
|
"dialogue_utterances": dialogue,
|
|
"chapters": chapters,
|
|
"store_counts": {
|
|
"speakers_json": len(profiles),
|
|
"voice_profiles_json": voice_profiles,
|
|
"voice_ref_wavs": refs,
|
|
"vmatrix_pt": vmatrix,
|
|
},
|
|
"unresolved_speaker_names": unresolved,
|
|
"narrator_speaker_names": narrator,
|
|
"source_unit_keys": key_report(con),
|
|
}
|
|
finally:
|
|
con.close()
|
|
|
|
|
|
def phase_2() -> dict[str, Any]:
|
|
per_project = [
|
|
_project_phase_2(str(Path(p["slug"])))
|
|
for p in list_projects()
|
|
if (Path(p["slug"]) / ".umbra" / "project.db").exists()
|
|
]
|
|
evidence = {"per_project": per_project}
|
|
findings: list[dict[str, Any]] = []
|
|
|
|
# The three-way disagreement is a KNOWN finding (manifest.md §15.2). Report it, do
|
|
# not resolve it, and do not let it masquerade as a pass.
|
|
for row in per_project:
|
|
stores = row["store_counts"]
|
|
counts = {stores["speakers_json"], stores["voice_profiles_json"], stores["voice_ref_wavs"]}
|
|
if len(counts) > 1:
|
|
findings.append(
|
|
_finding(
|
|
f"C-{row['project']}",
|
|
"major",
|
|
f"Speaker stores disagree for {row['project']}: speakers.json="
|
|
f"{stores['speakers_json']}, voice_profiles.json="
|
|
f"{stores['voice_profiles_json']}, ref WAVs={stores['voice_ref_wavs']}. "
|
|
"Known finding (manifest.md §15.2), not resolved by this node.",
|
|
["characters", "speakers"],
|
|
)
|
|
)
|
|
if row["unresolved_speaker_names"]:
|
|
findings.append(
|
|
_finding(
|
|
f"U-{row['project']}",
|
|
"major",
|
|
f"{len(row['unresolved_speaker_names'])} distinct speaker_name value(s) in "
|
|
f"{row['project']} resolve to no profile: "
|
|
f"{row['unresolved_speaker_names']}. Recorded, never discarded.",
|
|
["speaker_identity"],
|
|
)
|
|
)
|
|
if row["narrator_speaker_names"]:
|
|
findings.append(
|
|
_finding(
|
|
f"N-{row['project']}",
|
|
"minor",
|
|
f"{row['project']} still has dialogue attributed to 'narrator'.",
|
|
["speaker_identity"],
|
|
)
|
|
)
|
|
|
|
# Phase 2's job is to RECONCILE cardinality. Our own stores do not agree, so the
|
|
# check does not hold: that is a fail with evidence, not a pass. A pass here would be
|
|
# exactly the "check that passed because its mechanism did nothing" the controller
|
|
# warns about (spec §6).
|
|
if not per_project:
|
|
return _envelope(2, "blocked", evidence, findings)
|
|
verdict = "fail" if findings else "pass"
|
|
return _envelope(2, verdict, evidence, findings)
|
|
|
|
|
|
# --------------------------------------------------------------------------- phase 3
|
|
def _probe_keys(project_dir: str, stride: int, cap: int) -> tuple[list[str], int]:
|
|
con = _ro(project_dir)
|
|
try:
|
|
# A project whose schema predates the key must report blocked, not crash.
|
|
cols = {r[1] for r in con.execute("PRAGMA table_info(utterances)")}
|
|
if "source_unit_key" not in cols:
|
|
return [], 0
|
|
rows = con.execute(
|
|
"SELECT source_unit_key FROM utterances "
|
|
"WHERE source_unit_key IS NOT NULL ORDER BY source_unit_key ASC"
|
|
).fetchall()
|
|
finally:
|
|
con.close()
|
|
keys = [r[0] for r in rows]
|
|
return keys[::stride][:cap], len(keys)
|
|
|
|
|
|
def _resolve(project_dir: str, keys: list[str]) -> list[dict[str, Any]]:
|
|
con = _ro(project_dir)
|
|
alias, _ = _alias_map(project_dir)
|
|
out: list[dict[str, Any]] = []
|
|
try:
|
|
for key in keys:
|
|
row = con.execute(
|
|
"SELECT speaker_name, type, chapter_id, utterance_index, text "
|
|
"FROM utterances WHERE source_unit_key = ?",
|
|
(key,),
|
|
).fetchone()
|
|
if row is None:
|
|
out.append(
|
|
{"unit_key": key, "resolved": False, "character": None,
|
|
"speaker": None, "lang": "en", "used_index": False}
|
|
)
|
|
continue
|
|
speaker, utype, _chapter, _index, _text = row
|
|
character = alias.get((speaker or "").casefold(), speaker)
|
|
out.append(
|
|
{
|
|
"unit_key": key,
|
|
"resolved": True,
|
|
"character": character,
|
|
"speaker": character if (utype or "") == "dialogue" else None,
|
|
"lang": "en",
|
|
# Resolved by primary-key-equivalent lookup on the content key.
|
|
# utterance_index was never consulted.
|
|
"used_index": False,
|
|
}
|
|
)
|
|
finally:
|
|
con.close()
|
|
return out
|
|
|
|
|
|
def phase_3(project: str) -> dict[str, Any]:
|
|
project_dir = _resolve_project_dir(project)
|
|
keys, total = _probe_keys(project_dir, PROBE_STRIDE, PROBE_CAP)
|
|
resolutions = _resolve(project_dir, keys)
|
|
evidence = {
|
|
"resolutions": resolutions,
|
|
"selection": {
|
|
"project": Path(project_dir).name,
|
|
"total_keys": total,
|
|
"stride": PROBE_STRIDE,
|
|
"cap": PROBE_CAP,
|
|
"selected": len(keys),
|
|
"rule": "sort source_unit_key ASC (byte order), take every 17th from index 0, cap 200",
|
|
},
|
|
}
|
|
findings: list[dict[str, Any]] = []
|
|
used_index = [r for r in resolutions if r["used_index"]]
|
|
if used_index:
|
|
findings.append(
|
|
_finding(
|
|
"P3-01",
|
|
"blocker",
|
|
f"{len(used_index)} unit(s) could only be resolved using utterance_index.",
|
|
["source_unit_key"],
|
|
)
|
|
)
|
|
if not keys:
|
|
findings.append(
|
|
_finding(
|
|
"P3-02",
|
|
"blocker",
|
|
"No source_unit_key values present. Run "
|
|
"scripts/backfill_source_unit_keys.py --write before the identity probe.",
|
|
["source_unit_key"],
|
|
)
|
|
)
|
|
return _envelope(3, "blocked", evidence, findings)
|
|
|
|
verdict = "fail" if used_index else "pass"
|
|
return _envelope(3, verdict, evidence, findings)
|
|
|
|
|
|
def phase_3_set(project: str) -> dict[str, Any]:
|
|
"""The probe set itself, so the controller can send back exactly these units."""
|
|
project_dir = _resolve_project_dir(project)
|
|
keys, total = _probe_keys(project_dir, PROBE_STRIDE, PROBE_CAP)
|
|
evidence = {
|
|
"selection": {
|
|
"project": Path(project_dir).name,
|
|
"total_keys": total,
|
|
"stride": PROBE_STRIDE,
|
|
"cap": PROBE_CAP,
|
|
"selected": len(keys),
|
|
"rule": "sort source_unit_key ASC (byte order), take every 17th from index 0, cap 200",
|
|
"selection_sha256": hashlib.sha256("\n".join(keys).encode()).hexdigest(),
|
|
"pre_resolved": len(_resolve(project_dir, keys)),
|
|
"pre_resolved_using_index": 0,
|
|
},
|
|
"unit_keys": keys,
|
|
}
|
|
findings: list[dict[str, Any]] = []
|
|
if not keys:
|
|
findings.append(
|
|
_finding(
|
|
"P3S-01",
|
|
"blocker",
|
|
"Cannot select a probe set: no source_unit_key values. Backfill first.",
|
|
["source_unit_key"],
|
|
)
|
|
)
|
|
return _envelope("3-set", "blocked" if not keys else "pass", evidence, findings)
|
|
|
|
|
|
# --------------------------------------------------------------------- phases 4-6c, 7
|
|
def phase_needs_partner(phase: Any, what: str, why: str) -> dict[str, Any]:
|
|
findings = [_finding(f"X-{phase}", "major", why, [what])]
|
|
return _envelope(phase, "blocked", {}, findings)
|
|
|
|
|
|
def phase_6b() -> dict[str, Any]:
|
|
"""Persistence and partial-capture. This node has neither, and must say so."""
|
|
evidence = {"characters": [], "supported": False}
|
|
findings = [
|
|
_finding(
|
|
"B-01",
|
|
"blocker",
|
|
"characters.persistence (book_local | series_persistent) is NOT captured anywhere "
|
|
"in this product. Controller D11 requires it to be DECLARED, never inferred. "
|
|
"speakers.json has no such field, so every character reports null persistence.",
|
|
["character_persistence"],
|
|
),
|
|
_finding(
|
|
"B-02",
|
|
"major",
|
|
"Absence is currently NULL, not typed (controller D13). For example "
|
|
"speakers.json demographics.gender is often null, which is ambiguous between "
|
|
"not_attempted / not_applicable / in_progress / superseded.",
|
|
["attribute_absence_reason"],
|
|
),
|
|
_finding(
|
|
"B-03",
|
|
"major",
|
|
"No personality artifact exists (controller D4/D7). What this product has today is "
|
|
"speakers.json vocal_traits / communication_style / description: prose written by "
|
|
"LLM character discovery, with NO version, NO model record and NO prompt version. "
|
|
"It cannot yet serve as a versioned personality artifact.",
|
|
["character_profile"],
|
|
),
|
|
]
|
|
return _envelope("6b", "blocked", evidence, findings)
|
|
|
|
|
|
def phase_6c() -> dict[str, Any]:
|
|
"""Editions. This product has no edition concept."""
|
|
evidence = {"editions": [], "supported": False}
|
|
findings = [
|
|
_finding(
|
|
"E-01",
|
|
"major",
|
|
"No editions table or concept. projects.status='released' is the only release "
|
|
"signal and it is per-work, not per-edition (controller D14/D15). The published "
|
|
"DealHunter export cannot be expressed as edition 1 + its artifact set.",
|
|
["editions"],
|
|
)
|
|
]
|
|
return _envelope("6c", "blocked", evidence, findings)
|
|
|
|
|
|
# ----------------------------------------------------------------------------- driver
|
|
def _resolve_project_dir(project: str) -> str:
|
|
slugs = {p["slug"] for p in list_projects()}
|
|
if project in slugs:
|
|
return project
|
|
for slug in slugs:
|
|
if slug.casefold() == project.casefold():
|
|
return slug
|
|
raise SystemExit(f"Unknown project {project!r}. Known: {sorted(slugs)}")
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument(
|
|
"--phase",
|
|
required=True,
|
|
help="1 | 2 | 3 | 3-set | 4 | 5 | 6 | 6b | 6c | 7 | all",
|
|
)
|
|
parser.add_argument("--project", default="Last Dwarf", help="project for phases 3/3-set")
|
|
parser.add_argument("--out-dir", default=None, help="write one JSON file per phase")
|
|
parser.add_argument(
|
|
"--write-definitions",
|
|
metavar="PATH",
|
|
default=None,
|
|
help="also write the published definition set to PATH",
|
|
)
|
|
args = parser.parse_args()
|
|
|
|
phases = {
|
|
"1": lambda: phase_1(),
|
|
"2": phase_2,
|
|
"3": lambda: phase_3(args.project),
|
|
"3-set": lambda: phase_3_set(args.project),
|
|
"4": lambda: phase_needs_partner(
|
|
4, "artifact_lineage", "Needs the controller's chosen artifact; this node must "
|
|
"state its chain for the same one. Cannot start unilaterally."
|
|
),
|
|
"5": lambda: phase_needs_partner(
|
|
5, "character_attribute_states", "No time-scoped attribute model exists in this "
|
|
"product yet; requires the timeline design to be implemented locally first."
|
|
),
|
|
"6": lambda: phase_needs_partner(
|
|
6, "voice_matrix_values", "A voice matrix is not captured as structured values; "
|
|
"only prose profiles and a name-keyed vmatrix exist. Needs voice.capture_character."
|
|
),
|
|
"6b": phase_6b,
|
|
"6c": phase_6c,
|
|
"7": lambda: phase_needs_partner(
|
|
7, "round_trip", "Round trip needs the ingest API, which does not exist yet "
|
|
"(controller §4 item 4). Deliberately not attempted."
|
|
),
|
|
}
|
|
|
|
selected = list(phases) if args.phase == "all" else [args.phase]
|
|
out_dir = Path(args.out_dir) if args.out_dir else None
|
|
if out_dir:
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
if args.write_definitions:
|
|
defs_path = Path(args.write_definitions)
|
|
defs_path.parent.mkdir(parents=True, exist_ok=True)
|
|
defs_path.write_text(
|
|
json.dumps(declaration(), indent=2, sort_keys=True) + "\n", encoding="utf-8"
|
|
)
|
|
print(f"wrote definition set -> {defs_path}")
|
|
|
|
for name in selected:
|
|
if name not in phases:
|
|
raise SystemExit(f"Unknown phase {name!r}. Choose from {sorted(phases)} or 'all'.")
|
|
report = phases[name]()
|
|
filename = f"alignment_phase_{str(name).replace('-', '_')}.json"
|
|
if out_dir:
|
|
(out_dir / filename).write_text(
|
|
json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8"
|
|
)
|
|
print(f"phase {name}: {report['verdict']} -> {out_dir / filename}")
|
|
else:
|
|
print(json.dumps(report, indent=2, sort_keys=True))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|