Private
Public Access
0
0
Files
manual_slop/src/fuzzy_anchor.py
T
ed ba3eb0c090 refactor(multiple): continue Phase 6 Optional[T] elimination (batch 2)
Phase 6: Eliminate Optional[T] returns - BATCH 2 of 7
Before: 7 more Optional[T] returns removed
After:  0 in command_palette.py, diff_viewer.py, fuzzy_anchor.py,
        multi_agent_conductor.py, patch_modal.py, app_controller.py
Delta:  -7 sites (cumulative: -15 of 30)

Specific changes:
- src/command_palette.py:50: CommandRegistry.get() returns Command (zero-init
  sentinel: id="", title="", category="uncategorized", action=lambda: None)
- src/diff_viewer.py:117: get_line_color returns "" when no marker prefix
- src/fuzzy_anchor.py:40: FuzzyAnchor.resolve_slice returns (-1, -1) sentinel
  (replaced 3x `return None` with `return (-1, -1)`)
- src/multi_agent_conductor.py:64: WorkerPool.spawn returns threading.Thread()
  (empty sentinel, not started) when pool is full
- src/patch_modal.py:33: PatchModalManager.get_pending_patch returns
  PendingPatch; class has EMPTY_PATCH sentinel; field type changed from
  Optional[PendingPatch] to PendingPatch; 2x `= None` reset replaced with
  `= EMPTY_PATCH`
- src/app_controller.py:4414: _confirm_and_run returns "" when not approved
  (was Optional[str] returning None)

Test updates:
- tests/test_diff_viewer.py:95: get_line_color(" context") == ""
- tests/test_fuzzy_anchor.py:42,59: assert result == (-1, -1)
- tests/test_parallel_execution.py:31: t3 sentinel is now unstarted thread
  (check via not t3.is_alive())
- tests/test_patch_modal.py:9,31,78: get_pending_patch() == "" sentinel check

Verification:
- audit_weak_types --strict: OK (107 <= 112 baseline)
- 22+ tests pass (test_diff_viewer, test_fuzzy_anchor,
  test_parallel_execution, test_patch_modal, test_command_palette)
- py_check_syntax: OK on all changed files

REMAINING: ~15 Optional[T] returns in:
- src/external_editor.py (3)
- src/file_cache.py (7)
- src/diff_viewer.py: parse_hunk_header (1)
- src/models.py: ExternalEditorConfig.get_default (1)
- src/project_manager.py: load_track_state (1)
- src/session_logger.py: log_tool_call (1)
- src/app_controller.py: _pending_mma_spawn, _pending_mma_approval (2)
2026-06-26 05:07:35 -04:00

93 lines
3.6 KiB
Python

import hashlib
import re
from typing import Optional, Tuple
class FuzzyAnchor:
@staticmethod
def get_context(lines: list[str], index: int, count: int, direction: int) -> list[str]:
context = []
curr = index
while len(context) < count and 0 <= curr < len(lines):
line = lines[curr].strip()
if line:
context.append(line)
curr += direction
return context
@classmethod
def create_slice(cls, text: str, start_line: int, end_line: int) -> dict:
"""
start_line and end_line are 1-based.
[C: src/gui_2.py:App._populate_auto_slices, src/gui_2.py:App._render_text_viewer_window, tests/test_fuzzy_anchor.py:TestFuzzyAnchor.test_create_slice_basic, tests/test_fuzzy_anchor.py:TestFuzzyAnchor.test_resolve_slice_anchor_mismatch_returns_none, tests/test_fuzzy_anchor.py:TestFuzzyAnchor.test_resolve_slice_exact_match, tests/test_fuzzy_anchor.py:TestFuzzyAnchor.test_resolve_slice_line_deleted_before_returns_none, tests/test_fuzzy_anchor.py:TestFuzzyAnchor.test_resolve_slice_line_inserted_before, tests/test_fuzzy_anchor.py:TestFuzzyAnchor.test_resolve_slice_multiple_lines_changed, tests/test_slice_editor_behavior.py:test_add_slice_with_annotations]
"""
lines = text.splitlines()
s_idx = max(0, start_line - 1)
e_idx = min(len(lines), end_line)
slice_lines = lines[s_idx:e_idx]
slice_text = "\n".join(slice_lines)
return {
"start_line": start_line,
"end_line": end_line,
"start_context": cls.get_context(lines, s_idx, 3, 1),
"end_context": cls.get_context(lines, e_idx - 1, 3, -1)[::-1], # Reverse back to normal order
"content_hash": hashlib.mdsafe(slice_text.encode()).hexdigest() if hasattr(hashlib, 'mdsafe') else hashlib.md5(slice_text.encode()).hexdigest()
}
@classmethod
def resolve_slice(cls, text: str, slice_data: dict) -> Tuple[int, int]:
"""Returns (start_line, end_line) on success, or (-1, -1) if unresolved."""
result: Tuple[int, int] = (-1, -1)
"""
[C: tests/test_fuzzy_anchor.py:TestFuzzyAnchor.test_resolve_slice_anchor_mismatch_returns_none, tests/test_fuzzy_anchor.py:TestFuzzyAnchor.test_resolve_slice_exact_match, tests/test_fuzzy_anchor.py:TestFuzzyAnchor.test_resolve_slice_line_deleted_before_returns_none, tests/test_fuzzy_anchor.py:TestFuzzyAnchor.test_resolve_slice_line_inserted_before, tests/test_fuzzy_anchor.py:TestFuzzyAnchor.test_resolve_slice_multiple_lines_changed]
"""
lines = text.splitlines()
# 1. Try exact match
s_idx = slice_data["start_line"] - 1
e_idx = slice_data["end_line"]
if 0 <= s_idx < len(lines) and e_idx <= len(lines):
current_text = "\n".join(lines[s_idx:e_idx])
curr_hash = hashlib.md5(current_text.encode()).hexdigest()
if curr_hash == slice_data["content_hash"]:
return (slice_data["start_line"], slice_data["end_line"])
# 2. Fuzzy match
start_ctx = slice_data["start_context"]
end_ctx = slice_data["end_context"]
if not start_ctx or not end_ctx: return (-1, -1)
# Search for start_ctx
best_s = -1
for i in range(len(lines)):
match = True
for j, ctx_line in enumerate(start_ctx):
if i+j >= len(lines) or lines[i+j].strip() != ctx_line:
match = False
break
if match:
best_s = i
break
if best_s == -1: return (-1, -1)
# Search for end_ctx after start_ctx
best_e = -1
for i in range(best_s, len(lines)):
match = True
for j, ctx_line in enumerate(end_ctx):
# end_ctx is the LAST 3 lines. So we match backwards from i.
idx = i - (len(end_ctx) - 1) + j
if idx < 0 or idx >= len(lines) or lines[idx].strip() != ctx_line:
match = False
break
if match:
best_e = i + 1
break
if best_e != -1:
return (best_s + 1, best_e)
return (-1, -1)