From 9dfa7915eec39b74fb19a14ee90db05a9a06973d Mon Sep 17 00:00:00 2001 From: "B.D. Murphy" <161863146+bdmurphy73@users.noreply.github.com> Date: Fri, 16 Oct 2026 00:27:24 -0700 Subject: [PATCH] Probe suite: T06/T07 now test, T08 names its real blocker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The dry run passes since routes_compute.py:1792 was aligned with read_bundle (separators + ensure_ascii=False), so three tests had to catch up. All three carried narratives written when the bundle was rejected, and each kept asserting a reason that had stopped being true — a BLOCKED test is a finding, so a stale BLOCKED reason is a finding about the wrong world. T06 re-runs the dry run and compares responses, proving validation is deterministic. It says plainly that re-import suppressing stored rows is NOT covered: that path needs rows that persist, and the import still builds InMemoryAudioImportStore. T07 reconciles the controller's counts against the bytes sent. 8884 records in, 8883 accounted for, plus the book record it does not count — so a controller that silently dropped a record kind now fails here instead of passing quietly. T08 keeps BLOCKED, but for the reason that is actually true: no Postgres AudioImportStore adapter exists, so even a successful commit would return a uuid4 project_id while translation_utterances stayed empty. Its text names InMemoryAudioImportStore on purpose, so wiring an adapter fails a test and forces the verdict to be re-decided. Suite: 9 PASS / 0 FAIL / 1 BLOCKED (T08). Tests 30 -> 40. --- scripts/audio_controller_connect.py | 234 ++++++++++++++++++++----- tests/test_audio_controller_connect.py | 158 +++++++++++++++++ 2 files changed, 351 insertions(+), 41 deletions(-) diff --git a/scripts/audio_controller_connect.py b/scripts/audio_controller_connect.py index 864d067..eba3850 100644 --- a/scripts/audio_controller_connect.py +++ b/scripts/audio_controller_connect.py @@ -24,6 +24,7 @@ Exit 0 when no test failed. BLOCKED does not fail the run; it is a finding. from __future__ import annotations import argparse +import hashlib import json import os import ssl @@ -408,6 +409,44 @@ def t04_ingest_exists(client: Client, **csrf_note}) +def _load_bundle(bundle_dir: str) -> tuple[dict[str, Any] | None, + list[dict[str, Any]] | None, + str | None]: + """Read bundle.json + manifest.jsonl, or say why they are unusable.""" + + root = Path(bundle_dir) + envelope_path = root / "bundle.json" + manifest_path = root / "manifest.jsonl" + if not envelope_path.is_file() or not manifest_path.is_file(): + return None, None, f"{bundle_dir} is missing bundle.json or manifest.jsonl" + envelope = json.loads(envelope_path.read_text(encoding="utf-8")) + records = [json.loads(line) for line in + manifest_path.read_text(encoding="utf-8").splitlines() + if line.strip()] + return envelope, records, None + + +def _dry_run_once(client: Client, + bundle_dir: str, + headers: dict[str, str]) -> tuple[int, Any, int]: + """One dry run. Returns (status, body, duration_ms).""" + + envelope, records, _problem = _load_bundle(bundle_dir) + return client.request( + "POST", "/api/node/audio/import", headers=headers, + body={"envelope": envelope, "records": records, + "dry_run": True, "confirm": None}) + + +def _digest(body: Any) -> str | None: + """A comparable fingerprint of a dry-run response.""" + + if not isinstance(body, dict): + return None + return hashlib.sha256( + json.dumps(body, sort_keys=True).encode("utf-8")).hexdigest() + + def t05_dry_run(client: Client, bundle_dir: str | None, headers: dict[str, str]) -> Result: @@ -427,23 +466,12 @@ def t05_dry_run(client: Client, "against; pass --bundle to run a real dry run", {"depends_on": None, "ran": False}) - root = Path(bundle_dir) - envelope_path = root / "bundle.json" - manifest_path = root / "manifest.jsonl" - if not envelope_path.is_file() or not manifest_path.is_file(): - return Result("T05", "Dry-run validation", "BLOCKED", - f"{bundle_dir} is missing bundle.json or manifest.jsonl", + envelope, records, problem = _load_bundle(bundle_dir) + if problem: + return Result("T05", "Dry-run validation", "BLOCKED", problem, {"ran": False}) - envelope = json.loads(envelope_path.read_text(encoding="utf-8")) - records = [json.loads(line) for line in - manifest_path.read_text(encoding="utf-8").splitlines() - if line.strip()] - - status, body, duration = client.request( - "POST", "/api/node/audio/import", headers=headers, - body={"envelope": envelope, "records": records, - "dry_run": True, "confirm": None}) + status, body, duration = _dry_run_once(client, bundle_dir, headers) summary = _summarise(body) evidence: dict[str, Any] = { @@ -461,9 +489,22 @@ def t05_dry_run(client: Client, if status in (200, 201): evidence["reported_status"] = body.get("status") if isinstance(body, dict) else None evidence["content_hash"] = body.get("content_hash") if isinstance(body, dict) else None + evidence["counts"] = body.get("counts") if isinstance(body, dict) else None + evidence["warning_count"] = (len(body.get("warnings") or []) + if isinstance(body, dict) else None) + evidence["response_digest"] = _digest(body) + declared = (envelope or {}).get("content_hash", {}) + declared_value = declared.get("value") if isinstance(declared, dict) else None + evidence["declared_content_hash"] = declared_value + evidence["hash_round_trips"] = ( + evidence["content_hash"] is not None + and evidence["content_hash"] == declared_value + ) return Result("T05", "Dry-run validation", "PASS", f"controller accepted the bundle in dry-run mode ({status}); " - f"reported status={evidence['reported_status']}", evidence) + f"reported status={evidence['reported_status']}, " + f"content_hash round-trips={evidence['hash_round_trips']}", + evidence) detail = (f"dry run rejected ({status}); " + (str(summary)[:300] if summary else "empty body")) @@ -475,33 +516,144 @@ def t05_dry_run(client: Client, return Result("T05", "Dry-run validation", "FAIL", detail, evidence) -def t06_idempotency(client: Client) -> Result: - return Result("T06", "Idempotency check", "BLOCKED", - "needs T05 to produce a content_hash to compare against, and " - "T05 does not get one while the bundle is rejected. The " - "controller also reports its accepted manifest format only " - "after it receives a bundle, so this is sequenced, not broken.", - {"depends_on": "T05", - "client_side_equivalent": "the bundle build is deterministic: " - "the same project yields the same manifest.jsonl bytes and the " - "same content_hash, so re-shipping is detectable without a server"}) +def t06_idempotency(client: Client, + bundle_dir: str | None, + headers: dict[str, str], + t05: Result | None) -> Result: + """Two dry runs must agree; re-import behaviour still needs a real commit. + + Historically this returned a narrative written for the era when T05 was + rejected ("T05 does not get one while the bundle is rejected"), so it stayed + BLOCKED with a reason that had stopped being true. It now runs the dry run a + second time and compares responses. + + What this proves: the controller's validation is deterministic, so the same + bundle always produces the same plan and the same content_hash. That is one + half of idempotency. + + What it cannot prove: that re-importing the same hash is a no-op. That + assertion lives in ``import_bundle``'s ``last_content_hash`` check and only + means anything once rows exist, which needs a store that persists. See T08. + """ + + if not bundle_dir: + return Result("T06", "Idempotency check", "BLOCKED", + "no --bundle given, so there is nothing to re-run", + {"depends_on": "T05", "ran": False}) + if t05 is None or t05.status != "PASS": + return Result("T06", "Idempotency check", "BLOCKED", + "T05 did not pass, so there is no first response to " + "compare a second one against", + {"depends_on": "T05", "ran": False}) + + first_digest = t05.evidence.get("response_digest") + status, body, duration = _dry_run_once(client, bundle_dir, headers) + second_digest = _digest(body) + + evidence: dict[str, Any] = { + "ran": True, + "first_digest": first_digest, + "second_digest": second_digest, + "second_status_code": status, + "second_duration_ms": duration, + "second_content_hash": (body.get("content_hash") + if isinstance(body, dict) else None), + "wrote": False, + } + + if status not in (200, 201): + return Result("T06", "Idempotency check", "FAIL", + f"the second dry run was rejected ({status}) while the " + f"first passed, so validation is not repeatable", + evidence) + if first_digest != second_digest: + return Result("T06", "Idempotency check", "FAIL", + "two identical dry runs returned different bodies, so the " + "plan is not deterministic", + evidence) + + evidence["untested"] = ("re-import of an already-stored content_hash " + "returning no_change") + return Result("T06", "Idempotency check", "PASS", + "two consecutive dry runs returned byte-identical responses, " + "so the plan and content_hash are deterministic. Re-import " + "suppressing already-stored rows is NOT covered here: that " + "path needs rows that persist, and the import currently " + "builds an in-memory store", + evidence) -def t07_schema_contract(client: Client) -> Result: - return Result("T07", "Schema contract", "BLOCKED", - "the controller has not yet issued a manifest format to accept; " - "its request says the mapping and the accepted format come after " - "the bundle. A contract is proposed in the report, not assumed.", - {"depends_on": "T04", - "bundle_validates_locally": True}) +def t07_schema_contract(t05: Result | None, + bundle_records: int | None) -> Result: + """Did the controller's counts reconcile with the bytes we actually sent? + + This used to say "the controller has not yet issued a manifest format to + accept" — true before the first accepted dry run, stale afterwards. An + accepted response IS the format confirmation: the controller parsed all + 8,884 records and reported counts for them. + """ + + if t05 is None or t05.status != "PASS": + return Result("T07", "Schema contract", "BLOCKED", + "the controller has not yet accepted a bundle, so no " + "format has been confirmed by it", + {"depends_on": "T04", "bundle_validates_locally": True}) + + counts = t05.evidence.get("counts") or {} + accounted = sum(counts.get(k, 0) + for k in ("chapters", "speakers", "utterances", "artifacts")) + evidence: dict[str, Any] = { + "ran": True, + "controller_counts": counts, + "records_sent": bundle_records, + "records_accounted": accounted, + "note": ("the book record is the one kind the controller does not " + "report as a count, so records_sent - records_accounted should " + "be exactly 1"), + } + if bundle_records is None: + evidence["delta"] = None + return Result("T07", "Schema contract", "PASS", + "the controller accepted and parsed the manifest format", + evidence) + + delta = bundle_records - accounted + evidence["delta"] = delta + if delta != 1: + return Result("T07", "Schema contract", "FAIL", + f"the controller accounted for {accounted} of " + f"{bundle_records} records sent (delta {delta}, expected " + f"1 for the uncounted book record) — some record kind is " + f"being dropped silently", + evidence) + return Result("T07", "Schema contract", "PASS", + f"the controller parsed all {bundle_records} records: its " + f"counts account for {accounted}, plus 1 book record it does " + f"not count separately", + evidence) def t08_audit_round_trip(client: Client) -> Result: + """Read back what was written, from the controller's side. + + Its old text said "no ingest event can be created without an ingest + endpoint". The endpoint has existed since T04 went green, so that reason was + stale. The real blocker is that nothing is written: the import builds + ``InMemoryAudioImportStore``, whose ``commit`` appends to a list and mints + ``uuid4()`` ids, so there would be no row to read back even after a + successful commit. + """ + return Result("T08", "Bidirectional audit", "BLOCKED", - "no ingest event can be created without an ingest endpoint; " - "the controller does expose an audit table but only to " - "authenticated principals", - {"depends_on": "T04"}) + "needs a committed import to read back from. The endpoint " + "routes and validates (T04, T05), but routes_compute builds " + "InMemoryAudioImportStore, whose commit() writes to a list and " + "not to Postgres — so a commit would return a uuid4 project_id " + "while translation_utterances stayed empty, and there would be " + "nothing to audit. Blocked on a persistence adapter, not on " + "the endpoint", + {"depends_on": "T04", + "needs": "a Postgres AudioImportStore adapter"}) def t09_error_handling(client: Client) -> Result: @@ -803,14 +955,15 @@ def main() -> int: bundle_records = sum(1 for line in manifest.read_text().splitlines() if line.strip()) bundle_bytes = manifest.stat().st_size + t05_result = t05_dry_run(client, args.bundle, headers) results = [ t01_health(client), t02_auth(client, headers, how), t03_version(client), t04_ingest_exists(client, headers), - t05_dry_run(client, args.bundle, headers), - t06_idempotency(client), - t07_schema_contract(client), + t05_result, + t06_idempotency(client, args.bundle, headers, t05_result), + t07_schema_contract(t05_result, bundle_records), t08_audit_round_trip(client), t09_error_handling(client), t10_payload_readiness(bundle_records, bundle_bytes), @@ -829,7 +982,6 @@ def main() -> int: transport = "plain HTTP (uvicorn); no TLS on this port" \ if client.base_url.startswith("http") else "HTTPS" - t05_result = next((r for r in results if r.test_id == "T05"), None) verdict, rationale = _readiness(routed_ingest, missing_ingest, t05_result) report: dict[str, Any] = { diff --git a/tests/test_audio_controller_connect.py b/tests/test_audio_controller_connect.py index b8e2061..28ff8d4 100644 --- a/tests/test_audio_controller_connect.py +++ b/tests/test_audio_controller_connect.py @@ -516,3 +516,161 @@ def test_dry_run_passes_only_on_a_2xx(tmp_path: Path) -> None: assert passed.evidence["content_hash"] == "a" * 64 assert failed.status == "FAIL" assert failed.evidence["status_code"] == 500 + + +# --------------------------------------------------------------------------- +# What T06/T07/T08 claim, once a dry run can actually pass +# --------------------------------------------------------------------------- +# +# These tests exist because all three carried narratives written for the era +# when the bundle was rejected, and each kept asserting a reason that had +# stopped being true: T06 said "T05 does not get [a content_hash] while the +# bundle is rejected", T07 said the controller "has not yet issued a manifest +# format", T08 said there was no ingest endpoint. Every one of those was once +# accurate, and every one survived its own refutation. A BLOCKED test is a +# finding, so a stale BLOCKED reason is a finding about the wrong world. + + +def _write_bundle(tmp_path: Path, records: int = 1) -> None: + (tmp_path / "bundle.json").write_text( + json.dumps({"bundle_format_version": 1, + "content_hash": {"value": "0" * 64}}), encoding="utf-8") + (tmp_path / "manifest.jsonl").write_text( + "".join(json.dumps({"id": f"b-{i}", "kind": "book"}) + "\n" + for i in range(records)), encoding="utf-8") + + +def _passing_t05(digest: str = "d" * 64, + counts: dict[str, int] | None = None) -> Any: + return tool.Result( + "T05", "Dry-run validation", "PASS", "accepted", + {"status_code": 200, "reported_status": "dry_run", + "content_hash": "0" * 64, "response_digest": digest, + "counts": counts if counts is not None + else {"chapters": 1, "speakers": 0, "utterances": 0, "artifacts": 0}}) + + +def test_t06_does_not_run_a_canned_reason_when_t05_failed(tmp_path: Path) -> None: + """The old bug: T06 narrated why it could not run, whatever T05 said.""" + + _write_bundle(tmp_path) + rejected = tool.Result("T05", "Dry-run validation", "FAIL", "500", {}) + + result = tool.t06_idempotency(FakeClient({}), str(tmp_path), {}, rejected) + + assert result.status == "BLOCKED" + assert result.evidence.get("ran") is False + + +def test_t06_blocked_without_a_bundle_or_a_first_response() -> None: + assert tool.t06_idempotency(FakeClient({}), None, {}, None).evidence["ran"] is False + assert tool.t06_idempotency( + FakeClient({}), None, {}, _passing_t05()).evidence["ran"] is False + + +def test_t06_passes_when_two_dry_runs_agree(tmp_path: Path) -> None: + _write_bundle(tmp_path) + body = {"status": "dry_run", "content_hash": "0" * 64, + "counts": {"chapters": 1}} + first = tool.t05_dry_run( + FakeClient({("POST", NODE_IMPORT): 200}, + bodies={("POST", NODE_IMPORT): body}), str(tmp_path), {}) + + second = tool.t06_idempotency( + FakeClient({("POST", NODE_IMPORT): 200}, + bodies={("POST", NODE_IMPORT): body}), + str(tmp_path), {}, first) + + assert first.status == "PASS" + assert second.status == "PASS" + assert second.evidence["first_digest"] == second.evidence["second_digest"] + + +def test_t06_fails_when_two_identical_dry_runs_disagree(tmp_path: Path) -> None: + """Non-determinism means the plan is not a function of the bundle.""" + + _write_bundle(tmp_path) + first = _passing_t05(digest="d" * 64) + + drifted = tool.t06_idempotency( + FakeClient({("POST", NODE_IMPORT): 200}, + bodies={("POST", NODE_IMPORT): {"status": "dry_run", + "content_hash": "1" * 64}}), + str(tmp_path), {}, first) + + assert drifted.status == "FAIL" + + +def test_t06_names_what_it_cannot_prove(tmp_path: Path) -> None: + """Passing on determinism must not read as proof of re-import safety.""" + + _write_bundle(tmp_path) + body = {"status": "dry_run", "content_hash": "0" * 64} + first = tool.t05_dry_run( + FakeClient({("POST", NODE_IMPORT): 200}, + bodies={("POST", NODE_IMPORT): body}), str(tmp_path), {}) + + result = tool.t06_idempotency( + FakeClient({("POST", NODE_IMPORT): 200}, + bodies={("POST", NODE_IMPORT): body}), + str(tmp_path), {}, first) + + assert result.status == "PASS" + assert "no_change" not in result.detail.split(".")[0] + assert "NOT covered" in result.detail + assert "in-memory store" in result.detail + + +def test_t07_reconciles_counts_and_allows_for_the_uncounted_book_record() -> None: + """32+55+4319+4477 = 8883, and the 8884th record is the book.""" + + result = tool.t07_schema_contract( + _passing_t05(counts={"chapters": 32, "speakers": 55, + "utterances": 4319, "artifacts": 4477}), + 8884) + + assert result.status == "PASS" + assert result.evidence["delta"] == 1 + + +def test_t07_fails_when_the_controller_drops_records(tmp_path: Path) -> None: + """A controller that silently ignores a record kind would otherwise pass.""" + + result = tool.t07_schema_contract( + _passing_t05(counts={"chapters": 1, "speakers": 0, + "utterances": 0, "artifacts": 0}), + 5) + + assert result.status == "FAIL" + assert result.evidence["delta"] == 4 + + +def test_t07_blocked_until_a_bundle_is_accepted() -> None: + result = tool.t07_schema_contract( + tool.Result("T05", "Dry-run validation", "FAIL", "500", {}), 1) + + assert result.status == "BLOCKED" + + +def test_t08_no_longer_blames_an_endpoint_that_routs() -> None: + """The specific stale claim T08 used to make.""" + + result = tool.t08_audit_round_trip(FakeClient({})) + + assert result.status == "BLOCKED" + assert "without an ingest endpoint" not in result.detail + assert "depends on T04" not in result.detail + + +def test_t08_names_the_actual_blocker_so_a_fix_has_to_update_it() -> None: + """Wiring a persistence adapter must force this text to change. + + If someone implements a Postgres AudioImportStore, this test fails and + makes them re-decide T08 rather than leave a blocked verdict describing a + store that no longer exists. + """ + + result = tool.t08_audit_round_trip(FakeClient({})) + + assert "InMemoryAudioImportStore" in result.detail + assert "uuid4" in result.detail