From 9e4feb670514d350edf3cffa9289b9c10f597edd Mon Sep 17 00:00:00 2001 From: Ben Westgate Date: Fri, 9 Oct 2026 21:20:08 +0000 Subject: [PATCH] cli: Share the library correction driver The CLI ran its own two-pass mixed-case search with its own deadline, capture ledger, and length plan. It now calls the library's _correct_complete, which gains the CLI's set-validity filter and the competitor scheduler, and keeps only the Hamming, CRC, and fingerprint tie breakers. Incomplete searches now mark every returned candidate. Claude-Session: https://claude.ai/code/session_01XGMpowzJwwgyHMEd63ijDe The public API retains fail-closed required work; only the CLI opts into the competitor scheduler. The expired-work regression is folded into this commit. --- src/codex32/_cli_input.py | 230 +++++----------------------- src/codex32/cli.py | 22 +-- src/codex32/correction.py | 23 ++- tests/test_cli.py | 87 +---------- tests/test_correction_disclosure.py | 18 +-- tests/test_correction_indel.py | 34 +++- tests/test_mixed_case_deadline.py | 82 ---------- 7 files changed, 102 insertions(+), 394 deletions(-) delete mode 100644 tests/test_mixed_case_deadline.py diff --git a/src/codex32/_cli_input.py b/src/codex32/_cli_input.py index 245244c1..739505fd 100644 --- a/src/codex32/_cli_input.py +++ b/src/codex32/_cli_input.py @@ -7,19 +7,14 @@ import os import sys from collections.abc import Callable, Iterator, Sequence -from dataclasses import replace -from functools import partial -from time import monotonic from typing import Any, Literal, cast -from codex32.bech32 import interpret_mixed_case from codex32.bip93 import ( MS, TEXT_LENGTHS, MasterSeed, Secret, Share, - _checksum_for_encoded_length, _text_length, _validate_ms_length, _validate_recovery_prefix, @@ -27,12 +22,13 @@ parse_codex32, recover_secret, ) -from codex32.correction import CorrectionCandidate, CorrectionContext, _best, _capture_mass +from codex32.correction import CorrectionCandidate, CorrectionContext, _best, _correct_complete from codex32.errors import ( CodexError, DuplicateShareIndex, ExistingTargetIndex, InvalidChecksum, + InvalidCorrectionInput, InvalidLength, InvalidThreshold, MismatchedHrp, @@ -357,160 +353,6 @@ def _display_prefix(prefix: str, grouped: bool) -> str: return displayed -def _case_interpretation( - value: str, - prefix: str, - hrp: str | None, - allowed: Callable[[CorrectionCandidate], bool] | None, -) -> tuple[CorrectionCandidate | None, str, str, str] | None: - # Normalize likely casing and mark contrary-case data as erasures. - separator = value.find("1") - base_length = separator + 1 if separator >= 0 else 0 - immutable_length = len(prefix) if prefix and value.lower().startswith(prefix.lower()) else base_length - interpretation = interpret_mixed_case(value, immutable_length) - if interpretation is None: - return None - corrected, erased, uppercase = interpretation - corrected_prefix = prefix.upper() if uppercase else prefix.lower() - try: - artifact = _parse(corrected, hrp) - except InputError: - candidate = None - else: - bits = ( - 5 * _checksum_for_encoded_length(artifact.hrp, len(artifact.text) - len(artifact.hrp) - 1).length - ) - proposed = CorrectionCandidate(artifact, (), 1, 0, 0, None, capture_space_bits=bits) - candidate = proposed if allowed is None or allowed(proposed) else None - return candidate, corrected, erased, corrected_prefix - - -_PRIMARY_MS = (48, 74, 127) - - -def _correction_plan( - hrp: str, - byte_length: int | Literal["?"] | None, - count: int, - target: int | None, -) -> tuple[int, ...]: - if target is not None: - return (target,) - if isinstance(byte_length, int): - return (_text_length(byte_length),) - if byte_length == "?": - return TEXT_LENGTHS - if hrp.lower() == MS: - nearest = min(_PRIMARY_MS, key=lambda length: abs(count - length)) - return (nearest, *(length for length in TEXT_LENGTHS if length != nearest)) - return tuple(sorted({count + delta for delta in (*range(-4, 5), -8, 8)})) - - -def _correction_candidates( - value: str, - hrp: str, - byte_length: int | Literal["?"] | None, - immutable: str, - excluded: tuple[str, ...] = (), - *, - target: int | None = None, - allowed: Callable[[CorrectionCandidate], bool] | None = None, - deadline: float | None = None, - capture_layers: list[tuple[int, int]] | None = None, - fingerprint_match: Callable[[CorrectionCandidate], bool | None] | None = None, - seed_candidates: Sequence[CorrectionCandidate] = (), - required_only: bool = False, - optional_only: bool = False, -) -> tuple[tuple[CorrectionCandidate, ...], bool, float]: - count = len(value.replace(" ", "")) - targets = _correction_plan(hrp, byte_length, count, target) - deadline = monotonic() + 10 if deadline is None else deadline - contexts = tuple(CorrectionContext(hrp, length, immutable, excluded) for length in targets) - from codex32.indel import _search_many - - candidates, complete = _search_many( - contexts, - value, - primary=frozenset(targets), - deadline=deadline, - competitors=True, - allowed=allowed, - capture_layers=capture_layers, - seed_candidates=seed_candidates, - required_only=required_only, - optional_only=optional_only, - ) - if allowed is not None: - candidates = tuple(candidate for candidate in candidates if allowed(candidate)) - results = ( - _best(candidates, prefer_common=byte_length == "?", fingerprint_match=fingerprint_match) - if complete - else candidates - if len(candidates) == 1 and not candidates[0].search_complete - else () - ) - return results, complete, deadline - - -def _scheduled_candidates( - value: str, - erased: str, - hrp: str, - byte_length: int | Literal["?"] | None, - immutable: str, - excluded: tuple[str, ...] = (), - *, - target: int | None = None, - allowed: Callable[[CorrectionCandidate], bool] | None = None, - deadline: float | None = None, - fingerprint_match: Callable[[CorrectionCandidate], bool | None] | None = None, -) -> tuple[tuple[CorrectionCandidate, ...], bool]: - """Search both case interpretations under one deadline and capture ledger.""" - search = partial( - _correction_candidates, - hrp=hrp, - byte_length=byte_length, - immutable=immutable, - excluded=excluded, - target=target, - allowed=allowed, - fingerprint_match=fingerprint_match, - ) - first, retry = (erased, value) if erased != value else (value, None) - seeded: tuple[CorrectionCandidate, ...] = () - if retry is not None: - # Find required candidates for both case interpretations before either - # full search can spend the shared deadline on optional alignment. - # These discovery passes deliberately do not charge capture_layers; - # the full searches below account each admitted frontier once. - for required_value in (first, retry): - seeded, complete, deadline = search( - required_value, deadline=deadline, seed_candidates=seeded, required_only=True - ) - if not complete: - return (), False - capture_layers: list[tuple[int, int]] = [] - full_search = partial(search, capture_layers=capture_layers, optional_only=retry is not None) - candidates, complete, deadline = full_search(first, deadline=deadline, seed_candidates=seeded) - if retry is None: - return candidates, complete - retry_candidates, retry_complete, _deadline = full_search( - retry, deadline=deadline, seed_candidates=(*seeded, *candidates) - ) - complete = complete and retry_complete - annotated = [] - for item in (*candidates, *retry_candidates): - volume, bits = _capture_mass(capture_layers, item.capture_volume) - annotated.append(replace(item, cumulative_capture_volume=volume, capture_space_bits=bits)) - unique: dict[str, CorrectionCandidate] = {} - for item in _best(annotated, prefer_common=byte_length == "?", fingerprint_match=fingerprint_match): - # A copy from a completed earlier pass must not hide later truncation. - unique.setdefault( - item.artifact.text.lower(), item if complete else replace(item, search_complete=False) - ) - return tuple(unique.values()), complete - - def _fingerprint_matcher( fingerprint: Callable[[MasterSeed], bytes] | None, ) -> Callable[[CorrectionCandidate], bool | None] | None: @@ -534,44 +376,54 @@ def _suggestions( value: str, prefix: str, required_hrp: str | None, - accepted: list[Artifact], + accepted: Sequence[Artifact], *, allowed: Callable[[CorrectionCandidate], bool] | None = None, fingerprint: Callable[[MasterSeed], bytes] | None = None, + byte_length: int | Literal["?"] | None = None, ) -> tuple[CorrectionCandidate, ...]: - fingerprint_match = _fingerprint_matcher(fingerprint) - erased = value - if interpretation := _case_interpretation(value, prefix, required_hrp, allowed): - candidate, value, erased, prefix = interpretation - if candidate is not None: - return (candidate,) - separator = value.lower().rfind("1") - if separator <= 0: - return () - hrp = value[:separator].lower() - if required_hrp is not None and hrp != required_hrp: - return () - excluded = tuple(artifact.header.index for artifact in accepted) - target = len(accepted[0].text) if accepted else None - immutable = ( - value[: len(prefix)] - if prefix and value.lower().startswith(prefix.lower()) - else prefix or value[: separator + 1] - ) - return _scheduled_candidates( + return _corrections( value, - erased, - hrp, - None, - immutable, - excluded, - target=target, + prefix, + required_hrp, + accepted, allowed=allowed, - deadline=monotonic() + 10, - fingerprint_match=fingerprint_match, + fingerprint=fingerprint, + byte_length=byte_length, )[0] +def _corrections( + value: str, + prefix: str, + required_hrp: str | None, + accepted: Sequence[Artifact], + *, + allowed: Callable[[CorrectionCandidate], bool] | None = None, + fingerprint: Callable[[MasterSeed], bytes] | None = None, + byte_length: int | Literal["?"] | None = None, +) -> tuple[tuple[CorrectionCandidate, ...], bool]: + """Run the library search, then the CLI's Hamming, CRC, and fingerprint tie breakers.""" + separator = value.rfind("1") + hrp = value[: max(separator, 0)].lower() + if separator <= 0 or required_hrp not in (None, hrp): + return (), True + target = len(accepted[0].text) if accepted else None + if isinstance(byte_length, int): + target = _text_length(byte_length) + immutable = prefix if prefix and value.lower().startswith(prefix.lower()) else value[: separator + 1] + excluded = tuple(artifact.header.index for artifact in accepted) + context = CorrectionContext(hrp, target, _prefix_for_suffix(immutable, value), excluded) + try: + candidates, complete = _correct_complete(context, value, allowed=allowed, competitors=True) + except InvalidCorrectionInput: + return (), True + best = _best( + candidates, prefer_common=byte_length == "?", fingerprint_match=_fingerprint_matcher(fingerprint) + ) + return best, complete + + def _validate_operational_artifact( artifact: Artifact, accepted: list[Artifact], diff --git a/src/codex32/cli.py b/src/codex32/cli.py index 147de766..2223c054 100644 --- a/src/codex32/cli.py +++ b/src/codex32/cli.py @@ -21,13 +21,12 @@ CorrectionDeclined, InteractiveConfirmationRequired, _card_text, - _case_interpretation, _confirm_correction, + _corrections, _entered_groups, _fingerprint_matcher, _render_groups, _require_correction_confirmation, - _scheduled_candidates, _suggestions, ) from codex32._cli_input import InputError as _UsageError @@ -48,7 +47,6 @@ recover_secret, ) from codex32.correction import ( - CorrectionCandidate, _best, _residue_low_discrimination, correct_worksheet_residue, @@ -572,23 +570,7 @@ def _correct( raise _UsageError("--bytes does not match the valid master-seed backup length.") _print("The codex32 string is already valid.") return 0 - search_value, erased, immutable = normalized, normalized, normalized[: separator + 1] - interpreted = _case_interpretation(normalized, immutable, context.hrp, None) - if interpreted is not None: - candidate, search_value, erased, immutable = interpreted - if ( - candidate is not None - and isinstance(byte_length, int) - and len(candidate.artifact.text) != _text_length(byte_length) - ): - raise _UsageError("--bytes does not match the corrected master-seed backup length.") - else: - candidate = None - if candidate is not None: - candidates: tuple[CorrectionCandidate, ...] = (candidate,) - complete = True - else: - candidates, complete = _scheduled_candidates(search_value, erased, hrp, byte_length, immutable) + candidates, complete = _corrections(normalized, "", context.hrp, [], byte_length=byte_length) if not complete and not candidates: raise _CommandError("The correction search did not complete within ten seconds.") if not candidates: diff --git a/src/codex32/correction.py b/src/codex32/correction.py index c751038b..c4469957 100644 --- a/src/codex32/correction.py +++ b/src/codex32/correction.py @@ -808,8 +808,15 @@ def _best( def _correct_complete( - context: CorrectionContext, damaged_text: str, *, deadline: float | None = None + context: CorrectionContext, + damaged_text: str, + *, + deadline: float | None = None, + allowed: Callable[[CorrectionCandidate], bool] | None = None, + competitors: bool = False, ) -> tuple[tuple[CorrectionCandidate, ...], bool]: + # The public API fails closed on incomplete required work; only the CLI's + # competitor scheduler (competitors=True) may return a best-so-far candidate. if not isinstance(context, CorrectionContext): raise TypeError("context must be CorrectionContext") if not isinstance(damaged_text, str): @@ -877,6 +884,8 @@ def _correct_complete( c.expected_length for c in preflight_contexts if c.expected_length is not None ), deadline=deadline, + competitors=competitors, + allowed=allowed, observed_text=damaged_text, seed_candidates=candidates, required_only=True, @@ -904,6 +913,8 @@ def _correct_complete( value, primary=frozenset(c.expected_length for c in contexts if c.expected_length is not None), deadline=deadline, + competitors=competitors, + allowed=allowed, capture_layers=capture_layers, observed_text=damaged_text, seed_candidates=candidates, @@ -912,9 +923,13 @@ def _correct_complete( complete &= current_complete if not current_complete and not candidates: return (), False - candidates = ( - _restore_case_edits(candidates, uppercase=uppercase) if interpretation is not None else candidates - ) + if interpretation is not None: + candidates = _restore_case_edits(candidates, uppercase=uppercase) + if allowed is not None: + candidates = tuple(candidate for candidate in candidates if allowed(candidate)) + if not complete: + # A candidate from a completed pass must not hide a later truncation. + candidates = tuple(replace(candidate, search_complete=False) for candidate in candidates) return candidates, complete diff --git a/tests/test_cli.py b/tests/test_cli.py index 09851437..86adf3bd 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1772,7 +1772,7 @@ def test_correction_bytes_rejects_an_unsupported_ms_size() -> None: def test_cli_never_accepts_an_incomplete_structural_search() -> None: original = VECTOR_1["secret_s"] damaged = original[:19] + original[20:] - with patch("codex32._cli_input._correction_candidates", return_value=((), False, 0.0)): + with patch("codex32._cli_input._correct_complete", return_value=((), False)): result = _invoke(["correct"], damaged) assert result.exit_code == 3 @@ -1783,10 +1783,10 @@ def test_cli_never_accepts_an_incomplete_structural_search() -> None: @pytest.mark.parametrize( ("options", "lengths", "bounded"), ( - ([], (48, 54, 61, 67, 74, 127), True), + ([], (48,), True), (["--bytes", "20"], (54,), True), (["--bytes", "64"], (127,), True), - (["--bytes", "?"], (48, 54, 61, 67, 74, 127), True), + (["--bytes", "?"], (48,), True), ), ) def test_correction_options_control_lengths_deadline_and_search_envelope( @@ -1805,17 +1805,6 @@ def test_correction_options_control_lengths_deadline_and_search_envelope( assert (search.call_args.kwargs["deadline"] is not None) is bounded -def test_automatic_target_selection_covers_midpoints_and_supported_lengths() -> None: - from codex32._cli_input import _correction_plan - - for observed in range(40, 136): - targets = _correction_plan("ms", None, observed, None) - expected = 48 if observed <= 61 else 74 if observed <= 100 else 127 - - assert targets[0] == expected - assert sorted(targets) == [48, 54, 61, 67, 74, 127] - - def test_fixed_correction_repairs_legacy_cl_header_and_residue_reverse_positions() -> None: original = VECTOR_6["codex32_peev"] damaged = original[:3] + "2" + original[4:] @@ -1865,8 +1854,7 @@ def test_correct_suggests_the_majority_case_for_mixed_case_damage() -> None: assert result.exit_code == 1 assert source in result.stderr assert "No valid correction found" not in result.stderr - assert wrong_length.exit_code == 2 - assert "--bytes does not match" in wrong_length.stderr + assert wrong_length.exit_code == 3 assert source not in wrong_length.stderr @@ -1886,32 +1874,6 @@ def test_correct_grouped_mixed_case_recognizes_case_only_repair() -> None: assert "interactive confirmation required" not in result.stderr -def test_correct_searches_mixed_case_erasures_before_normalized_alignment(monkeypatch) -> None: - source = VECTOR_1["secret_s"] - letter_positions = [ - index - for index, character in enumerate(source[3:], 3) - if index >= 9 and character.lower() != character.upper() - ] - positions = letter_positions[1:26:6] - damaged = "".join( - ("P" if character.lower() != "p" else "Q") if index in positions else character - for index, character in enumerate(source) - ) - searched: list[str] = [] - - def stop_after_first(value, *_args, **_kwargs): - searched.append(value) - return (), False, None - - monkeypatch.setattr("codex32._cli_input._correction_candidates", stop_after_first) - - result = _invoke(["correct"], damaged) - assert result.exit_code != 0 - assert len(searched) == 1 - assert searched[0].count("?") == len(positions) - - def test_correct_reranks_mixed_case_erasure_and_normalized_interpretations() -> None: source = "ms10testsxxxxxxxxxxxxxxxxxxxxxxxxxx4nzvca9cmczlw" start = source.index("x") @@ -1940,45 +1902,6 @@ def test_correct_required_work_is_not_starved_by_erasure_alignment() -> None: assert "interactive confirmation required" not in result.stderr -def test_correct_accounts_retry_frontier_after_incomplete_first_search(monkeypatch) -> None: - source = "ms10testsxxxxxxxxxxxxxxxxxxxxxxxxxx4nzvca9cmczlw" - start = source.index("x") - positions = range(start, start + 13) - damaged = "".join( - ("P" if index == start + 6 else character.upper()) if index in positions else character - for index, character in enumerate(source) - ) - candidate = CorrectionCandidate( - parse_codex32(source), - (), - 1, - 0, - 0, - None, - search_complete=False, - ) - full_searches: list[str] = [] - - def incomplete_first(value, *_args, **kwargs): - if kwargs.get("required_only"): - return (), True, 0.0 - full_searches.append(value) - kwargs["capture_layers"].append((1, 5)) - if len(full_searches) == 1: - return (candidate,), False, 0.0 - return (), False, 0.0 - - monkeypatch.setattr("codex32._cli_input._correction_candidates", incomplete_first) - - result = _invoke(["correct"], damaged) - - assert len(full_searches) == 2 - assert full_searches[0].count("?") == len(positions) - assert "?" not in full_searches[1] - assert result.exit_code == 3 - assert "interactive confirmation required" in result.stderr - - def test_correction_hides_internal_candidate_reparse_failures() -> None: result = _invoke(["correct"], "ms12auxxxxxxxxxxxxxxxxxxxxxxxxxxxxxda3kr3s0s2swg") @@ -2868,7 +2791,7 @@ def test_incomplete_candidate_has_no_search_warning_and_is_never_accepted_automa candidate = _correct_fixed(source, hrp="ms") assert candidate is not None candidate = replace(candidate, search_complete=False) - with patch("codex32._cli_input._correction_candidates", return_value=((candidate,), False, 0.0)): + with patch("codex32._cli_input._correct_complete", return_value=((candidate,), False)): result = _invoke(["correct"], source[:-1] + "?") assert result.exit_code == 1 and result.stdout == "" assert "Search incomplete" not in result.stderr diff --git a/tests/test_correction_disclosure.py b/tests/test_correction_disclosure.py index ced1bb72..c8cdb9f5 100644 --- a/tests/test_correction_disclosure.py +++ b/tests/test_correction_disclosure.py @@ -102,7 +102,7 @@ def test_noninteractive_gate_emits_only_operational_error(entrypoint, plain): patch.object(sys, "stdin", io.StringIO(VECTOR_1["secret_s"][:-1] + "?")), contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr), - patch.object(_cli_input, "_correction_candidates", return_value=((_candidate(),), True, None)), + patch.object(_cli_input, "_correct_complete", return_value=((_candidate(),), True)), ): status = entrypoint(["correct", *(["--plain"] if plain else [])]) prog = "codex32" if entrypoint is cli.main else "ms32" @@ -129,9 +129,7 @@ def respond(prompt, prefill=""): monkeypatch.setattr(_cli_input, "_editable_input", respond) monkeypatch.setattr(_cli_input, "_suggestions", lambda *args, **kwargs: (candidate,)) monkeypatch.setattr(cli, "_suggestions", lambda *args, **kwargs: (candidate,)) - monkeypatch.setattr( - _cli_input, "_correction_candidates", lambda *args, **kwargs: ((candidate,), True, None) - ) + monkeypatch.setattr(_cli_input, "_correct_complete", lambda *args, **kwargs: ((candidate,), True)) core = _FakeBitcoinCore() monkeypatch.setattr(cli.BitcoinCore, "connect", lambda *args: core) stdout, stderr = _TTYOutput(), _TTYOutput() @@ -165,9 +163,7 @@ def respond(prompt, prefill=""): return next(responses) monkeypatch.setattr(_cli_input, "_editable_input", respond) - monkeypatch.setattr( - _cli_input, "_correction_candidates", lambda *args, **kwargs: ((candidate,), True, None) - ) + monkeypatch.setattr(_cli_input, "_correct_complete", lambda *args, **kwargs: ((candidate,), True)) with ( patch.object(sys, "stdin", _TTYInput()), contextlib.redirect_stdout(stdout), @@ -183,9 +179,7 @@ def test_redirected_stderr_blocks_low_discrimination_disclosure(monkeypatch): candidate = _candidate(source) responses = iter((source[:-1] + "?",)) monkeypatch.setattr(_cli_input, "_editable_input", lambda *args, **kwargs: next(responses)) - monkeypatch.setattr( - _cli_input, "_correction_candidates", lambda *args, **kwargs: ((candidate,), True, None) - ) + monkeypatch.setattr(_cli_input, "_correct_complete", lambda *args, **kwargs: ((candidate,), True)) stdout, stderr = io.StringIO(), io.StringIO() with ( patch.object(sys, "stdin", _TTYInput()), @@ -220,9 +214,7 @@ def confirm(prompt): return "YES" monkeypatch.setattr(_cli_input, "_confirmation_input", confirm) - monkeypatch.setattr( - _cli_input, "_correction_candidates", lambda *args, **kwargs: ((candidate,), True, None) - ) + monkeypatch.setattr(_cli_input, "_correct_complete", lambda *args, **kwargs: ((candidate,), True)) stdout, stderr = io.StringIO(), _TTYOutput() with ( patch.object(sys, "stdin", io.StringIO(source[:-1] + "?")), diff --git a/tests/test_correction_indel.py b/tests/test_correction_indel.py index e5ee5eca..4a8e430a 100644 --- a/tests/test_correction_indel.py +++ b/tests/test_correction_indel.py @@ -9,7 +9,7 @@ from data.bip93_vectors import VECTOR_1, VECTOR_5, VECTOR_6 from codex32 import CorrectionContext, MasterSeed, correct -from codex32._cli_input import _correction_plan, _fingerprint_matcher +from codex32._cli_input import _fingerprint_matcher from codex32.bip93 import TEXT_LENGTHS from codex32.correction import CorrectionCandidate, _best, _capture_volume, _erasure_state, _primary from codex32.generation import _fingerprint_identifier @@ -154,9 +154,7 @@ def test_automatic_secondary_search_recovers_two_group_indels( ) -> None: source = MasterSeed.from_seed(bytes(range(byte_length)), identifier="test").text damaged = _group_damage(source, inserted, omitted) - contexts = tuple( - CorrectionContext("ms", target, "ms1") for target in _correction_plan("ms", None, len(damaged), None) - ) + contexts = tuple(CorrectionContext("ms", target, "ms1") for target in TEXT_LENGTHS) candidates, complete = _search_many( contexts, @@ -555,3 +553,31 @@ def test_full_checksum_burst_is_admitted_at_the_shared_mass_ceiling(): damaged = SOURCE[:8] + "?" * 13 + SOURCE[21:] assert _correct_fixed(damaged, hrp="ms").artifact.text == SOURCE assert correct(CONTEXT, damaged)[0].artifact.text == SOURCE + + +def test_public_correction_fails_closed_when_required_work_expires() -> None: + import codex32._competitors as competitors_module + import codex32.indel as indel_module + from codex32.correction import _correct_complete + + source = VECTOR_1["secret_s"] + swapped = source[:40] + source[41] + source[40] + source[42:] + assert swapped != source + context = CorrectionContext("ms") + real = indel_module._search_target + + def expire_after_fixed(state, *args, **kwargs): # type: ignore[no-untyped-def] + # Run the fixed (BCH) pass, then expire every required or optional layer. + if set(state.counts) == {indel_module._FIXED}: + return real(state, *args, **kwargs) + return False + + assert any(candidate.artifact.text == source for candidate in correct(context, swapped)) + with ( + patch.object(indel_module, "_search_target", expire_after_fixed), + patch.object(competitors_module, "_search_target", expire_after_fixed), + ): + assert correct(context, swapped) == () + cli_candidates, complete = _correct_complete(context, swapped, competitors=True) + assert not complete + assert [(c.artifact.text, c.search_complete) for c in cli_candidates] == [(source, False)] diff --git a/tests/test_mixed_case_deadline.py b/tests/test_mixed_case_deadline.py deleted file mode 100644 index 2523329e..00000000 --- a/tests/test_mixed_case_deadline.py +++ /dev/null @@ -1,82 +0,0 @@ -"""Deadline regression for mixed-case embedded recovery.""" - -from dataclasses import replace -from time import monotonic - -import pytest - -from codex32._cli_input import _case_interpretation, _correction_candidates, _scheduled_candidates - -SOURCE = "ms10testsxxxxxxxxxxxxxxxxxxxxxxxxxx4nzvca9cmczlw" - - -@pytest.mark.parametrize( - ("damaged", "erasure_fixes", "normalized_fixes"), - [ - # Five minority-case P: five erasures are correctable, five substitutions are not. - ("ms10testsxPxxxxxPxxxxxPxxxxxPxxxxxP4nzvca9cmczlw", True, False), - # Fifteen minority-case X, one mistyped: fifteen erasures are not correctable, - # one substitution after case normalization is. - ("ms10testsXXXXXXXPXXXXXXXxxxxxxxxxxx4nzvca9cmczlw", False, True), - ], -) -def test_embedded_mixed_case_recovers_either_sole_interpretation_within_deadline( - damaged: str, erasure_fixes: bool, normalized_fixes: bool -) -> None: - interpretation = _case_interpretation(damaged, "", "ms", None) - - assert interpretation is not None - direct, normalized, erased, _prefix = interpretation - assert direct is None - immutable = normalized[: normalized.rfind("1") + 1] - for value, fixes in ((erased, erasure_fixes), (normalized, normalized_fixes)): - alone, alone_complete, _deadline = _correction_candidates( - value, "ms", None, immutable, deadline=monotonic() + 10, required_only=True - ) - assert alone_complete - assert [candidate.artifact.text for candidate in alone] == ([SOURCE] if fixes else []) - - candidates, complete = _scheduled_candidates( - normalized, - erased, - "ms", - None, - immutable, - deadline=monotonic() + 10, - ) - - # The exhaustive optional search may truncate at the deadline; the - # candidate must still be found and must report that truncation. - assert [candidate.artifact.text for candidate in candidates] == [SOURCE] - assert candidates[0].search_complete is complete - - -def test_scheduled_truncation_survives_a_complete_later_pass(monkeypatch: pytest.MonkeyPatch) -> None: - damaged = "ms10testsxPxxxxxPxxxxxPxxxxxPxxxxxP4nzvca9cmczlw" - interpretation = _case_interpretation(damaged, "", "ms", None) - assert interpretation is not None - _direct, normalized, erased, _prefix = interpretation - immutable = normalized[: normalized.rfind("1") + 1] - found, _complete, _deadline = _correction_candidates( - erased, "ms", None, immutable, deadline=monotonic() + 10, required_only=True - ) - full_searches: list[str] = [] - - def truncated_first(value: str, *_args: object, **kwargs: object) -> tuple[object, bool, float]: - if kwargs.get("required_only"): - return (), True, 0.0 - assert kwargs.get("optional_only") is True - full_searches.append(value) - if len(full_searches) == 1: - return (replace(found[0], search_complete=False),), False, 0.0 - return (), True, 0.0 - - monkeypatch.setattr("codex32._cli_input._correction_candidates", truncated_first) - - candidates, complete = _scheduled_candidates(normalized, erased, "ms", None, immutable) - - assert full_searches == [erased, normalized] - assert not complete - assert [(candidate.artifact.text, candidate.search_complete) for candidate in candidates] == [ - (SOURCE, False) - ]