From 0f67da8acf696aace9f7e2f3a4c5b277ae321e79 Mon Sep 17 00:00:00 2001 From: Dmitry Dementiev Date: Tue, 1 Sep 2026 22:32:21 +0300 Subject: [PATCH 1/7] =?UTF-8?q?test(diarization):=20=D0=B8=D0=B7=D0=BE?= =?UTF-8?q?=D0=BB=D0=B8=D1=80=D0=BE=D0=B2=D0=B0=D0=BD=D0=B0=20=D0=BF=D1=80?= =?UTF-8?q?=D0=BE=D0=B2=D0=B5=D1=80=D0=BA=D0=B0=20=D0=BA=D0=B5=D1=88=D0=B0?= =?UTF-8?q?=20=D0=BC=D0=BE=D0=B4=D0=B5=D0=BB=D0=B5=D0=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Зачем: - детерминированный gate эксперимента не должен загружать native runtime в checksum-only тесте. - Что: - в тест проверки поврежденного кеша добавлена структурная подмена sherpa_onnx. - Проверка: - uv run pytest -q tests/test_streaming_sortformer_experiment.py tests/test_benchmark_experiment.py tests/test_speaker_diarizer.py tests/test_diarization.py. --- tests/test_speaker_diarizer.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/test_speaker_diarizer.py b/tests/test_speaker_diarizer.py index 589d5d0..e39200b 100644 --- a/tests/test_speaker_diarizer.py +++ b/tests/test_speaker_diarizer.py @@ -222,6 +222,7 @@ def test_load_speaker_diarizer_keeps_corrupt_cache_when_download_is_invalid( with ( patch("huggingface_hub.cached_assets_path", return_value=tmp_path), + patch.dict(sys.modules, {"sherpa_onnx": SimpleNamespace()}), patch("httpx.stream", return_value=FakeResponse()), pytest.raises(RuntimeError, match="Контрольная сумма"), ): From e71b2d16688780ca0172adf0da8e9b6b1222cf05 Mon Sep 17 00:00:00 2001 From: Dmitry Dementiev Date: Wed, 2 Sep 2026 03:20:42 +0300 Subject: [PATCH 2/7] =?UTF-8?q?feat(experiment):=20=D0=B4=D0=BE=D0=B1?= =?UTF-8?q?=D0=B0=D0=B2=D0=BB=D0=B5=D0=BD=20gate=20=D0=BF=D1=80=D0=B8?= =?UTF-8?q?=D0=BD=D1=8F=D1=82=D0=B8=D1=8F=20=D0=BE=D0=BF=D0=BE=D1=80=D1=8B?= =?UTF-8?q?=20=D0=B2=D0=BB=D0=B0=D0=B4=D0=B5=D0=BB=D1=8C=D1=86=D0=B5=D0=BC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Зачем: - не допустить загрузку W0 и Sortformer без отдельного принятия трёх референсов владельцем - связать manifest с точными байтами listening package и owner-authored record Что: - добавлены строгие схемы трёх reference rows и четырёх решений Step 3 - все внешние файлы проверяются через held fd без symlink и TOCTOU - добавлены negative tests для подмен, неточных JSON-типов и небезопасных путей Проверка: - uv run pytest -q (532 passed, 2 skipped) - uv run ruff check src/local_transcriber/streaming_sortformer_experiment.py tests/test_streaming_sortformer_experiment.py - git diff --check --- .../streaming_sortformer_experiment.py | 397 +++++++- tests/test_streaming_sortformer_experiment.py | 933 +++++++++++++++++- 2 files changed, 1296 insertions(+), 34 deletions(-) diff --git a/src/local_transcriber/streaming_sortformer_experiment.py b/src/local_transcriber/streaming_sortformer_experiment.py index 47f7f2b..efefe5b 100644 --- a/src/local_transcriber/streaming_sortformer_experiment.py +++ b/src/local_transcriber/streaming_sortformer_experiment.py @@ -5,11 +5,13 @@ from __future__ import annotations import argparse import copy import ctypes +import errno import hashlib import json import math import os import re +import stat import statistics import subprocess import sys @@ -44,10 +46,51 @@ TOP_LEVEL_KEYS = { "schedule", "rss_sample_interval_ms", } -FRESH_BASIS_KEYS = {"marker", "approval"} -APPROVAL_KEYS = {"decision", "timestamp"} +FRESH_BASIS_KEYS = {"marker", "owner_package", "approval"} +OWNER_PACKAGE_LINK_KEYS = {"path", "sha256", "generated_at"} +APPROVAL_LINK_KEYS = { + "path", + "sha256", + "actor", + "status", + "decision", + "timestamp", +} +OWNER_PACKAGE_SCHEMA = "local-transcriber.owner-listening-package.v1" +OWNER_PACKAGE_KEYS = {"schema", "generated_at", "references"} +OWNER_PACKAGE_REFERENCE_KEYS = { + "neutral_alias", + "canonical_fragment_audio_sha256", + "final_reference_sha256", + "full_recording_sha256", +} +REFERENCE_ACCEPTANCE_SCHEMA = "local-transcriber.reference-owner-acceptance.v1" +REFERENCE_ACCEPTANCE_KEYS = { + "schema", + "actor", + "status", + "decision", + "timestamp", + "owner_package_sha256", + "references", + "step3", +} +REFERENCE_ACCEPTANCE_ROW_KEYS = OWNER_PACKAGE_REFERENCE_KEYS | { + "fragment_expected_cluster_count", + "full_recording_expected_cluster_count", + "fragment_review_complete", + "full_recording_count_review_complete", + "reference_accepted", +} +STEP3_KEYS = { + "recording_use_approved", + "inputs_and_counts_approved", + "full_recordings_available", + "final_hearing_gate_committed", +} SOURCE_KEYS = {"commit"} INPUT_KEYS = { + "neutral_alias", "source_path", "source_sha256", "source_size_bytes", @@ -312,45 +355,294 @@ def build_full_recording_schedule( return result -def _resolve(root: Path, value: Any) -> Path: +def _aware_datetime(value: Any) -> datetime: + if not isinstance(value, str): + raise PreflightError("approval-timestamp") + try: + parsed = datetime.fromisoformat(value) + except ValueError as exc: + raise PreflightError("approval-timestamp") from exc + if parsed.tzinfo is None or parsed.utcoffset() is None: + raise PreflightError("approval-timestamp") + return parsed + + +def _safe_relative_parts(value: Any) -> tuple[str, ...]: if not isinstance(value, str) or not value: raise PreflightError("unsafe-relative-path") pure = PurePosixPath(value) - if pure.is_absolute() or ".." in pure.parts or "\\" in value: + if ( + pure.is_absolute() + or not pure.parts + or ".." in pure.parts + or "\\" in value + or "\x00" in value + ): raise PreflightError("path-escape") - base = root.resolve() - resolved = (base / Path(value)).resolve() - if not resolved.is_relative_to(base): - raise PreflightError("path-escape") - return resolved + return pure.parts + + +def _same_file_state(before: os.stat_result, after: os.stat_result) -> bool: + return ( + before.st_dev, + before.st_ino, + before.st_mode, + before.st_nlink, + before.st_size, + before.st_mtime_ns, + before.st_ctime_ns, + ) == ( + after.st_dev, + after.st_ino, + after.st_mode, + after.st_nlink, + after.st_size, + after.st_mtime_ns, + after.st_ctime_ns, + ) + + +@dataclass(frozen=True) +class _BoundFile: + sha256: str + size_bytes: int + payload: bytes | None + + +def _read_bound_regular_file( + root: Path, path_value: Any, *, capture_payload: bool +) -> _BoundFile: + """Читает файл относительно доверенного неизменяемого root dirfd.""" + parts = _safe_relative_parts(path_value) + directory_flags = ( + os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC + ) + opened_fds: list[int] = [] + fd_states: list[tuple[int, os.stat_result]] = [] + namespace_entries: list[tuple[int, str, os.stat_result]] = [] + try: + directory_fd = os.open(root, directory_flags) + opened_fds.append(directory_fd) + root_state = os.fstat(directory_fd) + if not stat.S_ISDIR(root_state.st_mode): + raise PreflightError("file-binding") + fd_states.append((directory_fd, root_state)) + for part in parts[:-1]: + parent_fd = directory_fd + directory_fd = os.open(part, directory_flags, dir_fd=parent_fd) + opened_fds.append(directory_fd) + directory_state = os.fstat(directory_fd) + if not stat.S_ISDIR(directory_state.st_mode): + raise PreflightError("file-binding") + fd_states.append((directory_fd, directory_state)) + namespace_entries.append((parent_fd, part, directory_state)) + file_fd = os.open( + parts[-1], + os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC, + dir_fd=directory_fd, + ) + opened_fds.append(file_fd) + file_state = os.fstat(file_fd) + if not stat.S_ISREG(file_state.st_mode) or file_state.st_nlink != 1: + raise PreflightError("file-binding") + fd_states.append((file_fd, file_state)) + namespace_entries.append((directory_fd, parts[-1], file_state)) + + digest = hashlib.sha256() + size_bytes = 0 + chunks: list[bytes] | None = [] if capture_payload else None + while chunk := os.read(file_fd, 64 * 1024): + digest.update(chunk) + size_bytes += len(chunk) + if chunks is not None: + chunks.append(chunk) + + for file_descriptor, original_state in fd_states: + if not _same_file_state(original_state, os.fstat(file_descriptor)): + raise PreflightError("file-binding") + for parent_fd, name, original_state in namespace_entries: + current_state = os.stat(name, dir_fd=parent_fd, follow_symlinks=False) + if not _same_file_state(original_state, current_state): + raise PreflightError("file-binding") + if size_bytes != file_state.st_size: + raise PreflightError("file-binding") + return _BoundFile( + sha256=digest.hexdigest(), + size_bytes=size_bytes, + payload=b"".join(chunks) if chunks is not None else None, + ) + except PreflightError: + raise + except ValueError as exc: + raise PreflightError("path-escape") from exc + except OSError as exc: + reason = ( + "path-escape" + if exc.errno in {errno.ELOOP, errno.ENOTDIR} + else "file-binding" + ) + raise PreflightError(reason) from exc + finally: + for opened_fd in reversed(opened_fds): + try: + os.close(opened_fd) + except OSError: + pass def _verify(root: Path, path_value: Any, expected_hash: Any, size: Any | None) -> None: - path = _resolve(root, path_value) expected = _sha(expected_hash, "invalid-sha256") - if not path.is_file(): - raise PreflightError("file-unavailable") - if size is not None and path.stat().st_size != _positive_int(size, "invalid-size"): + expected_size = None if size is None else _positive_int(size, "invalid-size") + bound = _read_bound_regular_file(root, path_value, capture_payload=False) + if expected_size is not None and bound.size_bytes != expected_size: raise PreflightError("size-mismatch") - if file_sha256(path) != expected: + if bound.sha256 != expected: raise PreflightError("hash-mismatch") -def _validate_approval(value: Any) -> bool: - if value is None: - return False - approval = _exact(value, APPROVAL_KEYS, "approval-schema") - if approval["decision"] != "approved": - return False - timestamp = approval["timestamp"] - if not isinstance(timestamp, str): - raise PreflightError("approval-timestamp") +def _strict_json_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + value: dict[str, Any] = {} + for key, item in pairs: + if key in value: + raise PreflightError("owner-acceptance-record") + value[key] = item + return value + + +def _reject_json_constant(_value: str) -> None: + raise PreflightError("owner-acceptance-record") + + +def _bound_json( + root: Path, link: Mapping[str, Any], link_keys: set[str] +) -> Mapping[str, Any]: + value = _exact(link, link_keys, "owner-acceptance-link") + expected_sha256 = _sha(value["sha256"], "owner-acceptance-hash") + bound = _read_bound_regular_file(root, value["path"], capture_payload=True) + if bound.sha256 != expected_sha256: + raise PreflightError("owner-acceptance-hash") + if bound.payload is None: + raise PreflightError("owner-acceptance-record") try: - parsed = datetime.fromisoformat(timestamp) - except ValueError as exc: - raise PreflightError("approval-timestamp") from exc - if parsed.tzinfo is None: - raise PreflightError("approval-timestamp") + loaded = json.loads( + bound.payload, + object_pairs_hook=_strict_json_object, + parse_constant=_reject_json_constant, + ) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise PreflightError("owner-acceptance-record") from exc + if not isinstance(loaded, Mapping): + raise PreflightError("owner-acceptance-record") + return loaded + + +def _validate_owner_package_row(row: Any) -> None: + value = _exact(row, OWNER_PACKAGE_REFERENCE_KEYS, "owner-package-reference") + alias = value["neutral_alias"] + if ( + not isinstance(alias, str) + or re.fullmatch(r"[a-z][a-z0-9-]{0,63}", alias) is None + ): + raise PreflightError("owner-package-reference") + for key in OWNER_PACKAGE_REFERENCE_KEYS - {"neutral_alias"}: + _sha(value[key], "owner-package-reference") + + +def _validate_acceptance_row(row: Any) -> None: + value = _exact( + row, + REFERENCE_ACCEPTANCE_ROW_KEYS, + "owner-acceptance-reference", + ) + _validate_owner_package_row( + {key: value[key] for key in OWNER_PACKAGE_REFERENCE_KEYS} + ) + for key in ( + "fragment_expected_cluster_count", + "full_recording_expected_cluster_count", + ): + count = value[key] + if type(count) is not int or not 2 <= count <= 4: + raise PreflightError("owner-acceptance-reference") + for key in ( + "fragment_review_complete", + "full_recording_count_review_complete", + "reference_accepted", + ): + if value[key] is not True: + raise PreflightError("owner-acceptance-reference") + + +def _validate_reference_owner_acceptance( + *, + root: Path, + basis: Mapping[str, Any], + expected_package_rows: Sequence[Mapping[str, Any]], + expected_acceptance_rows: Sequence[Mapping[str, Any]], +) -> bool: + try: + package_link = _exact( + basis["owner_package"], + OWNER_PACKAGE_LINK_KEYS, + "owner-package-link", + ) + package = _exact( + _bound_json(root, package_link, OWNER_PACKAGE_LINK_KEYS), + OWNER_PACKAGE_KEYS, + "owner-package-schema", + ) + if package["schema"] != OWNER_PACKAGE_SCHEMA: + raise PreflightError("owner-package-schema") + package_generated_at = _aware_datetime(package["generated_at"]) + if package["generated_at"] != package_link["generated_at"]: + raise PreflightError("owner-package-timestamp") + package_rows = package["references"] + if not isinstance(package_rows, list) or len(package_rows) != 3: + raise PreflightError("owner-package-references") + for row in package_rows: + _validate_owner_package_row(row) + if package_rows != list(expected_package_rows): + raise PreflightError("owner-package-binding") + + approval_link = _exact( + basis["approval"], APPROVAL_LINK_KEYS, "owner-acceptance-link" + ) + acceptance = _exact( + _bound_json(root, approval_link, APPROVAL_LINK_KEYS), + REFERENCE_ACCEPTANCE_KEYS, + "owner-acceptance-schema", + ) + if acceptance["schema"] != REFERENCE_ACCEPTANCE_SCHEMA: + raise PreflightError("owner-acceptance-schema") + for key in ("actor", "status", "decision", "timestamp"): + if acceptance[key] != approval_link[key]: + raise PreflightError("owner-acceptance-link-mismatch") + if ( + acceptance["actor"] != "owner" + or acceptance["status"] != "approved" + or acceptance["decision"] != "approved" + ): + raise PreflightError("owner-acceptance-decision") + accepted_at = _aware_datetime(acceptance["timestamp"]) + if accepted_at <= package_generated_at: + raise PreflightError("owner-acceptance-timestamp") + if ( + _sha(acceptance["owner_package_sha256"], "owner-package-binding") + != package_link["sha256"] + ): + raise PreflightError("owner-package-binding") + acceptance_rows = acceptance["references"] + if not isinstance(acceptance_rows, list) or len(acceptance_rows) != 3: + raise PreflightError("owner-acceptance-references") + for row in acceptance_rows: + _validate_acceptance_row(row) + if acceptance_rows != list(expected_acceptance_rows): + raise PreflightError("owner-acceptance-binding") + step3 = _exact(acceptance["step3"], STEP3_KEYS, "owner-acceptance-step3") + if any(step3[key] is not True for key in STEP3_KEYS): + raise PreflightError("owner-acceptance-step3") + except (KeyError, OSError, TypeError, PreflightError): + return False return True @@ -457,10 +749,12 @@ def validate_manifest( raise PreflightError("legacy-identity") raise PreflightError("legacy-root") - basis = _exact(value["fresh_basis"], FRESH_BASIS_KEYS, "fresh-basis-schema") - if basis["marker"] != FRESH_BASIS_MARKER: + basis = value["fresh_basis"] + if not isinstance(basis, Mapping): + raise PreflightError("fresh-basis-schema") + if basis.get("marker") != FRESH_BASIS_MARKER: raise PreflightError("fresh-basis-marker") - approved = _validate_approval(basis["approval"]) + basis_schema_valid = set(basis) == FRESH_BASIS_KEYS source = _exact(value["source"], SOURCE_KEYS, "source-schema") source_commit = _commit(source["commit"], "source-commit") if set(environment) != {"commit", "clean"}: @@ -486,8 +780,18 @@ def validate_manifest( raise PreflightError("source-hashes-not-unique") fragment_hashes = [] fragment_sidecars = [] + package_rows = [] + acceptance_rows = [] + aliases = [] for item in inputs: entry = _exact(item, INPUT_KEYS, "input-schema") + alias = entry["neutral_alias"] + if ( + not isinstance(alias, str) + or re.fullmatch(r"[a-z][a-z0-9-]{0,63}", alias) is None + ): + raise PreflightError("neutral-alias") + aliases.append(alias) if verify_files: _verify( root, @@ -522,11 +826,40 @@ def validate_manifest( record["reference_sha256"], None, ) + package_row = { + "neutral_alias": alias, + "canonical_fragment_audio_sha256": entry["fragment"]["sha256"], + "final_reference_sha256": entry["fragment"]["reference_sha256"], + "full_recording_sha256": entry["full_recording"]["sha256"], + } + package_rows.append(package_row) + acceptance_rows.append( + { + **package_row, + "fragment_expected_cluster_count": entry["fragment"][ + "expected_cluster_count" + ], + "full_recording_expected_cluster_count": entry["full_recording"][ + "expected_cluster_count" + ], + "fragment_review_complete": True, + "full_recording_count_review_complete": True, + "reference_accepted": True, + } + ) + if len(set(aliases)) != 3: + raise PreflightError("neutral-alias") _validate_schedule(value["schedule"], fragment_hashes, fragment_sidecars) _positive_int(value["rss_sample_interval_ms"], "rss-sample-interval") + approved = basis_schema_valid and _validate_reference_owner_acceptance( + root=root, + basis=basis, + expected_package_rows=package_rows, + expected_acceptance_rows=acceptance_rows, + ) return PreflightOutcome( state="valid" if approved else "needs-user", - reason=None if approved else "owner-approval-required", + reason=None if approved else "reference-owner-acceptance", fragment_count=3, full_recording_count=3, source_commit_sha256=hashlib.sha256(source_commit.encode("ascii")).hexdigest(), diff --git a/tests/test_streaming_sortformer_experiment.py b/tests/test_streaming_sortformer_experiment.py index 93d15df..071d63d 100644 --- a/tests/test_streaming_sortformer_experiment.py +++ b/tests/test_streaming_sortformer_experiment.py @@ -2,6 +2,7 @@ import copy import ctypes import hashlib import json +import os import tomllib from datetime import UTC, datetime from pathlib import Path @@ -32,6 +33,14 @@ def _write(root: Path, name: str, payload: bytes) -> dict[str, object]: } +def _write_json(root: Path, name: str, value: object) -> dict[str, object]: + return _write( + root, + name, + (json.dumps(value, ensure_ascii=True, sort_keys=True) + "\n").encode(), + ) + + def _record( root: Path, prefix: str, *, duration: float, reference: bool ) -> dict[str, object]: @@ -61,6 +70,7 @@ def _manifest(root: Path) -> dict[str, object]: source = _write(root, f"sources/{index}.wav", f"source-{index}".encode()) inputs.append( { + "neutral_alias": f"control-{index + 1}", "source_path": source["path"], "source_sha256": source["sha256"], "source_size_bytes": source["size_bytes"], @@ -72,13 +82,74 @@ def _manifest(root: Path) -> dict[str, object]: ), } ) + package_generated_at = "2026-09-01T08:00:00+00:00" + package_rows = [ + { + "neutral_alias": item["neutral_alias"], + "canonical_fragment_audio_sha256": item["fragment"]["sha256"], + "final_reference_sha256": item["fragment"]["reference_sha256"], + "full_recording_sha256": item["full_recording"]["sha256"], + } + for item in inputs + ] + package = _write_json( + root, + "owner/listening-package.json", + { + "schema": "local-transcriber.owner-listening-package.v1", + "generated_at": package_generated_at, + "references": package_rows, + }, + ) + approval_timestamp = "2026-09-01T09:00:00+00:00" + acceptance = { + "schema": "local-transcriber.reference-owner-acceptance.v1", + "actor": "owner", + "status": "approved", + "decision": "approved", + "timestamp": approval_timestamp, + "owner_package_sha256": package["sha256"], + "references": [ + { + **package_row, + "fragment_expected_cluster_count": item["fragment"][ + "expected_cluster_count" + ], + "full_recording_expected_cluster_count": item["full_recording"][ + "expected_cluster_count" + ], + "fragment_review_complete": True, + "full_recording_count_review_complete": True, + "reference_accepted": True, + } + for item, package_row in zip(inputs, package_rows, strict=True) + ], + "step3": { + "recording_use_approved": True, + "inputs_and_counts_approved": True, + "full_recordings_available": True, + "final_hearing_gate_committed": True, + }, + } + acceptance_file = _write_json( + root, "owner/reference-acceptance.json", acceptance + ) return { "schema": "local-transcriber.streaming-sortformer-experiment.v1", "fresh_basis": { "marker": "issue-46-fresh-owner-reviewed-basis", + "owner_package": { + "path": package["path"], + "sha256": package["sha256"], + "generated_at": package_generated_at, + }, "approval": { + "path": acceptance_file["path"], + "sha256": acceptance_file["sha256"], + "actor": "owner", + "status": "approved", "decision": "approved", - "timestamp": "2026-09-01T09:00:00+00:00", + "timestamp": approval_timestamp, }, }, "source": {"commit": "a" * 40}, @@ -194,6 +265,232 @@ def _environment() -> dict[str, object]: return {"commit": "a" * 40, "clean": True} +def _read_linked_json(root: Path, link: dict[str, object]) -> dict[str, object]: + return json.loads((root / str(link["path"])).read_text(encoding="utf-8")) + + +def _rewrite_linked_json( + root: Path, link: dict[str, object], value: object +) -> None: + payload = (json.dumps(value, ensure_ascii=True, sort_keys=True) + "\n").encode() + (root / str(link["path"])).write_bytes(payload) + link["sha256"] = hashlib.sha256(payload).hexdigest() + + +def _rewrite_linked_bytes( + root: Path, link: dict[str, object], payload: bytes +) -> None: + (root / str(link["path"])).write_bytes(payload) + link["sha256"] = hashlib.sha256(payload).hexdigest() + + +def _media_path(manifest: dict[str, object], binding: str) -> str: + item = manifest["inputs"][0] + paths = { + "source": item["source_path"], + "fragment": item["fragment"]["path"], + "fragment-sidecar": item["fragment"]["sidecar_path"], + "reference": item["fragment"]["reference_path"], + "full-recording": item["full_recording"]["path"], + "full-sidecar": item["full_recording"]["sidecar_path"], + } + return str(paths[binding]) + + +def _set_binding_path( + manifest: dict[str, object], binding: str, value: str +) -> None: + if binding in {"owner-package", "owner-acceptance"}: + link = "owner_package" if binding == "owner-package" else "approval" + manifest["fresh_basis"][link]["path"] = value + return + item = manifest["inputs"][0] + if binding == "source": + item["source_path"] = value + elif binding == "fragment": + item["fragment"]["path"] = value + elif binding == "fragment-sidecar": + item["fragment"]["sidecar_path"] = value + elif binding == "reference": + item["fragment"]["reference_path"] = value + elif binding == "full-recording": + item["full_recording"]["path"] = value + elif binding == "full-sidecar": + item["full_recording"]["sidecar_path"] = value + else: + raise AssertionError(f"unknown binding: {binding}") + + +def _mutate_owner_acceptance( + manifest: dict[str, object], root: Path, case: str +) -> None: + basis = manifest["fresh_basis"] + approval_link = basis["approval"] + package_link = basis["owner_package"] + acceptance = _read_linked_json(root, approval_link) + package = _read_linked_json(root, package_link) + + def write_acceptance() -> None: + _rewrite_linked_json(root, approval_link, acceptance) + + def write_package(*, sync_acceptance: bool = True) -> None: + _rewrite_linked_json(root, package_link, package) + if sync_acceptance: + acceptance["owner_package_sha256"] = package_link["sha256"] + write_acceptance() + + if case == "approval-missing": + del basis["approval"] + elif case == "approval-null": + basis["approval"] = None + elif case == "approval-link-unknown": + approval_link["count_review_status"] = "approved" + elif case == "record-missing": + (root / str(approval_link["path"])).unlink() + elif case == "record-path-missing": + approval_link["path"] = "owner/missing-acceptance.json" + elif case == "record-hash-mismatch": + approval_link["sha256"] = "f" * 64 + elif case == "record-json-invalid": + payload = b"{\n" + (root / str(approval_link["path"])).write_bytes(payload) + approval_link["sha256"] = hashlib.sha256(payload).hexdigest() + elif case == "record-unknown-key": + acceptance["count_review_status"] = "approved" + write_acceptance() + elif case == "record-count-review-schema": + acceptance = { + "schema": "local-transcriber.reference-count-review.v1", + "status": "approved", + } + write_acceptance() + elif case in {"actor", "status", "decision"}: + replacement = { + "actor": "automation", + "status": "draft", + "decision": "rejected", + }[case] + approval_link[case] = replacement + acceptance[case] = replacement + write_acceptance() + elif case == "timestamp-naive": + approval_link["timestamp"] = "2026-09-01T09:00:00" + acceptance["timestamp"] = approval_link["timestamp"] + write_acceptance() + elif case == "timestamp-early": + approval_link["timestamp"] = package_link["generated_at"] + acceptance["timestamp"] = approval_link["timestamp"] + write_acceptance() + elif case == "timestamp-link-mismatch": + acceptance["timestamp"] = "2026-09-01T10:00:00+00:00" + write_acceptance() + elif case == "package-missing": + del basis["owner_package"] + elif case == "package-link-unknown": + package_link["status"] = "approved" + elif case == "package-file-missing": + (root / str(package_link["path"])).unlink() + elif case == "package-hash-mismatch": + package_link["sha256"] = "e" * 64 + elif case == "package-timestamp-mismatch": + package_link["generated_at"] = "2026-09-01T07:00:00+00:00" + elif case == "package-unknown-key": + package["status"] = "approved" + write_package() + elif case == "package-row-binding": + package["references"][0]["final_reference_sha256"] = "d" * 64 + write_package() + elif case == "package-row-omitted": + package["references"].pop() + write_package() + elif case == "package-row-duplicate": + package["references"][2] = copy.deepcopy(package["references"][1]) + write_package() + elif case == "acceptance-row-omitted": + acceptance["references"].pop() + write_acceptance() + elif case == "acceptance-row-duplicate": + acceptance["references"][2] = copy.deepcopy(acceptance["references"][1]) + write_acceptance() + elif case in { + "neutral_alias", + "canonical_fragment_audio_sha256", + "final_reference_sha256", + "full_recording_sha256", + }: + acceptance["references"][0][case] = ( + "different-control" if case == "neutral_alias" else "c" * 64 + ) + write_acceptance() + elif case in { + "fragment_expected_cluster_count", + "full_recording_expected_cluster_count", + }: + acceptance["references"][0][case] = 4 + write_acceptance() + elif case == "fragment-count-float": + acceptance["references"][0]["fragment_expected_cluster_count"] = 2.0 + write_acceptance() + elif case == "fragment-count-bool": + acceptance["references"][0]["fragment_expected_cluster_count"] = True + write_acceptance() + elif case == "fragment-count-string": + acceptance["references"][0]["fragment_expected_cluster_count"] = "2" + write_acceptance() + elif case == "full-count-float": + acceptance["references"][0][ + "full_recording_expected_cluster_count" + ] = 2.0 + write_acceptance() + elif case == "package-alias-type": + package["references"][0]["neutral_alias"] = 1 + write_package() + elif case == "package-hash-type": + package["references"][0]["canonical_fragment_audio_sha256"] = 1 + write_package() + elif case == "acceptance-hash-type": + acceptance["references"][0]["canonical_fragment_audio_sha256"] = 1 + write_acceptance() + elif case in { + "fragment_review_complete", + "full_recording_count_review_complete", + "reference_accepted", + }: + acceptance["references"][0][case] = False + write_acceptance() + elif case == "fragment-review-int": + acceptance["references"][0]["fragment_review_complete"] = 1 + write_acceptance() + elif case == "full-review-int": + acceptance["references"][0][ + "full_recording_count_review_complete" + ] = 1 + write_acceptance() + elif case == "reference-accepted-int": + acceptance["references"][0]["reference_accepted"] = 1 + write_acceptance() + elif case == "acceptance-row-unknown": + acceptance["references"][0]["status"] = "approved" + write_acceptance() + elif case == "owner-package-binding": + acceptance["owner_package_sha256"] = "b" * 64 + write_acceptance() + elif case == "step3-int": + acceptance["step3"]["recording_use_approved"] = 1 + write_acceptance() + elif case.startswith("step3-"): + acceptance["step3"][case.removeprefix("step3-")] = False + write_acceptance() + elif case == "step3-missing": + del acceptance["step3"]["recording_use_approved"] + write_acceptance() + elif case == "step3-unknown": + acceptance["step3"]["count_review_approved"] = True + write_acceptance() + else: + raise AssertionError(f"unknown case: {case}") + + def test_preflight_validates_fresh_exact_schema_before_loading(tmp_path): manifest = _manifest(tmp_path) calls = [] @@ -213,6 +510,605 @@ def test_preflight_validates_fresh_exact_schema_before_loading(tmp_path): assert calls == [] +def test_complete_reference_owner_acceptance_is_required_before_loading(tmp_path): + manifest = _manifest(tmp_path) + calls = [] + + outcome = experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert outcome.state == "valid" + assert calls == ["w0", "candidate"] + + +@pytest.mark.parametrize( + "case", + [ + "approval-missing", + "approval-null", + "approval-link-unknown", + "record-missing", + "record-path-missing", + "record-hash-mismatch", + "record-json-invalid", + "record-unknown-key", + "record-count-review-schema", + "actor", + "status", + "decision", + "timestamp-naive", + "timestamp-early", + "timestamp-link-mismatch", + "package-missing", + "package-link-unknown", + "package-file-missing", + "package-hash-mismatch", + "package-timestamp-mismatch", + "package-unknown-key", + "package-row-binding", + "package-row-omitted", + "package-row-duplicate", + "acceptance-row-omitted", + "acceptance-row-duplicate", + "neutral_alias", + "canonical_fragment_audio_sha256", + "final_reference_sha256", + "full_recording_sha256", + "fragment_expected_cluster_count", + "full_recording_expected_cluster_count", + "fragment-count-float", + "fragment-count-bool", + "fragment-count-string", + "full-count-float", + "package-alias-type", + "package-hash-type", + "acceptance-hash-type", + "fragment_review_complete", + "full_recording_count_review_complete", + "reference_accepted", + "fragment-review-int", + "full-review-int", + "reference-accepted-int", + "acceptance-row-unknown", + "owner-package-binding", + "step3-recording_use_approved", + "step3-inputs_and_counts_approved", + "step3-full_recordings_available", + "step3-final_hearing_gate_committed", + "step3-int", + "step3-missing", + "step3-unknown", + ], +) +def test_reference_owner_acceptance_failures_need_user_before_loaders( + tmp_path, case +): + manifest = _manifest(tmp_path) + _mutate_owner_acceptance(manifest, tmp_path, case) + calls = [] + + outcome = experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert outcome.state == "needs-user" + assert outcome.reason == "reference-owner-acceptance" + assert calls == [] + + +@pytest.mark.parametrize( + ("record", "original", "replacement"), + [ + ( + "approval", + b'"status": "approved"', + b'"status": "draft", "status": "approved"', + ), + ( + "approval", + b'"reference_accepted": true', + b'"reference_accepted": false, "reference_accepted": true', + ), + ( + "approval", + b'"recording_use_approved": true', + b'"recording_use_approved": false, "recording_use_approved": true', + ), + ( + "owner_package", + b'"neutral_alias": "control-1"', + b'"neutral_alias": "other", "neutral_alias": "control-1"', + ), + ( + "approval", + b'"fragment_expected_cluster_count": 2', + b'"fragment_expected_cluster_count": NaN', + ), + ], +) +def test_bound_owner_json_rejects_ambiguous_json_before_loaders( + tmp_path, record, original, replacement +): + manifest = _manifest(tmp_path) + link = manifest["fresh_basis"][record] + path = tmp_path / str(link["path"]) + payload = path.read_bytes().replace(original, replacement, 1) + assert payload != path.read_bytes() + _rewrite_linked_bytes(tmp_path, link, payload) + if record == "owner_package": + acceptance_link = manifest["fresh_basis"]["approval"] + acceptance = _read_linked_json(tmp_path, acceptance_link) + acceptance["owner_package_sha256"] = link["sha256"] + _rewrite_linked_json(tmp_path, acceptance_link, acceptance) + calls = [] + + outcome = experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert outcome.state == "needs-user" + assert outcome.reason == "reference-owner-acceptance" + assert calls == [] + + +@pytest.mark.parametrize("symlink_kind", ["final", "intermediate"]) +def test_bound_owner_json_rejects_symlinks_before_loaders( + tmp_path, symlink_kind +): + manifest = _manifest(tmp_path) + if symlink_kind == "final": + link = manifest["fresh_basis"]["approval"] + path = tmp_path / str(link["path"]) + target = path.with_name("acceptance-target.json") + path.rename(target) + path.symlink_to(target.name) + else: + owner = tmp_path / "owner" + target = tmp_path / "owner-target" + owner.rename(target) + owner.symlink_to(target.name, target_is_directory=True) + calls = [] + + outcome = experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert outcome.state == "needs-user" + assert outcome.reason == "reference-owner-acceptance" + assert calls == [] + + +def test_bound_owner_json_cannot_hash_one_version_and_parse_another( + tmp_path, monkeypatch +): + manifest = _manifest(tmp_path) + link = manifest["fresh_basis"]["approval"] + path = tmp_path / str(link["path"]) + approved_payload = path.read_bytes() + rejected_payload = approved_payload.replace( + b'"status": "approved"', b'"status": "draft"', 1 + ) + path.write_bytes(rejected_payload) + link["sha256"] = hashlib.sha256(rejected_payload).hexdigest() + original_file_sha256 = experiment.file_sha256 + replaced = False + + def replace_after_hash(candidate): + nonlocal replaced + digest = original_file_sha256(candidate) + if candidate == path and not replaced: + path.write_bytes(approved_payload) + replaced = True + return digest + + monkeypatch.setattr(experiment, "file_sha256", replace_after_hash) + calls = [] + + outcome = experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert outcome.state == "needs-user" + assert outcome.reason == "reference-owner-acceptance" + assert calls == [] + + +def test_bound_owner_json_rejects_path_replacement_during_held_read( + tmp_path, monkeypatch +): + manifest = _manifest(tmp_path) + link = manifest["fresh_basis"]["approval"] + path = tmp_path / str(link["path"]) + target_inode = path.stat().st_ino + replacement = path.with_name("replacement.json") + payload = path.read_bytes() + original_read = os.read + replaced = False + + def replace_path_after_read(file_descriptor, count): + nonlocal replaced + chunk = original_read(file_descriptor, count) + if os.fstat(file_descriptor).st_ino == target_inode and not replaced: + replacement.write_bytes(payload) + os.replace(replacement, path) + replaced = True + return chunk + + monkeypatch.setattr(experiment.os, "read", replace_path_after_read) + calls = [] + + outcome = experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert replaced is True + assert outcome.state == "needs-user" + assert outcome.reason == "reference-owner-acceptance" + assert calls == [] + + +def test_bound_owner_json_rejects_in_place_mutation_during_held_read( + tmp_path, monkeypatch +): + manifest = _manifest(tmp_path) + link = manifest["fresh_basis"]["approval"] + path = tmp_path / str(link["path"]) + json_size = len(path.read_bytes().rstrip(b"\n")) + payload = path.read_bytes().rstrip(b"\n") + (b" " * 70_000) + b"\n" + _rewrite_linked_bytes(tmp_path, link, payload) + target_inode = path.stat().st_ino + original_read = os.read + mutated = False + + def mutate_after_first_chunk(file_descriptor, count): + nonlocal mutated + chunk = original_read(file_descriptor, count) + if os.fstat(file_descriptor).st_ino == target_inode and not mutated: + writer = os.open(path, os.O_WRONLY) + try: + os.pwrite(writer, b"\t", json_size + 66_000) + os.fsync(writer) + finally: + os.close(writer) + mutated = True + return chunk + + monkeypatch.setattr(experiment.os, "read", mutate_after_first_chunk) + calls = [] + + outcome = experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert mutated is True + assert outcome.state == "needs-user" + assert outcome.reason == "reference-owner-acceptance" + assert calls == [] + + +def test_bound_owner_json_rejects_intermediate_directory_swap_during_read( + tmp_path, monkeypatch +): + manifest = _manifest(tmp_path) + link = manifest["fresh_basis"]["owner_package"] + path = tmp_path / str(link["path"]) + target_inode = path.stat().st_ino + parent = path.parent + moved_parent = parent.with_name("owner-original") + original_read = os.read + swapped = False + + def swap_directory_after_read(file_descriptor, count): + nonlocal swapped + chunk = original_read(file_descriptor, count) + if os.fstat(file_descriptor).st_ino == target_inode and not swapped: + parent.rename(moved_parent) + parent.mkdir() + (parent / path.name).write_bytes(b"{}\n") + swapped = True + return chunk + + monkeypatch.setattr(experiment.os, "read", swap_directory_after_read) + calls = [] + + outcome = experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert swapped is True + assert outcome.state == "needs-user" + assert outcome.reason == "reference-owner-acceptance" + assert calls == [] + + +@pytest.mark.parametrize( + "binding", + [ + "source", + "fragment", + "fragment-sidecar", + "reference", + "full-recording", + "full-sidecar", + ], +) +@pytest.mark.parametrize("symlink_kind", ["final", "intermediate"]) +def test_manifest_file_bindings_reject_internal_symlinks_before_loaders( + tmp_path, binding, symlink_kind +): + manifest = _manifest(tmp_path) + path = tmp_path / _media_path(manifest, binding) + if symlink_kind == "final": + target = path.with_name(f"{path.name}.target") + path.rename(target) + path.symlink_to(target.name) + else: + parent = path.parent + target = parent.with_name(f"{parent.name}-target") + parent.rename(target) + parent.symlink_to(target.name, target_is_directory=True) + calls = [] + + with pytest.raises(experiment.PreflightError) as raised: + experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert str(tmp_path) not in str(raised.value) + assert calls == [] + + +def test_manifest_file_binding_rejects_final_replacement_during_read( + tmp_path, monkeypatch +): + manifest = _manifest(tmp_path) + path = tmp_path / _media_path(manifest, "fragment") + target_inode = path.stat().st_ino + replacement = path.with_name("replacement.wav") + original_read = os.read + replaced = False + + def replace_path_after_read(file_descriptor, count): + nonlocal replaced + chunk = original_read(file_descriptor, count) + if os.fstat(file_descriptor).st_ino == target_inode and not replaced: + replacement.write_bytes(b"different-fragment-bytes") + os.replace(replacement, path) + replaced = True + return chunk + + monkeypatch.setattr(experiment.os, "read", replace_path_after_read) + calls = [] + + with pytest.raises(experiment.PreflightError): + experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert replaced is True + assert path.read_bytes() == b"different-fragment-bytes" + assert hashlib.sha256(path.read_bytes()).hexdigest() != manifest["inputs"][0][ + "fragment" + ]["sha256"] + assert calls == [] + + +def test_manifest_file_binding_rejects_intermediate_directory_swap_during_read( + tmp_path, monkeypatch +): + manifest = _manifest(tmp_path) + path = tmp_path / _media_path(manifest, "fragment") + target_inode = path.stat().st_ino + parent = path.parent + moved_parent = parent.with_name("media-original") + original_read = os.read + swapped = False + + def swap_directory_after_read(file_descriptor, count): + nonlocal swapped + chunk = original_read(file_descriptor, count) + if os.fstat(file_descriptor).st_ino == target_inode and not swapped: + parent.rename(moved_parent) + parent.mkdir() + (parent / path.name).write_bytes(b"replacement-fragment") + swapped = True + return chunk + + monkeypatch.setattr(experiment.os, "read", swap_directory_after_read) + calls = [] + + with pytest.raises(experiment.PreflightError): + experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert swapped is True + assert calls == [] + + +def test_manifest_file_binding_rejects_in_place_mutation_during_read( + tmp_path, monkeypatch +): + manifest = _manifest(tmp_path) + path = tmp_path / _media_path(manifest, "fragment") + target_inode = path.stat().st_ino + original_read = os.read + mutated = False + + def mutate_file_after_read(file_descriptor, count): + nonlocal mutated + chunk = original_read(file_descriptor, count) + if os.fstat(file_descriptor).st_ino == target_inode and not mutated: + path.write_bytes(b"mutated-fragment-with-different-size") + mutated = True + return chunk + + monkeypatch.setattr(experiment.os, "read", mutate_file_after_read) + calls = [] + + with pytest.raises(experiment.PreflightError): + experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert mutated is True + assert calls == [] + + +def test_bound_file_verification_closes_descriptors_on_success_and_failure( + tmp_path +): + proc_fds = Path("/proc/self/fd") + if not proc_fds.is_dir(): + pytest.skip("fd accounting requires procfs") + + manifest = _manifest(tmp_path) + before_success = len(list(proc_fds.iterdir())) + for _ in range(20): + assert experiment.validate_manifest( + manifest, root=tmp_path, environment=_environment() + ).state == "valid" + assert len(list(proc_fds.iterdir())) == before_success + + failed_root = tmp_path / "failed" + failed_manifest = _manifest(failed_root) + path = failed_root / _media_path(failed_manifest, "fragment") + target = path.with_name("fragment-target.wav") + path.rename(target) + path.symlink_to(target.name) + before_failure = len(list(proc_fds.iterdir())) + for _ in range(20): + with pytest.raises(experiment.PreflightError): + experiment.validate_manifest( + failed_manifest, + root=failed_root, + environment=_environment(), + ) + assert len(list(proc_fds.iterdir())) == before_failure + + +@pytest.mark.parametrize( + "binding", + [ + "source", + "fragment", + "fragment-sidecar", + "reference", + "full-recording", + "full-sidecar", + "owner-package", + "owner-acceptance", + ], +) +def test_manifest_file_bindings_reject_nul_paths_before_loaders( + tmp_path, binding +): + manifest = _manifest(tmp_path) + _set_binding_path(manifest, binding, "private\x00binding") + calls = [] + + if binding.startswith("owner-"): + outcome = experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + assert outcome.state == "needs-user" + assert outcome.reason == "reference-owner-acceptance" + else: + with pytest.raises(experiment.PreflightError, match="path-escape"): + experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert calls == [] + + +def test_manifest_file_binding_redacts_path_layer_value_error( + tmp_path, monkeypatch +): + manifest = _manifest(tmp_path) + original_open = os.open + + def reject_component(path, flags, *, dir_fd=None): + if path == "sources": + raise ValueError("private path detail") + if dir_fd is None: + return original_open(path, flags) + return original_open(path, flags, dir_fd=dir_fd) + + monkeypatch.setattr(experiment.os, "open", reject_component) + calls = [] + + with pytest.raises(experiment.PreflightError) as raised: + experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert str(raised.value) == "path-escape" + assert "private" not in str(raised.value) + assert calls == [] + + def test_missing_owner_approval_is_needs_user_and_never_loads(tmp_path): manifest = _manifest(tmp_path) manifest["fresh_basis"]["approval"] = None @@ -227,7 +1123,7 @@ def test_missing_owner_approval_is_needs_user_and_never_loads(tmp_path): ) assert outcome.state == "needs-user" - assert outcome.reason == "owner-approval-required" + assert outcome.reason == "reference-owner-acceptance" assert calls == [] @@ -244,6 +1140,10 @@ def test_missing_owner_approval_is_needs_user_and_never_loads(tmp_path): lambda value: value["inputs"][0]["full_recording"].update( expected_cluster_count=5 ), + lambda value: value["inputs"][0].update(neutral_alias=""), + lambda value: value["inputs"][1].update( + neutral_alias=value["inputs"][0]["neutral_alias"] + ), lambda value: value["source"].update(commit="b" * 40), ], ) @@ -1307,6 +2207,35 @@ def test_invalid_cli_output_has_one_trailing_newline(tmp_path, capsys): assert not output.endswith("\n\n") +def test_validate_only_cli_redacts_nul_path_failure( + tmp_path, capsys, monkeypatch +): + manifest = _manifest(tmp_path) + manifest["inputs"][0]["source_path"] = "private\x00source.wav" + manifest_path = tmp_path / "manifest.json" + manifest_path.write_text(json.dumps(manifest), encoding="utf-8") + monkeypatch.setattr( + experiment, + "git_environment_probe", + lambda _repository_root: _environment(), + ) + + with pytest.raises(SystemExit) as raised: + experiment.main(["--manifest", str(manifest_path), "--validate-only"]) + + captured = capsys.readouterr() + assert raised.value.code == 2 + assert json.loads(captured.out) == { + "reason": "path-escape", + "state": "invalid", + } + assert captured.err == "" + assert "Traceback" not in captured.out + assert "\x00" not in captured.out + assert "private" not in captured.out + assert str(tmp_path) not in captured.out + + def test_child_process_request_result_seam_uses_atomic_private_files(tmp_path): request = { "semantic_id": "cell-safe", From 16e5372f2d77c6f0304099cde49d5054cd2b1666 Mon Sep 17 00:00:00 2001 From: Dmitry Dementiev Date: Fri, 4 Sep 2026 16:01:39 +0300 Subject: [PATCH 3/7] =?UTF-8?q?chore(repo):=20=D0=BD=D0=BE=D0=B2=D1=8B?= =?UTF-8?q?=D0=B5=20=D0=BC=D0=B0=D1=82=D0=B5=D1=80=D0=B8=D0=B0=D0=BB=D1=8B?= =?UTF-8?q?=20.scratch=20=D0=B8=D1=81=D0=BA=D0=BB=D1=8E=D1=87=D0=B5=D0=BD?= =?UTF-8?q?=D1=8B=20=D0=B8=D0=B7=20Git?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Зачем: - хранить локальные handoff и launcher в постоянном каталоге без риска случайного коммита - не терять временно закрепленный стенд диаризации между компьютерами Что: - весь корневой .scratch добавлен в ignore - существующие диагностические скрипты оставлены отслеживаемыми - описание локальных и переносимых артефактов приведено к новой политике Проверка: - git check-ignore -v --no-index .scratch/new-local-artifact - git ls-files .scratch - git diff --check --- .gitignore | 2 +- .scratch/diarization/README.md | 5 +++++ docs/specs/2026-08-30-cpu-diarization-experiment.md | 7 ++++--- .../2026-09-01-streaming-sortformer-cpu-experiment.md | 8 ++++---- 4 files changed, 14 insertions(+), 8 deletions(-) diff --git a/.gitignore b/.gitignore index 49d5592..5b86eab 100644 --- a/.gitignore +++ b/.gitignore @@ -7,4 +7,4 @@ dist/ .codex .qwen/ tmp -.scratch/handoffs/ +/.scratch/ diff --git a/.scratch/diarization/README.md b/.scratch/diarization/README.md index ce16722..fb6e555 100644 --- a/.scratch/diarization/README.md +++ b/.scratch/diarization/README.md @@ -5,6 +5,11 @@ Не часть пакета: они опираются на `sherpa-onnx`, которого нет в зависимостях проекта, и живут в `.scratch/`, а не в `src/`. +Код этого каталога временно закреплен в Git: его используют несколько задач на +нескольких компьютерах, пока идет исследование диаризации. Общее правило +`/.scratch/` относится к новым локальным материалам; уже отслеживаемые файлы +этого стенда продолжают версионироваться. + Результаты первого прогона описаны в [разведочном замере](../../docs/benchmarks/2026-08-12-diarization-feasibility.md). diff --git a/docs/specs/2026-08-30-cpu-diarization-experiment.md b/docs/specs/2026-08-30-cpu-diarization-experiment.md index 042dde1..ea7ea56 100644 --- a/docs/specs/2026-08-30-cpu-diarization-experiment.md +++ b/docs/specs/2026-08-30-cpu-diarization-experiment.md @@ -151,9 +151,10 @@ За пределами Git остаются manifest, записи, WAV-кеш, ASR-sidecar, опорные транскрипты, сырые интервалы, результаты, worker requests, журналы и материалы -слуховой проверки. Рабочий каталог стенда нельзя размещать в обычной -`.scratch`: репозиторий игнорирует там только служебный каталог -`.scratch/handoffs/`, не связанный с переносом эксперимента между ноутбуками. +слуховой проверки. Рабочий каталог стенда нельзя размещать в `.scratch`: этот +каталог локален и целиком игнорируется, поэтому эксперимент из него не +переносится между ноутбуками. Временно закрепленные диагностические скрипты уже +отслеживаются Git и не меняют этого правила для новых материалов. Публичный отчет содержит хеши, агрегаты, конфигурацию и выводы. В него не входят локальные пути, распознанный текст, временные интервалы, имена участников и diff --git a/docs/specs/2026-09-01-streaming-sortformer-cpu-experiment.md b/docs/specs/2026-09-01-streaming-sortformer-cpu-experiment.md index c8e7a4b..51dbacb 100644 --- a/docs/specs/2026-09-01-streaming-sortformer-cpu-experiment.md +++ b/docs/specs/2026-09-01-streaming-sortformer-cpu-experiment.md @@ -380,10 +380,10 @@ RSS выше 2 ГиБ на машине с 16 ГиБ требует отдель ## Артефакты и приватность Checkout и дерево сборки NeMo-Speech.cpp, скачанные веса и новое закрытое -хранилище размещаются вне рабочей копии local-transcriber. Обычная `.scratch` -внутри репозитория не подходит: `.gitignore` покрывает там только -`.scratch/handoffs/`. Ни один runtime/model/build artifact или закрытый вход не -попадает в Git. +хранилище размещаются вне рабочей копии local-transcriber. `.scratch` внутри +репозитория целиком игнорируется, но остается локальной частью конкретного +checkout и не подходит для закрытого переносимого хранилища. Ни один +runtime/model/build artifact или закрытый вход не попадает в Git. Новое закрытое хранилище содержит: From d71aec4ab2af829ce5b3e601f1a701d73bf7feea Mon Sep 17 00:00:00 2001 From: Dmitry Dementiev Date: Fri, 4 Sep 2026 17:48:36 +0300 Subject: [PATCH 4/7] =?UTF-8?q?fix(experiment):=20=D1=83=D1=81=D1=82=D1=80?= =?UTF-8?q?=D0=B0=D0=BD=D0=B5=D0=BD=20=D0=BF=D0=BE=D0=B2=D1=82=D0=BE=D1=80?= =?UTF-8?q?=D0=BD=D1=8B=D0=B9=20gate=20=D0=BF=D1=80=D0=B8=D0=BD=D1=8F?= =?UTF-8?q?=D1=82=D0=B8=D1=8F=20=D0=BE=D0=BF=D0=BE=D1=80=D1=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Зачем: - канонический Ryzen-прогон не должен повторно запрашивать уже данное решение владельца до порога 30%. - Что: - восстановлен короткий контракт approval и удалена лишняя схема owner-package. - разрешена парная nullable-опора для диагностик без ослабления обязательных gates. - NUL в путях отклоняется обезличенной ошибкой до загрузки моделей. - Проверка: - uv run pytest -q: 457 passed, 2 skipped. - обязательный набор эксперимента: 110 passed; целевой Ruff пройден. - независимые code и task/taste review закрыты. --- ...-01-streaming-sortformer-cpu-experiment.md | 39 +- .../streaming_sortformer_experiment.py | 410 +------- tests/test_streaming_sortformer_experiment.py | 922 +----------------- 3 files changed, 99 insertions(+), 1272 deletions(-) diff --git a/docs/specs/2026-09-01-streaming-sortformer-cpu-experiment.md b/docs/specs/2026-09-01-streaming-sortformer-cpu-experiment.md index 51dbacb..0e6f5fb 100644 --- a/docs/specs/2026-09-01-streaming-sortformer-cpu-experiment.md +++ b/docs/specs/2026-09-01-streaming-sortformer-cpu-experiment.md @@ -186,20 +186,25 @@ mono/16 kHz float32. Публичные документы получают то после единственного закрепленного преобразования, а ни один путь, ID или интервал не попадает в stdout режима `--validate-only`. -### Шаг 2. Создать ASR-sidecar и опорную разметку +### Шаг 2. Подготовить ASR-sidecar и доступную опорную разметку Действующий ASR выполняется один раз для каждой полной записи до измерений. Sidecar хранит канонические слова с временной привязкой и нормализованный текст; fragment-sidecar получается детерминированным срезом той же версии. Затем -человек прослушивает каждый фрагмент целиком, фиксирует ожидаемое число -голосовых кластеров для фрагментов и полных записей и создает опорную разметку, -достаточную для mapped purity и выбора мест прослушивания. Опорная разметка -остается приближенной и не объявляется истиной границ. +человек прослушивает каждый фрагмент целиком и фиксирует ожидаемое число +голосовых кластеров для фрагментов и полных записей. Опорная разметка фрагмента +необязательна: `reference_path` и `reference_sha256` либо заполнены вместе, либо +оба равны `null`. При наличии разметки стенд считает mapped purity и выбирает +места прослушивания; сама разметка остается приближенной и не объявляется +истиной границ. Критерий завершения: для каждого фрагмента и каждой полной записи есть подтвержденное человеком `expected_cluster_count` в диапазоне `2..4`, один -неизменяемый sidecar и SHA-256; для фрагментов также есть опорная разметка. -Числа старых контролей не копируются. +неизменяемый sidecar и SHA-256. Отсутствие опорной разметки не блокирует +измерение и не ослабляет остальные gates. Для канонического Ryzen-прогона +используются подтвержденные владельцем числа `3 / 2 / 2`; их привязывают к +источникам по хешам, а не по порядку нейтральных имен. Повторное прослушивание +для подсчета голосов не требуется. ### Шаг 3. Пройти пользовательский gate @@ -216,6 +221,10 @@ fragment-sidecar получается детерминированным сре Критерий завершения: в закрытом manifest есть timestamped decision `approved`. Без него исход `needs-user`; стенд не загружает модели. +В каноническом Ryzen-прогоне решением `approved` служит датированный handoff +владельца в комментарии к задаче #46. Нового подтверждения до достижения +30%-го порога не требуется. + ## Измеряемая величина и A/B Время W0 - wall вокруг одного `diarizer.process(samples)`. Время кандидата - @@ -333,9 +342,10 @@ hashes и один pair ID; рассчитаны ускорение, CPU и RSS - каноническая нумерация меток детерминирована на повторном расчете метрик; - все inputs, параметры и output artifacts имеют SHA-256. -Purity и число слов без говорящего сравниваются с новым `W0-FRESH`. Ухудшение -не дает автоматического `fail`, но обязательно добавляет все затронутые места в -слуховой gate. Старые абсолютные значения не участвуют. +Purity для фрагментов с опорной разметкой и число слов без говорящего +сравниваются с новым `W0-FRESH`. Ухудшение не дает автоматического `fail`, но +обязательно добавляет все затронутые места в слуховой gate. Старые абсолютные +значения не участвуют. Открытый диагностикой слуховой gate не прерывает шаги 5-6. Стенд накапливает затронутые места и добавляет их в материал шага 7. Состояние `needs-user` из-за @@ -353,7 +363,7 @@ Purity и число слов без говорящего сравниваютс - каждая итоговая speaker label принадлежит `1..4`; - число и суммарная длительность интервалов по каждому каналу; - число коротких интервалов, overlap ratio, unmatched words, малые/остаточные - кластеры и mapped purity относительно свежей опорной разметки; + кластеры и mapped purity для фрагментов с опорной разметкой; - wall отдельно для model inference и segment post-processing. Любой неверный shape/value/label или эффективная конфигурация, отличная от @@ -390,7 +400,7 @@ runtime/model/build artifact или закрытый вход не попада - versioned manifest со schema, clean local-transcriber source commit, source/model/runtime/build hashes и расписанием; - исходные ссылки, подготовленные WAV и recipe преобразования; -- ASR-sidecar и опорную разметку; +- ASR-sidecar и имеющуюся опорную разметку; - raw per-fragment results, intervals, word assignments и probability diagnostics; - machine/power/runtime guards и stderr/stdout logs; @@ -458,8 +468,9 @@ privacy scan. Реализация стенда фиксируется чист ## Критерии завершения эксперимента -- [ ] Свежая опора подтверждена владельцем и не наследует старые контрольные - значения. +- [ ] Свежая опора собрана в новом закрытом хранилище; подтвержденные входы, + доступная разметка и числа голосов связаны по хешам, а старые manifest, + расписание и результаты не наследуются. - [ ] Runtime, модель, сборка и оба рецепта закреплены и проверены до моделей. - [ ] Clean local-transcriber source commit сохранен в закрытом хранилище и публичном отчете, а `--validate-only` проверяет его до моделей. diff --git a/src/local_transcriber/streaming_sortformer_experiment.py b/src/local_transcriber/streaming_sortformer_experiment.py index efefe5b..ded571c 100644 --- a/src/local_transcriber/streaming_sortformer_experiment.py +++ b/src/local_transcriber/streaming_sortformer_experiment.py @@ -5,13 +5,11 @@ from __future__ import annotations import argparse import copy import ctypes -import errno import hashlib import json import math import os import re -import stat import statistics import subprocess import sys @@ -46,51 +44,10 @@ TOP_LEVEL_KEYS = { "schedule", "rss_sample_interval_ms", } -FRESH_BASIS_KEYS = {"marker", "owner_package", "approval"} -OWNER_PACKAGE_LINK_KEYS = {"path", "sha256", "generated_at"} -APPROVAL_LINK_KEYS = { - "path", - "sha256", - "actor", - "status", - "decision", - "timestamp", -} -OWNER_PACKAGE_SCHEMA = "local-transcriber.owner-listening-package.v1" -OWNER_PACKAGE_KEYS = {"schema", "generated_at", "references"} -OWNER_PACKAGE_REFERENCE_KEYS = { - "neutral_alias", - "canonical_fragment_audio_sha256", - "final_reference_sha256", - "full_recording_sha256", -} -REFERENCE_ACCEPTANCE_SCHEMA = "local-transcriber.reference-owner-acceptance.v1" -REFERENCE_ACCEPTANCE_KEYS = { - "schema", - "actor", - "status", - "decision", - "timestamp", - "owner_package_sha256", - "references", - "step3", -} -REFERENCE_ACCEPTANCE_ROW_KEYS = OWNER_PACKAGE_REFERENCE_KEYS | { - "fragment_expected_cluster_count", - "full_recording_expected_cluster_count", - "fragment_review_complete", - "full_recording_count_review_complete", - "reference_accepted", -} -STEP3_KEYS = { - "recording_use_approved", - "inputs_and_counts_approved", - "full_recordings_available", - "final_hearing_gate_committed", -} +FRESH_BASIS_KEYS = {"marker", "approval"} +APPROVAL_KEYS = {"decision", "timestamp"} SOURCE_KEYS = {"commit"} INPUT_KEYS = { - "neutral_alias", "source_path", "source_sha256", "source_size_bytes", @@ -355,294 +312,45 @@ def build_full_recording_schedule( return result -def _aware_datetime(value: Any) -> datetime: - if not isinstance(value, str): - raise PreflightError("approval-timestamp") - try: - parsed = datetime.fromisoformat(value) - except ValueError as exc: - raise PreflightError("approval-timestamp") from exc - if parsed.tzinfo is None or parsed.utcoffset() is None: - raise PreflightError("approval-timestamp") - return parsed - - -def _safe_relative_parts(value: Any) -> tuple[str, ...]: +def _resolve(root: Path, value: Any) -> Path: if not isinstance(value, str) or not value: raise PreflightError("unsafe-relative-path") pure = PurePosixPath(value) - if ( - pure.is_absolute() - or not pure.parts - or ".." in pure.parts - or "\\" in value - or "\x00" in value - ): + if pure.is_absolute() or ".." in pure.parts or "\\" in value or "\x00" in value: raise PreflightError("path-escape") - return pure.parts - - -def _same_file_state(before: os.stat_result, after: os.stat_result) -> bool: - return ( - before.st_dev, - before.st_ino, - before.st_mode, - before.st_nlink, - before.st_size, - before.st_mtime_ns, - before.st_ctime_ns, - ) == ( - after.st_dev, - after.st_ino, - after.st_mode, - after.st_nlink, - after.st_size, - after.st_mtime_ns, - after.st_ctime_ns, - ) - - -@dataclass(frozen=True) -class _BoundFile: - sha256: str - size_bytes: int - payload: bytes | None - - -def _read_bound_regular_file( - root: Path, path_value: Any, *, capture_payload: bool -) -> _BoundFile: - """Читает файл относительно доверенного неизменяемого root dirfd.""" - parts = _safe_relative_parts(path_value) - directory_flags = ( - os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC - ) - opened_fds: list[int] = [] - fd_states: list[tuple[int, os.stat_result]] = [] - namespace_entries: list[tuple[int, str, os.stat_result]] = [] - try: - directory_fd = os.open(root, directory_flags) - opened_fds.append(directory_fd) - root_state = os.fstat(directory_fd) - if not stat.S_ISDIR(root_state.st_mode): - raise PreflightError("file-binding") - fd_states.append((directory_fd, root_state)) - for part in parts[:-1]: - parent_fd = directory_fd - directory_fd = os.open(part, directory_flags, dir_fd=parent_fd) - opened_fds.append(directory_fd) - directory_state = os.fstat(directory_fd) - if not stat.S_ISDIR(directory_state.st_mode): - raise PreflightError("file-binding") - fd_states.append((directory_fd, directory_state)) - namespace_entries.append((parent_fd, part, directory_state)) - file_fd = os.open( - parts[-1], - os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC, - dir_fd=directory_fd, - ) - opened_fds.append(file_fd) - file_state = os.fstat(file_fd) - if not stat.S_ISREG(file_state.st_mode) or file_state.st_nlink != 1: - raise PreflightError("file-binding") - fd_states.append((file_fd, file_state)) - namespace_entries.append((directory_fd, parts[-1], file_state)) - - digest = hashlib.sha256() - size_bytes = 0 - chunks: list[bytes] | None = [] if capture_payload else None - while chunk := os.read(file_fd, 64 * 1024): - digest.update(chunk) - size_bytes += len(chunk) - if chunks is not None: - chunks.append(chunk) - - for file_descriptor, original_state in fd_states: - if not _same_file_state(original_state, os.fstat(file_descriptor)): - raise PreflightError("file-binding") - for parent_fd, name, original_state in namespace_entries: - current_state = os.stat(name, dir_fd=parent_fd, follow_symlinks=False) - if not _same_file_state(original_state, current_state): - raise PreflightError("file-binding") - if size_bytes != file_state.st_size: - raise PreflightError("file-binding") - return _BoundFile( - sha256=digest.hexdigest(), - size_bytes=size_bytes, - payload=b"".join(chunks) if chunks is not None else None, - ) - except PreflightError: - raise - except ValueError as exc: - raise PreflightError("path-escape") from exc - except OSError as exc: - reason = ( - "path-escape" - if exc.errno in {errno.ELOOP, errno.ENOTDIR} - else "file-binding" - ) - raise PreflightError(reason) from exc - finally: - for opened_fd in reversed(opened_fds): - try: - os.close(opened_fd) - except OSError: - pass + base = root.resolve() + resolved = (base / Path(value)).resolve() + if not resolved.is_relative_to(base): + raise PreflightError("path-escape") + return resolved def _verify(root: Path, path_value: Any, expected_hash: Any, size: Any | None) -> None: + path = _resolve(root, path_value) expected = _sha(expected_hash, "invalid-sha256") - expected_size = None if size is None else _positive_int(size, "invalid-size") - bound = _read_bound_regular_file(root, path_value, capture_payload=False) - if expected_size is not None and bound.size_bytes != expected_size: + if not path.is_file(): + raise PreflightError("file-unavailable") + if size is not None and path.stat().st_size != _positive_int(size, "invalid-size"): raise PreflightError("size-mismatch") - if bound.sha256 != expected: + if file_sha256(path) != expected: raise PreflightError("hash-mismatch") -def _strict_json_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]: - value: dict[str, Any] = {} - for key, item in pairs: - if key in value: - raise PreflightError("owner-acceptance-record") - value[key] = item - return value - - -def _reject_json_constant(_value: str) -> None: - raise PreflightError("owner-acceptance-record") - - -def _bound_json( - root: Path, link: Mapping[str, Any], link_keys: set[str] -) -> Mapping[str, Any]: - value = _exact(link, link_keys, "owner-acceptance-link") - expected_sha256 = _sha(value["sha256"], "owner-acceptance-hash") - bound = _read_bound_regular_file(root, value["path"], capture_payload=True) - if bound.sha256 != expected_sha256: - raise PreflightError("owner-acceptance-hash") - if bound.payload is None: - raise PreflightError("owner-acceptance-record") - try: - loaded = json.loads( - bound.payload, - object_pairs_hook=_strict_json_object, - parse_constant=_reject_json_constant, - ) - except (UnicodeDecodeError, json.JSONDecodeError) as exc: - raise PreflightError("owner-acceptance-record") from exc - if not isinstance(loaded, Mapping): - raise PreflightError("owner-acceptance-record") - return loaded - - -def _validate_owner_package_row(row: Any) -> None: - value = _exact(row, OWNER_PACKAGE_REFERENCE_KEYS, "owner-package-reference") - alias = value["neutral_alias"] - if ( - not isinstance(alias, str) - or re.fullmatch(r"[a-z][a-z0-9-]{0,63}", alias) is None - ): - raise PreflightError("owner-package-reference") - for key in OWNER_PACKAGE_REFERENCE_KEYS - {"neutral_alias"}: - _sha(value[key], "owner-package-reference") - - -def _validate_acceptance_row(row: Any) -> None: - value = _exact( - row, - REFERENCE_ACCEPTANCE_ROW_KEYS, - "owner-acceptance-reference", - ) - _validate_owner_package_row( - {key: value[key] for key in OWNER_PACKAGE_REFERENCE_KEYS} - ) - for key in ( - "fragment_expected_cluster_count", - "full_recording_expected_cluster_count", - ): - count = value[key] - if type(count) is not int or not 2 <= count <= 4: - raise PreflightError("owner-acceptance-reference") - for key in ( - "fragment_review_complete", - "full_recording_count_review_complete", - "reference_accepted", - ): - if value[key] is not True: - raise PreflightError("owner-acceptance-reference") - - -def _validate_reference_owner_acceptance( - *, - root: Path, - basis: Mapping[str, Any], - expected_package_rows: Sequence[Mapping[str, Any]], - expected_acceptance_rows: Sequence[Mapping[str, Any]], -) -> bool: - try: - package_link = _exact( - basis["owner_package"], - OWNER_PACKAGE_LINK_KEYS, - "owner-package-link", - ) - package = _exact( - _bound_json(root, package_link, OWNER_PACKAGE_LINK_KEYS), - OWNER_PACKAGE_KEYS, - "owner-package-schema", - ) - if package["schema"] != OWNER_PACKAGE_SCHEMA: - raise PreflightError("owner-package-schema") - package_generated_at = _aware_datetime(package["generated_at"]) - if package["generated_at"] != package_link["generated_at"]: - raise PreflightError("owner-package-timestamp") - package_rows = package["references"] - if not isinstance(package_rows, list) or len(package_rows) != 3: - raise PreflightError("owner-package-references") - for row in package_rows: - _validate_owner_package_row(row) - if package_rows != list(expected_package_rows): - raise PreflightError("owner-package-binding") - - approval_link = _exact( - basis["approval"], APPROVAL_LINK_KEYS, "owner-acceptance-link" - ) - acceptance = _exact( - _bound_json(root, approval_link, APPROVAL_LINK_KEYS), - REFERENCE_ACCEPTANCE_KEYS, - "owner-acceptance-schema", - ) - if acceptance["schema"] != REFERENCE_ACCEPTANCE_SCHEMA: - raise PreflightError("owner-acceptance-schema") - for key in ("actor", "status", "decision", "timestamp"): - if acceptance[key] != approval_link[key]: - raise PreflightError("owner-acceptance-link-mismatch") - if ( - acceptance["actor"] != "owner" - or acceptance["status"] != "approved" - or acceptance["decision"] != "approved" - ): - raise PreflightError("owner-acceptance-decision") - accepted_at = _aware_datetime(acceptance["timestamp"]) - if accepted_at <= package_generated_at: - raise PreflightError("owner-acceptance-timestamp") - if ( - _sha(acceptance["owner_package_sha256"], "owner-package-binding") - != package_link["sha256"] - ): - raise PreflightError("owner-package-binding") - acceptance_rows = acceptance["references"] - if not isinstance(acceptance_rows, list) or len(acceptance_rows) != 3: - raise PreflightError("owner-acceptance-references") - for row in acceptance_rows: - _validate_acceptance_row(row) - if acceptance_rows != list(expected_acceptance_rows): - raise PreflightError("owner-acceptance-binding") - step3 = _exact(acceptance["step3"], STEP3_KEYS, "owner-acceptance-step3") - if any(step3[key] is not True for key in STEP3_KEYS): - raise PreflightError("owner-acceptance-step3") - except (KeyError, OSError, TypeError, PreflightError): +def _validate_approval(value: Any) -> bool: + if value is None: return False + approval = _exact(value, APPROVAL_KEYS, "approval-schema") + if approval["decision"] != "approved": + return False + timestamp = approval["timestamp"] + if not isinstance(timestamp, str): + raise PreflightError("approval-timestamp") + try: + parsed = datetime.fromisoformat(timestamp) + except ValueError as exc: + raise PreflightError("approval-timestamp") from exc + if parsed.tzinfo is None: + raise PreflightError("approval-timestamp") return True @@ -749,12 +457,10 @@ def validate_manifest( raise PreflightError("legacy-identity") raise PreflightError("legacy-root") - basis = value["fresh_basis"] - if not isinstance(basis, Mapping): - raise PreflightError("fresh-basis-schema") - if basis.get("marker") != FRESH_BASIS_MARKER: + basis = _exact(value["fresh_basis"], FRESH_BASIS_KEYS, "fresh-basis-schema") + if basis["marker"] != FRESH_BASIS_MARKER: raise PreflightError("fresh-basis-marker") - basis_schema_valid = set(basis) == FRESH_BASIS_KEYS + approved = _validate_approval(basis["approval"]) source = _exact(value["source"], SOURCE_KEYS, "source-schema") source_commit = _commit(source["commit"], "source-commit") if set(environment) != {"commit", "clean"}: @@ -780,18 +486,8 @@ def validate_manifest( raise PreflightError("source-hashes-not-unique") fragment_hashes = [] fragment_sidecars = [] - package_rows = [] - acceptance_rows = [] - aliases = [] for item in inputs: entry = _exact(item, INPUT_KEYS, "input-schema") - alias = entry["neutral_alias"] - if ( - not isinstance(alias, str) - or re.fullmatch(r"[a-z][a-z0-9-]{0,63}", alias) is None - ): - raise PreflightError("neutral-alias") - aliases.append(alias) if verify_files: _verify( root, @@ -813,53 +509,29 @@ def validate_manifest( media_hash = _sha(record["sha256"], "record-hash") sidecar_hash = _sha(record["sidecar_sha256"], "sidecar-hash") if role == "fragment": - _sha(record["reference_sha256"], "reference-hash") + reference_path = record["reference_path"] + reference_hash = record["reference_sha256"] + if (reference_path is None) != (reference_hash is None): + raise PreflightError("reference-binding") + if reference_hash is not None: + _sha(reference_hash, "reference-hash") fragment_hashes.append(media_hash) fragment_sidecars.append(sidecar_hash) if verify_files: _verify(root, record["path"], media_hash, record["size_bytes"]) _verify(root, record["sidecar_path"], sidecar_hash, None) - if role == "fragment": + if role == "fragment" and reference_hash is not None: _verify( root, - record["reference_path"], - record["reference_sha256"], + reference_path, + reference_hash, None, ) - package_row = { - "neutral_alias": alias, - "canonical_fragment_audio_sha256": entry["fragment"]["sha256"], - "final_reference_sha256": entry["fragment"]["reference_sha256"], - "full_recording_sha256": entry["full_recording"]["sha256"], - } - package_rows.append(package_row) - acceptance_rows.append( - { - **package_row, - "fragment_expected_cluster_count": entry["fragment"][ - "expected_cluster_count" - ], - "full_recording_expected_cluster_count": entry["full_recording"][ - "expected_cluster_count" - ], - "fragment_review_complete": True, - "full_recording_count_review_complete": True, - "reference_accepted": True, - } - ) - if len(set(aliases)) != 3: - raise PreflightError("neutral-alias") _validate_schedule(value["schedule"], fragment_hashes, fragment_sidecars) _positive_int(value["rss_sample_interval_ms"], "rss-sample-interval") - approved = basis_schema_valid and _validate_reference_owner_acceptance( - root=root, - basis=basis, - expected_package_rows=package_rows, - expected_acceptance_rows=acceptance_rows, - ) return PreflightOutcome( state="valid" if approved else "needs-user", - reason=None if approved else "reference-owner-acceptance", + reason=None if approved else "owner-approval-required", fragment_count=3, full_recording_count=3, source_commit_sha256=hashlib.sha256(source_commit.encode("ascii")).hexdigest(), diff --git a/tests/test_streaming_sortformer_experiment.py b/tests/test_streaming_sortformer_experiment.py index 071d63d..5905e43 100644 --- a/tests/test_streaming_sortformer_experiment.py +++ b/tests/test_streaming_sortformer_experiment.py @@ -2,7 +2,6 @@ import copy import ctypes import hashlib import json -import os import tomllib from datetime import UTC, datetime from pathlib import Path @@ -33,14 +32,6 @@ def _write(root: Path, name: str, payload: bytes) -> dict[str, object]: } -def _write_json(root: Path, name: str, value: object) -> dict[str, object]: - return _write( - root, - name, - (json.dumps(value, ensure_ascii=True, sort_keys=True) + "\n").encode(), - ) - - def _record( root: Path, prefix: str, *, duration: float, reference: bool ) -> dict[str, object]: @@ -70,7 +61,6 @@ def _manifest(root: Path) -> dict[str, object]: source = _write(root, f"sources/{index}.wav", f"source-{index}".encode()) inputs.append( { - "neutral_alias": f"control-{index + 1}", "source_path": source["path"], "source_sha256": source["sha256"], "source_size_bytes": source["size_bytes"], @@ -82,74 +72,13 @@ def _manifest(root: Path) -> dict[str, object]: ), } ) - package_generated_at = "2026-09-01T08:00:00+00:00" - package_rows = [ - { - "neutral_alias": item["neutral_alias"], - "canonical_fragment_audio_sha256": item["fragment"]["sha256"], - "final_reference_sha256": item["fragment"]["reference_sha256"], - "full_recording_sha256": item["full_recording"]["sha256"], - } - for item in inputs - ] - package = _write_json( - root, - "owner/listening-package.json", - { - "schema": "local-transcriber.owner-listening-package.v1", - "generated_at": package_generated_at, - "references": package_rows, - }, - ) - approval_timestamp = "2026-09-01T09:00:00+00:00" - acceptance = { - "schema": "local-transcriber.reference-owner-acceptance.v1", - "actor": "owner", - "status": "approved", - "decision": "approved", - "timestamp": approval_timestamp, - "owner_package_sha256": package["sha256"], - "references": [ - { - **package_row, - "fragment_expected_cluster_count": item["fragment"][ - "expected_cluster_count" - ], - "full_recording_expected_cluster_count": item["full_recording"][ - "expected_cluster_count" - ], - "fragment_review_complete": True, - "full_recording_count_review_complete": True, - "reference_accepted": True, - } - for item, package_row in zip(inputs, package_rows, strict=True) - ], - "step3": { - "recording_use_approved": True, - "inputs_and_counts_approved": True, - "full_recordings_available": True, - "final_hearing_gate_committed": True, - }, - } - acceptance_file = _write_json( - root, "owner/reference-acceptance.json", acceptance - ) return { "schema": "local-transcriber.streaming-sortformer-experiment.v1", "fresh_basis": { "marker": "issue-46-fresh-owner-reviewed-basis", - "owner_package": { - "path": package["path"], - "sha256": package["sha256"], - "generated_at": package_generated_at, - }, "approval": { - "path": acceptance_file["path"], - "sha256": acceptance_file["sha256"], - "actor": "owner", - "status": "approved", "decision": "approved", - "timestamp": approval_timestamp, + "timestamp": "2026-09-01T09:00:00+00:00", }, }, "source": {"commit": "a" * 40}, @@ -265,232 +194,6 @@ def _environment() -> dict[str, object]: return {"commit": "a" * 40, "clean": True} -def _read_linked_json(root: Path, link: dict[str, object]) -> dict[str, object]: - return json.loads((root / str(link["path"])).read_text(encoding="utf-8")) - - -def _rewrite_linked_json( - root: Path, link: dict[str, object], value: object -) -> None: - payload = (json.dumps(value, ensure_ascii=True, sort_keys=True) + "\n").encode() - (root / str(link["path"])).write_bytes(payload) - link["sha256"] = hashlib.sha256(payload).hexdigest() - - -def _rewrite_linked_bytes( - root: Path, link: dict[str, object], payload: bytes -) -> None: - (root / str(link["path"])).write_bytes(payload) - link["sha256"] = hashlib.sha256(payload).hexdigest() - - -def _media_path(manifest: dict[str, object], binding: str) -> str: - item = manifest["inputs"][0] - paths = { - "source": item["source_path"], - "fragment": item["fragment"]["path"], - "fragment-sidecar": item["fragment"]["sidecar_path"], - "reference": item["fragment"]["reference_path"], - "full-recording": item["full_recording"]["path"], - "full-sidecar": item["full_recording"]["sidecar_path"], - } - return str(paths[binding]) - - -def _set_binding_path( - manifest: dict[str, object], binding: str, value: str -) -> None: - if binding in {"owner-package", "owner-acceptance"}: - link = "owner_package" if binding == "owner-package" else "approval" - manifest["fresh_basis"][link]["path"] = value - return - item = manifest["inputs"][0] - if binding == "source": - item["source_path"] = value - elif binding == "fragment": - item["fragment"]["path"] = value - elif binding == "fragment-sidecar": - item["fragment"]["sidecar_path"] = value - elif binding == "reference": - item["fragment"]["reference_path"] = value - elif binding == "full-recording": - item["full_recording"]["path"] = value - elif binding == "full-sidecar": - item["full_recording"]["sidecar_path"] = value - else: - raise AssertionError(f"unknown binding: {binding}") - - -def _mutate_owner_acceptance( - manifest: dict[str, object], root: Path, case: str -) -> None: - basis = manifest["fresh_basis"] - approval_link = basis["approval"] - package_link = basis["owner_package"] - acceptance = _read_linked_json(root, approval_link) - package = _read_linked_json(root, package_link) - - def write_acceptance() -> None: - _rewrite_linked_json(root, approval_link, acceptance) - - def write_package(*, sync_acceptance: bool = True) -> None: - _rewrite_linked_json(root, package_link, package) - if sync_acceptance: - acceptance["owner_package_sha256"] = package_link["sha256"] - write_acceptance() - - if case == "approval-missing": - del basis["approval"] - elif case == "approval-null": - basis["approval"] = None - elif case == "approval-link-unknown": - approval_link["count_review_status"] = "approved" - elif case == "record-missing": - (root / str(approval_link["path"])).unlink() - elif case == "record-path-missing": - approval_link["path"] = "owner/missing-acceptance.json" - elif case == "record-hash-mismatch": - approval_link["sha256"] = "f" * 64 - elif case == "record-json-invalid": - payload = b"{\n" - (root / str(approval_link["path"])).write_bytes(payload) - approval_link["sha256"] = hashlib.sha256(payload).hexdigest() - elif case == "record-unknown-key": - acceptance["count_review_status"] = "approved" - write_acceptance() - elif case == "record-count-review-schema": - acceptance = { - "schema": "local-transcriber.reference-count-review.v1", - "status": "approved", - } - write_acceptance() - elif case in {"actor", "status", "decision"}: - replacement = { - "actor": "automation", - "status": "draft", - "decision": "rejected", - }[case] - approval_link[case] = replacement - acceptance[case] = replacement - write_acceptance() - elif case == "timestamp-naive": - approval_link["timestamp"] = "2026-09-01T09:00:00" - acceptance["timestamp"] = approval_link["timestamp"] - write_acceptance() - elif case == "timestamp-early": - approval_link["timestamp"] = package_link["generated_at"] - acceptance["timestamp"] = approval_link["timestamp"] - write_acceptance() - elif case == "timestamp-link-mismatch": - acceptance["timestamp"] = "2026-09-01T10:00:00+00:00" - write_acceptance() - elif case == "package-missing": - del basis["owner_package"] - elif case == "package-link-unknown": - package_link["status"] = "approved" - elif case == "package-file-missing": - (root / str(package_link["path"])).unlink() - elif case == "package-hash-mismatch": - package_link["sha256"] = "e" * 64 - elif case == "package-timestamp-mismatch": - package_link["generated_at"] = "2026-09-01T07:00:00+00:00" - elif case == "package-unknown-key": - package["status"] = "approved" - write_package() - elif case == "package-row-binding": - package["references"][0]["final_reference_sha256"] = "d" * 64 - write_package() - elif case == "package-row-omitted": - package["references"].pop() - write_package() - elif case == "package-row-duplicate": - package["references"][2] = copy.deepcopy(package["references"][1]) - write_package() - elif case == "acceptance-row-omitted": - acceptance["references"].pop() - write_acceptance() - elif case == "acceptance-row-duplicate": - acceptance["references"][2] = copy.deepcopy(acceptance["references"][1]) - write_acceptance() - elif case in { - "neutral_alias", - "canonical_fragment_audio_sha256", - "final_reference_sha256", - "full_recording_sha256", - }: - acceptance["references"][0][case] = ( - "different-control" if case == "neutral_alias" else "c" * 64 - ) - write_acceptance() - elif case in { - "fragment_expected_cluster_count", - "full_recording_expected_cluster_count", - }: - acceptance["references"][0][case] = 4 - write_acceptance() - elif case == "fragment-count-float": - acceptance["references"][0]["fragment_expected_cluster_count"] = 2.0 - write_acceptance() - elif case == "fragment-count-bool": - acceptance["references"][0]["fragment_expected_cluster_count"] = True - write_acceptance() - elif case == "fragment-count-string": - acceptance["references"][0]["fragment_expected_cluster_count"] = "2" - write_acceptance() - elif case == "full-count-float": - acceptance["references"][0][ - "full_recording_expected_cluster_count" - ] = 2.0 - write_acceptance() - elif case == "package-alias-type": - package["references"][0]["neutral_alias"] = 1 - write_package() - elif case == "package-hash-type": - package["references"][0]["canonical_fragment_audio_sha256"] = 1 - write_package() - elif case == "acceptance-hash-type": - acceptance["references"][0]["canonical_fragment_audio_sha256"] = 1 - write_acceptance() - elif case in { - "fragment_review_complete", - "full_recording_count_review_complete", - "reference_accepted", - }: - acceptance["references"][0][case] = False - write_acceptance() - elif case == "fragment-review-int": - acceptance["references"][0]["fragment_review_complete"] = 1 - write_acceptance() - elif case == "full-review-int": - acceptance["references"][0][ - "full_recording_count_review_complete" - ] = 1 - write_acceptance() - elif case == "reference-accepted-int": - acceptance["references"][0]["reference_accepted"] = 1 - write_acceptance() - elif case == "acceptance-row-unknown": - acceptance["references"][0]["status"] = "approved" - write_acceptance() - elif case == "owner-package-binding": - acceptance["owner_package_sha256"] = "b" * 64 - write_acceptance() - elif case == "step3-int": - acceptance["step3"]["recording_use_approved"] = 1 - write_acceptance() - elif case.startswith("step3-"): - acceptance["step3"][case.removeprefix("step3-")] = False - write_acceptance() - elif case == "step3-missing": - del acceptance["step3"]["recording_use_approved"] - write_acceptance() - elif case == "step3-unknown": - acceptance["step3"]["count_review_approved"] = True - write_acceptance() - else: - raise AssertionError(f"unknown case: {case}") - - def test_preflight_validates_fresh_exact_schema_before_loading(tmp_path): manifest = _manifest(tmp_path) calls = [] @@ -510,7 +213,7 @@ def test_preflight_validates_fresh_exact_schema_before_loading(tmp_path): assert calls == [] -def test_complete_reference_owner_acceptance_is_required_before_loading(tmp_path): +def test_approved_fresh_basis_loads_models_after_preflight(tmp_path): manifest = _manifest(tmp_path) calls = [] @@ -526,70 +229,12 @@ def test_complete_reference_owner_acceptance_is_required_before_loading(tmp_path assert calls == ["w0", "candidate"] -@pytest.mark.parametrize( - "case", - [ - "approval-missing", - "approval-null", - "approval-link-unknown", - "record-missing", - "record-path-missing", - "record-hash-mismatch", - "record-json-invalid", - "record-unknown-key", - "record-count-review-schema", - "actor", - "status", - "decision", - "timestamp-naive", - "timestamp-early", - "timestamp-link-mismatch", - "package-missing", - "package-link-unknown", - "package-file-missing", - "package-hash-mismatch", - "package-timestamp-mismatch", - "package-unknown-key", - "package-row-binding", - "package-row-omitted", - "package-row-duplicate", - "acceptance-row-omitted", - "acceptance-row-duplicate", - "neutral_alias", - "canonical_fragment_audio_sha256", - "final_reference_sha256", - "full_recording_sha256", - "fragment_expected_cluster_count", - "full_recording_expected_cluster_count", - "fragment-count-float", - "fragment-count-bool", - "fragment-count-string", - "full-count-float", - "package-alias-type", - "package-hash-type", - "acceptance-hash-type", - "fragment_review_complete", - "full_recording_count_review_complete", - "reference_accepted", - "fragment-review-int", - "full-review-int", - "reference-accepted-int", - "acceptance-row-unknown", - "owner-package-binding", - "step3-recording_use_approved", - "step3-inputs_and_counts_approved", - "step3-full_recordings_available", - "step3-final_hearing_gate_committed", - "step3-int", - "step3-missing", - "step3-unknown", - ], -) -def test_reference_owner_acceptance_failures_need_user_before_loaders( - tmp_path, case -): +def test_fragment_without_reference_loads_models_after_preflight(tmp_path): manifest = _manifest(tmp_path) - _mutate_owner_acceptance(manifest, tmp_path, case) + fragment = manifest["inputs"][0]["fragment"] + (tmp_path / fragment["reference_path"]).unlink() + fragment["reference_path"] = None + fragment["reference_sha256"] = None calls = [] outcome = experiment.preflight_and_load( @@ -600,289 +245,17 @@ def test_reference_owner_acceptance_failures_need_user_before_loaders( candidate_loader=lambda: calls.append("candidate"), ) - assert outcome.state == "needs-user" - assert outcome.reason == "reference-owner-acceptance" - assert calls == [] + assert outcome.state == "valid" + assert calls == ["w0", "candidate"] -@pytest.mark.parametrize( - ("record", "original", "replacement"), - [ - ( - "approval", - b'"status": "approved"', - b'"status": "draft", "status": "approved"', - ), - ( - "approval", - b'"reference_accepted": true', - b'"reference_accepted": false, "reference_accepted": true', - ), - ( - "approval", - b'"recording_use_approved": true', - b'"recording_use_approved": false, "recording_use_approved": true', - ), - ( - "owner_package", - b'"neutral_alias": "control-1"', - b'"neutral_alias": "other", "neutral_alias": "control-1"', - ), - ( - "approval", - b'"fragment_expected_cluster_count": 2', - b'"fragment_expected_cluster_count": NaN', - ), - ], -) -def test_bound_owner_json_rejects_ambiguous_json_before_loaders( - tmp_path, record, original, replacement -): +@pytest.mark.parametrize("field", ["reference_path", "reference_sha256"]) +def test_fragment_rejects_half_present_reference_before_loaders(tmp_path, field): manifest = _manifest(tmp_path) - link = manifest["fresh_basis"][record] - path = tmp_path / str(link["path"]) - payload = path.read_bytes().replace(original, replacement, 1) - assert payload != path.read_bytes() - _rewrite_linked_bytes(tmp_path, link, payload) - if record == "owner_package": - acceptance_link = manifest["fresh_basis"]["approval"] - acceptance = _read_linked_json(tmp_path, acceptance_link) - acceptance["owner_package_sha256"] = link["sha256"] - _rewrite_linked_json(tmp_path, acceptance_link, acceptance) + manifest["inputs"][0]["fragment"][field] = None calls = [] - outcome = experiment.preflight_and_load( - manifest, - root=tmp_path, - environment_probe=_environment, - w0_loader=lambda: calls.append("w0"), - candidate_loader=lambda: calls.append("candidate"), - ) - - assert outcome.state == "needs-user" - assert outcome.reason == "reference-owner-acceptance" - assert calls == [] - - -@pytest.mark.parametrize("symlink_kind", ["final", "intermediate"]) -def test_bound_owner_json_rejects_symlinks_before_loaders( - tmp_path, symlink_kind -): - manifest = _manifest(tmp_path) - if symlink_kind == "final": - link = manifest["fresh_basis"]["approval"] - path = tmp_path / str(link["path"]) - target = path.with_name("acceptance-target.json") - path.rename(target) - path.symlink_to(target.name) - else: - owner = tmp_path / "owner" - target = tmp_path / "owner-target" - owner.rename(target) - owner.symlink_to(target.name, target_is_directory=True) - calls = [] - - outcome = experiment.preflight_and_load( - manifest, - root=tmp_path, - environment_probe=_environment, - w0_loader=lambda: calls.append("w0"), - candidate_loader=lambda: calls.append("candidate"), - ) - - assert outcome.state == "needs-user" - assert outcome.reason == "reference-owner-acceptance" - assert calls == [] - - -def test_bound_owner_json_cannot_hash_one_version_and_parse_another( - tmp_path, monkeypatch -): - manifest = _manifest(tmp_path) - link = manifest["fresh_basis"]["approval"] - path = tmp_path / str(link["path"]) - approved_payload = path.read_bytes() - rejected_payload = approved_payload.replace( - b'"status": "approved"', b'"status": "draft"', 1 - ) - path.write_bytes(rejected_payload) - link["sha256"] = hashlib.sha256(rejected_payload).hexdigest() - original_file_sha256 = experiment.file_sha256 - replaced = False - - def replace_after_hash(candidate): - nonlocal replaced - digest = original_file_sha256(candidate) - if candidate == path and not replaced: - path.write_bytes(approved_payload) - replaced = True - return digest - - monkeypatch.setattr(experiment, "file_sha256", replace_after_hash) - calls = [] - - outcome = experiment.preflight_and_load( - manifest, - root=tmp_path, - environment_probe=_environment, - w0_loader=lambda: calls.append("w0"), - candidate_loader=lambda: calls.append("candidate"), - ) - - assert outcome.state == "needs-user" - assert outcome.reason == "reference-owner-acceptance" - assert calls == [] - - -def test_bound_owner_json_rejects_path_replacement_during_held_read( - tmp_path, monkeypatch -): - manifest = _manifest(tmp_path) - link = manifest["fresh_basis"]["approval"] - path = tmp_path / str(link["path"]) - target_inode = path.stat().st_ino - replacement = path.with_name("replacement.json") - payload = path.read_bytes() - original_read = os.read - replaced = False - - def replace_path_after_read(file_descriptor, count): - nonlocal replaced - chunk = original_read(file_descriptor, count) - if os.fstat(file_descriptor).st_ino == target_inode and not replaced: - replacement.write_bytes(payload) - os.replace(replacement, path) - replaced = True - return chunk - - monkeypatch.setattr(experiment.os, "read", replace_path_after_read) - calls = [] - - outcome = experiment.preflight_and_load( - manifest, - root=tmp_path, - environment_probe=_environment, - w0_loader=lambda: calls.append("w0"), - candidate_loader=lambda: calls.append("candidate"), - ) - - assert replaced is True - assert outcome.state == "needs-user" - assert outcome.reason == "reference-owner-acceptance" - assert calls == [] - - -def test_bound_owner_json_rejects_in_place_mutation_during_held_read( - tmp_path, monkeypatch -): - manifest = _manifest(tmp_path) - link = manifest["fresh_basis"]["approval"] - path = tmp_path / str(link["path"]) - json_size = len(path.read_bytes().rstrip(b"\n")) - payload = path.read_bytes().rstrip(b"\n") + (b" " * 70_000) + b"\n" - _rewrite_linked_bytes(tmp_path, link, payload) - target_inode = path.stat().st_ino - original_read = os.read - mutated = False - - def mutate_after_first_chunk(file_descriptor, count): - nonlocal mutated - chunk = original_read(file_descriptor, count) - if os.fstat(file_descriptor).st_ino == target_inode and not mutated: - writer = os.open(path, os.O_WRONLY) - try: - os.pwrite(writer, b"\t", json_size + 66_000) - os.fsync(writer) - finally: - os.close(writer) - mutated = True - return chunk - - monkeypatch.setattr(experiment.os, "read", mutate_after_first_chunk) - calls = [] - - outcome = experiment.preflight_and_load( - manifest, - root=tmp_path, - environment_probe=_environment, - w0_loader=lambda: calls.append("w0"), - candidate_loader=lambda: calls.append("candidate"), - ) - - assert mutated is True - assert outcome.state == "needs-user" - assert outcome.reason == "reference-owner-acceptance" - assert calls == [] - - -def test_bound_owner_json_rejects_intermediate_directory_swap_during_read( - tmp_path, monkeypatch -): - manifest = _manifest(tmp_path) - link = manifest["fresh_basis"]["owner_package"] - path = tmp_path / str(link["path"]) - target_inode = path.stat().st_ino - parent = path.parent - moved_parent = parent.with_name("owner-original") - original_read = os.read - swapped = False - - def swap_directory_after_read(file_descriptor, count): - nonlocal swapped - chunk = original_read(file_descriptor, count) - if os.fstat(file_descriptor).st_ino == target_inode and not swapped: - parent.rename(moved_parent) - parent.mkdir() - (parent / path.name).write_bytes(b"{}\n") - swapped = True - return chunk - - monkeypatch.setattr(experiment.os, "read", swap_directory_after_read) - calls = [] - - outcome = experiment.preflight_and_load( - manifest, - root=tmp_path, - environment_probe=_environment, - w0_loader=lambda: calls.append("w0"), - candidate_loader=lambda: calls.append("candidate"), - ) - - assert swapped is True - assert outcome.state == "needs-user" - assert outcome.reason == "reference-owner-acceptance" - assert calls == [] - - -@pytest.mark.parametrize( - "binding", - [ - "source", - "fragment", - "fragment-sidecar", - "reference", - "full-recording", - "full-sidecar", - ], -) -@pytest.mark.parametrize("symlink_kind", ["final", "intermediate"]) -def test_manifest_file_bindings_reject_internal_symlinks_before_loaders( - tmp_path, binding, symlink_kind -): - manifest = _manifest(tmp_path) - path = tmp_path / _media_path(manifest, binding) - if symlink_kind == "final": - target = path.with_name(f"{path.name}.target") - path.rename(target) - path.symlink_to(target.name) - else: - parent = path.parent - target = parent.with_name(f"{parent.name}-target") - parent.rename(target) - parent.symlink_to(target.name, target_is_directory=True) - calls = [] - - with pytest.raises(experiment.PreflightError) as raised: + with pytest.raises(experiment.PreflightError, match="reference-binding"): experiment.preflight_and_load( manifest, root=tmp_path, @@ -891,221 +264,6 @@ def test_manifest_file_bindings_reject_internal_symlinks_before_loaders( candidate_loader=lambda: calls.append("candidate"), ) - assert str(tmp_path) not in str(raised.value) - assert calls == [] - - -def test_manifest_file_binding_rejects_final_replacement_during_read( - tmp_path, monkeypatch -): - manifest = _manifest(tmp_path) - path = tmp_path / _media_path(manifest, "fragment") - target_inode = path.stat().st_ino - replacement = path.with_name("replacement.wav") - original_read = os.read - replaced = False - - def replace_path_after_read(file_descriptor, count): - nonlocal replaced - chunk = original_read(file_descriptor, count) - if os.fstat(file_descriptor).st_ino == target_inode and not replaced: - replacement.write_bytes(b"different-fragment-bytes") - os.replace(replacement, path) - replaced = True - return chunk - - monkeypatch.setattr(experiment.os, "read", replace_path_after_read) - calls = [] - - with pytest.raises(experiment.PreflightError): - experiment.preflight_and_load( - manifest, - root=tmp_path, - environment_probe=_environment, - w0_loader=lambda: calls.append("w0"), - candidate_loader=lambda: calls.append("candidate"), - ) - - assert replaced is True - assert path.read_bytes() == b"different-fragment-bytes" - assert hashlib.sha256(path.read_bytes()).hexdigest() != manifest["inputs"][0][ - "fragment" - ]["sha256"] - assert calls == [] - - -def test_manifest_file_binding_rejects_intermediate_directory_swap_during_read( - tmp_path, monkeypatch -): - manifest = _manifest(tmp_path) - path = tmp_path / _media_path(manifest, "fragment") - target_inode = path.stat().st_ino - parent = path.parent - moved_parent = parent.with_name("media-original") - original_read = os.read - swapped = False - - def swap_directory_after_read(file_descriptor, count): - nonlocal swapped - chunk = original_read(file_descriptor, count) - if os.fstat(file_descriptor).st_ino == target_inode and not swapped: - parent.rename(moved_parent) - parent.mkdir() - (parent / path.name).write_bytes(b"replacement-fragment") - swapped = True - return chunk - - monkeypatch.setattr(experiment.os, "read", swap_directory_after_read) - calls = [] - - with pytest.raises(experiment.PreflightError): - experiment.preflight_and_load( - manifest, - root=tmp_path, - environment_probe=_environment, - w0_loader=lambda: calls.append("w0"), - candidate_loader=lambda: calls.append("candidate"), - ) - - assert swapped is True - assert calls == [] - - -def test_manifest_file_binding_rejects_in_place_mutation_during_read( - tmp_path, monkeypatch -): - manifest = _manifest(tmp_path) - path = tmp_path / _media_path(manifest, "fragment") - target_inode = path.stat().st_ino - original_read = os.read - mutated = False - - def mutate_file_after_read(file_descriptor, count): - nonlocal mutated - chunk = original_read(file_descriptor, count) - if os.fstat(file_descriptor).st_ino == target_inode and not mutated: - path.write_bytes(b"mutated-fragment-with-different-size") - mutated = True - return chunk - - monkeypatch.setattr(experiment.os, "read", mutate_file_after_read) - calls = [] - - with pytest.raises(experiment.PreflightError): - experiment.preflight_and_load( - manifest, - root=tmp_path, - environment_probe=_environment, - w0_loader=lambda: calls.append("w0"), - candidate_loader=lambda: calls.append("candidate"), - ) - - assert mutated is True - assert calls == [] - - -def test_bound_file_verification_closes_descriptors_on_success_and_failure( - tmp_path -): - proc_fds = Path("/proc/self/fd") - if not proc_fds.is_dir(): - pytest.skip("fd accounting requires procfs") - - manifest = _manifest(tmp_path) - before_success = len(list(proc_fds.iterdir())) - for _ in range(20): - assert experiment.validate_manifest( - manifest, root=tmp_path, environment=_environment() - ).state == "valid" - assert len(list(proc_fds.iterdir())) == before_success - - failed_root = tmp_path / "failed" - failed_manifest = _manifest(failed_root) - path = failed_root / _media_path(failed_manifest, "fragment") - target = path.with_name("fragment-target.wav") - path.rename(target) - path.symlink_to(target.name) - before_failure = len(list(proc_fds.iterdir())) - for _ in range(20): - with pytest.raises(experiment.PreflightError): - experiment.validate_manifest( - failed_manifest, - root=failed_root, - environment=_environment(), - ) - assert len(list(proc_fds.iterdir())) == before_failure - - -@pytest.mark.parametrize( - "binding", - [ - "source", - "fragment", - "fragment-sidecar", - "reference", - "full-recording", - "full-sidecar", - "owner-package", - "owner-acceptance", - ], -) -def test_manifest_file_bindings_reject_nul_paths_before_loaders( - tmp_path, binding -): - manifest = _manifest(tmp_path) - _set_binding_path(manifest, binding, "private\x00binding") - calls = [] - - if binding.startswith("owner-"): - outcome = experiment.preflight_and_load( - manifest, - root=tmp_path, - environment_probe=_environment, - w0_loader=lambda: calls.append("w0"), - candidate_loader=lambda: calls.append("candidate"), - ) - assert outcome.state == "needs-user" - assert outcome.reason == "reference-owner-acceptance" - else: - with pytest.raises(experiment.PreflightError, match="path-escape"): - experiment.preflight_and_load( - manifest, - root=tmp_path, - environment_probe=_environment, - w0_loader=lambda: calls.append("w0"), - candidate_loader=lambda: calls.append("candidate"), - ) - - assert calls == [] - - -def test_manifest_file_binding_redacts_path_layer_value_error( - tmp_path, monkeypatch -): - manifest = _manifest(tmp_path) - original_open = os.open - - def reject_component(path, flags, *, dir_fd=None): - if path == "sources": - raise ValueError("private path detail") - if dir_fd is None: - return original_open(path, flags) - return original_open(path, flags, dir_fd=dir_fd) - - monkeypatch.setattr(experiment.os, "open", reject_component) - calls = [] - - with pytest.raises(experiment.PreflightError) as raised: - experiment.preflight_and_load( - manifest, - root=tmp_path, - environment_probe=_environment, - w0_loader=lambda: calls.append("w0"), - candidate_loader=lambda: calls.append("candidate"), - ) - - assert str(raised.value) == "path-escape" - assert "private" not in str(raised.value) assert calls == [] @@ -1123,7 +281,7 @@ def test_missing_owner_approval_is_needs_user_and_never_loads(tmp_path): ) assert outcome.state == "needs-user" - assert outcome.reason == "reference-owner-acceptance" + assert outcome.reason == "owner-approval-required" assert calls == [] @@ -1140,10 +298,6 @@ def test_missing_owner_approval_is_needs_user_and_never_loads(tmp_path): lambda value: value["inputs"][0]["full_recording"].update( expected_cluster_count=5 ), - lambda value: value["inputs"][0].update(neutral_alias=""), - lambda value: value["inputs"][1].update( - neutral_alias=value["inputs"][0]["neutral_alias"] - ), lambda value: value["source"].update(commit="b" * 40), ], ) @@ -1191,6 +345,25 @@ def test_preflight_rejects_hash_path_and_symlink_escape_without_leaking_path(tmp experiment.validate_manifest(manifest, root=tmp_path, environment=_environment()) +def test_preflight_rejects_nul_path_without_leaking_or_loading(tmp_path): + manifest = _manifest(tmp_path) + manifest["inputs"][0]["source_path"] = "private\x00source.wav" + calls = [] + + with pytest.raises(experiment.PreflightError) as raised: + experiment.preflight_and_load( + manifest, + root=tmp_path, + environment_probe=_environment, + w0_loader=lambda: calls.append("w0"), + candidate_loader=lambda: calls.append("candidate"), + ) + + assert str(raised.value) == "path-escape" + assert "private" not in str(raised.value) + assert calls == [] + + def test_preflight_rejects_legacy_cell_names_roots_and_recipe_pin_mismatch(tmp_path): manifest = _manifest(tmp_path) manifest["schedule"][0]["semantic_id"] = "41-W0" @@ -2207,35 +1380,6 @@ def test_invalid_cli_output_has_one_trailing_newline(tmp_path, capsys): assert not output.endswith("\n\n") -def test_validate_only_cli_redacts_nul_path_failure( - tmp_path, capsys, monkeypatch -): - manifest = _manifest(tmp_path) - manifest["inputs"][0]["source_path"] = "private\x00source.wav" - manifest_path = tmp_path / "manifest.json" - manifest_path.write_text(json.dumps(manifest), encoding="utf-8") - monkeypatch.setattr( - experiment, - "git_environment_probe", - lambda _repository_root: _environment(), - ) - - with pytest.raises(SystemExit) as raised: - experiment.main(["--manifest", str(manifest_path), "--validate-only"]) - - captured = capsys.readouterr() - assert raised.value.code == 2 - assert json.loads(captured.out) == { - "reason": "path-escape", - "state": "invalid", - } - assert captured.err == "" - assert "Traceback" not in captured.out - assert "\x00" not in captured.out - assert "private" not in captured.out - assert str(tmp_path) not in captured.out - - def test_child_process_request_result_seam_uses_atomic_private_files(tmp_path): request = { "semantic_id": "cell-safe", From 4c4a8be07b3653cede35b0435e6c9ef6cb3fb87d Mon Sep 17 00:00:00 2001 From: Dmitry Dementiev Date: Fri, 4 Sep 2026 18:50:31 +0300 Subject: [PATCH 5/7] =?UTF-8?q?fix(experiment):=20=D1=83=D1=82=D0=BE=D1=87?= =?UTF-8?q?=D0=BD=D0=B5=D0=BD=D0=BE=20=D0=BE=D0=BF=D1=80=D0=B5=D0=B4=D0=B5?= =?UTF-8?q?=D0=BB=D0=B5=D0=BD=D0=B8=D0=B5=20=D0=B0=D0=BA=D1=82=D0=B8=D0=B2?= =?UTF-8?q?=D0=BD=D0=BE=D0=B9=20=D0=BF=D0=BE=D0=B4=D0=BA=D0=B0=D1=87=D0=BA?= =?UTF-8?q?=D0=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Зачем: - занятые, но неизменные swap-страницы ошибочно блокировали канонический Ryzen-прогон. - Что: - active swap определяется изменением счетчиков sin/sout во время ячейки. - сохранены memory gates для OOM, RSS plateau, линейного роста и пикового RSS. - Проверка: - uv run pytest -q tests/test_streaming_sortformer_experiment.py tests/test_benchmark_experiment.py tests/test_speaker_diarizer.py tests/test_diarization.py. --- .../streaming_sortformer_experiment.py | 7 +-- tests/test_streaming_sortformer_experiment.py | 58 +++++++++++++++---- 2 files changed, 47 insertions(+), 18 deletions(-) diff --git a/src/local_transcriber/streaming_sortformer_experiment.py b/src/local_transcriber/streaming_sortformer_experiment.py index ded571c..bfa2575 100644 --- a/src/local_transcriber/streaming_sortformer_experiment.py +++ b/src/local_transcriber/streaming_sortformer_experiment.py @@ -1095,8 +1095,6 @@ def memory_gate( for value in integer_values ): raise ValueError("memory-value") - if swap_before_bytes > 0 or swap_after_bytes > 0: - return {"state": "stop", "reason": "active-swap", **base} if stage == "fragment": if len(rss_after_inputs) != 3: return {"state": "stop", "reason": "rss-no-plateau", **base} @@ -1446,13 +1444,10 @@ def _guard_transition_reasons( def _swap_active(before: Mapping[str, Any], after: Mapping[str, Any] | None = None) -> bool: - if before["swap_used_bytes"] > 0: - return True if after is None: return False return bool( - after["swap_used_bytes"] > 0 - or after["swap_sin_bytes"] != before["swap_sin_bytes"] + after["swap_sin_bytes"] != before["swap_sin_bytes"] or after["swap_sout_bytes"] != before["swap_sout_bytes"] ) diff --git a/tests/test_streaming_sortformer_experiment.py b/tests/test_streaming_sortformer_experiment.py index 5905e43..daf0dcf 100644 --- a/tests/test_streaming_sortformer_experiment.py +++ b/tests/test_streaming_sortformer_experiment.py @@ -774,7 +774,7 @@ def test_memory_gate_uses_frozen_plateau_and_linear_growth_policy_boundaries(): peak_rss_bytes=300 * mib, rss_after_inputs=[100 * mib, 120 * mib, 121 * mib], machine_ram_bytes=16 * 1024**3, - )["reason"] == "active-swap" + )["reason"] is None assert experiment.memory_gate( stage="fragment", oom=True, @@ -786,6 +786,23 @@ def test_memory_gate_uses_frozen_plateau_and_linear_growth_policy_boundaries(): )["reason"] == "oom" +def test_memory_gate_allows_preoccupied_swap() -> None: + mib = 1024**2 + + result = experiment.memory_gate( + stage="fragment", + oom=False, + swap_before_bytes=1_730_000_000, + swap_after_bytes=1_730_000_000, + peak_rss_bytes=300 * mib, + rss_after_inputs=[100 * mib, 120 * mib, 121 * mib], + machine_ram_bytes=16 * 1024**3, + ) + + assert result["state"] == "pass" + assert result["reason"] is None + + def test_memory_gate_marks_two_gibibyte_review_as_needs_user(): result = experiment.memory_gate( stage="fragment", @@ -1084,7 +1101,7 @@ def test_guarded_cell_accepts_stable_manifest_declared_performance_mode(): assert len(result["attempts"]) == 1 -def test_second_guard_invalidation_stops_with_first_reason_and_active_swap_never_retries(): +def test_second_guard_invalidation_stops_with_first_reason(): throttled = iter( [ _guard(), @@ -1104,17 +1121,33 @@ def test_second_guard_invalidation_stops_with_first_reason_and_active_swap_never assert second["reason"] == "new-throttling" assert len(second["attempts"]) == 2 - calls = [] - swapped = experiment.run_guarded_cell( + +def test_guarded_cell_allows_preoccupied_swap_when_counters_stay_stable(): + snapshots = iter( + [ + _guard( + swap_used_bytes=1_730_000_000, + swap_sin_bytes=100, + swap_sout_bytes=200, + ), + _guard( + swap_used_bytes=1_730_000_000, + swap_sin_bytes=100, + swap_sout_bytes=200, + ), + ] + ) + + result = experiment.run_guarded_cell( {"semantic_id": "safe-cell", "recipe": "W0-FRESH"}, - runner=lambda _request: calls.append("run") or {"wall_seconds": 1.0}, - guard_reader=lambda: _guard(swap_used_bytes=1), - stabilize=lambda: calls.append("stabilize"), + runner=lambda _request: {"wall_seconds": 1.0}, + guard_reader=lambda: next(snapshots), + stabilize=lambda: None, expected_machine=_machine(), ) - assert swapped["state"] == "stop" - assert swapped["reason"] == "active-swap" - assert calls == [] + + assert result["state"] == "accepted" + assert len(result["attempts"]) == 1 def test_oom_stops_without_retry_and_private_result_is_written_atomically(tmp_path): @@ -1196,8 +1229,9 @@ def test_guarded_cell_does_not_retry_hard_mismatch_and_sanitizes_exceptions(): assert calls == [] -def test_guarded_cell_hashes_result_rejected_by_after_run_swap(): - snapshots = iter([_guard(), _guard(swap_sin_bytes=1)]) +@pytest.mark.parametrize("counter", ["swap_sin_bytes", "swap_sout_bytes"]) +def test_guarded_cell_hashes_result_rejected_by_after_run_swap(counter): + snapshots = iter([_guard(), _guard(**{counter: 1})]) raw_result = {"wall_seconds": 1.0, "private": "not-published"} result = experiment.run_guarded_cell( {"semantic_id": "safe-cell", "recipe": "W0-FRESH"}, From b8bb78103a4017300720a5904561fdf9efa537dc Mon Sep 17 00:00:00 2001 From: Dmitry Dementiev Date: Fri, 4 Sep 2026 21:38:42 +0300 Subject: [PATCH 6/7] =?UTF-8?q?docs(benchmark):=20=D0=B7=D0=B0=D1=84=D0=B8?= =?UTF-8?q?=D0=BA=D1=81=D0=B8=D1=80=D0=BE=D0=B2=D0=B0=D0=BD=20stop=20Ryzen?= =?UTF-8?q?-=D0=BF=D1=80=D0=BE=D0=B3=D0=BE=D0=BD=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Зачем: - задача #46 должна завершиться воспроизводимым решением по каноническому Ryzen-эксперименту. - Что: - записан stop из-за несовпадения числа кластеров свежего W0. - сохранены безопасные агрегаты W0 и вторичный провал проверки swap. - отмечено, что Sortformer не запускался и данных об ускорении нет. - Проверка: - uv run pytest -q. - независимая сверка отчета с закрытым результатом и проверка литературного русского. --- .../2026-09-04-streaming-sortformer-ryzen.md | 73 +++++++++++++++++++ 1 file changed, 73 insertions(+) create mode 100644 docs/benchmarks/2026-09-04-streaming-sortformer-ryzen.md diff --git a/docs/benchmarks/2026-09-04-streaming-sortformer-ryzen.md b/docs/benchmarks/2026-09-04-streaming-sortformer-ryzen.md new file mode 100644 index 0000000..078ca5b --- /dev/null +++ b/docs/benchmarks/2026-09-04-streaming-sortformer-ryzen.md @@ -0,0 +1,73 @@ +# Streaming Sortformer: формальный `stop` на Ryzen + +**Дата:** 2026-09-04 + +**Статус:** эксперимент завершен как `stop`. Для трех объектов закрытого +манифеста были закреплены количества кластеров `2 / 2 / 3`, но W0 выдал +`2 / 2 / 2`. Кандидат не запускался. + +## Решение + +Свежий W0 не прошел обязательную проверку числа говорящих на одном из трех +300-секундных фрагментов. По условиям +[эксперимента](../specs/2026-09-01-streaming-sortformer-cpu-experiment.md) +такая база сравнения непригодна. Выполнена одна из двенадцати запланированных +ячеек. Sortformer и остальные ячейки не запускались, ускорение не считалось. +Данных о скорости или качестве кандидата этот прогон не дал. + +Во время W0 также зафиксированы чтение из swap и запись в него. Это независимый +провал проверки среды и памяти, но первой причиной `stop` остается +`cluster-count-mismatch`: проверка качества завершилась раньше итогового снимка +системных счетчиков. + +## Результат W0 + +| Показатель | Значение | +| --- | ---: | +| Суммарный wall, с | `103,242485` | +| Ожидаемые кластеры | `2 / 2 / 3` | +| Полученные кластеры | `2 / 2 / 2` | +| Пиковый RSS | `340 340 736` байт (`324,57` МиБ) | +| Прочитано из swap за ячейку | `503 808` байт (`492` КиБ) | +| Записано в swap за ячейку | `151 552` байт (`148` КиБ) | + +Первые два фрагмента прошли обязательные проверки числа кластеров, текста, +порядка слов и интервалов. Третий фрагмент содержал подтвержденные три голоса, +но W0 сформировал два кластера. Повтор не назначался: несовпадение числа +кластеров само по себе завершает шаг приемки W0. + +## Условия + +W0 выполнялся по закрепленному рецепту `W0-FRESH`: + +- `sherpa-onnx/core 1.13.6`, CPU provider; +- Pyannote Segmentation 3.0 FP32 и WeSpeaker ResNet34 LM FP32; +- один процесс, одна сессия, восемь физических ядер, + `intra_op_threads=8`, `inter_op_threads=1`; +- `window_shift_ratio=0.1`, `FastClustering(auto, 0.89)`, + `min_duration_on=0.3`, `min_duration_off=0.5`; +- Windows Balanced, boost mode `2`; Linux cpufreq/governor в WSL2 не + предоставляется. + +До измерения успешно завершились обязательные детерминированные тесты, запуск +с `--validate-only` и предварительная проверка C ABI и конфигурации кандидата. +Последняя подтвердила четыре выходных канала, шаг `0,08` с и закрепленную +CPU-конфигурацию, но не обрабатывала аудио и не является проверкой качества +Sortformer. + +## Техническая привязка + +- коммит исходного кода `local-transcriber`: + `4c4a8be07b3653cede35b0435e6c9ef6cb3fb87d`; +- коммит NeMo-Speech.cpp: + `4f9676226f667d14608487df744f375db87127f8`; +- модель Streaming Sortformer v2 Q8_0: + `0679cfeb1ce356d0dea9470b31274f4bfc7eb927497d82005483770666da998a`; +- закрытый итоговый файл W0: + `ad32b3fe8a204e2fe6fbb6a13142e22c7e00ce9f0d816c3b0b8889fcab898e89`. + +Последний SHA-256 нужен только для проверки неизменности закрытого результата: +он не оценивает диаризацию и не участвует в решении `stop`. Закрытая основа и +результаты сохраняются для следующих исследований. Их повторное использование +должно быть прямо предусмотрено новым экспериментом; вердикт и расписание +задачи #46 автоматически не переносятся. From 78960120fd14d3a7a771b2297ca9c17cdd88deac Mon Sep 17 00:00:00 2001 From: Dmitry Dementiev Date: Fri, 4 Sep 2026 22:00:31 +0300 Subject: [PATCH 7/7] =?UTF-8?q?docs(benchmark):=20=D0=B4=D0=BE=D0=B1=D0=B0?= =?UTF-8?q?=D0=B2=D0=BB=D0=B5=D0=BD=D0=BE=20=D1=81=D1=80=D0=B0=D0=B2=D0=BD?= =?UTF-8?q?=D0=B5=D0=BD=D0=B8=D0=B5=20W0=20=D0=B8=20Sortformer?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Зачем: - формальный stop W0 не отвечал на практический вопрос о скорости и качестве кандидата. - Что: - добавлены результаты одной диагностической пары на трех фрагментах. - сопоставлены скорость, число кластеров, purity, слова без говорящего и память. - границы вывода отделены от формального расписания эксперимента. - Проверка: - сравнительные агрегаты сверены с закрытыми результатами обеих моделей. - git diff --check. --- .../2026-09-04-streaming-sortformer-ryzen.md | 116 +++++++++--------- 1 file changed, 59 insertions(+), 57 deletions(-) diff --git a/docs/benchmarks/2026-09-04-streaming-sortformer-ryzen.md b/docs/benchmarks/2026-09-04-streaming-sortformer-ryzen.md index 078ca5b..353814d 100644 --- a/docs/benchmarks/2026-09-04-streaming-sortformer-ryzen.md +++ b/docs/benchmarks/2026-09-04-streaming-sortformer-ryzen.md @@ -1,73 +1,75 @@ -# Streaming Sortformer: формальный `stop` на Ryzen +# Streaming Sortformer: сравнение на Ryzen **Дата:** 2026-09-04 -**Статус:** эксперимент завершен как `stop`. Для трех объектов закрытого -манифеста были закреплены количества кластеров `2 / 2 / 3`, но W0 выдал -`2 / 2 / 2`. Кандидат не запускался. +Streaming Sortformer обработал три 300-секундных фрагмента за `67,433` с +против `103,242` с у свежего W0. Ускорение одной пары составило `34,68%`. +Кандидат правильно определил число голосов на всех трех фрагментах; W0 на +одном из них выделил два кластера вместо трех. -## Решение +У результата есть два заметных минуса. Sortformer оставил без говорящего +`92` слова против `73` у W0 и потребовал около `551` МиБ памяти против `325` +МиБ. Во время обеих ячеек была небольшая активность swap, поэтому одна пара +показывает направление, но не устойчивость ускорения. -Свежий W0 не прошел обязательную проверку числа говорящих на одном из трех -300-секундных фрагментов. По условиям -[эксперимента](../specs/2026-09-01-streaming-sortformer-cpu-experiment.md) -такая база сравнения непригодна. Выполнена одна из двенадцати запланированных -ячеек. Sortformer и остальные ячейки не запускались, ускорение не считалось. -Данных о скорости или качестве кандидата этот прогон не дал. +## Сравнение -Во время W0 также зафиксированы чтение из swap и запись в него. Это независимый -провал проверки среды и памяти, но первой причиной `stop` остается -`cluster-count-mismatch`: проверка качества завершилась раньше итогового снимка -системных счетчиков. +| Показатель | W0 | Sortformer | +| --- | ---: | ---: | +| Суммарный wall, с | `103,242` | `67,433` | +| Ускорение | - | `34,68%` | +| Ожидалось кластеров | `2 / 2 / 3` | `2 / 2 / 3` | +| Получено кластеров | `2 / 2 / 2` | `2 / 2 / 3` | +| Ошибок по числу кластеров | `1 из 3` | `0 из 3` | +| Mapped purity на двух размеченных фрагментах | `0,7655 / 0,8915` | `0,8392 / 0,9176` | +| Слов без говорящего | `73` | `92` | +| Пиковый RSS, МиБ | `324,57` | `551,07` | +| CPU time, с | `822,67` | `238,70` | +| Прочитано из swap, КиБ | `492` | `224` | +| Записано в swap, КиБ | `148` | `160` | -## Результат W0 +Обе модели получили одинаковые заранее подготовленные mono/16 kHz float32 +буферы и те же ASR-sidecar. W0 работал на восьми физических ядрах, Sortformer - +с четырьмя закрепленными внутренними потоками. Загрузка моделей, декодирование +аудио, чтение sidecar и расчет метрик в wall не входили. -| Показатель | Значение | -| --- | ---: | -| Суммарный wall, с | `103,242485` | -| Ожидаемые кластеры | `2 / 2 / 3` | -| Полученные кластеры | `2 / 2 / 2` | -| Пиковый RSS | `340 340 736` байт (`324,57` МиБ) | -| Прочитано из swap за ячейку | `503 808` байт (`492` КиБ) | -| Записано в swap за ячейку | `151 552` байт (`148` КиБ) | +Sortformer прошел автоматические проверки формы вероятностей, диапазона +значений, меток говорящих, текста, порядка слов и интервалов на всех трех +фрагментах. Его mapped purity выше W0 на обоих фрагментах с опорной разметкой. +При этом рост числа слов без говорящего требует слуховой проверки: по одной +автоматике нельзя заключить, что качество кандидата в целом лучше. -Первые два фрагмента прошли обязательные проверки числа кластеров, текста, -порядка слов и интервалов. Третий фрагмент содержал подтвержденные три голоса, -но W0 сформировал два кластера. Повтор не назначался: несовпадение числа -кластеров само по себе завершает шаг приемки W0. +## Как читать результат -## Условия +По исходному расписанию задача завершилась как `stop`, когда W0 не прошел +проверку числа кластеров. После этого была выполнена одна диагностическая +ячейка Sortformer на тех же входах, чтобы ответить на практический вопрос о +кандидате. Пять повторных пар, полные записи и слуховая проверка не выполнялись. -W0 выполнялся по закрепленному рецепту `W0-FRESH`: +По этой паре Sortformer выглядит перспективно: ускорение оказалось выше +целевого порога `30%`, модель правильно выделила три голоса на проблемном +фрагменте и показала лучшую mapped purity. Для решения о внедрении нужны +повторные парные замеры без активности swap и слуховая проверка мест, где +слова остались без говорящего. Текущий результат уже достаточен, чтобы не +отбрасывать кандидата и продолжить исследование отдельной задачей. -- `sherpa-onnx/core 1.13.6`, CPU provider; -- Pyannote Segmentation 3.0 FP32 и WeSpeaker ResNet34 LM FP32; -- один процесс, одна сессия, восемь физических ядер, - `intra_op_threads=8`, `inter_op_threads=1`; -- `window_shift_ratio=0.1`, `FastClustering(auto, 0.89)`, - `min_duration_on=0.3`, `min_duration_off=0.5`; -- Windows Balanced, boost mode `2`; Linux cpufreq/governor в WSL2 не - предоставляется. +## Рецепты и техническая привязка -До измерения успешно завершились обязательные детерминированные тесты, запуск -с `--validate-only` и предварительная проверка C ABI и конфигурации кандидата. -Последняя подтвердила четыре выходных канала, шаг `0,08` с и закрепленную -CPU-конфигурацию, но не обрабатывала аудио и не является проверкой качества -Sortformer. +W0 использовал `sherpa-onnx/core 1.13.6`, Pyannote Segmentation 3.0 FP32, +WeSpeaker ResNet34 LM FP32, CPU provider, `window_shift_ratio=0.1`, +`FastClustering(auto, 0.89)`, один процесс и одну сессию, +`intra_op_threads=8`, `inter_op_threads=1`, `min_duration_on=0.3` и +`min_duration_off=0.5`. -## Техническая привязка +Кандидат использовал NeMo-Speech.cpp commit +`4f9676226f667d14608487df744f375db87127f8`, модель Streaming Sortformer v2 +Q8_0 с SHA-256 +`0679cfeb1ce356d0dea9470b31274f4bfc7eb927497d82005483770666da998a`, CPU +backend, один процесс и один stream. Geometry и post-processing соответствовали +[спецификации](../specs/2026-09-01-streaming-sortformer-cpu-experiment.md). -- коммит исходного кода `local-transcriber`: - `4c4a8be07b3653cede35b0435e6c9ef6cb3fb87d`; -- коммит NeMo-Speech.cpp: - `4f9676226f667d14608487df744f375db87127f8`; -- модель Streaming Sortformer v2 Q8_0: - `0679cfeb1ce356d0dea9470b31274f4bfc7eb927497d82005483770666da998a`; -- закрытый итоговый файл W0: - `ad32b3fe8a204e2fe6fbb6a13142e22c7e00ce9f0d816c3b0b8889fcab898e89`. +Коммит исходного кода, выполнившего обе ячейки: +`4c4a8be07b3653cede35b0435e6c9ef6cb3fb87d`. -Последний SHA-256 нужен только для проверки неизменности закрытого результата: -он не оценивает диаризацию и не участвует в решении `stop`. Закрытая основа и -результаты сохраняются для следующих исследований. Их повторное использование -должно быть прямо предусмотрено новым экспериментом; вердикт и расписание -задачи #46 автоматически не переносятся. +Закрытая основа и результаты сохраняются для следующих исследований. Их +повторное использование должно быть явно описано в новой задаче.