bisect.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """Bisect engine — binary search through commit history to locate regressions. |
| 2 | |
| 3 | ``muse bisect`` is a power-tool for both human engineers and autonomous agents. |
| 4 | Given a known-bad commit and a known-good commit, it performs a binary search |
| 5 | through the commits between them, asking at each midpoint "is this good or |
| 6 | bad?" until the first bad commit is isolated. |
| 7 | |
| 8 | Agent-safety |
| 9 | ------------ |
| 10 | The ``run`` subcommand accepts an arbitrary shell command. The command is |
| 11 | executed in a subprocess; the exit code determines the verdict: |
| 12 | |
| 13 | 0 → good (the bug is NOT present) |
| 14 | 125 → skip (cannot test this commit — e.g. build fails) |
| 15 | any other → bad (the bug IS present) |
| 16 | |
| 17 | This mirrors Git's bisect protocol so any existing test harness works without |
| 18 | modification. |
| 19 | |
| 20 | Symbol-scoped bisect |
| 21 | -------------------- |
| 22 | Pass ``symbol_filter`` to ``start_bisect`` to restrict the candidate commit |
| 23 | list to only those commits whose ``structured_delta`` contains an op for the |
| 24 | given symbol address (or any symbol whose address starts with that prefix, |
| 25 | enabling class-level filtering like ``billing.py::Invoice``). |
| 26 | |
| 27 | This typically reduces the search space from hundreds of commits to fewer |
| 28 | than ten — fewer binary-search steps, and each displayed midpoint shows |
| 29 | exactly which ops on that symbol occurred. |
| 30 | |
| 31 | State file |
| 32 | ---------- |
| 33 | Bisect state is stored at ``.muse/BISECT_STATE.toml``:: |
| 34 | |
| 35 | bad_id = "<sha256>" |
| 36 | good_ids = ["<sha256>", …] |
| 37 | skipped_ids = ["<sha256>", …] |
| 38 | remaining = ["<sha256>", …] # sorted oldest-first |
| 39 | log = ["<sha256> <verdict> <ts>", …] |
| 40 | symbol_filter = "billing.py::Invoice.compute_total" # optional |
| 41 | branch = "feat/my-feature" # optional |
| 42 | |
| 43 | The remaining list is rebuilt at every step so it tolerates interruptions. |
| 44 | |
| 45 | Security model |
| 46 | -------------- |
| 47 | - **TOML injection**: All user-controlled string fields written to the state |
| 48 | file (``branch``, ``symbol_filter``) are passed through ``_toml_escape`` |
| 49 | before serialisation. Commit IDs are always SHA-256 hex strings and need |
| 50 | no escaping. |
| 51 | - **Symlink guard**: ``_load_state`` rejects a symlink at the state file path |
| 52 | to prevent path-traversal attacks that could redirect reads or writes to |
| 53 | files outside the repository. |
| 54 | - **Size cap**: ``_load_state`` refuses to read a state file larger than |
| 55 | ``_MAX_STATE_BYTES`` (4 MiB) to prevent memory exhaustion from a tampered |
| 56 | or corrupt state. |
| 57 | """ |
| 58 | |
| 59 | from __future__ import annotations |
| 60 | |
| 61 | import datetime |
| 62 | import logging |
| 63 | import pathlib |
| 64 | import subprocess |
| 65 | from dataclasses import dataclass, field |
| 66 | from typing import TypedDict |
| 67 | |
| 68 | logger = logging.getLogger(__name__) |
| 69 | |
| 70 | _BISECT_STATE_FILE = ".muse/BISECT_STATE.toml" |
| 71 | _SKIP_EXIT_CODE = 125 |
| 72 | |
| 73 | # Guard against pathologically large or tampered state files consuming memory. |
| 74 | _MAX_STATE_BYTES: int = 4 * 1024 * 1024 # 4 MiB |
| 75 | |
| 76 | |
| 77 | # --------------------------------------------------------------------------- |
| 78 | # State persistence |
| 79 | # --------------------------------------------------------------------------- |
| 80 | |
| 81 | |
| 82 | class BisectStateDict(TypedDict, total=False): |
| 83 | """On-disk shape of the bisect state.""" |
| 84 | |
| 85 | bad_id: str |
| 86 | good_ids: list[str] |
| 87 | skipped_ids: list[str] |
| 88 | remaining: list[str] |
| 89 | log: list[str] |
| 90 | branch: str |
| 91 | symbol_filter: str |
| 92 | |
| 93 | |
| 94 | def _state_path(repo_root: pathlib.Path) -> pathlib.Path: |
| 95 | return repo_root / ".muse" / "BISECT_STATE.toml" |
| 96 | |
| 97 | |
| 98 | def _load_state(repo_root: pathlib.Path) -> BisectStateDict | None: |
| 99 | """Return the bisect state if a session is active, else None. |
| 100 | |
| 101 | Safety guards applied before any read: |
| 102 | |
| 103 | - Symlink check: a symlink at the state path could redirect writes to |
| 104 | arbitrary locations or reads to sensitive files outside the repo. |
| 105 | - Size cap (``_MAX_STATE_BYTES``): a tampered or corrupt state file cannot |
| 106 | be used to exhaust memory. |
| 107 | """ |
| 108 | import tomllib |
| 109 | |
| 110 | path = _state_path(repo_root) |
| 111 | if not path.exists(): |
| 112 | return None |
| 113 | # Reject symlinks before any read/write to prevent path-traversal attacks. |
| 114 | if path.is_symlink(): |
| 115 | logger.warning("⚠️ Bisect state file is a symlink — ignoring to prevent path traversal") |
| 116 | return None |
| 117 | try: |
| 118 | size = path.stat().st_size |
| 119 | if size > _MAX_STATE_BYTES: |
| 120 | logger.warning( |
| 121 | "⚠️ Bisect state file is %.1f MiB — exceeds cap of %d MiB; ignoring", |
| 122 | size / (1024 * 1024), |
| 123 | _MAX_STATE_BYTES // (1024 * 1024), |
| 124 | ) |
| 125 | return None |
| 126 | raw = tomllib.loads(path.read_text(encoding="utf-8")) |
| 127 | except Exception as exc: |
| 128 | logger.warning("⚠️ Could not read bisect state: %s", exc) |
| 129 | return None |
| 130 | state: BisectStateDict = {} |
| 131 | if "bad_id" in raw and isinstance(raw["bad_id"], str): |
| 132 | state["bad_id"] = raw["bad_id"] |
| 133 | if "branch" in raw and isinstance(raw["branch"], str): |
| 134 | state["branch"] = raw["branch"] |
| 135 | if "symbol_filter" in raw and isinstance(raw["symbol_filter"], str): |
| 136 | state["symbol_filter"] = raw["symbol_filter"] |
| 137 | for key in ("good_ids", "skipped_ids", "remaining", "log"): |
| 138 | val = raw.get(key) |
| 139 | if isinstance(val, list) and all(isinstance(x, str) for x in val): |
| 140 | str_list: list[str] = list(val) |
| 141 | if key == "good_ids": |
| 142 | state["good_ids"] = str_list |
| 143 | elif key == "skipped_ids": |
| 144 | state["skipped_ids"] = str_list |
| 145 | elif key == "remaining": |
| 146 | state["remaining"] = str_list |
| 147 | elif key == "log": |
| 148 | state["log"] = str_list |
| 149 | return state |
| 150 | |
| 151 | |
| 152 | def _toml_escape(value: str) -> str: |
| 153 | """Escape *value* for safe embedding inside a TOML double-quoted string. |
| 154 | |
| 155 | Only backslash and double-quote need escaping in TOML basic strings. |
| 156 | Applied to all free-form string fields (branch name, symbol_filter) before |
| 157 | serialisation to prevent TOML injection via crafted branch names. |
| 158 | """ |
| 159 | return value.replace("\\", "\\\\").replace('"', '\\"') |
| 160 | |
| 161 | |
| 162 | def _save_state(repo_root: pathlib.Path, state: BisectStateDict) -> None: |
| 163 | """Write the bisect state to disk (TOML, atomic). |
| 164 | |
| 165 | All string values that may contain user-controlled content (branch name, |
| 166 | symbol_filter) are escaped with ``_toml_escape`` before serialisation. |
| 167 | Commit IDs are always 64-char hex strings and need no escaping. |
| 168 | """ |
| 169 | path = _state_path(repo_root) |
| 170 | lines = [f'bad_id = "{state.get("bad_id", "")}"'] |
| 171 | if "branch" in state: |
| 172 | # Branch names can contain backslashes and quotes — escape both. |
| 173 | lines.append(f'branch = "{_toml_escape(state["branch"])}"') |
| 174 | if "symbol_filter" in state and state["symbol_filter"]: |
| 175 | lines.append(f'symbol_filter = "{_toml_escape(state["symbol_filter"])}"') |
| 176 | for key, items in ( |
| 177 | ("good_ids", state.get("good_ids", [])), |
| 178 | ("skipped_ids", state.get("skipped_ids", [])), |
| 179 | ("remaining", state.get("remaining", [])), |
| 180 | ("log", state.get("log", [])), |
| 181 | ): |
| 182 | formatted = ", ".join(f'"{v}"' for v in items) |
| 183 | lines.append(f"{key} = [{formatted}]") |
| 184 | tmp = path.with_suffix(".tmp") |
| 185 | tmp.write_text("\n".join(lines) + "\n", encoding="utf-8") |
| 186 | tmp.replace(path) |
| 187 | |
| 188 | |
| 189 | # --------------------------------------------------------------------------- |
| 190 | # Commit graph helpers |
| 191 | # --------------------------------------------------------------------------- |
| 192 | |
| 193 | |
| 194 | def _ancestors( |
| 195 | repo_root: pathlib.Path, |
| 196 | start_id: str, |
| 197 | ) -> list[str]: |
| 198 | """Return commit IDs from start_id back to root, oldest first.""" |
| 199 | from muse.core.store import read_commit |
| 200 | |
| 201 | visited: list[str] = [] |
| 202 | queue = [start_id] |
| 203 | seen: set[str] = set() |
| 204 | while queue: |
| 205 | cid = queue.pop() |
| 206 | if cid in seen: |
| 207 | continue |
| 208 | seen.add(cid) |
| 209 | commit = read_commit(repo_root, cid) |
| 210 | if commit is None: |
| 211 | continue |
| 212 | visited.append(cid) |
| 213 | if commit.parent_commit_id: |
| 214 | queue.append(commit.parent_commit_id) |
| 215 | if commit.parent2_commit_id: |
| 216 | queue.append(commit.parent2_commit_id) |
| 217 | # Reverse to get oldest-first chronological order. |
| 218 | return list(reversed(visited)) |
| 219 | |
| 220 | |
| 221 | def _reachable_from_good( |
| 222 | repo_root: pathlib.Path, |
| 223 | good_ids: list[str], |
| 224 | ) -> set[str]: |
| 225 | """Return all commits reachable (inclusive) from any good commit.""" |
| 226 | from muse.core.store import read_commit |
| 227 | |
| 228 | reachable: set[str] = set() |
| 229 | queue = list(good_ids) |
| 230 | while queue: |
| 231 | cid = queue.pop() |
| 232 | if cid in reachable: |
| 233 | continue |
| 234 | reachable.add(cid) |
| 235 | commit = read_commit(repo_root, cid) |
| 236 | if commit is None: |
| 237 | continue |
| 238 | if commit.parent_commit_id: |
| 239 | queue.append(commit.parent_commit_id) |
| 240 | if commit.parent2_commit_id: |
| 241 | queue.append(commit.parent2_commit_id) |
| 242 | return reachable |
| 243 | |
| 244 | |
| 245 | def _build_remaining( |
| 246 | repo_root: pathlib.Path, |
| 247 | bad_id: str, |
| 248 | good_ids: list[str], |
| 249 | skipped_ids: list[str], |
| 250 | ) -> list[str]: |
| 251 | """Return commits strictly between good and bad (exclusive of both endpoints). |
| 252 | |
| 253 | Excludes ``bad_id`` itself — it is already the known-bad boundary. |
| 254 | Excludes good-reachable commits (already confirmed good). |
| 255 | Excludes skipped commits. |
| 256 | |
| 257 | When this list has length 0, ``bad_id`` is the first bad commit. |
| 258 | When it has length 1, there is exactly one candidate to test — if it is |
| 259 | bad, it becomes the new ``bad_id`` and remaining collapses to 0; if it is |
| 260 | good, ``bad_id`` is confirmed as first-bad. |
| 261 | """ |
| 262 | ancestors_of_bad = _ancestors(repo_root, bad_id) |
| 263 | good_reachable = _reachable_from_good(repo_root, good_ids) |
| 264 | skipped = set(skipped_ids) |
| 265 | return [ |
| 266 | cid for cid in ancestors_of_bad |
| 267 | if cid not in good_reachable and cid not in skipped and cid != bad_id |
| 268 | ] |
| 269 | |
| 270 | |
| 271 | def _midpoint(remaining: list[str]) -> str | None: |
| 272 | if not remaining: |
| 273 | return None |
| 274 | return remaining[len(remaining) // 2] |
| 275 | |
| 276 | |
| 277 | def _addr_matches(addr: str, symbol_filter: str) -> bool: |
| 278 | """Return True if *addr* is the filtered symbol or a child of it. |
| 279 | |
| 280 | Exact match covers ``file.py::func``. |
| 281 | Prefix match (``addr.startswith(symbol_filter + ".")``) covers class |
| 282 | methods when the filter is ``file.py::ClassName``. |
| 283 | """ |
| 284 | return addr == symbol_filter or addr.startswith(symbol_filter + ".") |
| 285 | |
| 286 | |
| 287 | def _commits_touching_symbol( |
| 288 | repo_root: pathlib.Path, |
| 289 | commit_ids: list[str], |
| 290 | symbol_filter: str, |
| 291 | ) -> list[str]: |
| 292 | """Return the subset of *commit_ids* whose structured_delta touched *symbol_filter*. |
| 293 | |
| 294 | Preserves the original order of *commit_ids*. Commits with no |
| 295 | ``structured_delta`` (genesis commits) are excluded. |
| 296 | """ |
| 297 | from muse.core.store import read_commit |
| 298 | from muse.plugins.code._query import flat_symbol_ops |
| 299 | |
| 300 | result: list[str] = [] |
| 301 | for cid in commit_ids: |
| 302 | commit = read_commit(repo_root, cid) |
| 303 | if commit is None or commit.structured_delta is None: |
| 304 | continue |
| 305 | ops_list = commit.structured_delta.get("ops", []) |
| 306 | for op in flat_symbol_ops(ops_list): |
| 307 | if _addr_matches(op.get("address", ""), symbol_filter): |
| 308 | result.append(cid) |
| 309 | break |
| 310 | return result |
| 311 | |
| 312 | |
| 313 | def _symbol_ops_in_commit( |
| 314 | repo_root: pathlib.Path, |
| 315 | commit_id: str, |
| 316 | symbol_filter: str, |
| 317 | ) -> list[str]: |
| 318 | """Return human-readable descriptions of ops for *symbol_filter* in *commit_id*. |
| 319 | |
| 320 | Used to display what changed in the midpoint commit before the user tests it. |
| 321 | Returns an empty list when the commit has no relevant delta. |
| 322 | """ |
| 323 | from muse.core.store import read_commit |
| 324 | from muse.plugins.code._query import flat_symbol_ops |
| 325 | |
| 326 | commit = read_commit(repo_root, commit_id) |
| 327 | if commit is None or commit.structured_delta is None: |
| 328 | return [] |
| 329 | |
| 330 | descriptions: list[str] = [] |
| 331 | ops_list = commit.structured_delta.get("ops", []) |
| 332 | for op in flat_symbol_ops(ops_list): |
| 333 | addr = op.get("address", "") |
| 334 | if not _addr_matches(addr, symbol_filter): |
| 335 | continue |
| 336 | op_type = op.get("op", "?") |
| 337 | old = op.get("old_summary", "") |
| 338 | new = op.get("new_summary", op.get("content_summary", "")) |
| 339 | if op_type == "replace": |
| 340 | descriptions.append(f" {addr} → {old!r} → {new!r}") |
| 341 | elif op_type == "insert": |
| 342 | descriptions.append(f" {addr} added {new!r}") |
| 343 | elif op_type == "delete": |
| 344 | descriptions.append(f" {addr} removed {old!r}") |
| 345 | else: |
| 346 | descriptions.append(f" {addr} {op_type}") |
| 347 | return descriptions |
| 348 | |
| 349 | |
| 350 | # --------------------------------------------------------------------------- |
| 351 | # Result type |
| 352 | # --------------------------------------------------------------------------- |
| 353 | |
| 354 | |
| 355 | @dataclass |
| 356 | class BisectResult: |
| 357 | """Result of a single bisect step.""" |
| 358 | |
| 359 | done: bool = False |
| 360 | """True when we have isolated the first bad commit.""" |
| 361 | |
| 362 | first_bad: str | None = None |
| 363 | """Commit ID of the isolated first-bad commit.""" |
| 364 | |
| 365 | next_to_test: str | None = None |
| 366 | """Commit ID to test next (None when done).""" |
| 367 | |
| 368 | remaining_count: int = 0 |
| 369 | """How many commits remain to test.""" |
| 370 | |
| 371 | steps_remaining: int = 0 |
| 372 | """Approximate remaining binary-search steps.""" |
| 373 | |
| 374 | verdict: str = "" |
| 375 | """The verdict just applied: 'good', 'bad', 'skip'.""" |
| 376 | |
| 377 | symbol_changes: list[str] = field(default_factory=list) |
| 378 | """Human-readable descriptions of symbol ops in *next_to_test*. |
| 379 | |
| 380 | Only populated when a symbol_filter is active. Empty list otherwise. |
| 381 | """ |
| 382 | |
| 383 | |
| 384 | # --------------------------------------------------------------------------- |
| 385 | # Public API |
| 386 | # --------------------------------------------------------------------------- |
| 387 | |
| 388 | |
| 389 | def start_bisect( |
| 390 | repo_root: pathlib.Path, |
| 391 | bad_id: str, |
| 392 | good_ids: list[str], |
| 393 | branch: str = "", |
| 394 | symbol_filter: str = "", |
| 395 | ) -> BisectResult: |
| 396 | """Start a new bisect session. |
| 397 | |
| 398 | Args: |
| 399 | repo_root: Repository root. |
| 400 | bad_id: Known-bad commit. |
| 401 | good_ids: One or more known-good commits. |
| 402 | branch: Current branch name (for display only). |
| 403 | symbol_filter: Optional symbol address (e.g. ``billing.py::Invoice.compute_total``). |
| 404 | When set, only commits whose structured_delta touched this |
| 405 | symbol are included in the search space. |
| 406 | """ |
| 407 | import math |
| 408 | |
| 409 | remaining = _build_remaining(repo_root, bad_id, good_ids, skipped_ids=[]) |
| 410 | |
| 411 | # Narrow the search space to commits that touched the requested symbol. |
| 412 | if symbol_filter: |
| 413 | remaining = _commits_touching_symbol(repo_root, remaining, symbol_filter) |
| 414 | |
| 415 | state: BisectStateDict = { |
| 416 | "bad_id": bad_id, |
| 417 | "good_ids": good_ids, |
| 418 | "skipped_ids": [], |
| 419 | "remaining": remaining, |
| 420 | "log": [ |
| 421 | f"{bad_id} bad {_ts()}", |
| 422 | *[f"{g} good {_ts()}" for g in good_ids], |
| 423 | ], |
| 424 | } |
| 425 | if branch: |
| 426 | state["branch"] = branch |
| 427 | if symbol_filter: |
| 428 | state["symbol_filter"] = symbol_filter |
| 429 | _save_state(repo_root, state) |
| 430 | |
| 431 | if not remaining: |
| 432 | # bad_id is immediately the first bad commit (no unknowns). |
| 433 | return BisectResult( |
| 434 | done=True, |
| 435 | first_bad=bad_id, |
| 436 | next_to_test=None, |
| 437 | remaining_count=0, |
| 438 | steps_remaining=0, |
| 439 | verdict="started", |
| 440 | ) |
| 441 | next_id = _midpoint(remaining) |
| 442 | steps = int(math.log2(len(remaining) + 1)) |
| 443 | changes: list[str] = [] |
| 444 | if symbol_filter and next_id: |
| 445 | changes = _symbol_ops_in_commit(repo_root, next_id, symbol_filter) |
| 446 | return BisectResult( |
| 447 | done=False, |
| 448 | first_bad=None, |
| 449 | next_to_test=next_id, |
| 450 | remaining_count=len(remaining), |
| 451 | steps_remaining=steps, |
| 452 | verdict="started", |
| 453 | symbol_changes=changes, |
| 454 | ) |
| 455 | |
| 456 | |
| 457 | def _ts() -> str: |
| 458 | return datetime.datetime.now(tz=datetime.timezone.utc).isoformat() |
| 459 | |
| 460 | |
| 461 | def _apply_verdict( |
| 462 | repo_root: pathlib.Path, |
| 463 | commit_id: str, |
| 464 | verdict: str, |
| 465 | ) -> BisectResult: |
| 466 | """Apply a 'good', 'bad', or 'skip' verdict and return the next step.""" |
| 467 | import math |
| 468 | |
| 469 | state = _load_state(repo_root) |
| 470 | if state is None: |
| 471 | raise RuntimeError("No bisect session in progress. Run 'muse bisect start' first.") |
| 472 | |
| 473 | bad_id = state.get("bad_id", "") |
| 474 | good_ids = list(state.get("good_ids", [])) |
| 475 | skipped_ids = list(state.get("skipped_ids", [])) |
| 476 | log = list(state.get("log", [])) |
| 477 | symbol_filter = state.get("symbol_filter", "") |
| 478 | |
| 479 | if verdict == "good": |
| 480 | good_ids.append(commit_id) |
| 481 | elif verdict == "bad": |
| 482 | bad_id = commit_id |
| 483 | else: |
| 484 | skipped_ids.append(commit_id) |
| 485 | |
| 486 | log.append(f"{commit_id} {verdict} {_ts()}") |
| 487 | |
| 488 | remaining = _build_remaining(repo_root, bad_id, good_ids, skipped_ids) |
| 489 | |
| 490 | # Re-apply the symbol filter so newly included commits are also filtered. |
| 491 | if symbol_filter: |
| 492 | remaining = _commits_touching_symbol(repo_root, remaining, symbol_filter) |
| 493 | |
| 494 | new_state: BisectStateDict = { |
| 495 | "bad_id": bad_id, |
| 496 | "good_ids": good_ids, |
| 497 | "skipped_ids": skipped_ids, |
| 498 | "remaining": remaining, |
| 499 | "log": log, |
| 500 | } |
| 501 | if "branch" in state: |
| 502 | new_state["branch"] = state["branch"] |
| 503 | if symbol_filter: |
| 504 | new_state["symbol_filter"] = symbol_filter |
| 505 | _save_state(repo_root, new_state) |
| 506 | |
| 507 | if len(remaining) == 0: |
| 508 | # No unknowns remain — bad_id is confirmed as the first bad commit. |
| 509 | return BisectResult( |
| 510 | done=True, |
| 511 | first_bad=bad_id, |
| 512 | next_to_test=None, |
| 513 | remaining_count=0, |
| 514 | steps_remaining=0, |
| 515 | verdict=verdict, |
| 516 | ) |
| 517 | |
| 518 | next_id = _midpoint(remaining) |
| 519 | steps = int(math.log2(len(remaining) + 1)) |
| 520 | changes: list[str] = [] |
| 521 | if symbol_filter and next_id: |
| 522 | changes = _symbol_ops_in_commit(repo_root, next_id, symbol_filter) |
| 523 | return BisectResult( |
| 524 | done=False, |
| 525 | first_bad=None, |
| 526 | next_to_test=next_id, |
| 527 | remaining_count=len(remaining), |
| 528 | steps_remaining=steps, |
| 529 | verdict=verdict, |
| 530 | symbol_changes=changes, |
| 531 | ) |
| 532 | |
| 533 | |
| 534 | def mark_good(repo_root: pathlib.Path, commit_id: str) -> BisectResult: |
| 535 | """Mark *commit_id* as good and advance the bisect.""" |
| 536 | return _apply_verdict(repo_root, commit_id, "good") |
| 537 | |
| 538 | |
| 539 | def mark_bad(repo_root: pathlib.Path, commit_id: str) -> BisectResult: |
| 540 | """Mark *commit_id* as bad and advance the bisect.""" |
| 541 | return _apply_verdict(repo_root, commit_id, "bad") |
| 542 | |
| 543 | |
| 544 | def skip_commit(repo_root: pathlib.Path, commit_id: str) -> BisectResult: |
| 545 | """Skip *commit_id* (e.g. build fails) and advance the bisect.""" |
| 546 | return _apply_verdict(repo_root, commit_id, "skip") |
| 547 | |
| 548 | |
| 549 | def reset_bisect(repo_root: pathlib.Path) -> None: |
| 550 | """End the bisect session and remove state.""" |
| 551 | path = _state_path(repo_root) |
| 552 | if path.exists(): |
| 553 | path.unlink() |
| 554 | |
| 555 | |
| 556 | def get_bisect_log(repo_root: pathlib.Path) -> list[str]: |
| 557 | """Return the bisect log entries, oldest first.""" |
| 558 | state = _load_state(repo_root) |
| 559 | if state is None: |
| 560 | return [] |
| 561 | return list(state.get("log", [])) |
| 562 | |
| 563 | |
| 564 | def get_bisect_next( |
| 565 | repo_root: pathlib.Path, |
| 566 | ) -> tuple[str | None, str]: |
| 567 | """Return ``(next_to_test, symbol_filter)`` from the current bisect state. |
| 568 | |
| 569 | This is the public alternative to importing the private ``_load_state`` |
| 570 | directly from the CLI layer. Returns ``(None, "")`` when no session is |
| 571 | active or remaining is empty. |
| 572 | """ |
| 573 | state = _load_state(repo_root) |
| 574 | if state is None: |
| 575 | return None, "" |
| 576 | remaining = state.get("remaining", []) |
| 577 | symbol_filter = state.get("symbol_filter", "") |
| 578 | if not remaining: |
| 579 | return None, symbol_filter |
| 580 | return _midpoint(remaining), symbol_filter |
| 581 | |
| 582 | |
| 583 | def is_bisect_active(repo_root: pathlib.Path) -> bool: |
| 584 | """Return True if a bisect session is in progress.""" |
| 585 | return _state_path(repo_root).exists() |
| 586 | |
| 587 | |
| 588 | def run_bisect_command( |
| 589 | repo_root: pathlib.Path, |
| 590 | command: str, |
| 591 | current_commit_id: str, |
| 592 | ) -> BisectResult: |
| 593 | """Run *command* in a shell, interpret exit code, and apply verdict. |
| 594 | |
| 595 | Exit codes:: |
| 596 | |
| 597 | 0 → good |
| 598 | 125 → skip |
| 599 | 1-124, 126-255 → bad |
| 600 | |
| 601 | The command is executed with the repository root as the working directory. |
| 602 | """ |
| 603 | result = subprocess.run( |
| 604 | command, |
| 605 | shell=True, |
| 606 | cwd=str(repo_root), |
| 607 | ) |
| 608 | code = result.returncode |
| 609 | if code == 0: |
| 610 | verdict = "good" |
| 611 | elif code == _SKIP_EXIT_CODE: |
| 612 | verdict = "skip" |
| 613 | else: |
| 614 | verdict = "bad" |
| 615 | return _apply_verdict(repo_root, current_commit_id, verdict) |