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