From ab49450467eae8eed3251adab736449e0c695613 Mon Sep 17 00:00:00 2001 From: Alexander Date: Mon, 14 Sep 2026 21:02:14 +0200 Subject: [PATCH] Compose fixes to a fixpoint and learn only from success New thefuck/compose.py resolve() iterates the auto-run sources (learned-db exact, error-suggestion, history, path guess, help) per round on the previous round's script until stable, a revisit aborts or no source has a safe candidate; dangerous candidates are declined mid-round and consultation continues. Output-aware sources get a synthesized Command wrapping the round's script with the original failure's output. Zero accepted steps return None so the original failing script is never auto-executed; no round cap (termination contract: visited-set + fixed-range/word-count-preserving sources, stated in the module docstring; a compose-local memo acts only as a determinism tripwire). fix_command runs the inline chain through the composer; under settings.repeat or a non-nushell shell it records immediately before the run unless the only step replayed a learned-db exact hit; under nushell it closes the learned db first and records on returncode 0, forgets on failure (poison self-heal; CorrectedCommand.run now stashes the nu returncode on the instance). learned.forget(original, corrected) is the exact inverse of record (cmd: + matching-count word:/part: deletes, assembly failures are a documented no-op); learned.close_db is idempotent and atexit now registers the stable close_db method exactly once across close/reopen cycles. Test migration (authorized): test_fix_command_learned.py rebased to composer semantics (per-round consultation incl. error-suggestion; input-aware fakes so round 2 declines, preserving the recordings==[] pin in test_exact_learned_wins_over_guess). --- .../task-3-intelligent-correction.txt | 134 ++++++++ tests/entrypoints/test_fix_command_learned.py | 319 ++++++++++++++---- tests/test_compose.py | 199 +++++++++++ tests/test_learned.py | 89 +++++ thefuck/compose.py | 119 +++++++ thefuck/entrypoints/fix_command.py | 66 ++-- thefuck/learned.py | 60 +++- thefuck/types.py | 3 +- 8 files changed, 895 insertions(+), 94 deletions(-) create mode 100644 .omo/evidence/task-3-intelligent-correction.txt create mode 100644 tests/test_compose.py create mode 100644 thefuck/compose.py diff --git a/.omo/evidence/task-3-intelligent-correction.txt b/.omo/evidence/task-3-intelligent-correction.txt new file mode 100644 index 0000000..15509bc --- /dev/null +++ b/.omo/evidence/task-3-intelligent-correction.txt @@ -0,0 +1,134 @@ +Task 3 — Fixpoint composer + outcome-gated learning +=================================================== +Plan: .omo/plans/intelligent-correction.md (todo 3), HEAD before: 58f6cca +Commit after: + +What landed +----------- +- thefuck/compose.py (new, 101 pure LOC): resolve(command) -> + (final_script_or_None, steps). Sources consulted per round in order + [learned-db exact, error-suggestion, history, path guess, help]; + first candidate passing danger.is_dangerous wins the round + (dangerous candidates declined, consultation continues). + Output-aware sources (error_suggestion, history) receive a + synthesized types.Command(script=round_script, output=original). + Acceptance: candidate == current -> STOP stable; candidate in + visited -> STOP-ABORT (None, steps); else advance. Round with no + acceptable candidate -> STOP. ZERO accepted steps (round-1 + no-candidate, round-1 candidate == original) -> (None, steps): + the original failing script is NEVER auto-executed. NO round cap. + Termination contract stated in the module docstring (visited-set + kills cycles; sources must be fixed-range or word-count-preserving, + appending sources forbidden). Compose-local memo dict per + (source, input-script) is a determinism TRIPWIRE (assert never hit) + — pinned by test_no_source_input_ever_repeats (4-round chain, all + (source, script) calls distinct). +- thefuck/entrypoints/fix_command.py: inline resolver chain replaced + by compose.resolve (learned-db lookup is source #1 INSIDE the + composer, no separate short-circuit). None -> rules+ask unchanged. + Belt-and-suspenders danger check kept before the auto-run. + _run_learned(): settings.repeat OR non-Nushell -> record(original, + final) exactly once immediately BEFORE run when >=1 step accepted, + except a win whose ONLY step was the learned-db exact hit (no + re-record; mirrors today, preserves test_exact_learned_wins_over_ + guess). Nushell + no-repeat -> learned.close_db(), run, + returncode==0 -> record else learned.forget(original, final) — + the poison self-heal path (no carve-out: a failing learned exact + hit MUST forget, pinned by test_learned_replay_failure_still_ + forgets). Ask-path record keeps today's record-immediately + behavior. Pre-existing hazard documented not fixed: under repeat + the spawned child may hit the parent's open shelve, silently + degraded via the {} fallback (learned.py _init_db). +- thefuck/types.py: CorrectedCommand.run nushell branch stashes + subprocess.run(...).returncode on the instance (2-line change). +- thefuck/learned.py: forget(original, corrected) — exact inverse of + record: deletes cmd:original; when token counts match deletes + word: and part:: for + diverged positions (part: keys require the corrected head — + hence arg 2); delete, not decrement; assembly failures (no cmd: + key) are a documented NO-OP (word:/part: mappings were validated + by a prior success of a different command and stand). close_db(): + idempotent, closes shelve if open, sets _db None ({} fallback has + no close — guarded). atexit now registers the STABLE close_db + METHOD once (instance flag) instead of the stale-prone + _db.close object. + +Authorized test migration (mandated by plan todo 3) +--------------------------------------------------- +tests/entrypoints/test_fix_command_learned.py rewritten: +- fixture patches compose.* source names (chain moved into the + composer) and fix_command.learned (Mock record/forget/close_db + with order tracking); danger patched in BOTH fix_command and + compose with the same fake, real_danger reinstalled in both for + the override tests (now parametrized over 'suggestion' too). +- fakes made INPUT-AWARE (constant candidates returned only for the + original script, None otherwise) so round 2 declines — exactly the + plan's prescribed fix for test_exact_learned_wins_over_guess + (recordings == [] preserved). +- TestResolverChainOrder rebased to per-round consultation incl. + error-suggestion: help-only hit -> [learned, error-suggestion, + history, path, help] x2 rounds; earlier hits -> round-1 prefix + + full round-2 decline; all-decline fallthrough -> 5 calls. + +Test results (all red-first: compose/learned/entrypoint suites +failed with ImportError/AttributeError before implementation) +----------------------------------------------------------------- +- tests/test_compose.py: 14 passed +- tests/test_learned.py: 57 passed (46 prior + 11 new: + TestForget x8, TestCloseDb x3) +- tests/entrypoints/test_fix_command_learned.py: 28 passed +- combined target: pytest tests/test_compose.py tests/test_learned.py + tests/entrypoints/ -q -> 183 passed +- FULL suite: pytest tests/ -q --ignore=tests/functional -> + 2200 passed, 16 skipped +- flake8 clean on all 7 touched files +- pure LOC: compose.py 101, fix_command.py 82, learned.py 201, + types.py 214 — all <= 250 +- py2.7 grammar proof: lib2to3 (py2 grammar) parses all 7 files OK + (no f-strings / py3-only syntax) + +Acceptance pins (plan todo 3, all present) +------------------------------------------ +- two-round composition -> single composite (test_compose + TestComposition + entrypoint TestComposition) +- round-1 NO candidate -> None -> reaches select_command, original + NOT executed (TestNoAutoRunOfOriginal + TestFallthrough + + entrypoint DangerOverride/fallthrough paths; most common path) +- round-1 candidate == original -> None (pinned) +- cycle fixture A<->B -> (None, steps) -> rules+ask, nothing + executed (TestAbort) +- original-script re-entry in round 2 -> abort-to-ask (pinned) +- synthesized-Command threading: mocked sources assert exact + Command(script=round_composite, output=original_output) sequence + per round (misleading_success_output class: exact call-arg asserts) +- returncode 1 -> forget removes seeded cmd: + derived word:/part:, + nothing recorded (entrypoint TestOutcomeLearning + learned + TestForget with db dumps) +- assembly failure (word:-only, no cmd:) -> forget no-op (pinned) +- returncode 0 -> recorded; non-nushell/repeat -> record regardless, + before run (pinned with order == ["record", "run"]) +- close_db idempotent + atexit registered once across reopens + (registration count asserted == [lc.close_db]) +- memo tripwire never hits (assert inside _round + dedicated test) +- stale_state: TestCloseDb uses fresh tmp XDG_CACHE_HOME per test + +Live sanity (real sources, seeded dict db, controlled PATH) +---------------------------------------------------------- +resolve(Command("gti psuh", git-not-a-command output)) with +cmd:gti psuh -> "git psuh" and part:git:psuh -> "push" seeded: +final == "git push", steps == [('learned', 'git psuh'), +('learned', 'git push')] — round 2 resolved via part:-assembly, +composition to a fixpoint confirmed with real resolver code. + +Risks / notes +------------- +- Under settings.repeat the child `thefuck --repeat` may hit the + parent's open shelve: pre-existing, silently degraded via the {} + fallback (documented, not fixed — per plan). +- forget's word:/part: deletion is exact-inverse-of-record; under + cross-record interference (a later record overwriting word: for + the same orig token) the later mapping is deleted too — accepted + by the plan's literal inverse semantics. +- Nushell outcome gating is new behavior: on success a learned-only + replay now re-records (count bump), on failure it forgets — + required for poison self-heal. diff --git a/tests/entrypoints/test_fix_command_learned.py b/tests/entrypoints/test_fix_command_learned.py index 4130dbf..9815f0e 100644 --- a/tests/entrypoints/test_fix_command_learned.py +++ b/tests/entrypoints/test_fix_command_learned.py @@ -2,61 +2,92 @@ import pytest from mock import Mock, patch from thefuck import danger as real_danger from thefuck.entrypoints.fix_command import fix_command -from thefuck.types import CorrectedCommand +from thefuck.shells import Nushell +from thefuck.types import Command, CorrectedCommand + +ORIGINAL = "git psuh origin main" +CORRECTED = "git push origin main" @pytest.fixture def mock_learned(monkeypatch): - state = {"correction": None, "guess": None, "history": None, - "help": None, "recordings": [], "calls": []} + state = {"correction": None, "suggestion": None, "history": None, + "guess": None, "help": None, "recordings": [], "calls": [], + "forgets": [], "order": [], "closes": 0} # The real gate fail-safes to True without bashlex, which would # make every auto-apply test platform-dependent; the danger - # override tests re-install the real module. + # override tests re-install the real module in both consumers. fake_danger = Mock() fake_danger.is_dangerous.return_value = False monkeypatch.setattr( "thefuck.entrypoints.fix_command.danger", fake_danger ) + monkeypatch.setattr("thefuck.compose.danger", fake_danger) - def fake_get_correction(script): - state["calls"].append("correction") - return state["correction"] + def scripted(key, script): + value = state[key] + if callable(value): + return value(script) + # Input-aware: a constant candidate is scripted for the + # original script only, so round 2 of the composition + # declines instead of accepting it again. + return value if script == ORIGINAL else None - def fake_history(command): - state["calls"].append("history") - return state["history"] + def script_source(key, label): + def fake(script): + state["calls"].append(label) + return scripted(key, script) + return fake - def fake_guess(script): - state["calls"].append("guess") - return state["guess"] - - def fake_help(script): - state["calls"].append("help") - return state["help"] + def command_source(key, label): + def fake(command): + state["calls"].append(label) + return scripted(key, command.script) + return fake def fake_record(original, corrected): + state["order"].append("record") state["recordings"].append((original, corrected)) + def fake_forget(original, corrected): + state["order"].append("forget") + state["forgets"].append((original, corrected)) + + def fake_close_db(): + state["order"].append("close") + state["closes"] += 1 + monkeypatch.setattr( - "thefuck.entrypoints.fix_command.get_correction", fake_get_correction + "thefuck.compose.get_correction", + script_source("correction", "learned") ) monkeypatch.setattr( - "thefuck.entrypoints.fix_command.get_history_correction", fake_history + "thefuck.compose.get_suggestion_candidates", + command_source("suggestion", "error-suggestion") ) monkeypatch.setattr( - "thefuck.entrypoints.fix_command.guess_from_path", fake_guess + "thefuck.compose.get_history_correction", + command_source("history", "history") ) monkeypatch.setattr( - "thefuck.entrypoints.fix_command.get_help_correction", fake_help + "thefuck.compose.guess_from_path", script_source("guess", "path") + ) + monkeypatch.setattr( + "thefuck.compose.get_help_correction", script_source("help", "help") + ) + fake_learned = Mock(record=fake_record, forget=fake_forget, + close_db=fake_close_db) + monkeypatch.setattr( + "thefuck.entrypoints.fix_command.learned", fake_learned ) - monkeypatch.setattr("thefuck.entrypoints.fix_command.record", fake_record) return state @pytest.fixture def known_args(): return Mock( - force_command="git psuh origin main", yes=False, debug=False, repeat=False + force_command="git psuh origin main", yes=False, debug=False, + repeat=False ) @@ -64,9 +95,10 @@ class TestLearnedAutoApply(object): def test_auto_applies_learned_correction( self, mock_learned, known_args, settings, monkeypatch ): - mock_learned["correction"] = "git push origin main" + mock_learned["correction"] = CORRECTED monkeypatch.setattr( - "thefuck.entrypoints.fix_command.get_corrected_commands", lambda _: iter([]) + "thefuck.entrypoints.fix_command.get_corrected_commands", + lambda _: iter([]) ) monkeypatch.setattr( "thefuck.entrypoints.fix_command.select_command", lambda _: None @@ -81,10 +113,11 @@ class TestLearnedAutoApply(object): def test_learned_skips_rule_matching( self, mock_learned, known_args, settings, monkeypatch ): - mock_learned["correction"] = "git push origin main" + mock_learned["correction"] = CORRECTED get_corrected = Mock() monkeypatch.setattr( - "thefuck.entrypoints.fix_command.get_corrected_commands", get_corrected + "thefuck.entrypoints.fix_command.get_corrected_commands", + get_corrected ) with patch("thefuck.types.CorrectedCommand.run"), patch( @@ -96,7 +129,7 @@ class TestLearnedAutoApply(object): def test_shows_corrected_command_on_auto_apply( self, mock_learned, known_args, settings, monkeypatch ): - mock_learned["correction"] = "git push origin main" + mock_learned["correction"] = CORRECTED with patch("thefuck.types.CorrectedCommand.run"), patch( "thefuck.logs.show_corrected_command" @@ -104,17 +137,18 @@ class TestLearnedAutoApply(object): fix_command(known_args) assert mock_show.call_count == 1 shown_cmd = mock_show.call_args[0][0] - assert shown_cmd.script == "git push origin main" + assert shown_cmd.script == CORRECTED class TestGuessAutoApply(object): def test_guess_records_and_auto_applies( self, mock_learned, known_args, settings, monkeypatch ): - mock_learned["guess"] = "git push origin main" + mock_learned["guess"] = CORRECTED get_corrected = Mock() monkeypatch.setattr( - "thefuck.entrypoints.fix_command.get_corrected_commands", get_corrected + "thefuck.entrypoints.fix_command.get_corrected_commands", + get_corrected ) with patch("thefuck.types.CorrectedCommand.run") as mock_run, patch( @@ -123,15 +157,13 @@ class TestGuessAutoApply(object): fix_command(known_args) mock_run.assert_called_once() get_corrected.assert_not_called() - assert mock_learned["recordings"] == [ - ("git psuh origin main", "git push origin main") - ] + assert mock_learned["recordings"] == [(ORIGINAL, CORRECTED)] def test_exact_learned_wins_over_guess( self, mock_learned, known_args, settings, monkeypatch ): mock_learned["correction"] = "git push origin dev" - mock_learned["guess"] = "git push origin main" + mock_learned["guess"] = CORRECTED with patch("thefuck.types.CorrectedCommand.run"), patch( "thefuck.logs.show_corrected_command" @@ -143,13 +175,18 @@ class TestGuessAutoApply(object): class TestResolverChainOrder(object): + # Round 1 consults [learned, error-suggestion, history, path, + # help] up to the round's winner; round 2 consults all five with + # the corrected script, the input-aware fakes decline and the + # composition ends at a stable point. def test_full_chain_consulted_in_order_when_only_help_hits( self, mock_learned, known_args, settings, monkeypatch ): - mock_learned["help"] = "git push origin main" + mock_learned["help"] = CORRECTED get_corrected = Mock() monkeypatch.setattr( - "thefuck.entrypoints.fix_command.get_corrected_commands", get_corrected + "thefuck.entrypoints.fix_command.get_corrected_commands", + get_corrected ) with patch("thefuck.types.CorrectedCommand.run"), patch( @@ -158,18 +195,24 @@ class TestResolverChainOrder(object): fix_command(known_args) assert mock_learned["calls"] == [ - "correction", "history", "guess", "help" + "learned", "error-suggestion", "history", "path", "help", + "learned", "error-suggestion", "history", "path", "help", ] get_corrected.assert_not_called() @pytest.mark.parametrize("hit_source,expected_calls", [ - ("history", ["correction", "history"]), - ("guess", ["correction", "history", "guess"]), + ("history", ["learned", "error-suggestion", "history", + "learned", "error-suggestion", "history", + "path", "help"]), + ("guess", ["learned", "error-suggestion", "history", "path", + "learned", "error-suggestion", "history", "path", + "help"]), ]) - def test_earlier_hit_stops_the_chain( - self, mock_learned, known_args, settings, hit_source, expected_calls + def test_earlier_hit_stops_the_round( + self, mock_learned, known_args, settings, monkeypatch, hit_source, + expected_calls ): - mock_learned[hit_source] = "git push origin main" + mock_learned[hit_source] = CORRECTED with patch("thefuck.types.CorrectedCommand.run"), patch( "thefuck.logs.show_corrected_command" @@ -183,7 +226,7 @@ class TestResolverChainOrder(object): ): history_mock = Mock(return_value=None) monkeypatch.setattr( - "thefuck.entrypoints.fix_command.get_history_correction", history_mock + "thefuck.compose.get_history_correction", history_mock ) monkeypatch.setattr( "thefuck.entrypoints.fix_command.get_corrected_commands", @@ -197,7 +240,7 @@ class TestResolverChainOrder(object): fix_command(known_args) assert history_mock.call_count == 1 - assert history_mock.call_args[0][0].script == "git psuh origin main" + assert history_mock.call_args[0][0].script == ORIGINAL class TestResolverAutoApply(object): @@ -205,10 +248,11 @@ class TestResolverAutoApply(object): def test_resolver_hit_records_and_auto_applies( self, mock_learned, known_args, settings, monkeypatch, source ): - mock_learned[source] = "git push origin main" + mock_learned[source] = CORRECTED get_corrected = Mock() monkeypatch.setattr( - "thefuck.entrypoints.fix_command.get_corrected_commands", get_corrected + "thefuck.entrypoints.fix_command.get_corrected_commands", + get_corrected ) with patch("thefuck.types.CorrectedCommand.run") as mock_run, patch( @@ -217,25 +261,44 @@ class TestResolverAutoApply(object): fix_command(known_args) mock_run.assert_called_once() get_corrected.assert_not_called() - assert mock_show.call_args[0][0].script == "git push origin main" - assert mock_learned["recordings"] == [ - ("git psuh origin main", "git push origin main") - ] + assert mock_show.call_args[0][0].script == CORRECTED + assert mock_learned["recordings"] == [(ORIGINAL, CORRECTED)] + + +class TestComposition(object): + def test_two_sources_compose_into_one_run( + self, mock_learned, known_args, settings, monkeypatch + ): + mock_learned["guess"] = "git psuh origin main" + mock_learned["history"] = ( + lambda script: CORRECTED if script == "git psuh origin main" + else None) + + with patch("thefuck.types.CorrectedCommand.run") as mock_run, patch( + "thefuck.logs.show_corrected_command" + ) as mock_show: + fix_command(known_args) + mock_run.assert_called_once() + assert mock_show.call_args[0][0].script == CORRECTED + assert mock_learned["recordings"] == [(ORIGINAL, CORRECTED)] class TestDangerOverride(object): @pytest.mark.parametrize( - "hit_source", ["correction", "history", "guess", "help"]) + "hit_source", + ["correction", "suggestion", "history", "guess", "help"]) def test_dangerous_hit_asks_instead_of_auto_running( self, mock_learned, known_args, settings, monkeypatch, hit_source ): # The real gate: a danger-flagged candidate from ANY source — - # a seeded learned-db exact hit included — reaches - # select_command and nothing auto-runs. The abort via the - # mocked selection keeps the pin free of terminal IO. + # a seeded learned-db exact hit included — is declined by the + # composer, falls to select_command and nothing auto-runs. The + # abort via the mocked selection keeps the pin free of + # terminal IO. monkeypatch.setattr( "thefuck.entrypoints.fix_command.danger", real_danger ) + monkeypatch.setattr("thefuck.compose.danger", real_danger) mock_learned[hit_source] = "rm -rf /" select = Mock(return_value=None) monkeypatch.setattr( @@ -262,27 +325,27 @@ class TestRecordOnSelection(object): self, mock_learned, known_args, settings, monkeypatch ): selected = CorrectedCommand( - script="git push origin main", side_effect=None, priority=100 + script=CORRECTED, side_effect=None, priority=100 ) monkeypatch.setattr( "thefuck.entrypoints.fix_command.get_corrected_commands", lambda _: iter([selected]), ) monkeypatch.setattr( - "thefuck.entrypoints.fix_command.select_command", lambda _: selected + "thefuck.entrypoints.fix_command.select_command", + lambda _: selected ) with patch("thefuck.types.CorrectedCommand.run"): fix_command(known_args) - assert mock_learned["recordings"] == [ - ("git psuh origin main", "git push origin main") - ] + assert mock_learned["recordings"] == [(ORIGINAL, CORRECTED)] def test_does_not_record_on_abort( self, mock_learned, known_args, settings, monkeypatch ): monkeypatch.setattr( - "thefuck.entrypoints.fix_command.get_corrected_commands", lambda _: iter([]) + "thefuck.entrypoints.fix_command.get_corrected_commands", + lambda _: iter([]) ) monkeypatch.setattr( "thefuck.entrypoints.fix_command.select_command", lambda _: None @@ -298,14 +361,15 @@ class TestFallthrough(object): self, mock_learned, known_args, settings, monkeypatch ): selected = CorrectedCommand( - script="git push origin main", side_effect=None, priority=100 + script=CORRECTED, side_effect=None, priority=100 ) monkeypatch.setattr( "thefuck.entrypoints.fix_command.get_corrected_commands", lambda _: iter([selected]), ) monkeypatch.setattr( - "thefuck.entrypoints.fix_command.select_command", lambda _: selected + "thefuck.entrypoints.fix_command.select_command", + lambda _: selected ) with patch("thefuck.types.CorrectedCommand.run") as mock_run: @@ -316,7 +380,7 @@ class TestFallthrough(object): self, mock_learned, known_args, settings, monkeypatch ): selected = CorrectedCommand( - script="git push origin main", side_effect=None, priority=100 + script=CORRECTED, side_effect=None, priority=100 ) select = Mock(return_value=selected) monkeypatch.setattr( @@ -332,6 +396,131 @@ class TestFallthrough(object): mock_run.assert_called_once() assert mock_learned["calls"] == [ - "correction", "history", "guess", "help" + "learned", "error-suggestion", "history", "path", "help" ] assert select.call_count == 1 + + +class TestOutcomeLearning(object): + @pytest.fixture + def nushell(self, monkeypatch): + monkeypatch.setattr( + "thefuck.entrypoints.fix_command.shell", Nushell()) + + def _run_with_returncode(self, order, returncode): + def run(cmd_self, old_cmd): + order.append("run") + cmd_self.returncode = returncode + return run + + def test_success_records_after_run( + self, mock_learned, known_args, settings, monkeypatch, nushell + ): + mock_learned["guess"] = CORRECTED + + with patch("thefuck.types.CorrectedCommand.run", + autospec=True) as mock_run, patch( + "thefuck.logs.show_corrected_command"): + mock_run.side_effect = self._run_with_returncode( + mock_learned["order"], 0) + fix_command(known_args) + + assert mock_learned["order"] == ["close", "run", "record"] + assert mock_learned["recordings"] == [(ORIGINAL, CORRECTED)] + assert mock_learned["forgets"] == [] + + def test_failure_forgets_after_run( + self, mock_learned, known_args, settings, monkeypatch, nushell + ): + mock_learned["guess"] = CORRECTED + + with patch("thefuck.types.CorrectedCommand.run", + autospec=True) as mock_run, patch( + "thefuck.logs.show_corrected_command"): + mock_run.side_effect = self._run_with_returncode( + mock_learned["order"], 1) + fix_command(known_args) + + assert mock_learned["order"] == ["close", "run", "forget"] + assert mock_learned["forgets"] == [(ORIGINAL, CORRECTED)] + assert mock_learned["recordings"] == [] + + def test_learned_replay_failure_still_forgets( + self, mock_learned, known_args, settings, monkeypatch, nushell + ): + # Poison self-heal: a seeded exact hit that fails on rerun is + # removed, so the no-rerecord carve-out must not skip the + # outcome gating. + mock_learned["correction"] = CORRECTED + + with patch("thefuck.types.CorrectedCommand.run", + autospec=True) as mock_run, patch( + "thefuck.logs.show_corrected_command"): + mock_run.side_effect = self._run_with_returncode( + mock_learned["order"], 1) + fix_command(known_args) + + assert mock_learned["order"] == ["close", "run", "forget"] + assert mock_learned["forgets"] == [(ORIGINAL, CORRECTED)] + + def test_repeat_records_before_run_without_outcome_gating( + self, mock_learned, known_args, settings, monkeypatch, nushell + ): + known_args.repeat = True + mock_learned["guess"] = CORRECTED + + with patch("thefuck.types.CorrectedCommand.run", + autospec=True) as mock_run, patch( + "thefuck.logs.show_corrected_command"): + mock_run.side_effect = self._run_with_returncode( + mock_learned["order"], 1) + fix_command(known_args) + + assert mock_learned["order"] == ["record", "run"] + assert mock_learned["recordings"] == [(ORIGINAL, CORRECTED)] + assert mock_learned["closes"] == 0 + + def test_other_shells_record_before_run( + self, mock_learned, known_args, settings, monkeypatch + ): + mock_learned["guess"] = CORRECTED + + with patch("thefuck.types.CorrectedCommand.run", + autospec=True) as mock_run, patch( + "thefuck.logs.show_corrected_command"): + mock_run.side_effect = self._run_with_returncode( + mock_learned["order"], 1) + fix_command(known_args) + + assert mock_learned["order"] == ["record", "run"] + assert mock_learned["closes"] == 0 + + def test_learned_only_win_does_not_rerecord_on_other_shells( + self, mock_learned, known_args, settings, monkeypatch + ): + mock_learned["correction"] = CORRECTED + + with patch("thefuck.types.CorrectedCommand.run", + autospec=True) as mock_run, patch( + "thefuck.logs.show_corrected_command"): + mock_run.side_effect = self._run_with_returncode( + mock_learned["order"], 0) + fix_command(known_args) + + assert mock_learned["order"] == ["run"] + assert mock_learned["recordings"] == [] + + +class TestNushellRunStashesReturncode(object): + def test_run_stashes_nu_returncode_on_the_instance( + self, settings, monkeypatch + ): + settings.alter_history = False + monkeypatch.setattr("thefuck.types.shell", Nushell()) + completed = Mock(returncode=3) + with patch("thefuck.types.subprocess.run", + return_value=completed) as mock_spawn: + corrected = CorrectedCommand("git push", None, 0) + corrected.run(Command("git psuh", "")) + mock_spawn.assert_called_once_with(["nu", "-c", "git push"]) + assert corrected.returncode == 3 diff --git a/tests/test_compose.py b/tests/test_compose.py new file mode 100644 index 0000000..98586aa --- /dev/null +++ b/tests/test_compose.py @@ -0,0 +1,199 @@ +import pytest +from mock import Mock + +from thefuck import compose +from thefuck.types import Command + + +@pytest.fixture +def src(monkeypatch): + """Scripts per-source results and records every consultation. + + A source's scripted entry is a constant candidate, None, or a + callable mapping the consulted script to a candidate (for + input-aware tests). `calls` keeps (source, input-script) pairs in + consultation order; `commands` keeps every Command handed to the + two output-aware sources. + """ + state = { + "learned": None, + "error-suggestion": None, + "history": None, + "path": None, + "help": None, + "calls": [], + "commands": [], + } + fake_danger = Mock() + fake_danger.is_dangerous.return_value = False + monkeypatch.setattr(compose, "danger", fake_danger) + state["danger"] = fake_danger + + def _scripted(name, script): + scripted = state[name] + if callable(scripted): + return scripted(script) + return scripted + + def _script_source(name): + def fake(script): + state["calls"].append((name, script)) + return _scripted(name, script) + return fake + + def _command_source(name): + def fake(command): + state["calls"].append((name, command.script)) + state["commands"].append(command) + return _scripted(name, command.script) + return fake + + monkeypatch.setattr(compose, "get_correction", + _script_source("learned")) + monkeypatch.setattr(compose, "get_suggestion_candidates", + _command_source("error-suggestion")) + monkeypatch.setattr(compose, "get_history_correction", + _command_source("history")) + monkeypatch.setattr(compose, "guess_from_path", + _script_source("path")) + monkeypatch.setattr(compose, "get_help_correction", + _script_source("help")) + return state + + +class TestComposition(object): + def test_two_rounds_compose_into_one_script(self, src): + src["path"] = lambda script: ( + "git psuh" if script == "gti psuh" else None) + src["history"] = ( + lambda script: "git push" if script == "git psuh" else None) + final, steps = compose.resolve(Command("gti psuh", "git output")) + assert final == "git push" + assert steps == [("path", "git psuh"), ("history", "git push")] + + def test_first_safe_source_wins_the_round(self, src): + src["learned"] = "git push" + src["history"] = "git commit" + final, steps = compose.resolve(Command("gti psuh", "")) + assert final == "git push" + assert steps == [("learned", "git push")] + + def test_chain_stops_when_candidate_is_stable(self, src): + src["history"] = "git push" + final, steps = compose.resolve(Command("gti psuh", "")) + assert final == "git push" + assert steps == [("history", "git push")] + + def test_three_round_chain_without_cap(self, src): + src["history"] = lambda script: { + "aa bb": "ab bb", "ab bb": "ac bb"}.get(script) + final, steps = compose.resolve(Command("aa bb", "")) + assert final == "ac bb" + assert steps == [("history", "ab bb"), ("history", "ac bb")] + + +class TestNoAutoRunOfOriginal(object): + def test_round1_no_candidate_returns_none(self, src): + final, steps = compose.resolve(Command("gti psuh", "")) + assert final is None + assert steps == [] + assert src["calls"] == [ + ("learned", "gti psuh"), + ("error-suggestion", "gti psuh"), + ("history", "gti psuh"), + ("path", "gti psuh"), + ("help", "gti psuh"), + ] + + def test_round1_candidate_equal_to_original_returns_none(self, src): + src["path"] = "gti psuh" + final, steps = compose.resolve(Command("gti psuh", "")) + assert final is None + assert steps == [] + + def test_empty_script_declined_not_echoed(self, src): + final, steps = compose.resolve(Command("", "")) + assert final is None + assert steps == [] + assert src["calls"][0] == ("learned", "") + + +class TestAbort(object): + def test_cycle_between_two_scripts_aborts(self, src): + src["history"] = lambda script: { + "alpha": "beta", "beta": "alpha"}[script] + final, steps = compose.resolve(Command("alpha", "")) + assert final is None + assert steps == [("history", "beta")] + + def test_original_reentry_in_round2_aborts(self, src): + src["path"] = "beta" + src["history"] = lambda script: ( + "alpha" if script == "beta" else None) + final, steps = compose.resolve(Command("alpha", "")) + assert final is None + assert steps == [("path", "beta")] + + +class TestSourceInputs(object): + def test_output_sources_get_synthesized_commands(self, src): + command = Command("gti psuh", "suggestion output") + src["path"] = lambda script: ( + "git psuh" if script == "gti psuh" else None) + src["history"] = ( + lambda script: "git push" if script == "git psuh" else None) + final, _ = compose.resolve(command) + assert final == "git push" + # Per round, both output-aware sources see the round's script: + # original, round-1 composite, round-2 composite (all decline). + assert src["commands"] == [ + Command("gti psuh", "suggestion output"), + Command("gti psuh", "suggestion output"), + Command("git psuh", "suggestion output"), + Command("git psuh", "suggestion output"), + Command("git push", "suggestion output"), + Command("git push", "suggestion output"), + ] + + def test_script_sources_get_bare_strings(self, src, monkeypatch): + calls = [] + + def fake_guess(script): + calls.append(("path", script)) + assert not isinstance(script, Command) + return None + monkeypatch.setattr(compose, "guess_from_path", fake_guess) + compose.resolve(Command("gti psuh", "")) + assert calls == [("path", "gti psuh")] + + +class TestDangerDecline(object): + def test_dangerous_candidate_skipped_next_source_wins(self, src): + src["learned"] = "rm -rf /" + src["danger"].is_dangerous.side_effect = ( + lambda script: script == "rm -rf /") + src["path"] = "git push" + final, steps = compose.resolve(Command("gti psuh", "")) + assert final == "git push" + assert steps == [("path", "git push")] + + def test_only_dangerous_candidates_declined(self, src): + src["learned"] = "rm -rf /" + src["danger"].is_dangerous.return_value = True + final, steps = compose.resolve(Command("gti psuh", "")) + assert final is None + assert steps == [] + + +class TestMemoTripwire(object): + def test_no_source_input_ever_repeats(self, src): + # Every accepted candidate is a new node, so no (source, + # script) pair is consulted twice; the composer's internal + # memo assert stays silent across a long chain. + src["history"] = lambda script: { + "aaaa": "aaab", "aaab": "aabb", + "aabb": "abbb", "abbb": "bbbb"}.get(script) + final, steps = compose.resolve(Command("aaaa", "")) + assert final == "bbbb" + assert len(steps) == 4 + assert len(src["calls"]) == len(set(src["calls"])) diff --git a/tests/test_learned.py b/tests/test_learned.py index 5ccfc12..5e031bf 100644 --- a/tests/test_learned.py +++ b/tests/test_learned.py @@ -261,6 +261,95 @@ class TestClear(object): assert learned.get_correction("git psuh") is None +class TestForget(object): + def test_removes_cmd_and_derived_part_entries(self, learned): + learned.record("git psuh origin main", "git push origin main") + learned.forget("git psuh origin main", "git push origin main") + assert "cmd:git psuh origin main" not in learned.db + assert "part:git:psuh" not in learned.db + + def test_removes_cmd_and_derived_word_entry(self, learned): + learned.record("pyhton script.py", "python script.py") + learned.forget("pyhton script.py", "python script.py") + assert "cmd:pyhton script.py" not in learned.db + assert "word:pyhton" not in learned.db + + def test_removes_every_diverged_position(self, learned): + learned.record("gti comit -m msg", "git commit -m msg") + learned.forget("gti comit -m msg", "git commit -m msg") + assert "cmd:gti comit -m msg" not in learned.db + assert "word:gti" not in learned.db + assert "part:git:comit" not in learned.db + + def test_deletes_rather_than_decrements(self, learned): + for _ in range(3): + learned.record("git psuh", "git push") + learned.forget("git psuh", "git push") + assert "cmd:git psuh" not in learned.db + assert "part:git:psuh" not in learned.db + + def test_keeps_unrelated_entries(self, learned): + learned.record("git psuh", "git push") + learned.record("pyhton x.py", "python x.py") + learned.forget("git psuh", "git push") + assert "cmd:pyhton x.py" in learned.db + assert "word:pyhton" in learned.db + + def test_mismatched_token_counts_forgets_cmd_only(self, learned): + learned.record("git push", "git push --set-upstream origin main") + learned.forget("git push", "git push --set-upstream origin main") + assert "cmd:git push" not in learned.db + assert not any(k.startswith(("word:", "part:")) + for k in learned.db) + + def test_assembly_failure_is_a_noop(self, learned): + # "gti comit" resolves by assembling word:gti alone (no cmd: + # entry of its own); those mappings were validated by an + # earlier success of a different command and must stand. + learned.record("gti push", "git push") + assert learned.get_correction("gti comit") == "git comit" + learned.forget("gti comit", "git comit") + assert "word:gti" in learned.db + assert "cmd:gti comit" not in learned.db + + def test_unknown_command_is_a_noop(self, learned): + learned.forget("never recorded", "anything at all") + assert len(learned.db) == 0 + + +class TestCloseDb(object): + def test_close_sets_db_none_and_is_idempotent(self, learned): + learned.record("git psuh", "git push") + learned.close_db() + assert learned._db is None + learned.close_db() + assert learned._db is None + + def test_reopen_after_close_still_finds_entries( + self, tmp_path, monkeypatch): + monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path)) + lc = LearnedCorrections() + lc.record("git psuh", "git push") + first_handle = lc._db + lc.close_db() + assert lc.get_correction("git psuh") == "git push" + assert lc._db is not None + assert lc._db is not first_handle + + def test_atexit_binds_close_db_once_across_reopens( + self, tmp_path, monkeypatch): + registrations = [] + monkeypatch.setattr( + "thefuck.learned.atexit.register", registrations.append) + monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path)) + lc = LearnedCorrections() + lc.record("git psuh", "git push") + lc.close_db() + lc.record("gti comit -m msg", "git commit -m msg") + lc.close_db() + assert registrations == [lc.close_db] + + class TestRoundTrip(object): def test_record_then_match(self, learned): learned.record("docker bilud .", "docker build .") diff --git a/thefuck/compose.py b/thefuck/compose.py new file mode 100644 index 0000000..cdde152 --- /dev/null +++ b/thefuck/compose.py @@ -0,0 +1,119 @@ +"""Fixpoint composition of the auto-run correction sources. + +Consults the learned db, the failing binary's own suggestions, +history, $PATH guesses and `--help` in a fixed order, applying each +round's accepted correction to the previous round's script until the +result is stable, revisits a script (cycle) or no source has a safe +candidate left. Only the final composite is returned for running. + +Interface contract for sources (termination rests on it): every +source MUST return either a string drawn from a finite fixed range +(learned-db entries, history lines, help/suggestion lists, +executables on $PATH) or a word-count-preserving splice of its input +script — the reachable script space is then finite (finite +vocabulary, bounded word count), so the chain cannot grow forever. +A source that APPENDS to or otherwise grows scripts unboundedly +would break termination and is FORBIDDEN. Cycles cannot loop: any +revisit of an already-seen script hits the visited-set and aborts +the composition. + +Personal fork only: depends on bashlex (GPL-3+), do not distribute. +""" +from . import danger, logs, types +from .learned import get_correction, guess_from_path +from .resolvers.error_suggestion import get_suggestion_candidates +from .resolvers.help_resolver import get_help_correction +from .resolvers.history_resolver import get_history_correction + +LEARNED_SOURCE = 'learned' + + +def resolve(command): + """Composes corrections from the sources into one fixed script. + + Returns `(final_script, steps)` where `steps` holds one + `(source, script)` pair per accepted round, or `(None, steps)` + when nothing was accepted — round one yielding no candidate or a + candidate equal to the original — so the caller falls back to + rules and asking: the original failing script is never + auto-run. A candidate revisiting an already-seen script aborts + the same way, with the steps accepted before the revisit. + + :type command: thefuck.types.Command + :rtype: (str | None, [(str, str)]) + """ + current = command.script + visited = set([current]) + steps = [] + memo = {} + while True: + name, candidate = _round(command, current, memo) + if name is None: + break + if candidate == current: + logs.debug(u'Correction stable at: {}'.format(candidate)) + break + if candidate in visited: + logs.debug(u'Correction cycles back to {}, aborting' + .format(candidate)) + return None, steps + visited.add(candidate) + steps.append((name, candidate)) + current = candidate + if not steps: + return None, steps + return current, steps + + +def _round(command, script, memo): + """Consults every source in order against `script`. + + Returns the first `(name, candidate)` whose candidate passes the + danger check (dangerous candidates are declined for auto-run and + consultation continues), or `(None, None)` when every source + declines. `memo` is a determinism tripwire, not the termination + guarantee: inputs never repeat within one resolve because every + accepted candidate is a new node, so a `(source, script)` hit in + the memo means a source or the loop broke that invariant. + """ + for name, consult in _sources(): + key = (name, script) + assert key not in memo, ( + u'tripwire: {} consulted twice with {!r}'.format(name, script)) + candidate = consult(command, script) + memo[key] = candidate + if candidate is None: + continue + if danger.is_dangerous(candidate): + logs.debug(u'Declining dangerous candidate from {}: {}' + .format(name, candidate)) + continue + return name, candidate + return None, None + + +def _sources(): + """The auto-run sources in consultation order. + + Each adapter maps `(command, script)` to a candidate script or + None; the two output-aware resolvers get a synthesized + `types.Command` wrapping the round's script with the original + failure's output, so later rounds still see what went wrong. + """ + return ( + (LEARNED_SOURCE, lambda command, script: get_correction(script)), + ('error-suggestion', _suggestion), + ('history', lambda command, script: get_history_correction( + types.Command(script=script, output=command.output))), + ('path', lambda command, script: guess_from_path(script)), + ('help', lambda command, script: get_help_correction(script)), + ) + + +def _suggestion(command, script): + # The failing output is only readable while it exists; a timed-out + # rerun has None and nothing can be extracted from it. + if command.output is None: + return None + return get_suggestion_candidates( + types.Command(script=script, output=command.output)) diff --git a/thefuck/entrypoints/fix_command.py b/thefuck/entrypoints/fix_command.py index dfd49bd..86bbb85 100644 --- a/thefuck/entrypoints/fix_command.py +++ b/thefuck/entrypoints/fix_command.py @@ -2,13 +2,11 @@ from pprint import pformat import os import sys from difflib import SequenceMatcher -from .. import danger, logs, types, const +from .. import compose, const, danger, learned, logs, types from ..conf import settings from ..corrector import get_corrected_commands from ..exceptions import EmptyCommand -from ..learned import get_correction, guess_from_path, record -from ..resolvers.help_resolver import get_help_correction -from ..resolvers.history_resolver import get_history_correction +from ..shells import Nushell, shell from ..ui import select_command from ..utils import get_alias, get_all_executables @@ -29,6 +27,33 @@ def _get_raw_command(known_args): return [] +def _run_learned(command, corrected, steps): + """Runs an auto-applied correction, learning from the outcome. + + Under `settings.repeat` or a non-nushell parent shell the outcome + cannot be observed (the script goes to the parent shell or a + child `thefuck --repeat` process), so today's behavior stands: + record immediately before the run, unless the composition only + replayed a learned-db exact hit. Under nushell the db is closed + before the run (so the child can safely open its own) and the + fix is remembered only when the command succeeds — a failing + fix is forgotten again. + """ + replayed_learned_only = (len(steps) == 1 + and steps[0][0] == compose.LEARNED_SOURCE) + if settings.repeat or not isinstance(shell, Nushell): + if not replayed_learned_only: + learned.record(command.script, corrected.script) + corrected.run(command) + return + learned.close_db() + corrected.run(command) + if corrected.returncode == 0: + learned.record(command.script, corrected.script) + else: + learned.forget(command.script, corrected.script) + + def fix_command(known_args): """Fixes previous command. Used when `thefuck` called without arguments.""" settings.init(known_args) @@ -42,42 +67,29 @@ def fix_command(known_args): logs.debug("Empty command, nothing to do") return - learned_script = get_correction(command.script) - if not learned_script: - learned_script = get_history_correction(command) - if learned_script: - logs.debug("Corrected from history: {}".format( - learned_script)) - record(command.script, learned_script) - if not learned_script: - learned_script = guess_from_path(command.script) - if learned_script: - logs.debug("Guessed correction from $PATH: {}".format( - learned_script)) - record(command.script, learned_script) - if not learned_script: - learned_script = get_help_correction(command.script) - if learned_script: - logs.debug("Corrected from binary help: {}".format( - learned_script)) - record(command.script, learned_script) - if learned_script and danger.is_dangerous(learned_script): + learned_script, steps = compose.resolve(command) + if learned_script is not None \ + and danger.is_dangerous(learned_script): + # Belt and suspenders: the composer already declines + # dangerous candidates; this guards against regressions. logs.debug("Refusing to auto-run dangerous correction, " "asking instead: {}".format(learned_script)) learned_script = None - if learned_script: + if learned_script is not None: + for source, step in steps: + logs.debug("Composed via {}: {}".format(source, step)) learned_cmd = types.CorrectedCommand( script=learned_script, side_effect=None, priority=0 ) logs.show_corrected_command(learned_cmd) - learned_cmd.run(command) + _run_learned(command, learned_cmd, steps) return corrected_commands = get_corrected_commands(command) selected_command = select_command(corrected_commands) if selected_command: - record(command.script, selected_command.script) + learned.record(command.script, selected_command.script) selected_command.run(command) else: sys.exit(1) diff --git a/thefuck/learned.py b/thefuck/learned.py index c78d561..30aa59a 100644 --- a/thefuck/learned.py +++ b/thefuck/learned.py @@ -22,6 +22,7 @@ except ImportError: class LearnedCorrections(object): def __init__(self): self._db = None + self._exit_registered = False def _init_db(self): try: @@ -42,7 +43,11 @@ class LearnedCorrections(object): if os.path.exists(path): os.remove(path) self._db = shelve.open(cache_path) - atexit.register(self._db.close) + # Bind the stable close_db method, not the current _db.close + # object: it would go stale after a close/reopen cycle. + if not self._exit_registered: + atexit.register(self.close_db) + self._exit_registered = True @staticmethod def _get_cache_dir(): @@ -173,6 +178,57 @@ class LearnedCorrections(object): return None return shell_ast.splice(script, replacements) + def forget(self, original_script, corrected_script): + """Removes what `record(original, corrected)` stored. + + The exact inverse of `record`: deletes the `cmd:` entry and, + when token counts match, the `word:`/`part:` mappings for + every diverged position (`part:` keys live under the + corrected head, hence the second argument). A correction that + was assembled from `word:`/`part:` lookups alone — no `cmd:` + entry of its own — is a documented no-op: those mappings were + validated by a prior success of a different command and + stand. + """ + db = self.db + full_key = "cmd:" + original_script + if full_key not in db: + return + del db[full_key] + + original_parts = original_script.split() + corrected_parts = corrected_script.split() + if not (original_parts and corrected_parts + and len(original_parts) == len(corrected_parts)): + self._sync() + return + for i, (orig_tok, corr_tok) in enumerate( + zip(original_parts, corrected_parts)): + if orig_tok == corr_tok: + continue + if i == 0: + key = "word:" + orig_tok + else: + key = "part:" + corrected_parts[0] + ":" + orig_tok + if key in db: + del db[key] + self._sync() + + def close_db(self): + """Closes the learned db when open; safe to call repeatedly. + + The next access reopens it, so the run of a corrected command + can happen with the shelve closed (a `--repeat` child process + would otherwise hit the parent's open handle). + """ + if self._db is not None: + try: + self._db.close() + except Exception: + # The degraded {} fallback has no close. + logs.debug("Unable to close learned-corrections db") + self._db = None + def clear(self): db = self.db for key in list(db.keys()): @@ -191,4 +247,6 @@ _learned = LearnedCorrections() record = _learned.record get_correction = _learned.get_correction guess_from_path = _learned.guess_from_path +forget = _learned.forget +close_db = _learned.close_db clear = _learned.clear diff --git a/thefuck/types.py b/thefuck/types.py index 2f16ea6..7da86d2 100644 --- a/thefuck/types.py +++ b/thefuck/types.py @@ -264,6 +264,7 @@ class CorrectedCommand(object): if isinstance(shell, Nushell): # TODO: fix for better option, the lack of '-l' flag # means we lose any ENV that could be needed - subprocess.run(["nu", "-c", script]) + self.returncode = subprocess.run( + ["nu", "-c", script]).returncode else: sys.stdout.write(script)