Skip to content

cli: Share the library correction driver #181

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
BenWestgate wants to merge 1 commit into from
Draft

cli: Share the library correction driver #181

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
230 changes: 41 additions & 189 deletions src/codex32/_cli_input.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,32 +7,28 @@
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,
_validate_share_set,
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,
Expand Down Expand Up @@ -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:
Expand All @@ -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)
Comment on lines +415 to +416

@chatgpt-codex-connector chatgpt-codex-connector Bot Oct 24, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Exclude only ordinary share indices from the correction context

When a share derivation basis already contains a secret (index s), this tuple includes s, but _correct_complete now calls _validate_context, which rejects non-ordinary excluded indices. The exception is swallowed, so every subsequent damaged share receives no correction suggestion. For example, accepting VECTOR_3["secret_s"] and then entering VECTOR_3["share_a"] with its last character replaced by ? returns no candidates, whereas the previous search repairs it. Filter secrets out of excluded_indices and leave whole-basis compatibility checks to the existing allowed predicate.

Useful? React with 👍 / 👎.

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],
Expand Down
22 changes: 2 additions & 20 deletions src/codex32/cli.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -48,7 +47,6 @@
recover_secret,
)
from codex32.correction import (
CorrectionCandidate,
_best,
_residue_low_discrimination,
correct_worksheet_residue,
Expand Down Expand Up @@ -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:
Expand Down
23 changes: 19 additions & 4 deletions src/codex32/correction.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand All @@ -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


Expand Down
Loading
Loading