umbragit/UmbraAudio
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add the publish flag: pinned release sets, immutability, the Deal Hunter lesson

Browse Source 
The operator clarified the release model: the system keeps current and previous
state, there must be an explicit publish flag the user sets, there are approvals
for a speaker voice but none for the audiobook or chapters, the chapters that
make the m4b are the versions to flag, and a fix found after assembly causes a
change rather than editing what shipped.

Deal Hunter is the precedent, and it is measurable rather than theoretical.
Against the real package (built_at 2026-09-09T11:21:11Z), chapter 6:

  ch6_full.wav                    2227.26 s @ 24000 Hz   what a reviewer listens to
  Deal_Hunter_01_Salvage.mp3      2231.26 s @ 44100 Hz   what a listener received
                                  +4.00 s = 2 s lead + 2 s trail

plus a resample and a -20.5 dB RMS normalization. Found on 32 of 32 chapters.
The approved artifact and the published artifact were different files, and
nothing in the system recorded either fact. ar_package.json is rewritten by
every rebuild, so it is a build status, not a release record.

umbraaudio/release.py
  DeliveryProfile.from_ar_package reads the profile the exporter actually used,
  so the record follows the code instead of a declaration that can rot.
  build_publication is read-only and pins every chapter wav, every mp3 and the
  package by full SHA-256, plus a per-chapter source_unit_key_digest so a release
  pins which source revision produced it, not merely which bytes.
  verify_publication returns ok / drift / warnings and never silently accepts a
  changed artifact. write_publication is atomic and refuses to overwrite an
  existing publication id - immutability enforced by the writer, not by
  convention. A later fix is edition n+1 with supersedes pointing at n.

The rule that prevents a repeat: you cannot approve a chapter and then pad it.
An approval must name the artifact a listener receives. Every record states
chapter_approved: false and audiobook_approved: false in its own words, because
that is what would have caught Deal Hunter.

The record also refuses to overstate approval. voice_status.json is an approval
LEDGER - it only holds speakers someone explicitly touched - so it has 24 entries
against 61 speakers in the profile. Reporting "24 approved" as if it meant "all
approved" would claim an approval nobody gave, so the counts are separate and the
38-speaker gap is named.

scripts/publish_release.py: dry run by default; --publish, --verify, --list,
--edition/--supersedes. Exits 2 while blockers exist.

Also found, reported and NOT fixed (specs/22 section 7 gap 1): export_m4b's
_stitch_chapter hard-codes 500 ms/0 ms silence while ChapterBuilder uses
AudioStitcher defaults and AudioConfig holds its own values, so the m4b path can
stitch a chapter with different silence than the approved one. Until that is
closed an m4b release record would pin the wrong bytes, so it is better not to
offer one yet.

And projects.status is never advanced to 'released' in any of the four live
projects - all sit at parsed/analyzed - so the publication record is currently the
only release signal in the system.

tests/test_release.py: 24 tests. Full suite 512 passed / 8 failed, the same 8
pre-existing failures. ruff clean. No export, TTS or project data was modified.
This commit is contained in:
B.D. Murphy
2026-10-03 06:20:22 -07:00
parent 829a03b89f
commit ed3e249d0b
5 changed files with 1256 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

52
AGENTS.md
View File

@@ -171,6 +171,58 @@ Each pass has quality gates. Quality is the primary concern; generation speed is
## Current Session Summary
### Session: The publish flag, pinned release sets, and the Deal Hunter lesson (spec 22)
- **Trigger**: operator clarification of the release model — the system stores current +
previous state like git, there must be an **explicit publish flag** the user sets, there
are approvals for a speaker voice but **none for the audiobook or chapters**, **the
chapters that make the m4b are the versions to flag**, and a fix found after assembly
**causes a change** rather than editing what shipped. Deal Hunter is the precedent:
2 s of dead sound was added to the start and end of every chapter *after* approval.
- **The precedent, measured on the real package** (`ar_package.json`,
`built_at 2026-09-09T11:21:11Z`), chapter 6:
`ch6_full.wav` **2227.26 s @ 24 kHz** (what a reviewer listens to) vs
`Deal_Hunter_01_Salvage.mp3` **2231.26 s @ 44.1 kHz** (what a listener received) =
**+4.00 s = 2 s lead + 2 s trail**, plus resample and −20.5 dB RMS normalization.
Detected on **32 of 32 chapters**. The approved artifact and the published artifact were
different files and the system recorded neither.
- **New `umbraaudio/release.py`**: `DeliveryProfile.from_ar_package` (reads the profile the
exporter *actually used*, measured not declared), `build_publication` (read-only; pins
every chapter wav + mp3 + package by full SHA-256, the per-chapter
`source_unit_key_digest`, the approvals, and the delivery profile),
`verify_publication` (→ `ok` / `drift` / `warnings`), `write_publication` (atomic
tmp+`os.replace`, **refuses to overwrite an existing publication id**),
`list_publications`. Immutability is enforced by the writer, not by convention.
- **New `scripts/publish_release.py`**: dry-run by default; `--publish`, `--verify`,
`--list`, `--edition N --supersedes <id>`. Exits 2 while `blockers` is non-empty.
Against the real DealHunter package it reports: publishable, 32 chapters pinned,
44100 Hz + 2 s lead silence, and **24 of 61 speakers have an approval record, 38 do not**
(`voice_status.json` is an approval *ledger* — only speakers someone explicitly touched —
so reporting "24 approved" as if it meant "all approved" would claim an approval that was
never given; the two counts are kept separate and the gap is named).
- **Rule that prevents a repeat: you cannot approve a chapter and then pad it.** An approval
must name the artifact a listener receives; every transform applied after chapter approval
is recorded in the release as part of it. R-6 records `chapter_approved: false` and
`audiobook_approved: false` explicitly in every record, because R-6 is what would have
caught Deal Hunter.
- **Second, still-latent instance found** (`specs/22` §7 gap 1, reported not fixed):
`export_m4b._stitch_chapter` hard-codes `silence_between_utterances_ms=500,
silence_between_chapters_ms=0` while `ChapterBuilder` uses `AudioStitcher()` defaults
(500/2000/4000) and `AudioConfig` holds its own values — so the m4b path can stitch a
chapter with different silence than the approved one. **This is the last real hole in the
pinned-set invariant, and until it is closed an m4b release record would pin the wrong
bytes.** Proposed fix: make `export_m4b` consume `ch{N}_full.wav` instead of re-stitching
from utterances.
- **`projects.status` is never advanced to `released`** in any of the four live projects
(all sit at `parsed`/`analyzed`). The publication record is currently the only release
signal in the system.
- Tests: `tests/test_release.py` — 24 tests (delivery profile, pinning, approval-gap
honesty, post-approval padding/resample detection with an epsilon for rounding, quality
gate surfacing, blockers, immutability + successor with `supersedes`, atomic write,
verification drift for changed/missing chapter, changed mp3, standing silence caveat).
Full suite **512 passed / 8 failed** — the same 8 pre-existing/parallel failures. ruff clean.
- No export, TTS, or project data was modified. `publish_release.py` is read-only unless
given `--publish`.
### Session: Umbra system node declaration + cross-machine contract (spec 21)
- **Context:** another machine/tab is moving UmbraTrans to the main Umbra system on
`oldbox` (192.168.123.14). Four questions were raised for this machine: (1) align the

177
scripts/publish_release.py Normal file
View File

@@ -0,0 +1,177 @@
#!/usr/bin/env python3
"""Publish a release: pin what shipped, then prove it (spec 22).
An approval exists for a speaker voice. There is no approval for a chapter and none for the
audiobook, and Deal Hunter already paid for that gap: 2 s of dead sound was added to the
start and end of every chapter *after* approval, so the approved artifact and the published
artifact were different files.
This script makes the release explicit.
# What would ship, and what is missing? Read-only. Always do this first.
python scripts/publish_release.py --project "DealHunterShip_4.0e_whole-to-audio"
# Write the immutable publication record (refuses to overwrite an existing one).
python scripts/publish_release.py --project "..." --publish --by worstauthor
# Re-verify a published record against the filesystem.
python scripts/publish_release.py --project "..." --verify pub-001-2026-09-09
# List published editions.
python scripts/publish_release.py --project "..." --list
A change after publication is a NEW edition (``--edition 2 --supersedes pub-...``), never an
edit of a released record.
"""
from __future__ import annotations
import argparse
import json
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
from umbraaudio.config import load_config
from umbraaudio.release import (
build_publication,
list_publications,
publication_path,
verify_publication,
write_publication,
)
def _git_commit() -> str | None:
import subprocess
try:
out = subprocess.run(
["git", "rev-parse", "--short", "HEAD"],
capture_output=True,
text=True,
timeout=10,
check=False,
)
return out.stdout.strip() or None
except (OSError, subprocess.SubprocessError):
return None
def _print_summary(record: dict) -> None:
profile = record["delivery_profile"]
print(f"publication_id {record['publication_id']}")
print(f"work / edition {record['work']} #{record['edition_number']}"
+ (f" (supersedes {record['supersedes']})" if record.get("supersedes") else ""))
print(f"chapters pinned {len(record['chapters'])}")
pkg = record["package"]
print(f"package {pkg['object_key']} {pkg['bytes']} bytes sha256="
f"{(pkg['sha256'] or '')[:16]}…")
print("delivery profile "
f"{profile['sample_rate']} Hz / {profile['channels']} ch"
+ (f" / {profile['bit_rate_kbps']} kbps" if profile.get("bit_rate_kbps") else "")
+ f" silence +{profile['lead_silence_s']}s lead, +{profile['trail_silence_s']}s trail"
f" rms {profile['target_rms_db']} dB")
approvals = record["approvals"]
print(f"approvals {approvals['voices_approved']} of "
f"{approvals['voices_in_profile']} speakers have an approval record; "
f"{approvals['voices_not_approved']} do not")
print(" chapter approved: no | audiobook approved: no "
"-> this record IS the release approval")
warned = [c for c in record["chapters"] if c["warnings"]]
print(f"\npost-approval deltas detected on {len(warned)}/{len(record['chapters'])} chapters")
for chapter in warned[:5]:
for warning in chapter["warnings"]:
print(f" ch{chapter['chapter_id']:>3} {warning}")
if len(warned) > 5:
print(f" ... and {len(warned) - 5} more")
if record["blockers"]:
print(f"\nBLOCKERS ({len(record['blockers'])}) - not publishable:")
for blocker in record["blockers"]:
print(f" [{blocker['id']}] {blocker['why']}")
else:
print("\nno blockers: this set is publishable")
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--project", default=None, help="project name or path")
parser.add_argument("--publish", action="store_true", help="write the immutable record")
parser.add_argument("--verify", metavar="PUB_ID", default=None, help="verify a record")
parser.add_argument("--list", action="store_true", help="list published editions")
parser.add_argument("--edition", type=int, default=1)
parser.add_argument("--supersedes", default=None, help="publication_id this replaces")
parser.add_argument("--by", default="operator", help="who is publishing")
parser.add_argument("--json", action="store_true")
args = parser.parse_args()
cfg = load_config(args.project)
data_dir = cfg.data_dir
work = cfg.project_dir.name
if args.list:
records = list_publications(data_dir)
if args.json:
print(json.dumps(records, indent=2))
elif not records:
print(f"no publications recorded for {work}")
else:
print(f"publications for {work}")
for r in records:
print(f" {r['publication_id']} edition {r['edition_number']} "
f"{r['published_at']} {r['chapters']} chapters"
+ (f" supersedes {r['supersedes']}" if r["supersedes"] else ""))
return 0
if args.verify:
path = publication_path(data_dir, args.verify)
if not path.exists():
print(f"no such publication: {path}", file=sys.stderr)
return 1
result = verify_publication(data_dir, json.loads(path.read_text()))
if args.json:
print(json.dumps(result, indent=2))
else:
print(f"verify {result['publication_id']}")
print(f" ok: {result['ok']}")
for d in result["drift"]:
print(f" DRIFT [{d['id']}] {d['why']}")
for w in result["warnings"][:10]:
print(f" note: {w}")
if len(result["warnings"]) > 10:
print(f" ... and {len(result['warnings']) - 10} more notes")
return 0 if result["ok"] else 2
record = build_publication(
data_dir,
work,
edition_number=args.edition,
published_by=args.by,
git_commit=_git_commit(),
supersedes=args.supersedes,
)
if not args.publish:
if args.json:
print(json.dumps(record, indent=2, sort_keys=True))
else:
_print_summary(record)
print("\nDRY RUN - nothing written. Re-run with --publish to record it.")
return 0 if record["publishable"] else 2
if not record["publishable"]:
print("refusing to publish: blockers present (see above)", file=sys.stderr)
_print_summary(record)
return 2
path = write_publication(data_dir, record)
print(f"published -> {path}")
print(f" this record is immutable. A later fix is edition "
f"{args.edition + 1} with --supersedes {record['publication_id']}.")
return 0
if __name__ == "__main__":
raise SystemExit(main())

199
specs/22-releases-and-the-publish-flag.md Normal file
View File

@@ -0,0 +1,199 @@
# Spec 22 — Releases: the Publish Flag, Pinned Artifact Sets, and Post-Publication Change
- **Status:** implemented on the UmbraAudio node (backend + CLI + tests). Awaiting
ratification by the controller.
- **Date:** 2026-10-03
- **Origin:** an operator clarification of the release model, prompted by Deal Hunter.
- **Companions:** `specs/21` (nodes, tasks, changes), `specs/20` §F (`editions`,
`edition_artifacts`, invariant I-H), controller decisions D9/D12/D14/D15.
---
## 1. The clarification, stated back
> The system, with time markers, acts as a git repository: it stores the current state and
> keeps the previous ones. There should be an explicit flag. The user tells the system *this
> is the version that was published*, making explicit all the items and approvals that make
> up the final state. There are approvals for items like a speaker voice, but **not** for the
> full audiobook and chapters. **The chapters that make an m4b are the versions to flag.**
> After assembly into final chapters and m4b, other issues may need fixing; the publishing
> requirements must then be met, and that **causes a change**. Deal Hunter experienced this
> when 2 seconds of dead sound had to be added to the beginning and end of each chapter,
> after approval.
Five consequences, each implemented:
1. **A release is a pinned set, not a build directory.** Every chapter artifact and the
package carry a full SHA-256 in the release record.
2. **The publish flag is the release record.** Writing it *is* the approval — because no
other artifact-level audiobook approval exists.
3. **The pinned set is the chapters that make the m4b**, not the loose utterances and not the
pre-encode intermediate.
4. **Everything applied after a chapter is approved is part of the release, and is recorded
in it.** Delivery profile: sample rate, channels, bitrate, CBR, leading/trailing silence,
loudness target, gain limit, peak ceiling.
5. **A change after publication is a successor, never an edit.** A published record is
immutable; a fix produces edition *n+1* with `supersedes` pointing at *n*.
## 2. What was already wrong, measured
Not hypothetical. On the published Deal Hunter package (`ar_package.json`,
`built_at 2026-09-09T11:21:11Z`), chapter 6:
| artifact | duration | rate | what it is |
|---|---|---|---|
| `chapter_audio/ch6_full.wav` | **2227.26 s** | 24 000 Hz | what a reviewer listens to |
| `export/ar_package/Deal_Hunter_01_Salvage.mp3` | **2231.26 s** | 44 100 Hz | what a listener received |
**+4.00 s = 2 s leading + 2 s trailing digital silence**, plus a resample and a loudness
normalization to −20.5 dB RMS. Applied to **32 of 32 chapters**.
The approved artifact and the published artifact were different files, and the system
recorded neither fact. `ar_package.json` is rewritten by every rebuild, so it is a build
status — not a release record.
A second, independent instance of the same class, still latent: `export_m4b._stitch_chapter`
hard-codes `silence_between_utterances_ms=500, silence_between_chapters_ms=0` while
`ChapterBuilder` uses `AudioStitcher()` defaults and `AudioConfig` holds its own values. The
m4b path can therefore stitch a chapter with different silence than the approved one. Reported,
not fixed — see §7.
## 3. The rule that actually prevents a repeat
> **You cannot approve a chapter and then pad it.**
An approval must name the artifact a listener receives. Every transform applied between
"chapter finished" and "bytes shipped" is part of the release and is recorded in it. Where a
transform necessarily changes the bytes — padding, resampling, loudness normalization — the
approval is taken on the *post-transform* artifact, or it is not an approval of what ships.
## 4. What is implemented
### 4.1 `umbraaudio/release.py`
| function | purpose |
|---|---|
| `build_publication(...)` | Captures the pinned set + approvals + delivery profile. **Read-only** — hashes files, writes nothing. |
| `DeliveryProfile.from_ar_package(...)` | Reads the profile the exporter *actually used* from its own manifest. Measured, not declared. |
| `verify_publication(...)` | Re-checks a record against the filesystem → `ok` / `drift` / `warnings`. |
| `write_publication(...)` | Atomic (tmp + `os.replace`), and **refuses to overwrite an existing publication id**. |
| `list_publications(...)` | Every published edition. |
A publication record:
```json
{
"publication_id": "pub-001-2026-09-09",
"work": "...", "edition_number": 1, "supersedes": null,
"published_at": "...", "published_by": "...", "git_commit": "...",
"delivery_profile": {
"sample_rate": 44100, "channels": 1, "bit_rate_kbps": 192, "cbr": true,
"lead_silence_s": 2.0, "trail_silence_s": 2.0,
"target_rms_db": -20.5, "max_gain_db": 12.0, "peak_ceiling_dbfs": -3.0
},
"package": { "object_key": "export/Deal_Hunter.m4b", "sha256": "...", "bytes": 325875309 },
"approvals": {
"voices_in_profile": 61, "voices_approved": 24, "voices_not_approved": 38,
"voice_status_digest": "...",
"chapter_approved": false, "audiobook_approved": false
},
"chapters": [
{ "chapter_id": 6, "seq": 6, "title": "1 Salvage",
"chapter_wav": { "object_key": "chapter_audio/ch6_full.wav",
"sha256": "...", "duration_s": 2227.26, "sample_rate": 24000 },
"mp3": { "object_key": "export/ar_package/Deal_Hunter_01_Salvage.mp3",
"sha256": "...", "duration_s": 2231.26,
"rms_db": -20.5, "peak_db": -3.0,
"loudness_ok": true, "peak_ok": true },
"source_unit_key_digest": "...", "utterances": 364,
"warnings": ["shipped track is 4.0s longer than the approved chapter wav ..."] }
],
"blockers": [], "publishable": true
}
```
Two fields carry most of the weight and are worth calling out:
- **`source_unit_key_digest`** — a digest of every `source_unit_key` in the chapter
(`specs/21` §10). A release therefore pins *which source revision* produced it, not just
which bytes. Audio and source are traceable to each other after the fact.
- **`voices_not_approved: 38`** — `voice_status.json` is an *approval ledger*: it only holds
speakers someone explicitly touched, so it has 24 entries against 61 speakers. Reporting
"24 approved" as if it meant "all approved" would claim an approval that was never given,
so the two counts are separate and the gap is named.
### 4.2 `scripts/publish_release.py`
```bash
python scripts/publish_release.py --project "DealHunterShip_4.0e_whole-to-audio" # dry run
python scripts/publish_release.py --project "..." --publish --by worstauthor # write
python scripts/publish_release.py --project "..." --verify pub-001-2026-09-09 # re-check
python scripts/publish_release.py --project "..." --list # editions
```
Dry run is the default and prints the pinned set, the delivery profile, the approval gap, and
**every post-approval delta it found**. `--publish` refuses to run while `blockers` is
non-empty. A second `--publish` against the same id raises `FileExistsError` — immutability
is enforced by the writer, not by convention.
## 5. Invariants
| ID | invariant | Enforced by |
|---|---|---|
| **R-1** | A published record is immutable; a later change is a successor edition with `supersedes` | `write_publication` refuses to overwrite |
| **R-2** | A release pins the chapters that make the package, each by SHA-256 | `build_publication`, `verify_publication` |
| **R-3** | No publication occurs while a blocker exists | `publish_release.py --publish` exits 2 |
| **R-4** | Any post-approval transform is recorded in the release | `delivery_profile` + per-chapter `warnings` |
| **R-5** | A changed artifact after publication is reported as drift, never silently accepted | `verify_publication` → `drift` |
| **R-6** | An approval is never inferred from the absence of a complaint | `approvals.chapter_approved = false` is recorded explicitly |
| **R-7** | The approval ledger's coverage is stated, not assumed | `voices_in_profile` vs `voices_approved` |
R-6 is the one that would have prevented Deal Hunter: the release says, in its own words,
that no chapter and no audiobook was approved.
## 6. Relationship to the controller's model
This is the local, immutable half of what the controller calls `editions` + `edition_artifacts`:
| local field | controller entity |
|---|---|
| `publication_id`, `edition_number` | `editions` |
| `supersedes` | `editions.supersedes_edition` |
| `chapters[].chapter_wav.sha256`, `mp3.sha256` | `edition_artifacts` (immutable member set, I-H) |
| `delivery_profile` | `derivation` on each artifact |
| `chapters[].source_unit_key_digest` | linkage back to `source_revisions` / `source_units` |
| `approvals` | the acceptance record behind a release decision |
| `published_at` / `published_by` | `released_at` / actor |
The record is shaped so ingest is a **copy**, not a transformation. Nothing the controller
would need is expressed in a local-only dialect.
## 7. Known gaps, stated rather than hidden
1. **The m4b path stitches differently from the approved chapter.**
`export_m4b._stitch_chapter` hard-codes 500 ms / 0 ms silence; `ChapterBuilder` uses
`AudioStitcher()` defaults (500/2000/4000) and `AudioConfig` carries its own values. Until
the m4b path consumes the *same* stitched chapter artifact the reviewer approved, a
chapter approval still does not cover the m4b. **This is the last real hole in R-2 and it
is not yet closed.** Proposed fix: make `export_m4b` consume `ch{N}_full.wav` directly
instead of re-stitching from utterances.
2. **No release-level approval in the web UI.** The flag exists as a CLI and a file. A
reviewer cannot yet press "publish" in the review interface, which is where they will
expect it.
3. **`projects.status` is never advanced to `released`.** All four live projects sit at
`parsed`/`analyzed`. The publication record is currently the only release signal.
4. **Not idempotent against a rebuild.** Re-running the exporter overwrites
`ar_package.json`; the publication record keeps the old hashes, so verification correctly
reports drift — but there is no automatic "this needs a new edition" workflow yet.
5. **Scope.** Only the AR MP3 package path is covered. A single-workbook `m4b` built by
`export_m4b` is not yet represented, for the reason in gap 1.
## 8. What to do next
- Close gap 1 first: make the m4b path consume the approved `ch{N}_full.wav`. Until then the
release record for an m4b would be pinning the wrong bytes, and it is better not to offer
one.
- Put the publish flag behind the review UI, gated on `verify_publication` returning `ok`.
- Set `projects.status = 'released'` from the same act that writes the record, so the two
cannot disagree.
- Send this shape to the controller as the ingest payload for `editions` / `edition_artifacts`.

293
tests/test_release.py Normal file
View File

@@ -0,0 +1,293 @@
"""Tests for the release/publish mechanism (spec 22).
The mechanism exists because Deal Hunter shipped audio that differed from what was
approved. These tests pin that lesson so it cannot regress silently.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
import numpy as np
import pytest
import soundfile as sf
from umbraaudio.release import (
DURATION_EPSILON_S,
DeliveryProfile,
build_publication,
file_sha256,
list_publications,
publication_path,
verify_publication,
write_publication,
)
def _wav(path: Path, seconds: float, rate: int = 24000) -> Path:
path.parent.mkdir(parents=True, exist_ok=True)
tone = np.sin(np.linspace(0, 440, int(rate * seconds), dtype="float32")) * 0.1
sf.write(str(path), tone, rate)
return path
def _project(tmp_path: Path, tracks: list[dict[str, Any]], fmt: dict | None = None) -> Path:
"""Build a minimal project tree with an exporter manifest."""
data_dir = tmp_path / ".umbra"
package_dir = data_dir / "export" / "ar_package"
package_dir.mkdir(parents=True, exist_ok=True)
for track in tracks:
source = track["source"]
chapter_id = int("".join(c for c in Path(source).stem if c.isdigit()))
_wav(data_dir / "chapter_audio" / source, track["chapter_seconds"])
mp3 = package_dir / track["file"]
size = int(track.get("size_bytes", 4096))
mp3.write_bytes(b"ID3" + b"x" * max(size - 3, 0))
track.setdefault("size_bytes", mp3.stat().st_size)
track.setdefault("chapter_id", chapter_id)
manifest = {
"base_name": "Book",
"built_at": "2026-09-09T11:21:11+00:00",
"format": fmt or {"codec": "mp3", "sample_rate": 44100, "channels": 1,
"bit_rate_kbps": 192, "cbr": True, "silence_s": 2.0},
"track_count": len(tracks),
"tracks": tracks,
}
(data_dir / "export" / "ar_package.json").write_text(json.dumps(manifest))
(data_dir / "speakers.json").write_text(
json.dumps({"speakers": [{"primary_name": "Hallr"}, {"primary_name": "Xylia"},
{"primary_name": "Unreviewed"}]})
)
(data_dir / "voice_status.json").write_text(json.dumps({"Hallr": "approved"}))
return data_dir
class TestDeliveryProfile:
def test_reads_the_profile_actually_used(self) -> None:
profile = DeliveryProfile.from_ar_package(
{"format": {"sample_rate": 44100, "channels": 1, "silence_s": 2.0}}
)
assert profile.sample_rate == 44100
assert profile.lead_silence_s == 2.0
assert profile.trail_silence_s == 2.0
def test_silence_is_symmetric_by_default(self) -> None:
"""The Deal Hunter fix was lead AND trail. Asymmetric padding must still be visible."""
profile = DeliveryProfile.from_ar_package(
{"format": {"sample_rate": 44100, "channels": 1, "silence_s": 1.5}}
)
assert profile.lead_silence_s == profile.trail_silence_s == 1.5
def test_missing_format_is_not_fatal(self) -> None:
profile = DeliveryProfile.from_ar_package({})
assert profile.sample_rate == 0
class TestBuildPublication:
def test_pins_chapters_and_package(self, tmp_path: Path) -> None:
data_dir = _project(tmp_path, [
{"seq": 1, "title": "One", "file": "01.mp3", "source": "ch1_full.wav",
"duration_s": 6.0, "chapter_seconds": 2.0, "rms_db": -20.5, "peak_db": -3.0,
"loudness_ok": True, "peak_ok": True},
])
record = build_publication(data_dir, "Book", now="2026-10-03T00:00:00+00:00")
assert record["publishable"] is True
assert len(record["chapters"]) == 1
chapter = record["chapters"][0]
assert chapter["chapter_wav"]["sha256"] == file_sha256(
data_dir / "chapter_audio" / "ch1_full.wav"
)
assert chapter["mp3"]["sha256"] == file_sha256(
data_dir / "export" / "ar_package" / "01.mp3"
)
assert record["package"]["track_count"] == 1
def test_approval_gap_is_reported_not_glossed(self, tmp_path: Path) -> None:
"""voice_status.json only holds speakers someone touched. Saying '3 approved'
would claim an approval that was never given."""
data_dir = _project(tmp_path, [
{"seq": 1, "title": "One", "file": "01.mp3", "source": "ch1_full.wav",
"duration_s": 2.0, "chapter_seconds": 2.0},
])
approvals = build_publication(data_dir, "Book")["approvals"]
assert approvals["voices_in_profile"] == 3
assert approvals["voices_approved"] == 1
assert approvals["voices_not_approved"] == 2
assert "Unreviewed" in approvals["voices_not_approved_names"]
def test_records_that_chapter_and_audiobook_are_unapproved(self, tmp_path: Path) -> None:
data_dir = _project(tmp_path, [
{"seq": 1, "title": "One", "file": "01.mp3", "source": "ch1_full.wav",
"duration_s": 2.0, "chapter_seconds": 2.0},
])
approvals = build_publication(data_dir, "Book")["approvals"]
assert approvals["chapter_approved"] is False
assert approvals["audiobook_approved"] is False
class TestPostApprovalDelta:
"""The Deal Hunter defect, made mechanical."""
def test_padding_after_approval_is_flagged(self, tmp_path: Path) -> None:
data_dir = _project(tmp_path, [
{"seq": 6, "title": "Salvage", "file": "06.mp3", "source": "ch6_full.wav",
# shipped is 4s longer: 2s lead + 2s trail
"duration_s": 6.0, "chapter_seconds": 2.0},
])
chapter = build_publication(data_dir, "Book")["chapters"][0]
assert any("4.0s longer" in w for w in chapter["warnings"])
def test_resample_after_approval_is_flagged(self, tmp_path: Path) -> None:
data_dir = _project(tmp_path, [
{"seq": 1, "title": "One", "file": "01.mp3", "source": "ch1_full.wav",
"duration_s": 2.0, "chapter_seconds": 2.0},
]) # chapter wav is 24000 Hz, profile says 44100
chapter = build_publication(data_dir, "Book")["chapters"][0]
assert any("24000 Hz" in w and "resampled" in w for w in chapter["warnings"])
def test_no_delta_means_no_warning(self, tmp_path: Path) -> None:
data_dir = _project(tmp_path, [
{"seq": 1, "title": "One", "file": "01.mp3", "source": "ch1_full.wav",
"duration_s": 2.0, "chapter_seconds": 2.0},
], fmt={"sample_rate": 24000, "channels": 1, "silence_s": 0.0})
chapter = build_publication(data_dir, "Book")["chapters"][0]
assert chapter["warnings"] == []
def test_epsilon_tolerates_rounding(self, tmp_path: Path) -> None:
data_dir = _project(tmp_path, [
{"seq": 1, "title": "One", "file": "01.mp3", "source": "ch1_full.wav",
"duration_s": 2.0 + DURATION_EPSILON_S / 2, "chapter_seconds": 2.0},
])
chapter = build_publication(data_dir, "Book")["chapters"][0]
assert not any("longer" in w for w in chapter["warnings"])
def test_failed_quality_gate_surfaces(self, tmp_path: Path) -> None:
data_dir = _project(tmp_path, [
{"seq": 1, "title": "One", "file": "01.mp3", "source": "ch1_full.wav",
"duration_s": 2.0, "chapter_seconds": 2.0, "loudness_ok": False},
])
chapter = build_publication(data_dir, "Book")["chapters"][0]
assert any("quality gate" in w for w in chapter["warnings"])
class TestBlockers:
def test_missing_artifact_blocks_publication(self, tmp_path: Path) -> None:
data_dir = _project(tmp_path, [
{"seq": 1, "title": "One", "file": "01.mp3", "source": "ch1_full.wav",
"duration_s": 2.0, "chapter_seconds": 2.0},
])
(data_dir / "export" / "ar_package" / "01.mp3").unlink()
record = build_publication(data_dir, "Book")
assert record["publishable"] is False
assert any(b["id"].startswith("missing-artifact") for b in record["blockers"])
def test_empty_package_blocks_publication(self, tmp_path: Path) -> None:
data_dir = tmp_path / ".umbra"
(data_dir / "export").mkdir(parents=True)
(data_dir / "export" / "ar_package.json").write_text(json.dumps({"tracks": []}))
record = build_publication(data_dir, "Book")
assert record["publishable"] is False
assert any(b["id"] == "no-package" for b in record["blockers"])
def test_no_package_file_at_all_blocks(self, tmp_path: Path) -> None:
data_dir = tmp_path / ".umbra"
data_dir.mkdir(parents=True)
record = build_publication(data_dir, "Book")
assert record["publishable"] is False
class TestImmutability:
def test_publication_cannot_be_overwritten(self, tmp_path: Path) -> None:
data_dir = _project(tmp_path, [
{"seq": 1, "title": "One", "file": "01.mp3", "source": "ch1_full.wav",
"duration_s": 2.0, "chapter_seconds": 2.0},
])
record = build_publication(data_dir, "Book", now="2026-10-03T00:00:00+00:00")
write_publication(data_dir, record)
with pytest.raises(FileExistsError, match="immutable"):
write_publication(data_dir, record)
def test_successor_records_what_it_supersedes(self, tmp_path: Path) -> None:
data_dir = _project(tmp_path, [
{"seq": 1, "title": "One", "file": "01.mp3", "source": "ch1_full.wav",
"duration_s": 2.0, "chapter_seconds": 2.0},
])
first = build_publication(data_dir, "Book", edition_number=1,
now="2026-10-03T00:00:00+00:00")
write_publication(data_dir, first)
second = build_publication(data_dir, "Book", edition_number=2,
supersedes=first["publication_id"],
now="2026-10-03T01:00:00+00:00")
assert second["edition_number"] == 2
assert second["supersedes"] == first["publication_id"]
write_publication(data_dir, second)
ids = [r["publication_id"] for r in list_publications(data_dir)]
assert len(ids) == 2
# the first record is untouched and still on disk
stored = json.loads(publication_path(data_dir, first["publication_id"]).read_text())
assert stored["edition_number"] == 1
def test_write_is_atomic_no_tmp_left_behind(self, tmp_path: Path) -> None:
data_dir = _project(tmp_path, [
{"seq": 1, "title": "One", "file": "01.mp3", "source": "ch1_full.wav",
"duration_s": 2.0, "chapter_seconds": 2.0},
])
record = build_publication(data_dir, "Book")
path = write_publication(data_dir, record)
assert path.exists()
assert list(path.parent.glob("*.tmp")) == []
class TestVerification:
def _record(self, tmp_path: Path) -> tuple[Path, dict[str, Any]]:
data_dir = _project(tmp_path, [
{"seq": 1, "title": "One", "file": "01.mp3", "source": "ch1_full.wav",
"duration_s": 2.0, "chapter_seconds": 2.0},
])
return data_dir, build_publication(data_dir, "Book")
def test_unchanged_tree_verifies(self, tmp_path: Path) -> None:
data_dir, record = self._record(tmp_path)
result = verify_publication(data_dir, record)
assert result["ok"] is True
assert result["drift"] == []
def test_chapter_changed_after_publication_is_drift(self, tmp_path: Path) -> None:
data_dir, record = self._record(tmp_path)
_wav(data_dir / "chapter_audio" / "ch1_full.wav", 3.0)
result = verify_publication(data_dir, record)
assert result["ok"] is False
assert any(d["id"] == "chapter-changed-ch1" for d in result["drift"])
def test_missing_chapter_is_drift(self, tmp_path: Path) -> None:
data_dir, record = self._record(tmp_path)
(data_dir / "chapter_audio" / "ch1_full.wav").unlink()
result = verify_publication(data_dir, record)
assert result["ok"] is False
assert any("missing-chapter-wav" in d["id"] for d in result["drift"])
def test_tampered_mp3_is_drift(self, tmp_path: Path) -> None:
data_dir, record = self._record(tmp_path)
(data_dir / "export" / "ar_package" / "01.mp3").write_bytes(b"tampered!!")
result = verify_publication(data_dir, record)
assert result["ok"] is False
assert any(d["id"] == "mp3-changed-ch1" for d in result["drift"])
def test_silence_is_surfaced_as_a_standing_caveat(self, tmp_path: Path) -> None:
"""Even a clean tree carries the Deal Hunter caveat into every verification."""
data_dir, record = self._record(tmp_path)
result = verify_publication(data_dir, record)
assert any("Approving a chapter does not approve" in w for w in result["warnings"])
def test_file_sha256_matches_hashlib(tmp_path: Path) -> None:
import hashlib
target = tmp_path / "big.bin"
target.write_bytes(b"x" * (3 * 1024 * 1024)) # spans several chunks
assert file_sha256(target) == hashlib.sha256(b"x" * (3 * 1024 * 1024)).hexdigest()
def test_file_sha256_on_missing_file() -> None:
assert file_sha256(Path("/nonexistent/nope.bin")) is None

535
umbraaudio/release.py Normal file
View File

@@ -0,0 +1,535 @@
"""Releases: a pinned, immutable record of what was actually published.
Why this module exists
----------------------
Approvals in this system exist per *speaker voice* (``voice_status.json``). There is no
approval for a chapter, and none for the audiobook. That is not a missing feature, it is a
defect with a measured cost, and this repository already contains the receipt.
Deal Hunter, chapter 6, as shipped on 2026-09-09:
ch6_full.wav 2227.26 s 24000 Hz mono <- what a reviewer listens to
Deal_Hunter_01_Salvage.mp3 2231.26 s 44100 Hz <- what a listener received
^^^^^^^
+4.00 s = 2 s lead + 2 s trail digital silence
Plus a 24 kHz -> 44.1 kHz resample and a loudness normalization to -20.5 dB RMS. **Four
seconds of audio were added after the chapters were approved**, by the export stage, which
is precisely the "2 second dead sound at the beginning and end of each chapter" fix. The
approved artifact and the published artifact are different files, and nothing in the system
recorded that.
The lesson, generalized:
* You cannot approve a chapter and then pad it. The approval must name the artifact a
listener receives.
* A release is a **pinned set of artifact hashes**, not a mutable build directory. Here the
export rewrites ``ar_package.json`` on every build, so it is a build status, not a release.
* A change **after** publication is not a fix to the publication. It is a successor
publication that points back at the one it replaces. That is the only way "what did we
ship" stays answerable forever.
What this module provides
-------------------------
``build_publication`` - capture the pinned set plus the approvals and profile in force.
``verify_publication`` - re-check a record against the filesystem and report drift, and
name every post-approval transformation it can see.
``write_publication`` - atomic, and refuses to overwrite an existing publication id.
A publication record is the local, immutable half of the controller's ``editions`` +
``edition_artifacts`` tables (spec 20 §F, controller D14/I-H). It is written so that ingest
is a copy, not a transformation.
"""
from __future__ import annotations
import hashlib
import json
import os
from dataclasses import dataclass, field
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
SCHEMA_VERSION = 1
# Any chapter whose shipped duration exceeds its source by more than this is treated as
# having post-approval padding rather than as a rounding artifact.
DURATION_EPSILON_S = 0.05
# Hashes are full SHA-256. Truncation is for staleness markers only.
_CHUNK = 1024 * 1024
def file_sha256(path: Path) -> str | None:
"""Full SHA-256 of a file, streamed so a 300 MB chapter does not land in memory."""
if not path.exists():
return None
digest = hashlib.sha256()
with path.open("rb") as handle:
while chunk := handle.read(_CHUNK):
digest.update(chunk)
return digest.hexdigest()
def _digest_of_strings(values: list[str]) -> str:
"""Order-independent digest of a set of strings (sorted, newline-joined)."""
return hashlib.sha256("\n".join(sorted(values)).encode()).hexdigest()
def _read_json(path: Path) -> dict[str, Any]:
if not path.exists():
return {}
try:
return json.loads(path.read_text(encoding="utf-8"))
except (OSError, ValueError):
return {}
@dataclass(frozen=True)
class DeliveryProfile:
"""Every transform applied AFTER chapter approval and BEFORE a listener hears bytes.
These are the parameters that made Deal Hunter's approved chapter differ from its
shipped track. They belong in the release record because they are part of the artifact,
not an implementation detail of the exporter.
"""
sample_rate: int
channels: int
bit_rate_kbps: int | None = None
cbr: bool | None = None
lead_silence_s: float = 0.0
trail_silence_s: float = 0.0
target_rms_db: float | None = None
max_gain_db: float | None = None
peak_ceiling_dbfs: float | None = None
notes: str = (
"These transforms change the bytes after chapter approval. A chapter approval "
"does NOT cover them; the release record does."
)
@classmethod
def from_ar_package(cls, package: dict[str, Any]) -> DeliveryProfile:
"""Read the profile that was ACTUALLY used, from the exporter's own manifest.
Measured, not declared: if the exporter changes, the record follows.
"""
fmt = package.get("format", {}) or {}
silence = float(fmt.get("silence_s", 0.0) or 0.0)
return cls(
sample_rate=int(fmt.get("sample_rate", 0) or 0),
channels=int(fmt.get("channels", 0) or 0),
bit_rate_kbps=fmt.get("bit_rate_kbps"),
cbr=fmt.get("cbr"),
lead_silence_s=silence,
trail_silence_s=silence,
target_rms_db=-20.5,
max_gain_db=12.0,
peak_ceiling_dbfs=-3.0,
)
def as_dict(self) -> dict[str, Any]:
return {
"sample_rate": self.sample_rate,
"channels": self.channels,
"bit_rate_kbps": self.bit_rate_kbps,
"cbr": self.cbr,
"lead_silence_s": self.lead_silence_s,
"trail_silence_s": self.trail_silence_s,
"target_rms_db": self.target_rms_db,
"max_gain_db": self.max_gain_db,
"peak_ceiling_dbfs": self.peak_ceiling_dbfs,
"notes": self.notes,
}
@dataclass
class ChapterRelease:
chapter_id: int
seq: int
title: str
source_wav: str
chapter_wav_sha256: str | None
chapter_wav_bytes: int | None
chapter_wav_duration_s: float | None
chapter_wav_sample_rate: int | None
mp3_object_key: str
mp3_sha256: str | None
mp3_bytes: int | None
mp3_duration_s: float | None
rms_db: float | None
peak_db: float | None
loudness_ok: bool | None
peak_ok: bool | None
source_unit_key_digest: str | None = None
utterances: int = 0
warnings: list[str] = field(default_factory=list)
def as_dict(self) -> dict[str, Any]:
return {
"chapter_id": self.chapter_id,
"seq": self.seq,
"title": self.title,
"source_wav": self.source_wav,
"chapter_wav": {
"object_key": f"chapter_audio/{self.source_wav}",
"sha256": self.chapter_wav_sha256,
"bytes": self.chapter_wav_bytes,
"duration_s": self.chapter_wav_duration_s,
"sample_rate": self.chapter_wav_sample_rate,
},
"mp3": {
"object_key": f"export/ar_package/{self.mp3_object_key}",
"sha256": self.mp3_sha256,
"bytes": self.mp3_bytes,
"duration_s": self.mp3_duration_s,
"rms_db": self.rms_db,
"peak_db": self.peak_db,
"loudness_ok": self.loudness_ok,
"peak_ok": self.peak_ok,
},
"source_unit_key_digest": self.source_unit_key_digest,
"utterances": self.utterances,
"warnings": self.warnings,
}
def _chapter_id_from_source(source_wav: str) -> int:
stem = Path(source_wav).stem # ch6_full
digits = "".join(ch for ch in stem if ch.isdigit())
return int(digits) if digits else -1
def _unit_key_digest(data_dir: Path, chapter_id: int) -> tuple[str | None, int]:
"""Digest of every source_unit_key in a chapter, plus the utterance count."""
import sqlite3
db_path = data_dir / "project.db"
if not db_path.exists() or chapter_id < 0:
return None, 0
con = sqlite3.connect(f"file:{db_path.resolve()}?mode=ro", uri=True)
try:
cols = {r[1] for r in con.execute("PRAGMA table_info(utterances)")}
if "source_unit_key" not in cols:
count = con.execute(
"SELECT COUNT(*) FROM utterances WHERE chapter_id=?", (chapter_id,)
).fetchone()[0]
return None, count
rows = con.execute(
"SELECT source_unit_key FROM utterances "
"WHERE chapter_id=? AND source_unit_key IS NOT NULL",
(chapter_id,),
).fetchall()
count = con.execute(
"SELECT COUNT(*) FROM utterances WHERE chapter_id=?", (chapter_id,)
).fetchone()[0]
finally:
con.close()
if not rows:
return None, count
return _digest_of_strings([r[0] for r in rows]), count
def _wav_info(path: Path) -> tuple[float | None, int | None]:
"""(duration_s, sample_rate) for a wav, or (None, None) if unreadable."""
if not path.exists():
return None, None
try:
import soundfile as sf
info = sf.info(str(path))
return round(float(info.duration), 3), int(info.samplerate)
except Exception: # noqa: BLE001 - a corrupt file must not abort a release build
return None, None
def build_publication(
data_dir: Path,
work: str,
edition_number: int = 1,
published_by: str = "operator",
git_commit: str | None = None,
supersedes: str | None = None,
now: str | None = None,
) -> dict[str, Any]:
"""Capture the pinned set, the approvals in force, and the delivery profile.
Read-only with respect to the project: hashes files, writes nothing.
"""
data_dir = Path(data_dir)
export_dir = data_dir / "export"
package = _read_json(export_dir / "ar_package.json")
profile = DeliveryProfile.from_ar_package(package)
blockers: list[dict[str, str]] = []
chapters: list[ChapterRelease] = []
for track in package.get("tracks", []) or []:
source_wav = track.get("source") or ""
chapter_id = _chapter_id_from_source(source_wav)
wav_path = data_dir / "chapter_audio" / source_wav
mp3_path = export_dir / "ar_package" / (track.get("file") or "")
duration_s, sample_rate = _wav_info(wav_path)
key_digest, utt_count = _unit_key_digest(data_dir, chapter_id)
release = ChapterRelease(
chapter_id=chapter_id,
seq=int(track.get("seq", 0) or 0),
title=str(track.get("title", "") or ""),
source_wav=source_wav,
chapter_wav_sha256=file_sha256(wav_path),
chapter_wav_bytes=wav_path.stat().st_size if wav_path.exists() else None,
chapter_wav_duration_s=duration_s,
chapter_wav_sample_rate=sample_rate,
mp3_object_key=str(track.get("file", "") or ""),
mp3_sha256=file_sha256(mp3_path),
mp3_bytes=int(track["size_bytes"]) if track.get("size_bytes") else None,
mp3_duration_s=track.get("duration_s"),
rms_db=track.get("rms_db"),
peak_db=track.get("peak_db"),
loudness_ok=track.get("loudness_ok"),
peak_ok=track.get("peak_ok"),
source_unit_key_digest=key_digest,
utterances=utt_count,
)
# The Deal Hunter check, made mechanical.
if (
duration_s is not None
and track.get("duration_s") is not None
and float(track["duration_s"]) - duration_s > DURATION_EPSILON_S
):
delta = round(float(track["duration_s"]) - duration_s, 3)
release.warnings.append(
f"shipped track is {delta}s longer than the approved chapter wav "
f"({duration_s}s -> {track['duration_s']}s): padding or normalization was "
"applied AFTER approval. The chapter approval does not cover these bytes."
)
if (
sample_rate is not None
and profile.sample_rate
and sample_rate != profile.sample_rate
):
release.warnings.append(
f"chapter wav is {sample_rate} Hz, delivered at {profile.sample_rate} Hz: "
"resampled after approval"
)
if release.loudness_ok is False or release.peak_ok is False:
release.warnings.append("delivery quality gate reported a failure")
if release.mp3_sha256 is None:
blockers.append(
{
"id": f"missing-artifact-ch{chapter_id}",
"why": f"{track.get('file')} is referenced by the manifest but absent",
}
)
chapters.append(release)
if not chapters:
blockers.append(
{
"id": "no-package",
"why": f"{export_dir / 'ar_package.json'} has no tracks; nothing to pin",
}
)
voice_status = _read_json(data_dir / "voice_status.json")
approvals = _approval_block(voice_status, data_dir)
package_file = export_dir / f"{package.get('base_name', work)}.m4b"
if not package_file.exists():
candidates = sorted(export_dir.glob("*.m4b"))
package_file = candidates[0] if candidates else package_file
stamp = now or datetime.now(UTC).isoformat(timespec="seconds")
return {
"publication_id": f"pub-{edition_number:03d}-{stamp[:10]}",
"schema_version": SCHEMA_VERSION,
"work": work,
"edition_number": edition_number,
"supersedes": supersedes,
"published_at": stamp,
"published_by": published_by,
"git_commit": git_commit,
"delivery_profile": profile.as_dict(),
"package": {
"object_key": (
f"export/{package_file.name}" if package_file.exists() else None
),
"sha256": file_sha256(package_file),
"bytes": package_file.stat().st_size if package_file.exists() else None,
"track_count": package.get("track_count"),
"built_at": package.get("built_at"),
},
"approvals": approvals,
"chapters": [c.as_dict() for c in chapters],
"blockers": blockers,
"publishable": not blockers,
}
def _approval_block(voice_status: dict[str, Any], data_dir: Path) -> dict[str, Any]:
"""The per-item approvals that were in force, pinned by digest so they are auditable.
``voice_status.json`` is an *approval ledger*: it only holds speakers someone explicitly
touched, so it is smaller than ``speakers.json``. Reporting its size as "voices approved"
would imply an approval that was never given, so the two counts are reported separately
and the gap is named.
"""
raw = voice_status.get("speakers", voice_status) if isinstance(voice_status, dict) else {}
entries: dict[str, Any] = raw if isinstance(raw, dict) else {}
statuses = {
name: (value.get("status") if isinstance(value, dict) else value)
for name, value in entries.items()
}
approved = sorted(n for n, s in statuses.items() if str(s).lower() == "approved")
profiles = _read_json(data_dir / "speakers.json").get("speakers", []) or []
in_profile = {
str(p.get("primary_name", ""))
for p in profiles
if isinstance(p, dict) and p.get("primary_name")
}
# A speaker counts as approved only if the ledger says so by primary name.
unapproved = sorted(in_profile - set(approved))
orphans = sorted(set(approved) - in_profile)
return {
"voices_in_profile": len(in_profile),
"voices_with_approval_record": len(approved),
"voices_approved": len(approved),
"voices_not_approved": len(unapproved),
"voices_not_approved_names": unapproved,
"approval_ledger_entries_without_profile": orphans,
"voice_status_digest": _digest_of_strings(
[f"{n}={s}" for n, s in sorted(statuses.items())]
),
"chapter_approved": False,
"audiobook_approved": False,
"_note": (
"voice_status.json holds per-speaker approvals only, and only for speakers "
"someone explicitly touched - it is smaller than speakers.json. There is no "
"chapter or audiobook approval in the system today; this record IS the "
"release approval."
),
}
def verify_publication(data_dir: Path, record: dict[str, Any]) -> dict[str, Any]:
"""Re-check a publication against the filesystem.
Returns ``ok``, ``drift`` (the pinned bytes are gone or changed) and ``warnings``
(things a reader should know but which do not invalidate the record).
"""
data_dir = Path(data_dir)
drift: list[dict[str, str]] = []
warnings: list[str] = []
for chapter in record.get("chapters", []) or []:
wav = data_dir / "chapter_audio" / chapter.get("source_wav", "")
pinned_wav = (chapter.get("chapter_wav") or {}).get("sha256")
if not wav.exists():
drift.append(
{"id": f"missing-chapter-wav-ch{chapter.get('chapter_id')}",
"why": f"{wav.name} no longer exists"}
)
elif pinned_wav and file_sha256(wav) != pinned_wav:
drift.append(
{"id": f"chapter-changed-ch{chapter.get('chapter_id')}",
"why": (
f"{wav.name} changed after publication. The published release is "
"still reproducible from its recorded hashes, but the working tree "
"has moved on. Publish a successor edition rather than editing this one."
)}
)
mp3 = data_dir / "chapter_audio" / ".." / "export" / "ar_package" / (
(chapter.get("mp3") or {}).get("object_key", "").split("/")[-1]
)
pinned_mp3 = (chapter.get("mp3") or {}).get("sha256")
if pinned_mp3 and mp3.exists() and file_sha256(mp3) != pinned_mp3:
drift.append(
{"id": f"mp3-changed-ch{chapter.get('chapter_id')}",
"why": f"{mp3.name} no longer matches the hash this release published"}
)
for warning in chapter.get("warnings", []) or []:
warnings.append(f"ch{chapter.get('chapter_id')}: {warning}")
pinned_pkg = (record.get("package") or {}).get("sha256")
pkg_key = (record.get("package") or {}).get("object_key")
if pinned_pkg and pkg_key:
pkg_path = data_dir / pkg_key
if not pkg_path.exists():
drift.append({"id": "missing-package", "why": f"{pkg_key} no longer exists"})
elif file_sha256(pkg_path) != pinned_pkg:
drift.append(
{
"id": "package-changed",
"why": f"{pkg_key} no longer matches the hash this release published",
}
)
profile = record.get("delivery_profile", {}) or {}
lead = float(profile.get("lead_silence_s", 0.0) or 0.0)
trail = float(profile.get("trail_silence_s", 0.0) or 0.0)
if lead or trail:
warnings.append(
f"delivery adds {lead}s of leading and {trail}s of trailing silence after "
"chapter approval. Approving a chapter does not approve the delivered track; "
"approve the released artifact."
)
return {
"publication_id": record.get("publication_id"),
"verified_at": datetime.now(UTC).isoformat(timespec="seconds"),
"ok": not drift,
"drift": drift,
"warnings": warnings,
}
def publication_path(data_dir: Path, publication_id: str) -> Path:
return Path(data_dir) / "publications" / f"{publication_id}.json"
def write_publication(data_dir: Path, record: dict[str, Any]) -> Path:
"""Write a publication record atomically. Refuses to overwrite an existing one.
Refusal is the point: a published release is immutable, so a change after publication
is a NEW edition with ``supersedes`` set, never an edit of this file.
"""
path = publication_path(data_dir, str(record["publication_id"]))
path.parent.mkdir(parents=True, exist_ok=True)
if path.exists():
raise FileExistsError(
f"{path.name} already exists. A publication is immutable - publish a new "
"edition with supersedes set instead of rewriting a released one."
)
tmp = path.with_suffix(".json.tmp")
tmp.write_text(json.dumps(record, indent=2, sort_keys=True) + "\n", encoding="utf-8")
os.replace(tmp, path)
return path
def list_publications(data_dir: Path) -> list[dict[str, Any]]:
"""Every publication on disk, newest first. Records are read, never rewritten."""
directory = Path(data_dir) / "publications"
if not directory.is_dir():
return []
records = []
for path in sorted(directory.glob("pub-*.json"), reverse=True):
record = _read_json(path)
if record:
records.append(
{
"publication_id": record.get("publication_id"),
"edition_number": record.get("edition_number"),
"published_at": record.get("published_at"),
"supersedes": record.get("supersedes"),
"chapters": len(record.get("chapters", []) or []),
"publishable": record.get("publishable"),
}
)
return records