From 40e645e56f9d51b02fba07e135cddebe19d581a2 Mon Sep 17 00:00:00 2001 From: Jer Miller Date: Wed, 29 Jul 2026 12:45:29 -0600 Subject: [PATCH] fix(speakers-analyze): let supervised descendants borrow the install generation Move the speakers-analyze entry path to owned-or-borrowed generation handles. Supervised descendants can reuse the proof only when they hold the inherited descriptor capability, while direct second entrants still fail closed instead of trusting copied environment strings. Co-Authored-By: Claude Opus 5 (1M context) --- solstone/observe/sense.py | 4 +- .../think/speakers_analyze_installation.py | 235 ++++++++++++--- tests/conftest.py | 25 +- tests/helpers/speakers_analyze.py | 56 ++++ tests/test_sense.py | 97 +++++- tests/test_speakers_analyze_installation.py | 281 +++++++++++++++++- 6 files changed, 624 insertions(+), 74 deletions(-) create mode 100644 tests/helpers/speakers_analyze.py diff --git a/solstone/observe/sense.py b/solstone/observe/sense.py index 583fd12c2..294e56335 100644 --- a/solstone/observe/sense.py +++ b/solstone/observe/sense.py @@ -1409,11 +1409,11 @@ def main(): sensor.register(f"*{ext}", "depict", ["journal", "depict", "{file}"]) from solstone.think.speakers_analyze_installation import ( - begin_speakers_analyze_generation, + enter_speakers_analyze_generation, ) try: - sensor._speakers_analyze_generation = begin_speakers_analyze_generation( + sensor._speakers_analyze_generation = enter_speakers_analyze_generation( journal_path=journal ) except Exception as exc: diff --git a/solstone/think/speakers_analyze_installation.py b/solstone/think/speakers_analyze_installation.py index 51ef77465..0336b3a75 100644 --- a/solstone/think/speakers_analyze_installation.py +++ b/solstone/think/speakers_analyze_installation.py @@ -8,6 +8,7 @@ from __future__ import annotations import hashlib import json import os +import secrets import sys import uuid from collections.abc import Callable @@ -27,9 +28,12 @@ from solstone.apps.speakers.encoder_config import ( from solstone.think import probe from solstone.think.journal_io import MalformedPolicy, atomic_replace, read_json from solstone.think.journal_io.lease import ( + BorrowedFileLease, FileLease, acquire_file_lease, - probe_file_lease_held, + adopt_inherited_file_lease_fd, + read_file_lease_fd, + set_file_lease_offset_token, ) from solstone.think.model_assets import ( ModelsDistributionUnavailable, @@ -45,7 +49,12 @@ ROOT_DIST_NAME = "solstone" INSTALL_GENERATION_SCHEMA = "solstone.speakers_analyze.install_generation.v1" PROOF_KEY_SCHEMA = "solstone.speakers_analyze.install_proof_key.v1" GENERATION_ENV_KEY = "SOL_SPEAKERS_ANALYZE_INSTALL_GENERATION_ID" +GENERATION_FD_ENV_KEY = "SOL_SPEAKERS_ANALYZE_INSTALL_GENERATION_FD" +GENERATION_TOKEN_ENV_KEY = "SOL_SPEAKERS_ANALYZE_INSTALL_GENERATION_TOKEN" GENERATION_MODE = 0o600 +GENERATION_FD_MIN = 3 +GENERATION_FD_MAX = 1_048_576 +GENERATION_TOKEN_MAX = (1 << 62) - 1 SPEAKERS_ANALYZE_REPAIR_TEXT = ( "Repair: reinstall the journal host stack with solstone-journal, or " @@ -87,9 +96,15 @@ class SpeakersAnalyzeInstallationResult: @dataclass(frozen=True) class SpeakersAnalyzeGeneration: generation_id: str - lease: FileLease + lease: FileLease | BorrowedFileLease def release(self) -> None: + if isinstance(self.lease, FileLease) and ( + os.environ.get(GENERATION_ENV_KEY) == self.generation_id + ): + os.environ.pop(GENERATION_ENV_KEY, None) + os.environ.pop(GENERATION_FD_ENV_KEY, None) + os.environ.pop(GENERATION_TOKEN_ENV_KEY, None) self.lease.release() def __enter__(self) -> SpeakersAnalyzeGeneration: @@ -164,7 +179,7 @@ def check_speakers_analyze_installation( return _digest_assets(proof_key) -def begin_speakers_analyze_generation( +def enter_speakers_analyze_generation( *, journal_path: str | Path | None = None, executable: str | Path | None = None, @@ -174,45 +189,63 @@ def begin_speakers_analyze_generation( ] = probe.current_solstone_core_platform, ) -> SpeakersAnalyzeGeneration: root = _journal_root(journal_path) - lease = acquire_file_lease( - _generation_lease_path(root), attempts=1, retry_max_seconds=0 - ) - if lease is None: - raise RuntimeError("speakers-analyze generation lease is already held") - cheap = _cheap_installation_result( + borrowed = _borrow_speakers_analyze_generation( + root=root, executable=executable, version_reader=version_reader, platform_reader=platform_reader, - platform_tag_reader=_packaging_platform_tags, - executable_predicate=lambda path: os.access(path, os.X_OK), ) - if not cheap.ok: - lease.release() - raise RuntimeError(cheap.message) - proof_key = _installation_proof_key( - executable=executable, - version_reader=version_reader, - platform_reader=platform_reader, + if borrowed is not None: + return borrowed + + lease = acquire_file_lease( + _generation_lease_path(root), attempts=1, retry_max_seconds=0 ) - result, observed_assets = _validated_asset_digests(proof_key) - if not result.ok: + if lease is None: + raise RuntimeError("speakers-analyze generation lease is already held") + try: + cheap = _cheap_installation_result( + executable=executable, + version_reader=version_reader, + platform_reader=platform_reader, + platform_tag_reader=_packaging_platform_tags, + executable_predicate=lambda path: os.access(path, os.X_OK), + ) + if not cheap.ok: + raise RuntimeError(cheap.message) + proof_key = _installation_proof_key( + executable=executable, + version_reader=version_reader, + platform_reader=platform_reader, + ) + result, observed_assets = _validated_asset_digests(proof_key) + if not result.ok: + raise RuntimeError(result.message) + generation_id = uuid.uuid4().hex + token = secrets.randbelow(GENERATION_TOKEN_MAX) + 1 + set_file_lease_offset_token(lease, token, _generation_lease_path(root)) + record = { + "schema": INSTALL_GENERATION_SCHEMA, + "generation_id": generation_id, + "created_at": _now_iso(), + "verified_at": _now_iso(), + "proof_key": proof_key, + "assets": observed_assets, + "helper": proof_key["helper"], + "packages": proof_key["packages"], + "platform": proof_key["platform"], + } + _write_generation_record(root, record) + os.environ[GENERATION_ENV_KEY] = generation_id + os.environ[GENERATION_FD_ENV_KEY] = str( + read_file_lease_fd(lease, _generation_lease_path(root)) + ) + os.environ[GENERATION_TOKEN_ENV_KEY] = str(token) + return SpeakersAnalyzeGeneration(generation_id=generation_id, lease=lease) + except BaseException: + _clear_generation_env() lease.release() - raise RuntimeError(result.message) - generation_id = uuid.uuid4().hex - record = { - "schema": INSTALL_GENERATION_SCHEMA, - "generation_id": generation_id, - "created_at": _now_iso(), - "verified_at": _now_iso(), - "proof_key": proof_key, - "assets": observed_assets, - "helper": proof_key["helper"], - "packages": proof_key["packages"], - "platform": proof_key["platform"], - } - _write_generation_record(root, record) - os.environ[GENERATION_ENV_KEY] = generation_id - return SpeakersAnalyzeGeneration(generation_id=generation_id, lease=lease) + raise def _cheap_installation_result( @@ -379,8 +412,81 @@ def _generation_proves_digest( if not generation_id: return False root = _journal_root(journal_path) - if not probe_file_lease_held(_generation_lease_path(root)): + candidate = _inherited_generation_candidate() + if candidate is None: return False + fd, token = candidate + borrowed = adopt_inherited_file_lease_fd( + _generation_lease_path(root), fd=fd, token=token + ) + if borrowed is None: + return False + try: + return _generation_record_proves_digest( + root=root, + generation_id=generation_id, + proof_key=proof_key, + ) + finally: + borrowed.release() + + +def _borrow_speakers_analyze_generation( + *, + root: Path, + executable: str | Path | None, + version_reader: Callable[[str], str], + platform_reader: Callable[[], probe.CorePlatform], +) -> SpeakersAnalyzeGeneration | None: + generation_id = os.environ.get(GENERATION_ENV_KEY) + if not any( + os.environ.get(key) + for key in ( + GENERATION_ENV_KEY, + GENERATION_FD_ENV_KEY, + GENERATION_TOKEN_ENV_KEY, + ) + ): + return None + candidate = _inherited_generation_candidate() + if generation_id is None or candidate is None: + _reject_generation_borrow() + return None + fd, token = candidate + borrowed = adopt_inherited_file_lease_fd( + _generation_lease_path(root), fd=fd, token=token + ) + if borrowed is None: + _reject_generation_borrow(candidate_fd=fd) + return None + try: + proof_key = _installation_proof_key( + executable=executable, + version_reader=version_reader, + platform_reader=platform_reader, + ) + if _generation_record_proves_digest( + root=root, + generation_id=generation_id, + proof_key=proof_key, + ): + return SpeakersAnalyzeGeneration( + generation_id=generation_id, lease=borrowed + ) + except BaseException: + borrowed.release() + raise + borrowed.release() + _reject_generation_borrow(candidate_fd=fd) + return None + + +def _generation_record_proves_digest( + *, + root: Path, + generation_id: str, + proof_key: dict[str, object], +) -> bool: raw = read_json( _generation_record_path(root), on_error=MalformedPolicy.WARN_AND_SKIP, @@ -405,6 +511,57 @@ def _generation_proves_digest( return True +def _inherited_generation_candidate() -> tuple[int, int] | None: + fd = _parse_generation_fd(os.environ.get(GENERATION_FD_ENV_KEY)) + token = _parse_generation_token(os.environ.get(GENERATION_TOKEN_ENV_KEY)) + if fd is None or token is None: + return None + try: + os.fstat(fd) + except OSError: + return None + return fd, token + + +def _parse_generation_fd(value: str | None) -> int | None: + if value is None: + return None + try: + fd = int(value) + except (TypeError, ValueError): + return None + if fd < GENERATION_FD_MIN or fd > GENERATION_FD_MAX: + return None + return fd + + +def _parse_generation_token(value: str | None) -> int | None: + if value is None: + return None + try: + token = int(value) + except (TypeError, ValueError): + return None + if token <= 0 or token > GENERATION_TOKEN_MAX: + return None + return token + + +def _reject_generation_borrow(candidate_fd: int | None = None) -> None: + if candidate_fd is not None: + try: + os.close(candidate_fd) + except OSError: + pass + _clear_generation_env() + + +def _clear_generation_env() -> None: + os.environ.pop(GENERATION_ENV_KEY, None) + os.environ.pop(GENERATION_FD_ENV_KEY, None) + os.environ.pop(GENERATION_TOKEN_ENV_KEY, None) + + def _required_assets() -> tuple[tuple[str, Path, str], ...]: return ( ("wespeaker", resolve_wespeaker_model(), WESPEAKER_MODEL_SHA256), @@ -445,7 +602,9 @@ def _now_iso() -> str: __all__ = [ + "GENERATION_FD_ENV_KEY", "GENERATION_ENV_KEY", + "GENERATION_TOKEN_ENV_KEY", "HELPER_BINARY_NAME", "HELPER_DIST_NAME", "MODELS_DIST_NAME", @@ -453,8 +612,8 @@ __all__ = [ "SPEAKERS_ANALYZE_REPAIR_TEXT", "SpeakersAnalyzeGeneration", "SpeakersAnalyzeInstallationResult", - "begin_speakers_analyze_generation", "check_speakers_analyze_installation", + "enter_speakers_analyze_generation", "runtime_has_speakers_analyze_wheel_coverage", "speakers_analyze_path_for_executable", ] diff --git a/tests/conftest.py b/tests/conftest.py index 31501bd13..b1f515a0e 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -191,7 +191,9 @@ def set_test_journal_path(monkeypatch, _isolate_os_environ): @pytest.fixture(autouse=True) def _speakers_analyze_startup_invariant_ready( - monkeypatch: pytest.MonkeyPatch, request: pytest.FixtureRequest + monkeypatch: pytest.MonkeyPatch, + request: pytest.FixtureRequest, + tmp_path: Path, ) -> None: """Keep unit tests independent of the installed native helper wheel. @@ -202,31 +204,14 @@ def _speakers_analyze_startup_invariant_ready( return from solstone.think import speakers_analyze_installation as installation - - class _NoopLease: - def release(self) -> None: - return None - - def begin_ready_generation( - **_kwargs: object, - ) -> installation.SpeakersAnalyzeGeneration: - generation_id = "test-speakers-analyze-generation" - os.environ[installation.GENERATION_ENV_KEY] = generation_id - return installation.SpeakersAnalyzeGeneration( - generation_id=generation_id, - lease=_NoopLease(), - ) + from tests.helpers.speakers_analyze import install_enter_generation_stub monkeypatch.setattr( installation, "check_speakers_analyze_installation", lambda **_kwargs: installation.SpeakersAnalyzeInstallationResult("ok"), ) - monkeypatch.setattr( - installation, - "begin_speakers_analyze_generation", - begin_ready_generation, - ) + install_enter_generation_stub(monkeypatch, tmp_path) @pytest.fixture(autouse=True) diff --git a/tests/helpers/speakers_analyze.py b/tests/helpers/speakers_analyze.py new file mode 100644 index 000000000..cfaaf7aa6 --- /dev/null +++ b/tests/helpers/speakers_analyze.py @@ -0,0 +1,56 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Test helpers for speakers-analyze generation startup seams.""" + +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + + +def install_enter_generation_stub( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + *, + generation_id: str = "test-speakers-analyze-generation", +) -> None: + from solstone.think import speakers_analyze_installation as installation + + class _NoopLease: + def __init__(self, fd: int) -> None: + self.fd = fd + + def release(self) -> None: + if os.environ.get(installation.GENERATION_ENV_KEY) == generation_id: + monkeypatch.delenv(installation.GENERATION_ENV_KEY, raising=False) + monkeypatch.delenv(installation.GENERATION_FD_ENV_KEY, raising=False) + monkeypatch.delenv(installation.GENERATION_TOKEN_ENV_KEY, raising=False) + try: + os.close(self.fd) + except OSError: + pass + + def enter_ready_generation(**_kwargs: object): + token = 1 + fd = os.open( + tmp_path / "speakers-analyze-generation.fake", + os.O_RDWR | os.O_CREAT, + 0o600, + ) + os.lseek(fd, token, os.SEEK_SET) + monkeypatch.setenv(installation.GENERATION_ENV_KEY, generation_id) + monkeypatch.setenv(installation.GENERATION_FD_ENV_KEY, str(fd)) + monkeypatch.setenv(installation.GENERATION_TOKEN_ENV_KEY, str(token)) + return installation.SpeakersAnalyzeGeneration( + generation_id=generation_id, + lease=_NoopLease(fd), + ) + + monkeypatch.setattr( + installation, + "enter_speakers_analyze_generation", + enter_ready_generation, + ) diff --git a/tests/test_sense.py b/tests/test_sense.py index ab9415cf3..73baaf26d 100644 --- a/tests/test_sense.py +++ b/tests/test_sense.py @@ -5,6 +5,7 @@ import json import logging +import os import signal import subprocess import sys @@ -67,7 +68,7 @@ def _speakers_analyze_generation_ready(monkeypatch): monkeypatch.setattr( "solstone.think.speakers_analyze_installation." - "begin_speakers_analyze_generation", + "enter_speakers_analyze_generation", lambda **_kwargs: Generation(), ) @@ -2890,7 +2891,7 @@ def test_main_speakers_analyze_failure_prints_canonical_message_once( monkeypatch.setattr( "solstone.think.speakers_analyze_installation." - "begin_speakers_analyze_generation", + "enter_speakers_analyze_generation", fail_generation, ) @@ -2903,6 +2904,43 @@ def test_main_speakers_analyze_failure_prints_canonical_message_once( assert message in captured.err +def test_main_live_generation_id_without_fd_still_prints_canonical_message_once( + tmp_path, monkeypatch, capsys +): + from solstone.observe import sense + + message = "speakers-analyze generation lease is already held" + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) + monkeypatch.setenv( + "SOL_SPEAKERS_ANALYZE_INSTALL_GENERATION_ID", + "live-generation", + ) + monkeypatch.setenv("SOL_SUPERVISOR_SPAWNED", "1") + monkeypatch.delenv("SOL_SPEAKERS_ANALYZE_INSTALL_GENERATION_FD", raising=False) + monkeypatch.delenv( + "SOL_SPEAKERS_ANALYZE_INSTALL_GENERATION_TOKEN", + raising=False, + ) + monkeypatch.setattr(sense, "require_solstone", lambda: None) + monkeypatch.setattr(sys, "argv", ["sense", "--day", "20250101"]) + + def fail_generation(**_kwargs): + raise RuntimeError(message) + + monkeypatch.setattr( + "solstone.think.speakers_analyze_installation." + "enter_speakers_analyze_generation", + fail_generation, + ) + + with pytest.raises(SystemExit) as exc_info: + sense.main() + + captured = capsys.readouterr() + assert exc_info.value.code == 78 + assert captured.err.count(message) == 1 + + def _registered_describe_commands(sensor: FileSensor) -> list[list[str]]: return [ command @@ -3098,6 +3136,61 @@ def test_queue_wait_ms_reaches_child_env(tmp_path, monkeypatch): assert "SOL_QUEUE_WAIT_MS" not in captured["env"] +def test_generation_env_reaches_event_and_batch_transcribe_children( + tmp_path, monkeypatch +): + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) + fd = os.open(tmp_path / "generation.lock", os.O_RDWR | os.O_CREAT, 0o600) + monkeypatch.setenv( + "SOL_SPEAKERS_ANALYZE_INSTALL_GENERATION_ID", + "generation", + ) + monkeypatch.setenv("SOL_SPEAKERS_ANALYZE_INSTALL_GENERATION_FD", str(fd)) + monkeypatch.setenv("SOL_SPEAKERS_ANALYZE_INSTALL_GENERATION_TOKEN", "123") + sensor = FileSensor(tmp_path) + test_file = make_segment_file(tmp_path, "audio.flac") + captured: list[dict[str, str]] = [] + + def fake_runner_spawn(cmd, ref, callosum, env, day): + captured.append(env) + return FakeManaged(FakeProcess(0), ref=ref) + + monkeypatch.setattr( + "solstone.observe.sense.RunnerManagedProcess.spawn", fake_runner_spawn + ) + + try: + sensor._spawn_managed_process( + ["journal", "transcribe", str(test_file)], + test_file, + "event-ref", + "143022_300", + None, + None, + None, + ) + sensor._spawn_managed_process( + ["journal", "transcribe", str(test_file)], + test_file, + "batch-ref", + "143022_300", + None, + None, + "20250101", + ) + finally: + os.close(fd) + + assert [ + ( + env["SOL_SPEAKERS_ANALYZE_INSTALL_GENERATION_ID"], + env["SOL_SPEAKERS_ANALYZE_INSTALL_GENERATION_FD"], + env["SOL_SPEAKERS_ANALYZE_INSTALL_GENERATION_TOKEN"], + ) + for env in captured + ] == [("generation", str(fd), "123"), ("generation", str(fd), "123")] + + def test_run_handler_passes_queue_wait_from_queued_at(tmp_path, monkeypatch): monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) sensor = FileSensor(tmp_path) diff --git a/tests/test_speakers_analyze_installation.py b/tests/test_speakers_analyze_installation.py index c936ce6c1..db4c41755 100644 --- a/tests/test_speakers_analyze_installation.py +++ b/tests/test_speakers_analyze_installation.py @@ -5,6 +5,7 @@ from __future__ import annotations +import json import os from importlib.metadata import PackageNotFoundError from pathlib import Path @@ -13,6 +14,13 @@ import pytest from solstone.think import probe from solstone.think import speakers_analyze_installation as installation +from solstone.think.journal_io.lease import ( + BorrowedFileLease, + FileLease, + acquire_file_lease, + read_file_lease_fd, + read_file_lease_offset_token, +) def _version_reader(dist_name: str) -> str: @@ -61,6 +69,34 @@ def _asset_fixtures(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: ) +def _entry_kwargs(tmp_path: Path, executable: Path) -> dict: + return { + "journal_path": tmp_path, + "executable": executable, + "version_reader": _version_reader, + "platform_reader": _platform_reader, + } + + +def _clear_generation_env() -> None: + os.environ.pop(installation.GENERATION_ENV_KEY, None) + os.environ.pop(installation.GENERATION_FD_ENV_KEY, None) + os.environ.pop(installation.GENERATION_TOKEN_ENV_KEY, None) + + +def _restore_generation_env( + monkeypatch: pytest.MonkeyPatch, generation_id: str, fd: int, token: int +) -> None: + monkeypatch.setenv(installation.GENERATION_ENV_KEY, generation_id) + monkeypatch.setenv(installation.GENERATION_FD_ENV_KEY, str(fd)) + monkeypatch.setenv(installation.GENERATION_TOKEN_ENV_KEY, str(token)) + + +def _assert_fd_closed(fd: int) -> None: + with pytest.raises(OSError): + os.fstat(fd) + + def test_coverage_gate_reads_helper_constants_not_core_constants(monkeypatch): core_platform = ("coreos", "core64") helper_platform = ("helperos", "helper64") @@ -225,11 +261,8 @@ def test_live_generation_record_reuses_digest_proof( ): executable = _helper(tmp_path) _asset_fixtures(tmp_path, monkeypatch) - generation = installation.begin_speakers_analyze_generation( - journal_path=tmp_path, - executable=executable, - version_reader=_version_reader, - platform_reader=_platform_reader, + generation = installation.enter_speakers_analyze_generation( + **_entry_kwargs(tmp_path, executable) ) calls = 0 @@ -251,7 +284,7 @@ def test_live_generation_record_reuses_digest_proof( ) finally: generation.release() - os.environ.pop(installation.GENERATION_ENV_KEY, None) + _clear_generation_env() assert result.status == "ok" assert calls == 0 @@ -262,15 +295,12 @@ def test_stale_generation_record_degrades_to_full_digest( ): executable = _helper(tmp_path) _asset_fixtures(tmp_path, monkeypatch) - generation = installation.begin_speakers_analyze_generation( - journal_path=tmp_path, - executable=executable, - version_reader=_version_reader, - platform_reader=_platform_reader, + generation = installation.enter_speakers_analyze_generation( + **_entry_kwargs(tmp_path, executable) ) generation_id = generation.generation_id generation.release() - os.environ.pop(installation.GENERATION_ENV_KEY, None) + _clear_generation_env() calls = 0 original_digest = installation._sha256_file @@ -292,3 +322,230 @@ def test_stale_generation_record_degrades_to_full_digest( assert result.status == "ok" assert calls == 2 + + +def test_owned_entry_publishes_fd_token_and_token_free_record( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +): + executable = _helper(tmp_path) + _asset_fixtures(tmp_path, monkeypatch) + + generation = installation.enter_speakers_analyze_generation( + **_entry_kwargs(tmp_path, executable) + ) + try: + assert isinstance(generation.lease, FileLease) + fd = int(os.environ[installation.GENERATION_FD_ENV_KEY]) + token = int(os.environ[installation.GENERATION_TOKEN_ENV_KEY]) + assert os.environ[installation.GENERATION_ENV_KEY] == generation.generation_id + assert token > 0 + assert read_file_lease_fd(generation.lease) == fd + assert read_file_lease_offset_token(fd) == token + + record = json.loads( + ( + tmp_path / "health" / "speakers-analyze" / "install-generation.json" + ).read_text(encoding="utf-8") + ) + assert record["schema"] == installation.INSTALL_GENERATION_SCHEMA + assert record["generation_id"] == generation.generation_id + assert "token" not in record + assert installation.GENERATION_TOKEN_ENV_KEY not in record + finally: + generation.release() + _clear_generation_env() + + +def test_borrowed_entry_reuses_proof_without_hash_and_keeps_owner_live( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +): + executable = _helper(tmp_path) + _asset_fixtures(tmp_path, monkeypatch) + owner = installation.enter_speakers_analyze_generation( + **_entry_kwargs(tmp_path, executable) + ) + owner_fd = int(os.environ[installation.GENERATION_FD_ENV_KEY]) + token = int(os.environ[installation.GENERATION_TOKEN_ENV_KEY]) + inherited_fd = os.dup(owner_fd) + _restore_generation_env(monkeypatch, owner.generation_id, inherited_fd, token) + calls = 0 + + def fail_digest(_path: Path) -> str: + nonlocal calls + calls += 1 + raise AssertionError("borrow must reuse the live proof") + + monkeypatch.setattr(installation, "_sha256_file", fail_digest) + borrower = installation.enter_speakers_analyze_generation( + **_entry_kwargs(tmp_path, executable) + ) + try: + assert isinstance(borrower.lease, BorrowedFileLease) + assert calls == 0 + borrower.release() + assert ( + acquire_file_lease( + tmp_path / "health" / "speakers-analyze" / "install-generation.lock", + attempts=1, + ) + is None + ) + _restore_generation_env(monkeypatch, owner.generation_id, inherited_fd, token) + assert ( + installation.check_speakers_analyze_installation( + journal_path=tmp_path, + executable=executable, + version_reader=_version_reader, + platform_reader=_platform_reader, + platform_tag_reader=_platform_tags, + ).status + == "ok" + ) + finally: + try: + os.close(inherited_fd) + except OSError: + pass + owner.release() + _clear_generation_env() + + +def test_copied_or_separately_opened_generation_env_cannot_borrow( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +): + executable = _helper(tmp_path) + _asset_fixtures(tmp_path, monkeypatch) + owner = installation.enter_speakers_analyze_generation( + **_entry_kwargs(tmp_path, executable) + ) + owner_fd = int(os.environ[installation.GENERATION_FD_ENV_KEY]) + token = int(os.environ[installation.GENERATION_TOKEN_ENV_KEY]) + lock_path = tmp_path / "health" / "speakers-analyze" / "install-generation.lock" + + cases: list[tuple[str, int | None]] = [] + cases.append(("missing-fd", None)) + unrelated_fd = os.open(tmp_path / "unrelated.lock", os.O_RDWR | os.O_CREAT, 0o600) + cases.append(("unrelated-fd", unrelated_fd)) + separate_fd = os.open(lock_path, os.O_RDWR) + os.lseek(separate_fd, token, os.SEEK_SET) + cases.append(("separate-same-path-fd", separate_fd)) + mismatched_id_fd = os.dup(owner_fd) + cases.append(("mismatched-id", mismatched_id_fd)) + + try: + for name, fd in cases: + _restore_generation_env( + monkeypatch, + "stale-generation" if name == "mismatched-id" else owner.generation_id, + fd if fd is not None else owner_fd, + token, + ) + if name == "missing-fd": + os.environ.pop(installation.GENERATION_FD_ENV_KEY, None) + monkeypatch.setenv("SOL_SUPERVISOR_SPAWNED", "1") + with pytest.raises(RuntimeError, match="generation lease is already held"): + installation.enter_speakers_analyze_generation( + **_entry_kwargs(tmp_path, executable) + ) + assert installation.GENERATION_ENV_KEY not in os.environ + assert installation.GENERATION_FD_ENV_KEY not in os.environ + assert installation.GENERATION_TOKEN_ENV_KEY not in os.environ + if fd is not None: + _assert_fd_closed(fd) + finally: + owner.release() + _clear_generation_env() + + +def test_final_duplicate_rejection_closes_candidate_before_fresh_entry( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +): + executable = _helper(tmp_path) + _asset_fixtures(tmp_path, monkeypatch) + owner = installation.enter_speakers_analyze_generation( + **_entry_kwargs(tmp_path, executable) + ) + old_fd = int(os.environ[installation.GENERATION_FD_ENV_KEY]) + token = int(os.environ[installation.GENERATION_TOKEN_ENV_KEY]) + final_duplicate_fd = os.dup(old_fd) + os.close(old_fd) + owner.lease._fd = None + _restore_generation_env(monkeypatch, "stale-generation", final_duplicate_fd, token) + + calls = 0 + original_digest = installation._sha256_file + + def counted_digest(path: Path) -> str: + nonlocal calls + calls += 1 + return original_digest(path) + + monkeypatch.setattr(installation, "_sha256_file", counted_digest) + fresh = installation.enter_speakers_analyze_generation( + **_entry_kwargs(tmp_path, executable) + ) + try: + assert fresh.generation_id != owner.generation_id + assert calls == 2 + _assert_fd_closed(final_duplicate_fd) + finally: + fresh.release() + _clear_generation_env() + + +@pytest.mark.parametrize( + "failure", ["proof-key", "validation", "token-init", "record-write"] +) +def test_owned_entry_failures_release_and_allow_reacquire( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, failure: str +): + executable = _helper(tmp_path) + _asset_fixtures(tmp_path, monkeypatch) + if failure == "proof-key": + monkeypatch.setattr( + installation, + "_installation_proof_key", + lambda **_kwargs: (_ for _ in ()).throw(RuntimeError("proof failed")), + ) + elif failure == "validation": + monkeypatch.setattr( + installation, + "_validated_asset_digests", + lambda _proof_key: ( + installation.SpeakersAnalyzeInstallationResult( + "asset-missing", "validation failed" + ), + [], + ), + ) + elif failure == "token-init": + monkeypatch.setattr( + installation, + "set_file_lease_offset_token", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + RuntimeError("token failed") + ), + ) + else: + monkeypatch.setattr( + installation, + "_write_generation_record", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + RuntimeError("write failed") + ), + ) + + with pytest.raises(RuntimeError): + installation.enter_speakers_analyze_generation( + **_entry_kwargs(tmp_path, executable) + ) + assert installation.GENERATION_ENV_KEY not in os.environ + assert installation.GENERATION_FD_ENV_KEY not in os.environ + assert installation.GENERATION_TOKEN_ENV_KEY not in os.environ + + lease = acquire_file_lease( + tmp_path / "health" / "speakers-analyze" / "install-generation.lock", + attempts=1, + ) + assert lease is not None + lease.release() -- 2.51.2