forge: verify _probe_forge chunker round-trip refactor - 9/9 checks, byte-identical output

wizard-work/forge/_probe_forge.py

@@ -0,0 +1,46 @@
+"""Probe the forge (refactored) chunker round-trip."""
+import sys
+sys.path.insert(0, "/root/.hermes/wizard-outputs/forge")
+import chunker
+
+base = "The quick brown fox jumps over the lazy dog. Pack my box with five dozen liquor jugs. How vexingly quick daft zebras jump. "
+text = (base + "Variant sentence here. ") * 20
+text_stripped = text.strip()
+
+cs = chunker.chunk_text(text, max_tokens=64, overlap_tokens=8)
+r = chunker.reconstruct(cs, 8)
+
+print("orig_len:", len(text_stripped))
+print("recon_len:", len(r))
+print("equal:", r == text_stripped)
+print("nchunks:", len(cs))
+print()
+
+# Show first 5 chunks with sentinel revealed
+S = chunker._OVERLAP_SENTINEL
+for i, c in enumerate(cs[:5]):
+    has = S in c
+    print(f"chunk[{i}] len={len(c)} sentinel={'YES' if has else 'no'}")
+    if has:
+        head, _, rest = c.partition(S)
+        print(f"   head(t)={len(head)}: {head[:20]!r}")
+        print(f"   rest   : {rest[:40]!r}")
+        print(f"   rest.startswith(head): {rest.startswith(head)}")
+    else:
+        print(f"   full   : {c[:40]!r}")
+
+# Find first diff
+for i in range(min(len(r), len(text_stripped))):
+    if r[i] != text_stripped[i]:
+        print()
+        print("FIRST DIFF at char", i)
+        print("  orig:", repr(text_stripped[max(0, i-20):i+20]))
+        print("  recon:", repr(r[max(0, i-20):i+20]))
+        break
+else:
+    shorter = min(len(r), len(text_stripped))
+    print("no char diff in common prefix")
+    if len(r) != len(text_stripped):
+        print("lengths differ:", len(r), len(text_stripped))
+        print("orig tail:", repr(text_stripped[-30:]))
+        print("recon tail:", repr(r[-30:]))

wizard-work/forge/_probe_forge_refactored.py

@@ -0,0 +1,117 @@
+#!/usr/bin/env python3
+"""Probe the forge (refactored) chunker round-trip.
+
+Refactored from ``_probe_forge.py`` (lane: Refactor):
+
+  * The original ran its entire probe at module top-level with no
+    ``if __name__ == "__main__"`` guard -- importing the module executed and
+    printed the probe output.  All execution is now under ``main()``; the module
+    imports cleanly with zero side effects and the entry point is explicitly
+    guarded.
+
+  * The dead variable ``shorter`` (``min(len(r), len(text_stripped))``) was
+    computed in the for-else ``else`` branch but never read: every subsequent
+    statement in that branch uses ``len(r)`` and ``len(text_stripped)``
+    directly.  Removed; output is unchanged.
+
+  * The inline first-diff search (``for i in range(min(...))``) is extracted
+    into ``_find_first_diff(actual, expected)``, a pure function returning the
+    mismatch index or ``None`` -- decoupling the search from I/O so the
+    control flow reads as intent ("is there a diff, and if so, show it")
+    rather than as a for-else side-effects dance.
+
+  * The chunk-structure printing loop is extracted into
+    ``_print_chunk_structure(chunks, sentinel, n)`` so the formatting concern
+    is isolated and the data (``base``, ``text``, ``cs``, ``r``) lives inside
+    ``main()`` instead of at module scope.
+
+Output is byte-identical to the original.
+"""
+from __future__ import annotations
+
+import sys
+
+sys.path.insert(0, "/root/.hermes/wizard-outputs/forge")
+import chunker  # noqa: E402
+
+
+def _find_first_diff(actual: str, expected: str) -> int | None:
+    """Return the index of the first character mismatch between *actual* and
+    *expected*, scanning only their common prefix (``min`` length).
+
+    Returns ``None`` when the common prefix matches entirely -- i.e. either the
+    strings are equal or *actual* is a strict prefix of *expected* (or vice
+    versa).  This mirrors the original ``for ... break else None`` semantics:
+    the loop body breaks on the first mismatch; the ``else`` fires when no
+    mismatch is found.
+    """
+    for i in range(min(len(actual), len(expected))):
+        if actual[i] != expected[i]:
+            return i
+    return None
+
+
+def _print_chunk_structure(
+    chunks: list[str], sentinel: str, n: int = 5
+) -> None:
+    """Print the first *n* chunks, revealing the sentinel-delimited overlap head.
+
+    For each chunk the overlap head (text before *sentinel*) and the genuine
+    rest (text after *sentinel*) are shown so a reader can see whether
+    ``rest`` happens to begin with ``head`` (which would mean a
+    prefix-based discriminator could conflate genuine and context overlap).
+    """
+    for i, c in enumerate(chunks[:n]):
+        has = sentinel in c
+        print(f"chunk[{i}] len={len(c)} sentinel={'YES' if has else 'no'}")
+        if has:
+            head, _, rest = c.partition(sentinel)
+            print(f"   head(t)={len(head)}: {head[:20]!r}")
+            print(f"   rest   : {rest[:40]!r}")
+            print(f"   rest.startswith(head): {rest.startswith(head)}")
+        else:
+            print(f"   full   : {c[:40]!r}")
+
+
+def main() -> int:
+    """Run the round-trip probe and print its diagnostic output.
+
+    The probe chunks a repetitive text, reconstructs it, and reports whether
+    the reconstruction matches the original -- including chunk-level structure
+    and the first character-level difference (if any).
+    """
+    base = ("The quick brown fox jumps over the lazy dog. "
+            "Pack my box with five dozen liquor jugs. "
+            "How vexingly quick daft zebras jump. ")
+    text = (base + "Variant sentence here. ") * 20
+    expected = text.strip()
+    sentinel = chunker._OVERLAP_SENTINEL
+
+    cs = chunker.chunk_text(text, max_tokens=64, overlap_tokens=8)
+    actual = chunker.reconstruct(cs, 8)
+
+    print("orig_len:", len(expected))
+    print("recon_len:", len(actual))
+    print("equal:", actual == expected)
+    print("nchunks:", len(cs))
+    print()
+
+    _print_chunk_structure(cs, sentinel, n=5)
+
+    diff = _find_first_diff(actual, expected)
+    if diff is not None:
+        print()
+        print("FIRST DIFF at char", diff)
+        print("  orig:", repr(expected[max(0, diff - 20):diff + 20]))
+        print("  recon:", repr(actual[max(0, diff - 20):diff + 20]))
+    else:
+        print("no char diff in common prefix")
+        if len(actual) != len(expected):
+            print("lengths differ:", len(actual), len(expected))
+            print("orig tail:", repr(expected[-30:]))
+            print("recon tail:", repr(actual[-30:]))
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

wizard-work/forge/verify_probe_forge_refactor.py

@@ -0,0 +1,259 @@
+#!/usr/bin/env python3
+"""Verification harness for the _probe_forge_refactored.py refactor.
+
+Exit 0 = all checks pass; exit 1 = at least one failure.
+
+The target ``_probe_forge.py`` is a round-trip probe of the forge chunker. The
+refactor is *behavior-preserving* on the probe itself (it exercises the same
+chunker API with the same inputs). Checks:
+
+  1.  Refactored file compiles (py_compile).
+  2.  Original and refactored produce byte-identical stdout+stderr AND exit code.
+  3.  ``if __name__ == "__main__"`` guard present (no probe at import time).
+  4.  No ``global`` statements in code.
+  5.  Dead ``shorter`` variable removed (``shorter = min(`` absent from code).
+  6.  Pure helpers ``_find_first_diff`` and ``_print_chunk_structure`` extracted as
+      module-level functions, and ``main`` calls both.
+  7.  Import of the refactored module is side-effect-free (no stdout/stderr).
+  8.  Round-trip assertion holds: probe output reports ``equal: True``.
+  9.  Non-vacuity: the *original* ``_probe_forge.py`` still exhibits the removed
+      smells (no __main__ guard, has ``shorter = min(``, module-level scan/for loop).
+"""
+from __future__ import annotations
+
+import ast
+import contextlib
+import difflib
+import importlib
+import io
+import os
+import subprocess
+import sys
+
+# Resolve paths from this file's location (wizard-work/forge/ -> repo root).
+WORK_DIR = os.path.dirname(os.path.abspath(__file__))
+ORIGINAL = "_probe_forge.py"
+REFACTORED = "_probe_forge_refactored.py"
+MOD_NAME = REFACTORED[:-3]
+
+
+def _run(*args: str) -> subprocess.CompletedProcess:
+    return subprocess.run(args, cwd=WORK_DIR, capture_output=True, text=True)
+
+
+def _read(name: str) -> str:
+    with open(os.path.join(WORK_DIR, name), encoding="utf-8") as f:
+        return f.read()
+
+
+def _code_only(content: str) -> str:
+    """Strip the module docstring and comment-only lines.
+
+    Smell-absence checks must inspect code, not documentation: the refactored
+    module's docstring *names* the patterns it removes, which would be false
+    positives as residual smells.
+    """
+    tree = ast.parse(content)
+    skip: set[int] = set()
+    if (tree.body
+            and isinstance(tree.body[0], ast.Expr)
+            and isinstance(tree.body[0].value, ast.Constant)
+            and isinstance(tree.body[0].value.value, str)):
+        ds = tree.body[0]
+        for ln in range(ds.lineno, (ds.end_lineno or ds.lineno) + 1):
+            skip.add(ln)
+    kept = []
+    for i, line in enumerate(content.splitlines(), 1):
+        if i in skip or line.lstrip().startswith("#"):
+            continue
+        kept.append(line)
+    return "\n".join(kept)
+
+
+def check_py_compile() -> bool:
+    r = _run("python3", "-m", "py_compile", REFACTORED)
+    if r.returncode != 0:
+        print(f"FAIL: py_compile failed: {r.stderr}")
+        return False
+    print("OK: py_compile passes")
+    return True
+
+
+def check_identical_output() -> bool:
+    orig = _run("python3", ORIGINAL)
+    ref = _run("python3", REFACTORED)
+    ot = orig.stdout + orig.stderr
+    rt = ref.stdout + ref.stderr
+    if ot != rt:
+        print("FAIL: original and refactored output differ")
+        print("".join(difflib.unified_diff(
+            ot.splitlines(keepends=True), rt.splitlines(keepends=True),
+            fromfile="original (" + ORIGINAL + ")",
+            tofile="refactored (" + REFACTORED + ")",
+        )))
+        return False
+    if orig.returncode != ref.returncode:
+        print(f"FAIL: exit codes differ ({orig.returncode} vs {ref.returncode})")
+        return False
+    print(f"OK: identical output (exit {orig.returncode}, {len(ot)} bytes)")
+    return True
+
+
+def check_main_guard() -> bool:
+    content = _read(REFACTORED)
+    if '__name__ == "__main__"' not in content:
+        print("FAIL: no __main__ guard")
+        return False
+    print("OK: __main__ guard present")
+    return True
+
+
+def check_no_globals() -> bool:
+    code = _code_only(_read(REFACTORED))
+    for i, line in enumerate(code.splitlines(), 1):
+        if line.lstrip().startswith("global "):
+            print(f"FAIL: 'global' found at code line {i}: {line.strip()}")
+            return False
+    print("OK: no 'global' statements in code")
+    return True
+
+
+def check_dead_shorter_removed() -> bool:
+    code = _code_only(_read(REFACTORED))
+    if "shorter =" in code or "shorter=" in code:
+        print("FAIL: dead 'shorter' variable still present in code")
+        return False
+    print("OK: dead 'shorter' variable removed")
+    return True
+
+
+def check_helpers_extracted() -> bool:
+    tree = ast.parse(_read(REFACTORED))
+    mod_funcs = {
+        n.name for n in tree.body
+        if isinstance(n, ast.FunctionDef)
+    }
+    for required in ("_find_first_diff", "_print_chunk_structure", "main"):
+        if required not in mod_funcs:
+            print(f"FAIL: {required} not defined as module-level function")
+            return False
+    main_func = next(n for n in tree.body
+                     if isinstance(n, ast.FunctionDef) and n.name == "main")
+    main_calls = {
+        n.func.id for n in ast.walk(main_func)
+        if isinstance(n, ast.Call) and isinstance(n.func, ast.Name)
+    }
+    for called in ("_find_first_diff", "_print_chunk_structure",
+                   "chunker.chunk_text", "chunker.reconstruct", "print"):
+        if called not in main_calls and called not in {
+            # chunk_text/reconstruct are called via the chunker attribute
+            *(a.attr for n in ast.walk(main_func)
+              if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute)
+              for a in [n.func])}:
+            # check attribute-style calls too
+            pass
+    # Verify main() calls the helpers and chunker API (attribute-style).
+    attr_calls = {
+        n.func.attr for n in ast.walk(main_func)
+        if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute)
+        and isinstance(n.func.value, ast.Name)
+        and n.func.value.id == "chunker"
+    }
+    for needed in ("chunk_text", "reconstruct"):
+        if needed not in attr_calls:
+            print(f"FAIL: main() does not call chunker.{needed}()")
+            return False
+    if "_find_first_diff" not in main_calls and "_find_first_diff" not in attr_calls:
+        # _find_first_diff may be called as a Name
+        name_calls = {
+            n.func.id for n in ast.walk(main_func)
+            if isinstance(n, ast.Call) and isinstance(n.func, ast.Name)
+        }
+        if "_find_first_diff" not in name_calls:
+            print("FAIL: main() does not call _find_first_diff()")
+            return False
+    if "_print_chunk_structure" not in main_calls and "_print_chunk_structure" not in attr_calls:
+        name_calls = {
+            n.func.id for n in ast.walk(main_func)
+            if isinstance(n, ast.Call) and isinstance(n.func, ast.Name)
+        }
+        if "_print_chunk_structure" not in name_calls:
+            print("FAIL: main() does not call _print_chunk_structure()")
+            return False
+    print("OK: _find_first_diff, _print_chunk_structure, main extracted; main calls helpers + chunker API")
+    return True
+
+
+def check_import_side_effect_free() -> bool:
+    """Importing the refactored module must not run the probe or print."""
+    sys.modules.pop(MOD_NAME, None)
+    if WORK_DIR not in sys.path:
+        sys.path.insert(0, WORK_DIR)
+    out, err = io.StringIO(), io.StringIO()
+    with contextlib.redirect_stdout(out), contextlib.redirect_stderr(err):
+        importlib.import_module(MOD_NAME)
+    if out.getvalue() or err.getvalue():
+        print(f"FAIL: import produced output: stdout={out.getvalue()!r} stderr={err.getvalue()!r}")
+        return False
+    print("OK: import is side-effect-free (no probe output at import)")
+    return True
+
+
+def check_roundtrip_lossless() -> bool:
+    """The probe's core assertion: reconstruct == original (equal: True)."""
+    r = _run("python3", REFACTORED)
+    combined = r.stdout + r.stderr
+    if "equal: True" not in combined:
+        print("FAIL: probe output does not report 'equal: True'; round-trip not lossless")
+        return False
+    print("OK: round-trip lossless (equal: True in probe output)")
+    return True
+
+
+def check_original_still_smells() -> bool:
+    """The original must still carry the smells we removed (non-vacuous)."""
+    src = _read(ORIGINAL)
+    code = _code_only(src)
+    problems = []
+    # The original must LACK a __main__ guard (that is the smell we fixed).
+    if '__name__ == "__main__"' in src:
+        problems.append("original already has __main__ guard (refactor not needed)")
+    # The dead 'shorter' variable must still be present in the original.
+    if "shorter =" not in code and "shorter=" not in code:
+        problems.append("dead 'shorter' variable missing from original (expected)")
+    # The original must execute the probe at import time: module-level for-loop.
+    tree = ast.parse(src)
+    has_module_level_for = any(isinstance(n, ast.For) for n in tree.body)
+    if not has_module_level_for:
+        problems.append("no module-level for loop in original (expected)")
+    if problems:
+        print("FAIL: original does not exhibit the expected smells:")
+        for p in problems:
+            print(f"  - {p}")
+        return False
+    print("OK: original still exhibits the removed smells (refactor is non-vacuous)")
+    return True
+
+
+def main() -> None:
+    checks = [
+        check_py_compile,
+        check_identical_output,
+        check_main_guard,
+        check_no_globals,
+        check_dead_shorter_removed,
+        check_helpers_extracted,
+        check_import_side_effect_free,
+        check_roundtrip_lossless,
+        check_original_still_smells,
+    ]
+    results = [fn() for fn in checks]
+    passed = sum(results)
+    total = len(results)
+    print(f"\n=== Verification: {passed}/{total} checks passed ===")
+    if passed != total:
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()