correct: Interpret mixed-case damage

Apply the established majority-case interpretation to standalone correction while preserving immutable context, entered edit semantics, and disclosure accounting. Account the normalized retry frontier even when the first optional search reaches its deadline after finding a candidate, so cumulative capture mass remains fail-closed.

Fixes #37.

Author: BenWestgate

src/codex32/_cli_input.py

@@ -6,10 +6,11 @@
 import difflib
 import os
 import sys
-from collections.abc import Callable, Iterator
+from collections.abc import Callable, Iterator, Sequence
 from time import monotonic
 from typing import Any, Literal, cast
 
+from codex32.bech32 import interpret_mixed_case
 from codex32.bip93 import (
     Secret,
     Share,
@@ -377,15 +378,14 @@ def _case_interpretation(
     profiles: tuple[Profile, ...] | None,
     allowed: Callable[[CorrectionCandidate], bool] | None,
 ) -> tuple[CorrectionCandidate | None, str, str, str] | None:
-    """Normalize likely casing and mark contrary-case data as erasures."""
-    if value.upper() == value or value.lower() == value:
-        return 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
-    letters = [character for character in value[immutable_length:] if character.lower() != character.upper()]
-    uppercase = sum(character.isupper() for character in letters) > len(letters) / 2
-    corrected = value.upper() if uppercase else value.lower()
+    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, profiles)
@@ -397,14 +397,6 @@ def _case_interpretation(
         )
         proposed = CorrectionCandidate(artifact, (), 1, 0, 0, None, capture_space_bits=bits)
         candidate = proposed if allowed is None or allowed(proposed) else None
-    erased = "".join(
-        corrected[index]
-        if index < immutable_length
-        or character.lower() == character.upper()
-        or character.isupper() == uppercase
-        else "?"
-        for index, character in enumerate(value)
-    )
     return candidate, corrected, erased, corrected_prefix
 
 
@@ -460,6 +452,8 @@ def _correction_candidates(
     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,
 ) -> tuple[tuple[CorrectionCandidate, ...], bool, float | None, bool]:
     count = len(value.replace(" ", ""))
     targets, primary, reduced, _timed = _correction_plan(profile, byte_length, count, target)
@@ -476,6 +470,8 @@ def _correction_candidates(
         competitors=True,
         allowed=allowed,
         capture_layers=capture_layers,
+        seed_candidates=seed_candidates,
+        required_only=required_only,
     )
     if allowed is not None:
         candidates = tuple(candidate for candidate in candidates if allowed(candidate))

src/codex32/_competitors.py

@@ -173,8 +173,10 @@ def _search_competitors(
     frontier: dict[_Layer, int],
     deadline: float,
     allowed: Callable[[CorrectionCandidate], bool] | None,
+    *,
+    seed_candidates: Sequence[CorrectionCandidate] = (),
 ) -> tuple[tuple[CorrectionCandidate, ...], bool]:
-    results: dict[str, CorrectionCandidate] = {}
+    results = {candidate.artifact.text.lower(): candidate for candidate in seed_candidates}
     fixed: dict[int, CorrectionCandidate | None] = {}
     completed: set[_Layer] = set()
     try:

src/codex32/bech32.py

@@ -53,6 +53,22 @@ def _validate_single_case_ascii(value: str) -> bool:
     return value.isupper()
 
 
+def interpret_mixed_case(value: str, immutable_length: int) -> tuple[str, str, bool] | None:
+    # Return majority-cased and minority-erased interpretations of mixed-case text.
+    if not value.isascii() or value.upper() == value or value.lower() == value:
+        return None
+    letters = [character for character in value[immutable_length:] if character.isalpha()]
+    uppercase = sum(character.isupper() for character in letters) > len(letters) / 2
+    normalized = value.upper() if uppercase else value.lower()
+    erased = "".join(
+        normalized[index]
+        if index < immutable_length or not character.isalpha() or character.isupper() == uppercase
+        else "?"
+        for index, character in enumerate(value)
+    )
+    return normalized, erased, uppercase
+
+
 def bech32_encode(hrp: str, data: list[int], spec: _Checksum) -> str:
     """Compute a Bech32 string given HRP and data values."""
     checksum = spec.create(bech32_hrp_expand(hrp) + list(data))

src/codex32/cli.py

@@ -6,13 +6,15 @@
 import json
 import sys
 from collections.abc import Callable, Sequence
+from dataclasses import replace
 from typing import Literal, NamedTuple, cast
 
 from codex32._bitcoin_core import BitcoinCore, BitcoinCoreError
 from codex32._cli_input import (
     CorrectionDeclined,
     InteractiveConfirmationRequired,
     _card_text,
+    _case_interpretation,
     _confirm_correction,
     _correction_candidates,
     _entered_groups,
@@ -35,7 +37,13 @@
     parse_codex32,
     recover_secret,
 )
-from codex32.correction import _best, _residue_low_discrimination, correct_worksheet_residue
+from codex32.correction import (
+    CorrectionCandidate,
+    _best,
+    _capture_mass,
+    _residue_low_discrimination,
+    correct_worksheet_residue,
+)
 from codex32.errors import CodexError, HeaderCollision, InvalidCorrectionInput
 from codex32.generation import (
     ConfirmationResult,
@@ -550,12 +558,75 @@ def _correct(
             raise _UsageError("--bytes does not match the valid master-seed backup length.")
         _print("The codex32 string is already valid.")
         return 0
-    candidates, complete, _deadline, ambiguous = _correction_candidates(
-        value,
-        hrp,
-        byte_length,
-        value[: separator + 1],
-    )
+    search_value, erased, immutable = normalized, normalized, normalized[: separator + 1]
+    interpreted = _case_interpretation(normalized, immutable, context.profiles, 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) != _ms_text_length(byte_length)
+        ):
+            raise _UsageError("--bytes does not match the corrected master-seed backup length.")
+    else:
+        candidate = None
+    capture_layers: list[tuple[int, int]] = []
+    if candidate is not None:
+        candidates: tuple[CorrectionCandidate, ...] = (candidate,)
+        complete, deadline, ambiguous = True, None, False
+    else:
+        first_search = erased if erased != search_value else search_value
+        retry_search = search_value if erased != search_value else None
+        seeded: tuple[CorrectionCandidate, ...] = ()
+        deadline = None
+        if retry_search is not None:
+            # Discover required candidates for both case interpretations before
+            # either full search can spend the shared deadline on optional
+            # alignment work.  Full searches below own capture accounting.
+            for required_value in (first_search, retry_search):
+                required, required_complete, deadline, _ = _correction_candidates(
+                    required_value,
+                    hrp,
+                    byte_length,
+                    immutable,
+                    deadline=deadline,
+                    seed_candidates=seeded,
+                    required_only=True,
+                )
+                if not required_complete:
+                    raise _CommandError("The correction search did not complete within ten seconds.")
+                seeded = required
+        candidates, complete, deadline, ambiguous = _correction_candidates(
+            first_search,
+            hrp,
+            byte_length,
+            immutable,
+            deadline=deadline,
+            capture_layers=capture_layers,
+            seed_candidates=seeded,
+        )
+        if retry_search is not None:
+            retry_candidates, complete, deadline, retry_ambiguous = _correction_candidates(
+                retry_search,
+                hrp,
+                byte_length,
+                immutable,
+                deadline=deadline,
+                capture_layers=capture_layers,
+                seed_candidates=(*seeded, *candidates),
+            )
+            ambiguous = ambiguous or retry_ambiguous
+            combined = (*candidates, *retry_candidates)
+            if combined:
+                annotated = []
+                for item in combined:
+                    volume, bits = _capture_mass(capture_layers, item.capture_volume)
+                    annotated.append(replace(item, cumulative_capture_volume=volume, capture_space_bits=bits))
+                ranked = _best(annotated, prefer_common=byte_length == "?")
+                unique: dict[str, CorrectionCandidate] = {}
+                for item in ranked:
+                    unique.setdefault(item.artifact.text.lower(), item)
+                candidates = tuple(unique.values())
     if not complete and not candidates:
         raise _CommandError("The correction search did not complete within ten seconds.")
     if ambiguous:

src/codex32/correction.py

@@ -34,6 +34,7 @@
     _u5_to_chars,
     _validate_single_case_ascii,
     bech32_hrp_expand,
+    interpret_mixed_case,
 )
 from codex32.bip93 import (
     IDX_SORT,
@@ -839,6 +840,17 @@ def _primary(
     )
 
 
+def _restore_case_edits(candidates: tuple[CorrectionCandidate, ...]) -> tuple[CorrectionCandidate, ...]:
+    # Hide erasures synthesized only to search minority-case symbols.
+    def restore(edit: CorrectionEdit) -> CorrectionEdit:
+        kind = "substitution" if edit.kind == "erasure" and edit.observed.lower() in CHARSET else edit.kind
+        return replace(edit, kind=kind) if kind != edit.kind else edit
+
+    return tuple(
+        replace(candidate, edits=tuple(restore(edit) for edit in candidate.edits)) for candidate in candidates
+    )
+
+
 def _best(
     candidates: Sequence[CorrectionCandidate],
     *,
@@ -874,30 +886,93 @@ def _correct_complete(
     # displayed strings are no longer than the largest expanded codeword.
     if len(damaged_text) > 2 * (_LONG_SPEC.period + 8):
         return (), True
-    from codex32.indel import _search_many
-
     deadline = monotonic() + 10 if deadline is None else deadline
-    contexts: tuple[CorrectionContext, ...]
-    if context.expected_length is not None:
-        contexts = (context,)
+    base = f"{context.hrp}1"
+    locked = context.immutable_prefix or base
+    immutable_length = len(locked) if damaged_text.lower().startswith(locked.lower()) else len(base)
+    interpretation = interpret_mixed_case(damaged_text, immutable_length)
+    inputs: tuple[tuple[CorrectionContext, str], ...]
+    if interpretation is None:
+        inputs = ((context, damaged_text),)
     else:
-        # Only lengths reachable by either disjoint family are eligible.
-        observed = len(damaged_text.replace(" ", ""))
-        contexts_list = []
-        for target in sorted({observed + delta for delta in (*range(-4, 5), -8, 8)}):
-            candidate_context = replace(context, expected_length=target)
-            try:
-                _validate_context(candidate_context)
-            except InvalidCorrectionInput:
-                continue
-            contexts_list.append(candidate_context)
-        contexts = tuple(contexts_list)
-    return _search_many(
-        contexts,
-        damaged_text,
-        primary=frozenset(c.expected_length for c in contexts if c.expected_length is not None),
-        deadline=deadline,
-    )
+        normalized, erased, uppercase = interpretation
+        normalized_prefix = locked.upper() if uppercase else locked.lower()
+        normalized_context = replace(
+            context, immutable_prefix=normalized_prefix if context.immutable_prefix is not None else None
+        )
+        # Minority-case symbols are explicit erasures, so search that stronger
+        # interpretation before optional alignment work on the normalized text
+        # can consume the shared correction deadline.
+        inputs = ((normalized_context, erased), (normalized_context, normalized))
+
+    from codex32.indel import _search_many
+
+    capture_layers: list[tuple[int, int]] = []
+    candidates: tuple[CorrectionCandidate, ...] = ()
+    complete = True
+    if interpretation is not None:
+        # Establish both interpretations' fixed/required candidates before
+        # either interpretation can spend the shared deadline on optional
+        # alignment work.  These discovery passes use a private accounting
+        # ledger; the full searches below account every admitted layer once.
+        for input_context, value in inputs:
+            preflight_contexts: tuple[CorrectionContext, ...]
+            if input_context.expected_length is not None:
+                preflight_contexts = (input_context,)
+            else:
+                observed = len(value.replace(" ", ""))
+                contexts_list = []
+                for target in sorted({observed + delta for delta in (*range(-4, 5), -8, 8)}):
+                    candidate_context = replace(input_context, expected_length=target)
+                    try:
+                        _validate_context(candidate_context)
+                    except InvalidCorrectionInput:
+                        continue
+                    contexts_list.append(candidate_context)
+                preflight_contexts = tuple(contexts_list)
+            candidates, current_complete = _search_many(
+                preflight_contexts,
+                value,
+                primary=frozenset(
+                    c.expected_length for c in preflight_contexts if c.expected_length is not None
+                ),
+                deadline=deadline,
+                observed_text=damaged_text,
+                seed_candidates=candidates,
+                required_only=True,
+            )
+            if not current_complete:
+                return (), False
+    for input_context, value in inputs:
+        contexts: tuple[CorrectionContext, ...]
+        if input_context.expected_length is not None:
+            contexts = (input_context,)
+        else:
+            # Only lengths reachable by either disjoint family are eligible.
+            observed = len(value.replace(" ", ""))
+            contexts_list = []
+            for target in sorted({observed + delta for delta in (*range(-4, 5), -8, 8)}):
+                candidate_context = replace(input_context, expected_length=target)
+                try:
+                    _validate_context(candidate_context)
+                except InvalidCorrectionInput:
+                    continue
+                contexts_list.append(candidate_context)
+            contexts = tuple(contexts_list)
+        candidates, current_complete = _search_many(
+            contexts,
+            value,
+            primary=frozenset(c.expected_length for c in contexts if c.expected_length is not None),
+            deadline=deadline,
+            capture_layers=capture_layers,
+            observed_text=damaged_text,
+            seed_candidates=candidates,
+        )
+        complete &= current_complete
+        if not current_complete and not candidates:
+            return (), False
+    candidates = _restore_case_edits(candidates) if interpretation is not None else candidates
+    return candidates, complete
 
 
 def correct(context: CorrectionContext, damaged_text: str) -> tuple[CorrectionCandidate, ...]: