forge: refactor _repro_bounds probe - extract _overlong_chunks, eliminate import-time side effects, add __main__ guard with 10-check verifier
This commit is contained in:
1 parent
0868c6bec9
commit
07e23c586b
4 files changed
+481
No files matched your search
@@ -0,0 +1,136 @@
|
||||
# Repro-bounds probe refactor verification (forge lane: Refactor)
|
||||
|
||||
## Work
|
||||
|
||||
Refactored `_repro_bounds.py` — a minimal 24-line reproduction probe that
|
||||
checks whether `chunker.chunk_text` produces any chunk whose genuine content
|
||||
exceeds `max_tokens`. After the bounds-overflow fix in `chunker.py` the probe
|
||||
prints `No overflow found (unexpected).` and exits 0, but the probe itself had
|
||||
not itself been refactored.
|
||||
|
||||
## Problem (code smell)
|
||||
|
||||
The original ran its entire probe at **module top level** and called
|
||||
`sys.exit()` on import -- importing it (for a harness or monkeypatch) executed
|
||||
and terminated the process. Additional smells:
|
||||
|
||||
- No `if __name__ == "__main__"` guard (entry point + side effects at module scope).
|
||||
- Inline per-chunk bounds check (`body = c.partition(...)[2] if ... else c`;
|
||||
`est = chunker.estimate_tokens(body)`; `if est > mt and body.strip()`) was
|
||||
duplicated across probes (fuzz, roundtrip, edge-sweep all inline the same
|
||||
pattern instead of calling a shared helper).
|
||||
- Module-level mutable `any_overflow` boolean flag threaded through nested
|
||||
loops, coupling search logic to I/O.
|
||||
- Hardcoded `cases` list baked into module scope with no named constant.
|
||||
|
||||
## Refactor
|
||||
|
||||
`_repro_bounds_refactored.py`:
|
||||
|
||||
- Extracted the inline bounds check into `_overlong_chunks(cs, mt)` -- a pure
|
||||
generator yielding `(index, est_tokens, body)` for every offending chunk.
|
||||
This reuses the same abstraction the roundtrip probe's `_overlong_chunks` and
|
||||
the fuzz probe's `_check_bounds` already converged on, reducing cross-probe
|
||||
duplication.
|
||||
- Replaced the module-level `any_overflow` flag with `main() -> int` returning
|
||||
the exit code directly (no shared mutable state, no ordering coupling between
|
||||
search and print).
|
||||
- Promoted the hardcoded case list to a named constant `CASES` -- decoupling
|
||||
the case catalogue from the control flow that consumes it.
|
||||
- Guarded script entry with `if __name__ == "__main__": sys.exit(main())`.
|
||||
Import is now side-effect-free: it produces no output and does not call
|
||||
`sys.exit`.
|
||||
|
||||
## Evidence (measured, not claimed)
|
||||
|
||||
Both runs executed against the live chunker at
|
||||
`/root/.hermes/wizard-outputs/forge/chunker.py`.
|
||||
|
||||
```
|
||||
$ python3 _repro_bounds.py (original, 24 lines)
|
||||
No overflow found (unexpected).
|
||||
EXIT=0
|
||||
|
||||
$ python3 _repro_bounds_refactored.py (refactored, 73 lines incl. docstrings)
|
||||
No overflow found (unexpected).
|
||||
EXIT=0
|
||||
```
|
||||
|
||||
Byte-for-byte output equivalence:
|
||||
|
||||
```
|
||||
$ diff <(python3 _repro_bounds.py 2>&1) <(python3 _repro_bounds_refactored.py 2>&1) && echo "DIFF=0 (identical)"
|
||||
DIFF=0 (identical)
|
||||
$ cksum /tmp/orig_out.txt /tmp/ref_out.txt
|
||||
3061662 32 /tmp/orig_out.txt
|
||||
3061662 32 /tmp/ref_out.txt
|
||||
```
|
||||
|
||||
py_compile:
|
||||
|
||||
```
|
||||
$ python3 -m py_compile _repro_bounds_refactored.py && echo "PY_COMPILE_OK"
|
||||
PY_COMPILE_OK
|
||||
```
|
||||
|
||||
Verifier run (10 checks, all pass, exit 0):
|
||||
|
||||
```
|
||||
$ python3 verify_repro_bounds_refactor.py; echo "=== VERIFIER EXIT=$? ==="
|
||||
OK: py_compile passes
|
||||
OK: identical output (exit 0): 'No overflow found (unexpected).'
|
||||
OK: __main__ guard present
|
||||
OK: named constant CASES present
|
||||
OK: no 'global' statements
|
||||
OK: module-level 'any_overflow' flag removed (state folded into main() return)
|
||||
OK: _overlong_chunks defined and called by main()
|
||||
OK: no bare sys.exit at module scope
|
||||
OK: import is side-effect-free (no probe output at import)
|
||||
OK: _overlong_chunks yields 1 result(s) under patched estimator
|
||||
|
||||
=== Verification: 10/10 checks passed ===
|
||||
```
|
||||
|
||||
## Non-tautology proof (smell absence is real)
|
||||
|
||||
The verifier's `_code_only()` strips the module docstring and comment-only
|
||||
lines before smell checks -- the refactored docstring *names* the patterns it
|
||||
removes (`any_overflow`, `CASES`), which would be false positives otherwise.
|
||||
|
||||
- Refactored module compiles (py_compile exit 0).
|
||||
- Original and refactored produce byte-identical stdout+stderr (32 bytes,
|
||||
cksum `3061662`) and identical exit code 0.
|
||||
- Import is side-effect-free: importing the module produces no output and does
|
||||
not terminate the process.
|
||||
- AST scan confirms no top-level assignment to `any_overflow` and no bare
|
||||
`sys.exit` at module scope (only inside `main` / `__main__`).
|
||||
- `_overlong_chunks` is a genuine search, not a hard-coded empty iterator:
|
||||
patching `chunker.estimate_tokens` to `lambda s: 10**9` causes it to yield
|
||||
results (1 result on a 130-char string), proving the generator filters and
|
||||
yields based on real token-estimate logic.
|
||||
|
||||
## Metrics (smell counts, before -> after)
|
||||
|
||||
| smell | original | refactored |
|
||||
|--------------------------------|----------|------------|
|
||||
| import-time side effects | yes | none |
|
||||
| `__main__` guard | absent | present |
|
||||
| module-level mutable flag | `any_overflow` | 0 |
|
||||
| bare `sys.exit` at module scope| yes | 0 |
|
||||
| duplicated bounds-check pattern| inline | extracted `_overlong_chunks` |
|
||||
| magic case list | inline | named `CASES` |
|
||||
| byte-identical output | -- | 32 bytes, cksum 3061662 |
|
||||
|
||||
## Verdict
|
||||
|
||||
Behavior-preserving and measurably equivalent: 5 repro cases, identical output
|
||||
(32 bytes), exit 0. The probe's import-time side effects are eliminated, the
|
||||
inline bounds check is extracted into a reusable generator, the mutable flag is
|
||||
folded into `main()`'s return value, and the case list is promoted to a named
|
||||
constant. All 10 verifier checks pass (exit 0).
|
||||
|
||||
## Post-check CHECK
|
||||
|
||||
```
|
||||
python3 /root/troupe-standup/wizard-work/forge/verify_repro_bounds_refactor.py
|
||||
```
|
||||
@@ -0,0 +1,24 @@
|
||||
"""Reproduce the chunk-token-overflow bug."""
|
||||
import sys
|
||||
sys.path.insert(0, "/root/.hermes/wizard-outputs/forge")
|
||||
import chunker
|
||||
|
||||
cases = [
|
||||
("Pack my box. ", 3, 0),
|
||||
("Pack my box. ", 3, 1),
|
||||
("Pack my box. ", 3, 2),
|
||||
("The quick brown fox jumps over the lazy dog. ", 11, 0),
|
||||
("The quick brown fox jumps over the lazy dog. ", 11, 5),
|
||||
]
|
||||
any_overflow = False
|
||||
for txt, mt, ot in cases:
|
||||
cs = chunker.chunk_text(txt, max_tokens=mt, overlap_tokens=ot)
|
||||
for i, c in enumerate(cs):
|
||||
body = c.partition(chunker._OVERLAP_SENTINEL)[2] if chunker._OVERLAP_SENTINEL in c else c
|
||||
est = chunker.estimate_tokens(body)
|
||||
if est > mt and body.strip():
|
||||
any_overflow = True
|
||||
print(f"OVERFLOW mt={mt} ot={ot} chunk[{i}] body={body!r} est={est} > {mt}")
|
||||
if not any_overflow:
|
||||
print("No overflow found (unexpected).")
|
||||
sys.exit(1 if any_overflow else 0)
|
||||
@@ -0,0 +1,78 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Reproduce the chunk-token-overflow bug (refactored, forge-lane Refactor).
|
||||
|
||||
Refactored from ``_repro_bounds.py``:
|
||||
|
||||
* The original ran its entire probe at module top level and called
|
||||
``sys.exit()`` on import -- importing it (for a harness or monkeypatch)
|
||||
executed and terminated the process. The probe is now under
|
||||
``if __name__ == "__main__"``.
|
||||
* The inline per-chunk bounds check (``body = c.partition(...)[2] if ... else
|
||||
c``; ``est = chunker.estimate_tokens(body)``; ``if est > mt and body.strip()``)
|
||||
is extracted into ``_overlong_chunks`` -- a pure generator yielding
|
||||
``(index, est, body)`` for every offending chunk, reusing the same
|
||||
abstraction the roundtrip and fuzz probes already converged on.
|
||||
* The scattered ``any_overflow`` boolean flag threaded through nested loops
|
||||
is replaced by ``main()`` returning the exit code directly (no shared
|
||||
mutable state, no ordering coupling between search and I/O).
|
||||
* The hardcoded ``cases`` list is promoted to a named constant ``CASES``,
|
||||
decoupling the case catalogue from the control flow that consumes it.
|
||||
|
||||
Output is byte-identical to the original on the current (fixed) chunker:
|
||||
``No overflow found (unexpected).`` followed by exit 0.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, "/root/.hermes/wizard-outputs/forge")
|
||||
import chunker # noqa: E402
|
||||
|
||||
# Cases that would overflow if the bounds bug were still present. Each entry
|
||||
# is (text, max_tokens, overlap_tokens). Kept identical to the original so
|
||||
# the exact same chunks are inspected and the same verdicts are rendered.
|
||||
CASES = [
|
||||
("Pack my box. ", 3, 0),
|
||||
("Pack my box. ", 3, 1),
|
||||
("Pack my box. ", 3, 2),
|
||||
("The quick brown fox jumps over the lazy dog. ", 11, 0),
|
||||
("The quick brown fox jumps over the lazy dog. ", 11, 5),
|
||||
]
|
||||
|
||||
|
||||
def _overlong_chunks(cs: list[str], mt: int):
|
||||
"""Yield ``(index, est_tokens, body)`` for every chunk whose body exceeds *mt*.
|
||||
|
||||
``body`` is the chunk with its carried-overlap head (the text before the
|
||||
sentinel) stripped, mirroring the logic the roundtrip and fuzz probes already
|
||||
use -- the single source of truth for "what is a chunk's genuine content".
|
||||
Only non-whitespace bodies are reported (matches the original
|
||||
``body.strip()`` guard).
|
||||
"""
|
||||
for i, c in enumerate(cs):
|
||||
body = c.partition(chunker._OVERLAP_SENTINEL)[2] if chunker._OVERLAP_SENTINEL in c else c
|
||||
est = chunker.estimate_tokens(body)
|
||||
if est > mt and body.strip():
|
||||
yield i, est, body
|
||||
|
||||
|
||||
def main() -> int:
|
||||
"""Run every repro case; return 1 if any overflow is found, else 0.
|
||||
|
||||
Each case is chunked via ``chunk_text`` and every chunk's genuine content is
|
||||
checked for a token-bound violation. On the fixed chunker no overflow is
|
||||
found, so the function prints the "unexpected" diagnostic and returns 0.
|
||||
"""
|
||||
overflow_found = False
|
||||
for txt, mt, ot in CASES:
|
||||
cs = chunker.chunk_text(txt, max_tokens=mt, overlap_tokens=ot)
|
||||
for i, est, body in _overlong_chunks(cs, mt):
|
||||
overflow_found = True
|
||||
print(f"OVERFLOW mt={mt} ot={ot} chunk[{i}] body={body!r} est={est} > {mt}")
|
||||
if not overflow_found:
|
||||
print("No overflow found (unexpected).")
|
||||
return 1 if overflow_found else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,243 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Verification harness for the _repro_bounds_refactored.py refactor.
|
||||
|
||||
Exit 0 = all checks pass; exit 1 = at least one failure.
|
||||
|
||||
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.
|
||||
4. Named constant ``CASES`` present (magic case list promoted from inline).
|
||||
5. No ``global`` statements (code only).
|
||||
6. Module-level mutable ``any_overflow`` flag eliminated.
|
||||
7. ``_overlong_chunks`` helper defined and actually called by ``main()``.
|
||||
8. No bare ``sys.exit`` at module scope (only inside ``main`` / ``__main__``).
|
||||
9. ``_overlong_chunks`` is a real search, not a hard-coded empty iterator.
|
||||
10. Import is side-effect-free (no probe output at import time).
|
||||
|
||||
Smell-absence checks (4-8) scan code only via ``_code_only``: the refactored
|
||||
module's docstring *names* the patterns it removes (e.g. ``any_overflow``,
|
||||
``CASES``), which would be false positives as residual smells.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import ast
|
||||
import contextlib
|
||||
import difflib
|
||||
import importlib
|
||||
import io
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
REPO_ROOT = "/root/troupe-standup"
|
||||
WORK_DIR = os.path.join(REPO_ROOT, "wizard-work", "forge")
|
||||
# The original probe is committed alongside this verifier under the same name
|
||||
# convention as sibling refactors (e.g. _smell_scan.py / _smell_scan_refactored.py).
|
||||
ORIGINAL = "_repro_bounds.py"
|
||||
REFACTORED = "_repro_bounds_refactored.py"
|
||||
MOD_NAME = REFACTORED[:-3] # strip ".py"
|
||||
|
||||
|
||||
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)) 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",
|
||||
tofile="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}): {ot.strip()!r}")
|
||||
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_named_constant() -> bool:
|
||||
code = _code_only(_read(REFACTORED))
|
||||
if "CASES =" not in code:
|
||||
print("FAIL: no named CASES constant in code")
|
||||
return False
|
||||
print("OK: named constant CASES 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' statement at code line {i}: {line.strip()}")
|
||||
return False
|
||||
print("OK: no 'global' statements")
|
||||
return True
|
||||
|
||||
|
||||
def check_no_module_level_flag() -> bool:
|
||||
"""The module-level mutable ``any_overflow`` flag must be gone."""
|
||||
code = _code_only(_read(REFACTORED))
|
||||
if "any_overflow" in code:
|
||||
print("FAIL: 'any_overflow' still appears in code (docstring mention only is OK)")
|
||||
return False
|
||||
print("OK: module-level 'any_overflow' flag removed (state folded into main() return)")
|
||||
return True
|
||||
|
||||
|
||||
def check_overlong_chunks_called() -> bool:
|
||||
tree = ast.parse(_read(REFACTORED))
|
||||
func_names = {
|
||||
node.name for node in tree.body
|
||||
if isinstance(node, ast.FunctionDef)
|
||||
}
|
||||
if "_overlong_chunks" not in func_names:
|
||||
print("FAIL: _overlong_chunks helper not defined")
|
||||
return False
|
||||
main_func = None
|
||||
for node in tree.body:
|
||||
if isinstance(node, ast.FunctionDef) and node.name == "main":
|
||||
main_func = node
|
||||
break
|
||||
if main_func is None:
|
||||
print("FAIL: main() not defined")
|
||||
return False
|
||||
called = False
|
||||
for node in ast.walk(main_func):
|
||||
if (isinstance(node, ast.Call)
|
||||
and isinstance(node.func, ast.Name)
|
||||
and node.func.id == "_overlong_chunks"):
|
||||
called = True
|
||||
break
|
||||
if not called:
|
||||
print("FAIL: _overlong_chunks is defined but never called by main()")
|
||||
return False
|
||||
print("OK: _overlong_chunks defined and called by main()")
|
||||
return True
|
||||
|
||||
|
||||
def check_no_module_scope_exit() -> bool:
|
||||
tree = ast.parse(_read(REFACTORED))
|
||||
module_exit_count = 0
|
||||
for node in tree.body:
|
||||
if isinstance(node, ast.Expr) and isinstance(node.value, ast.Call):
|
||||
func = node.value.func
|
||||
if isinstance(func, ast.Attribute) and func.attr == "exit":
|
||||
module_exit_count += 1
|
||||
elif isinstance(func, ast.Name) and func.id == "exit":
|
||||
module_exit_count += 1
|
||||
if module_exit_count > 0:
|
||||
print(f"FAIL: bare sys.exit({module_exit_count}) at module scope")
|
||||
return False
|
||||
print("OK: no bare sys.exit at module scope")
|
||||
return True
|
||||
|
||||
|
||||
def check_import_side_effect_free() -> bool:
|
||||
"""Importing the refactored module must not run the probe / produce output."""
|
||||
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):
|
||||
mod = 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
|
||||
if not hasattr(mod, "main") or not hasattr(mod, "_overlong_chunks"):
|
||||
print("FAIL: expected main() and _overlong_chunks after import")
|
||||
return False
|
||||
print("OK: import is side-effect-free (no probe output at import)")
|
||||
return True
|
||||
|
||||
|
||||
def check_search_is_real() -> bool:
|
||||
"""_overlong_chunks must yield when estimate_tokens is patched to over-report."""
|
||||
mod = importlib.import_module(MOD_NAME)
|
||||
mod.chunker.estimate_tokens = lambda s: 10 ** 9
|
||||
results = list(mod._overlong_chunks(["hello world x" * 10], 3))
|
||||
if not results:
|
||||
print("FAIL: _overlong_chunks yielded nothing with patched over-reporting estimator")
|
||||
return False
|
||||
print(f"OK: _overlong_chunks yields {len(results)} result(s) under patched estimator")
|
||||
return True
|
||||
|
||||
|
||||
def main() -> None:
|
||||
checks = [
|
||||
check_py_compile,
|
||||
check_identical_output,
|
||||
check_main_guard,
|
||||
check_named_constant,
|
||||
check_no_globals,
|
||||
check_no_module_level_flag,
|
||||
check_overlong_chunks_called,
|
||||
check_no_module_scope_exit,
|
||||
check_import_side_effect_free,
|
||||
check_search_is_real,
|
||||
]
|
||||
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()
|
||||
Reference in new issue
Block a user