From 8ed6a664a54dd0477e2a9baab47f038135407b73 Mon Sep 17 00:00:00 2001 From: auroter Date: Thu, 13 Aug 2026 18:22:54 -0700 Subject: [PATCH] Add minimal SourceHunt resume memoization --- clearwing/llm/budget.py | 121 +++++- clearwing/sourcehunt/findings_pool.py | 3 + clearwing/sourcehunt/pool.py | 91 ++++- clearwing/sourcehunt/resume.py | 334 ++++++++++++++++ clearwing/sourcehunt/runner.py | 151 +++++++- clearwing/ui/commands/sourcehunt.py | 22 ++ docs/architecture.md | 10 + docs/cli.md | 16 + tests/test_sourcehunt_resume.py | 537 ++++++++++++++++++++++++++ 9 files changed, 1249 insertions(+), 36 deletions(-) create mode 100644 clearwing/sourcehunt/resume.py create mode 100644 tests/test_sourcehunt_resume.py diff --git a/clearwing/llm/budget.py b/clearwing/llm/budget.py index 2d48b6e3..35ac5f0f 100644 --- a/clearwing/llm/budget.py +++ b/clearwing/llm/budget.py @@ -130,6 +130,7 @@ def __init__( default_max_output_tokens: int = DEFAULT_MAX_OUTPUT_TOKENS, manifest_filename: str = "manifest.json", endpoint: LLMEndpoint | None = None, + resume: bool = False, ) -> None: if not math.isfinite(limit_usd) or limit_usd < 0: raise BudgetConfigurationError("LLM budget must be a finite value >= 0") @@ -188,16 +189,133 @@ def __init__( self.ledger_path = session_dir / "spend-ledger.jsonl" self.manifest_path = session_dir / manifest_filename with self._lock: + if resume: + self._restore_history_locked() self._persist_event_locked( { - "event": "run_started", + "event": "run_resumed" if resume else "run_started", "session_id": self.session_id, "budget_usd": self.limit_usd, + "carried_forward_usd": self._spent_usd, "timestamp": self._timestamp(), } ) self._persist_snapshot_locked() + def _restore_history_locked(self) -> None: + """Restore settled calls and conservatively close interrupted reservations.""" + + if not self.ledger_path.is_file(): + raise BudgetConfigurationError( + f"Cannot resume session {self.session_id!r}: spend ledger is missing" + ) + try: + lines = self.ledger_path.read_text(encoding="utf-8").splitlines() + except (OSError, UnicodeDecodeError) as exc: + raise BudgetConfigurationError( + f"Cannot resume session {self.session_id!r}: spend ledger is unreadable" + ) from exc + + reservations: dict[str, dict[str, Any]] = {} + settlements: dict[str, dict[str, Any]] = {} + has_run_header = False + for index, line in enumerate(lines): + try: + event = json.loads(line) + except json.JSONDecodeError as exc: + if index == len(lines) - 1: + break + raise BudgetConfigurationError( + f"Cannot resume session {self.session_id!r}: spend ledger is corrupt" + ) from exc + if not isinstance(event, dict): + continue + if event.get("event") in {"run_started", "run_resumed"}: + has_run_header = has_run_header or event.get("session_id") == self.session_id + call_id = str(event.get("call_id") or "") + if not call_id: + continue + if event.get("event") == "call_reserved": + reservations.setdefault(call_id, event) + elif event.get("event") == "call_settled": + settlements.setdefault(call_id, event) + + if not has_run_header: + raise BudgetConfigurationError( + f"Cannot resume session {self.session_id!r}: spend ledger is corrupt" + ) + + for call_id in reservations.keys() | settlements.keys(): + if call_id in settlements: + self._restore_settlement_locked(settlements[call_id]) + else: + self._recover_reservation_locked(call_id, reservations[call_id]) + if self.enforcing and self._spent_usd >= self.limit_usd - self._EPSILON: + self._exhausted = True + self._status = "budget_exhausted" + + def _restore_settlement_locked(self, event: dict[str, Any]) -> None: + try: + cost = float(event["cost_usd"]) + input_tokens = int(event.get("input_tokens", 0)) + output_tokens = int(event.get("output_tokens", 0)) + cached_tokens = int(event.get("cached_input_tokens", 0)) + except (KeyError, TypeError, ValueError) as exc: + raise BudgetConfigurationError( + f"Cannot resume session {self.session_id!r}: spend ledger is corrupt" + ) from exc + if ( + not math.isfinite(cost) + or min(cost, input_tokens, output_tokens, cached_tokens) < 0 + ): + raise BudgetConfigurationError( + f"Cannot resume session {self.session_id!r}: spend ledger is corrupt" + ) + if not isinstance(event.get("metadata"), dict): + event = {**event, "metadata": {}} + self._records.append(event) + self._spent_usd += cost + self._input_tokens += input_tokens + self._output_tokens += output_tokens + self._cached_input_tokens += cached_tokens + + def _recover_reservation_locked( + self, call_id: str, reservation: dict[str, Any] + ) -> None: + try: + reserved_usd = float(reservation["reserved_usd"]) + except (KeyError, TypeError, ValueError) as exc: + raise BudgetConfigurationError( + f"Cannot resume session {self.session_id!r}: spend ledger is corrupt" + ) from exc + if not math.isfinite(reserved_usd) or reserved_usd < 0: + raise BudgetConfigurationError( + f"Cannot resume session {self.session_id!r}: spend ledger is corrupt" + ) + charged = reserved_usd if reservation.get("budget_enforcing", reserved_usd > 0) else 0.0 + event = { + "event": "call_settled", + "call_id": call_id, + "timestamp": self._timestamp(), + "stage": reservation.get("stage"), + "model": reservation.get("model"), + "provider": reservation.get("provider"), + "status": "recovered_ambiguous_failure", + "reserved_usd": reserved_usd, + "cost_usd": charged, + "cost_source": "reservation" if charged else "none", + "input_tokens": 0, + "cached_input_tokens": 0, + "output_tokens": 0, + "metadata": ( + reservation["metadata"] if isinstance(reservation.get("metadata"), dict) else {} + ), + "error": "Process exited before spend settlement; charged on resume", + } + self._records.append(event) + self._spent_usd += charged + self._persist_event_locked(event) + @property def enforcing(self) -> bool: """Whether this run has a finite non-zero dollar cap.""" @@ -349,6 +467,7 @@ def reserve_call( "stage": stage, "model": model, "provider": provider, + "budget_enforcing": self.enforcing, "reserved_usd": reserved_usd, "input_token_upper_bound": input_token_upper_bound, "max_output_tokens": effective_max_tokens, diff --git a/clearwing/sourcehunt/findings_pool.py b/clearwing/sourcehunt/findings_pool.py index 9f367a96..227bb101 100644 --- a/clearwing/sourcehunt/findings_pool.py +++ b/clearwing/sourcehunt/findings_pool.py @@ -191,6 +191,9 @@ def count(self) -> int: async def add(self, finding: Finding) -> Finding: """Add a finding, classify primitive, run dedup, return updated finding.""" async with self._lock: + existing = self._findings.get(finding.get("id", "")) + if existing is not None: + return existing if not finding.get("primitive_type"): finding.primitive_type = await self._classify_primitive(finding) diff --git a/clearwing/sourcehunt/pool.py b/clearwing/sourcehunt/pool.py index 89f44691..97a303b8 100644 --- a/clearwing/sourcehunt/pool.py +++ b/clearwing/sourcehunt/pool.py @@ -72,19 +72,33 @@ class WorkItem: seed_transcript: str | None = None entry_point: Any = None # EntryPoint | None — spec 004 seed_context: str | None = None # spec 004 seed corpus + context_id: str = "" # hash of per-file seeded-crash/static prompt inputs - def stable_identifier(self, run_id: str) -> str: + def stable_identifier(self, run_id: str, tier: str = "") -> str: entry_point = self.entry_point + entry_point_id = ( + { + "file": getattr(entry_point, "file_path", ""), + "function": getattr(entry_point, "function_name", ""), + "start": getattr(entry_point, "start_line", 0), + "end": getattr(entry_point, "end_line", 0), + "type": getattr(entry_point, "entry_type", ""), + } + if entry_point is not None + else None + ) return stable_run_id( "work", { "run_id": run_id, "file": self.file_target.get("path", ""), + "tier": tier, "band": self.band, "attempt": self.attempt, - "entry_point": ( - getattr(entry_point, "function_name", "") if entry_point is not None else "" - ), + "entry_point": entry_point_id, + "seed_context": self.seed_context, + "seed_transcript": self.seed_transcript, + "context_id": self.context_id, }, ) @@ -212,6 +226,7 @@ class HuntPoolConfig: findings_pool: Any = None # FindingsPool | None — spec 005 trajectory_root: str | Path | None = None instrumentation: Any = None # SourceHuntInstrumentation | None + checkpoint: Any = None # SourceHuntCheckpoint | None def _format_seed_context(entries: list) -> str | None: @@ -297,6 +312,11 @@ def _expand_to_work_items(self, files: list[FileTarget], band: str) -> list[Work else [] ) seed_entries = self.config.seed_corpus_by_file.get(file_path, []) + context = { + "seeded_crash": self.config.seeded_crashes_by_file.get(file_path), + "static_hints": self.config.semgrep_hints_by_file.get(file_path, []), + } + context_id = stable_run_id("context", context) if any(context.values()) else "" if entry_points: for ep in entry_points: @@ -314,6 +334,7 @@ def _expand_to_work_items(self, files: list[FileTarget], band: str) -> list[Work attempt=attempt, entry_point=ep, seed_context=seed_ctx, + context_id=context_id, ) ) else: @@ -325,6 +346,7 @@ def _expand_to_work_items(self, files: list[FileTarget], band: str) -> list[Work band=band, attempt=attempt, seed_context=seed_ctx, + context_id=context_id, ) ) return items @@ -448,7 +470,7 @@ async def _run_tier_phase( else None ) spent = 0.0 - in_flight: dict[asyncio.Task[TargetResult], WorkItem] = {} + in_flight: dict[asyncio.Future[TargetResult], tuple[WorkItem, str, bool]] = {} item_iter = iter(work_items) promotion_queue: list[WorkItem] = [] @@ -465,20 +487,35 @@ def _submit_next() -> bool: if wi is None: return False band_cost = self.config.band_budget.for_band(wi.band) - work_item_id = wi.stable_identifier(self.config.session_id_prefix) - task = asyncio.create_task( - self._run_file_task( - wi.file_target, - cost_limit=band_cost, - tier=tier, - band=wi.band, - seed_transcript=wi.seed_transcript, - entry_point=wi.entry_point, - seed_context=wi.seed_context, - work_item_id=work_item_id, + work_item_id = wi.stable_identifier(self.config.session_id_prefix, tier) + cached = self.config.checkpoint.load(work_item_id) if self.config.checkpoint else None + if cached is not None and ( + cached.target != wi.file_target.get("path", "") + or cached.tier != tier + or cached.band != wi.band + ): + logger.warning("Ignoring mismatched cached hunter work %s", work_item_id) + cached = None + if cached is not None: + logger.info("Reusing completed hunter work for %s", wi.file_target.get("path", "")) + task = asyncio.get_running_loop().create_future() + task.set_result(cached) + from_cache = True + else: + task = asyncio.create_task( + self._run_file_task( + wi.file_target, + cost_limit=band_cost, + tier=tier, + band=wi.band, + seed_transcript=wi.seed_transcript, + entry_point=wi.entry_point, + seed_context=wi.seed_context, + work_item_id=work_item_id, + ) ) - ) - in_flight[task] = wi + from_cache = False + in_flight[task] = (wi, work_item_id, from_cache) return True for _ in range(max(1, self.config.max_parallel)): @@ -498,7 +535,7 @@ def _submit_next() -> bool: timeout, len(in_flight), ) - for task, wi in list(in_flight.items()): + for task, (wi, _work_item_id, _from_cache) in list(in_flight.items()): task.cancel() key = wi.file_target.get("path", "") async with self._state_lock: @@ -513,7 +550,7 @@ def _submit_next() -> bool: return spent for task in done: - wi = in_flight.pop(task) + wi, work_item_id, from_cache = in_flight.pop(task) key = wi.file_target.get("path", "") try: result = await task @@ -544,6 +581,19 @@ def _submit_next() -> bool: tier=tier, band=wi.band, ) + if ( + result.status == "completed" + and not from_cache + and self.config.checkpoint is not None + ): + try: + self.config.checkpoint.save(work_item_id, result) + except Exception: + logger.warning( + "Could not checkpoint completed work %s", + work_item_id, + exc_info=True, + ) ep_suffix = f":{wi.entry_point.function_name}" if wi.entry_point else "" async with self._state_lock: self._results[f"{key}{ep_suffix}:{wi.band}:{wi.attempt}"] = result @@ -624,6 +674,7 @@ def _submit_next() -> bool: band=next_band, attempt=wi.attempt, seed_transcript=_extract_transcript(result), + context_id=wi.context_id, ) ) diff --git a/clearwing/sourcehunt/resume.py b/clearwing/sourcehunt/resume.py new file mode 100644 index 00000000..9f23f4b5 --- /dev/null +++ b/clearwing/sourcehunt/resume.py @@ -0,0 +1,334 @@ +"""Crash-safe memoization for standalone SourceHunt sessions.""" + +from __future__ import annotations + +import fcntl +import hashlib +import json +import math +import os +import re +import tempfile +from dataclasses import asdict, is_dataclass +from pathlib import Path +from typing import Any + +from clearwing.findings.types import Finding +from clearwing.runners.parallel.executor import TargetResult + +_SCHEMA_VERSION = 1 +_SESSION_ID = re.compile(r"^sh-[A-Za-z0-9_-]+$") +_WORK_ID = re.compile(r"^work-[a-f0-9]{16}$") +_RANK_FIELDS = ( + "surface", + "influence", + "reachability", + "priority", + "surface_rationale", + "influence_rationale", + "reachability_rationale", +) + + +class SourceHuntResumeError(ValueError): + """A requested resume session is invalid, incompatible, or busy.""" + + +def session_directory(output_dir: str | Path, session_id: str) -> Path: + """Resolve a safe, bare SourceHunt session ID below an output root.""" + + if not _SESSION_ID.fullmatch(session_id): + raise SourceHuntResumeError( + f"Invalid sourcehunt session ID {session_id!r}; expected a value like sh-535ed81b" + ) + return Path(output_dir).expanduser().resolve() / session_id + + +class SourceHuntSessionLock: + """Prevent two processes from writing one standalone session.""" + + def __init__(self, session_dir: str | Path): + self.path = Path(session_dir) / ".resume.lock" + self._stream: Any = None + + def acquire(self) -> None: + self.path.parent.mkdir(parents=True, exist_ok=True) + stream = open(self.path, "a+", encoding="utf-8") + try: + fcntl.flock(stream.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + except BlockingIOError as exc: + stream.close() + raise SourceHuntResumeError( + f"Sourcehunt session {self.path.parent.name!r} is already running" + ) from exc + stream.seek(0) + stream.truncate() + stream.write(f"pid={os.getpid()}\n") + stream.flush() + self._stream = stream + + def release(self) -> None: + if self._stream is None: + return + try: + fcntl.flock(self._stream.fileno(), fcntl.LOCK_UN) + finally: + self._stream.close() + self._stream = None + + +def fingerprint_source(repo_path: str | Path, files: list[dict[str, Any]]) -> str: + """Hash the selected relative paths and their complete file bytes.""" + + repository = Path(repo_path).resolve() + selected: list[tuple[str, Path]] = [] + for target in files: + relative = str(target.get("path") or "") + relative_path = Path(relative) + absolute = Path(str(target.get("absolute_path") or repository / relative_path)).resolve() + expected = (repository / relative_path).resolve() + if ( + not relative + or relative_path.is_absolute() + or ".." in relative_path.parts + or absolute != expected + or not absolute.is_relative_to(repository) + ): + raise SourceHuntResumeError(f"Selected source input {relative!r} is invalid") + selected.append((relative_path.as_posix(), absolute)) + if len(selected) != len({relative for relative, _ in selected}): + raise SourceHuntResumeError("Selected source inputs contain duplicate paths") + + digest = hashlib.sha256() + for relative, absolute in sorted(selected): + try: + content = absolute.read_bytes() + except OSError as exc: + raise SourceHuntResumeError( + f"Unable to fingerprint selected source input {relative!r}: {exc}" + ) from exc + for value in (relative.encode(), content): + digest.update(len(value).to_bytes(8, "big")) + digest.update(value) + return digest.hexdigest() + + +def fingerprint_invocation(options: dict[str, Any]) -> str: + """Hash behavior-affecting options without credentials or model routing.""" + + encoded = json.dumps(options, sort_keys=True, separators=(",", ":"), default=str).encode() + return hashlib.sha256(encoded).hexdigest() + + +class SourceHuntCheckpoint: + """One session manifest and immutable completed-work results.""" + + def __init__(self, session_dir: str | Path, *, resuming: bool): + self.session_dir = Path(session_dir) + self.path = self.session_dir / "checkpoint.json" + self.work_dir = self.session_dir / "work-results" + self.resuming = resuming + self._manifest: dict[str, Any] | None = None + + def prepare( + self, + *, + invocation_fingerprint: str, + source_fingerprint: str, + source_paths: list[str], + ) -> list[dict[str, Any]] | None: + """Create or validate the session and return a complete cached rank plan.""" + + if self.resuming: + manifest = _read_json(self.path) + if not self._valid_manifest(manifest): + raise SourceHuntResumeError( + f"Session {self.session_dir.name!r} has no valid resumable checkpoint" + ) + if manifest["invocation_fingerprint"] != invocation_fingerprint: + raise SourceHuntResumeError( + "Resume options do not match the original sourcehunt invocation" + ) + if manifest["source_fingerprint"] != source_fingerprint: + raise SourceHuntResumeError( + "Selected source inputs changed since this sourcehunt session began" + ) + if manifest["source_paths"] != sorted(source_paths): + raise SourceHuntResumeError( + "Selected source inputs changed since this sourcehunt session began" + ) + self._manifest = manifest + else: + if self.path.exists(): + raise SourceHuntResumeError(f"Session checkpoint already exists: {self.path}") + self._manifest = { + "schema_version": _SCHEMA_VERSION, + "session_id": self.session_dir.name, + "invocation_fingerprint": invocation_fingerprint, + "source_fingerprint": source_fingerprint, + "source_paths": sorted(source_paths), + "rank_plan": None, + } + _atomic_json(self.path, self._manifest) + return self._rank_plan(self._manifest.get("rank_plan")) + + def save_rank_plan(self, files: list[dict[str, Any]]) -> None: + if self._manifest is None: + raise SourceHuntResumeError("Sourcehunt checkpoint has not been prepared") + by_path = [] + for target in files: + item = {"path": target.get("path", "")} + item.update({field: target.get(field) for field in _RANK_FIELDS}) + by_path.append(item) + self._manifest["rank_plan"] = by_path + _atomic_json(self.path, self._manifest) + + def apply_rank_plan( + self, files: list[dict[str, Any]], plan: list[dict[str, Any]] + ) -> None: + by_path = {item["path"]: item for item in plan} + for target in files: + cached = by_path[str(target.get("path") or "")] + target.update({field: cached[field] for field in _RANK_FIELDS}) + + def load(self, work_id: str) -> TargetResult | None: + if not _WORK_ID.fullmatch(work_id): + return None + payload = _read_json(self.work_dir / f"{work_id}.json") + if ( + not isinstance(payload, dict) + or payload.get("schema_version") != _SCHEMA_VERSION + or payload.get("work_id") != work_id + or not isinstance(payload.get("result"), dict) + ): + return None + result = payload["result"] + allowed_fields = set(TargetResult.__dataclass_fields__) + if ( + result.get("status") != "completed" + or not isinstance(result.get("findings"), list) + or not set(result).issubset(allowed_fields) + or not all(field in result for field in ("target", "status", "tier", "band")) + ): + return None + try: + findings = [_finding(item) for item in result["findings"]] + if any(finding is None for finding in findings): + return None + target_result = TargetResult( + **{ + key: value + for key, value in result.items() + if key in TargetResult.__dataclass_fields__ and key != "findings" + }, + findings=[finding for finding in findings if finding is not None], + ) + if ( + target_result.status != "completed" + or target_result.tier not in {"A", "B", "C"} + or target_result.band not in {"fast", "standard", "deep"} + or not math.isfinite(float(target_result.cost_usd)) + or target_result.cost_usd < 0 + or target_result.tokens_used < 0 + ): + return None + return target_result + except (TypeError, ValueError): + return None + + def save(self, work_id: str, result: TargetResult) -> None: + if not _WORK_ID.fullmatch(work_id) or result.status != "completed": + return + path = self.work_dir / f"{work_id}.json" + if self.load(work_id) is not None: + return + _atomic_json( + path, + { + "schema_version": _SCHEMA_VERSION, + "work_id": work_id, + "result": _json_value(result), + }, + ) + + def _rank_plan(self, value: Any) -> list[dict[str, Any]] | None: + if not isinstance(value, list) or self._manifest is None: + return None + paths = self._manifest["source_paths"] + if len(value) != len(paths): + return None + plan: list[dict[str, Any]] = [] + for item in value: + if not isinstance(item, dict) or not isinstance(item.get("path"), str): + return None + if any(field not in item for field in _RANK_FIELDS): + return None + if any(not isinstance(item[field], int) for field in _RANK_FIELDS[:3]): + return None + priority = item["priority"] + if not isinstance(priority, (int, float)) or not math.isfinite(float(priority)): + return None + if any(not isinstance(item[field], str) for field in _RANK_FIELDS[4:]): + return None + plan.append(dict(item)) + if sorted(item["path"] for item in plan) != paths: + return None + return plan + + def _valid_manifest(self, value: Any) -> bool: + return ( + isinstance(value, dict) + and value.get("schema_version") == _SCHEMA_VERSION + and value.get("session_id") == self.session_dir.name + and isinstance(value.get("invocation_fingerprint"), str) + and isinstance(value.get("source_fingerprint"), str) + and isinstance(value.get("source_paths"), list) + and all(isinstance(path, str) and path for path in value["source_paths"]) + and value["source_paths"] == sorted(set(value["source_paths"])) + ) + + +def _finding(value: Any) -> Finding | None: + if not isinstance(value, dict) or not isinstance(value.get("id"), str) or not value["id"]: + return None + try: + return Finding( + **{key: item for key, item in value.items() if key in Finding.__dataclass_fields__} + ) + except (TypeError, ValueError): + return None + + +def _json_value(value: Any) -> Any: + if is_dataclass(value) and not isinstance(value, type): + return {key: _json_value(item) for key, item in asdict(value).items()} + if isinstance(value, dict): + return {str(key): _json_value(item) for key, item in value.items()} + if isinstance(value, (list, tuple)): + return [_json_value(item) for item in value] + if isinstance(value, Path): + return str(value) + return value + + +def _read_json(path: Path) -> dict[str, Any] | None: + try: + value = json.loads(path.read_text(encoding="utf-8")) + except (OSError, UnicodeDecodeError, json.JSONDecodeError): + return None + return value if isinstance(value, dict) else None + + +def _atomic_json(path: Path, value: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + descriptor, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) + try: + with os.fdopen(descriptor, "w", encoding="utf-8") as stream: + json.dump(value, stream, indent=2, sort_keys=True) + stream.write("\n") + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + finally: + if os.path.exists(temporary): + os.unlink(temporary) diff --git a/clearwing/sourcehunt/runner.py b/clearwing/sourcehunt/runner.py index 92659f7c..61246c95 100644 --- a/clearwing/sourcehunt/runner.py +++ b/clearwing/sourcehunt/runner.py @@ -53,6 +53,13 @@ from .pool import HunterPool, HuntPoolConfig, TierBudget from .preprocessor import Preprocessor, PreprocessResult from .ranker import Ranker, RankerConfig +from .resume import ( + SourceHuntCheckpoint, + SourceHuntSessionLock, + fingerprint_invocation, + fingerprint_source, + session_directory, +) from .state import ( EvidenceLevel, FileTarget, @@ -231,6 +238,7 @@ def __init__( exploiter_llm: Any = None, sandbox_factory: Any = None, # callable[[], SandboxContainer] parent_session_id: str | None = None, + resume_session_id: str | None = None, agent_mode: str = "auto", # "auto" | "constrained" | "deep" prompt_mode: str = "unconstrained", # "unconstrained" | "specialist" campaign_hint: str | None = None, @@ -442,6 +450,10 @@ def __init__( raise ValueError("sandbox_cpus must be a finite number greater than or equal to 0") if flow not in {"legacy", "proof"}: raise ValueError("flow must be 'legacy' or 'proof'") + if resume_session_id is not None and parent_session_id is not None: + raise ValueError("resume_session_id cannot be combined with parent_session_id") + if resume_session_id is not None and flow != "legacy": + raise ValueError("Sourcehunt resume currently supports only the legacy flow") if proof_max_actions < 1: raise ValueError("proof_max_actions must be positive") if proof_max_model_calls < 0 or proof_max_dynamic_actions < 0: @@ -511,7 +523,20 @@ def __init__( self.sandbox_factory = sandbox_factory self._sandbox_manager: HunterSandbox | None = None self._preprocessor: Preprocessor | None = None - self._session_id = parent_session_id or f"sh-{uuid.uuid4().hex[:8]}" + self._session_id = resume_session_id or parent_session_id or f"sh-{uuid.uuid4().hex[:8]}" + self._resume_session_id = resume_session_id + self._checkpoint: SourceHuntCheckpoint | None = None + self._session_lock: SourceHuntSessionLock | None = None + self._session_lock_held = False + if parent_session_id is None and flow == "legacy": + session_dir = session_directory(self.output_dir, self._session_id) + if resume_session_id is not None and not session_dir.is_dir(): + raise ValueError(f"Sourcehunt session {resume_session_id!r} does not exist") + self._checkpoint = SourceHuntCheckpoint( + session_dir, + resuming=resume_session_id is not None, + ) + self._session_lock = SourceHuntSessionLock(session_dir) self._agent_mode_override = agent_mode self._prompt_mode = prompt_mode self._campaign_hint = campaign_hint @@ -558,14 +583,70 @@ def __init__( self._retain_incomplete_certificates = retain_incomplete_certificates self._emit_rejection_certificates = emit_rejection_certificates self._falsify = falsify - self._instrumentation = SourceHuntInstrumentation( - Path(self.output_dir) / self._session_id, - self._session_id, - ) + self.__instrumentation: SourceHuntInstrumentation | None = None self._instrumentation_finalized = False self._last_reporting_error: dict[str, str] | None = None self._on_progress = on_progress + @property + def _instrumentation(self) -> SourceHuntInstrumentation: + if self.__instrumentation is None: + self.__instrumentation = SourceHuntInstrumentation( + Path(self.output_dir) / self._session_id, + self._session_id, + ) + return self.__instrumentation + + def _acquire_session_lock(self) -> bool: + if self._session_lock is None or self._session_lock_held: + return False + self._session_lock.acquire() + self._session_lock_held = True + return True + + def _release_session_lock(self) -> None: + if self._session_lock is not None and self._session_lock_held: + self._session_lock.release() + self._session_lock_held = False + + def _invocation_fingerprint(self) -> str: + local_path = str(Path(self.local_path).resolve()) if self.local_path else None + return fingerprint_invocation( + { + "repo_url": self.repo_url, + "branch": self.branch, + "local_path": local_path, + "depth": self.depth, + "budget_usd": self.budget_usd, + "max_parallel": self.max_parallel, + "tier_budget": { + "a": self.tier_budget.tier_a_fraction, + "b": self.tier_budget.tier_b_fraction, + "c": self.tier_budget.tier_c_fraction, + }, + "agent_mode": self._agent_mode_override, + "prompt_mode": self._prompt_mode, + "campaign_hint": self._campaign_hint, + "exploit_mode": self._exploit_mode, + "starting_band": self._starting_band, + "max_band": self._max_band, + "redundancy": self._redundancy_override, + "shard_entry_points": self._shard_entry_points, + "min_shard_rank": self._min_shard_rank, + "min_project_loc": self._min_project_loc, + "seed_corpus_sources": self._seed_corpus_sources, + "seed_harness_crashes": self._seed_harness_crashes, + "preprocessing": self._preprocessing, + "respect_gitignore": self._respect_gitignore, + "no_rank": self._no_rank, + "no_per_file_hunt": self._no_per_file_hunt, + "findings_pool": self._enable_findings_pool, + "mechanism_memory": self.enable_mechanism_memory, + "gvisor_runtime": self._gvisor_runtime, + "sandbox_cpus": self._sandbox_cpus, + } + ) + @staticmethod def _check_runtime_available(runtime: str | None) -> str | None: if not runtime: @@ -730,6 +811,7 @@ def _ensure_spend_ledger(self) -> SpendLedger: manifest_filename=( "spend-summary.json" if self._flow == "proof" else "manifest.json" ), + resume=self._resume_session_id is not None, ) return self._spend_ledger @@ -769,13 +851,18 @@ def _finalize_spend_ledger(self, status: str | None = None) -> dict[str, Any]: def run(self) -> SourceHuntResult: from clearwing.ui.llm_activity import llm_activity_panel - self._ensure_spend_ledger() - with llm_activity_panel( - live=self._live, - budget_usd=self.budget_usd or None, - spend_ledger=self._spend_ledger, - ): - return asyncio.run(self.arun()) + acquired = self._acquire_session_lock() + try: + self._ensure_spend_ledger() + with llm_activity_panel( + live=self._live, + budget_usd=self.budget_usd or None, + spend_ledger=self._spend_ledger, + ): + return asyncio.run(self.arun()) + finally: + if acquired: + self._release_session_lock() async def _arun_proof_flow(self) -> SourceHuntResult: """Run the proof-carrying engine and adapt its typed output.""" @@ -1003,6 +1090,15 @@ def _finalize_proof_manifest( proof_result.output_paths.update(outputs) async def arun(self) -> SourceHuntResult: + acquired = self._acquire_session_lock() + try: + self._ensure_spend_ledger() + return await self._arun() + finally: + if acquired: + self._release_session_lock() + + async def _arun(self) -> SourceHuntResult: if self._flow == "proof": try: return await self._arun_proof_flow() @@ -1035,12 +1131,20 @@ async def arun(self) -> SourceHuntResult: detail=f"Enumerated {files_ranked} files", files=stage_files, ) + rank_plan: list[dict[str, Any]] | None = None + if self._checkpoint is not None: + source_fingerprint = fingerprint_source(repo_path, files) + rank_plan = self._checkpoint.prepare( + invocation_fingerprint=self._invocation_fingerprint(), + source_fingerprint=source_fingerprint, + source_paths=stage_files, + ) self._ensure_sandbox_factory(repo_path, files) # 2. Rank — unless depth=quick AND no LLM available, or --no-rank ranker_llm = ( None - if self._no_rank + if self._no_rank or rank_plan is not None else self._get_native_client( "ranker", self.ranker_llm, @@ -1053,7 +1157,17 @@ async def arun(self) -> SourceHuntResult: detail=f"{len(files)} files", files=stage_files, ) - if self._no_rank: + if rank_plan is not None: + self._checkpoint.apply_rank_plan(files, rank_plan) + logger.info("Restored complete rank plan from session checkpoint") + pipeline_status.record_succeeded("ranker") + self._emit_stage( + "rank", + "completed", + detail=f"Restored ranks for {len(files)} files", + files=stage_files, + ) + elif self._no_rank: logger.info("Ranker skipped (--no-rank); assigning default priority scores") for ft in files: ft["surface"] = ft.get("surface") or 3 @@ -1152,6 +1266,8 @@ async def arun(self) -> SourceHuntResult: ft["priority"] = ft.get("priority") or ( ft["surface"] * 0.5 + ft["influence"] * 0.2 + ft["reachability"] * 0.3 ) + if self._checkpoint is not None and rank_plan is None: + self._checkpoint.save_rank_plan(files) # depth=quick exits here with the static_findings as-is if self.depth == "quick": @@ -1284,7 +1400,11 @@ async def arun(self) -> SourceHuntResult: from .historical_findings_db import HistoricalFindingsDB checkpoint_path = Path(self.output_dir) / self._session_id / "findings_pool.jsonl" - findings_pool = FindingsPool(checkpoint_path=checkpoint_path) + findings_pool = ( + FindingsPool.from_checkpoint(checkpoint_path) + if self._resume_session_id is not None + else FindingsPool(checkpoint_path=checkpoint_path) + ) try: historical_db = HistoricalFindingsDB(path=self._historical_db_path) prior = historical_db.query_prior(repo_url=self.repo_url) @@ -1358,6 +1478,7 @@ async def arun(self) -> SourceHuntResult: findings_pool=findings_pool, trajectory_root=(Path(self.output_dir) / self._session_id / "trajectories"), instrumentation=self._instrumentation, + checkpoint=self._checkpoint, ) ) try: diff --git a/clearwing/ui/commands/sourcehunt.py b/clearwing/ui/commands/sourcehunt.py index 5362713c..4d590f41 100644 --- a/clearwing/ui/commands/sourcehunt.py +++ b/clearwing/ui/commands/sourcehunt.py @@ -40,6 +40,12 @@ def add_parser(subparsers): help="Source-code vulnerability hunting (source-hunt pipeline)", ) parser.add_argument("repo", nargs="?", help="Git URL or local path to a repository") + parser.add_argument( + "--resume", + metavar="SESSION_ID", + default=None, + help="Reuse completed hunter work from an existing standalone session", + ) parser.add_argument("--machine-fd", type=int, help=argparse.SUPPRESS) parser.add_argument( "--flow", @@ -672,6 +678,21 @@ def handle(cli, args): raise SystemExit(_handle_machine(args.machine_fd)) if not args.repo: args._command_parser.error("the following arguments are required: repo") + if args.resume and args.flow != "legacy": + args._command_parser.error("--resume currently supports only --flow legacy") + if args.resume and any( + ( + args.retro_hunt, + args.nday, + getattr(args, "reveng", False), + args.elaborate, + args.elaborate_auto, + args.calibrate, + args.webhook, + args.watch, + ) + ): + args._command_parser.error("--resume cannot be combined with an alternate sourcehunt mode") from ...core.config import default_results_dir from ...providers import ProviderManager, resolve_llm_endpoint @@ -1163,6 +1184,7 @@ def handle(cli, args): disclosure_reporter_email=args.reporter_email, model_override=args.model, provider_manager=provider_manager, + resume_session_id=args.resume, agent_mode=args.agent_mode, prompt_mode=args.prompt_mode, campaign_hint=args.campaign_hint, diff --git a/docs/architecture.md b/docs/architecture.md index ba3bca6d..b6d4af62 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -132,6 +132,16 @@ input for the next: writes pre-filled MITRE CVE-request and HackerOne templates for every verified finding `>= root_cause_explained`. +Standalone legacy-flow resume is deliberately a memoization boundary, not a +second workflow: `HunterPool` maps a deterministic `WorkItem` to an immutable +completed `TargetResult`. Preprocessing runs normally, a valid complete rank +plan keeps scheduling stable, and missing/corrupt work is simply a cache miss. +Cached findings re-enter the ordinary findings pool and promotion logic; all +verification, exploitation, enrichment, and reporting stages run normally. +The append-only spend ledger supplies the lifetime session total and a session +lock prevents concurrent writers. Campaign and proof flows retain their own +orchestration and do not use this mechanism. + ## The shared Finding type `clearwing.findings.Finding` is the single canonical finding diff --git a/docs/cli.md b/docs/cli.md index d43665fd..0944ec1f 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -30,6 +30,7 @@ with `clearwing report` or `clearwing history`. ```bash clearwing sourcehunt + [--resume SESSION_ID] # reuse completed hunter work from this session [--branch BRANCH] # default: main [--depth quick|standard|deep] [--budget USD] # default: unlimited; 0 = unlimited @@ -53,6 +54,21 @@ clearwing sourcehunt See [LLM providers](providers.md) for the full precedence rules and provider-specific snippets. +To continue an interrupted standalone legacy run, use the same output directory, +repeat the original repository and hunt options, and add its session ID: + +```bash +clearwing sourcehunt --resume sh-deadbeef [original hunt options] +``` + +Preprocessing reruns and rejects changed source files or hunt behavior. Completed +file/band work (including zero-finding work) is reused; corrupt or unfinished work +runs again. Ranking is reused only when its complete checkpoint is valid, while +verification, exploitation, enrichment, and reports run normally. The original +budget remains the lifetime session cap. Provider credentials, endpoint, model, +and later-stage switches such as `--no-verify` and `--no-exploit` may change. +Resume is not available for proof-flow, campaign-owned, or alternate-mode runs. + Depths: - **`quick`** — preprocessor + ranker + static findings. No LLM hunters. Free. Useful as a sanity check or for CI. diff --git a/tests/test_sourcehunt_resume.py b/tests/test_sourcehunt_resume.py new file mode 100644 index 00000000..b2e4b081 --- /dev/null +++ b/tests/test_sourcehunt_resume.py @@ -0,0 +1,537 @@ +from __future__ import annotations + +import argparse +import json +from pathlib import Path +from unittest.mock import AsyncMock + +import pytest + +from clearwing.findings.types import Finding +from clearwing.llm.budget import BudgetConfigurationError, SpendLedger +from clearwing.runners.parallel.executor import TargetResult +from clearwing.sourcehunt.entry_points import EntryPoint +from clearwing.sourcehunt.findings_pool import FindingsPool +from clearwing.sourcehunt.pool import ( + HunterPool, + HuntPoolConfig, + WorkItem, + _extract_transcript, +) +from clearwing.sourcehunt.preprocessor import PreprocessResult +from clearwing.sourcehunt.resume import ( + SourceHuntCheckpoint, + SourceHuntResumeError, + SourceHuntSessionLock, + fingerprint_invocation, + fingerprint_source, + session_directory, +) +from clearwing.sourcehunt.runner import SourceHuntRunner +from clearwing.ui.commands import sourcehunt as sourcehunt_command + + +def _target(repo: Path, **overrides): + target = { + "path": "app.py", + "absolute_path": str(repo / "app.py"), + "surface": 4, + "influence": 4, + "reachability": 3, + "priority": 3.7, + "tier": "C", + "tags": ["attacker_reachable"], + "language": "python", + "loc": 1, + "surface_rationale": "ranked", + "influence_rationale": "ranked", + "reachability_rationale": "reachable", + "static_hint": 0, + "semgrep_hint": 0, + "taint_hits": 0, + "imports_by": 0, + "transitive_callers": 0, + "defines_constants": False, + "has_fuzz_entry_point": False, + "fuzz_harness_path": None, + } + target.update(overrides) + return target + + +def _prepared_store(tmp_path: Path): + repo = tmp_path / "repo" + repo.mkdir() + (repo / "app.py").write_text("print('safe')\n", encoding="utf-8") + target = _target(repo) + invocation = fingerprint_invocation({"depth": "standard"}) + source = fingerprint_source(repo, [target]) + store = SourceHuntCheckpoint(tmp_path / "out" / "sh-test", resuming=False) + assert ( + store.prepare( + invocation_fingerprint=invocation, + source_fingerprint=source, + source_paths=["app.py"], + ) + is None + ) + return repo, target, invocation, source, store + + +def test_checkpoint_restores_only_a_complete_compatible_rank_plan(tmp_path): + repo, target, invocation, source, store = _prepared_store(tmp_path) + store.save_rank_plan([target]) + + resumed = SourceHuntCheckpoint(store.session_dir, resuming=True) + plan = resumed.prepare( + invocation_fingerprint=invocation, + source_fingerprint=source, + source_paths=["app.py"], + ) + assert plan is not None + fresh_target = _target(repo, surface=0, priority=0.0, surface_rationale="") + resumed.apply_rank_plan([fresh_target], plan) + assert fresh_target["surface"] == 4 + assert fresh_target["priority"] == 3.7 + + payload = json.loads(store.path.read_text(encoding="utf-8")) + payload["rank_plan"] = [{"path": "app.py"}] + store.path.write_text(json.dumps(payload), encoding="utf-8") + assert ( + SourceHuntCheckpoint(store.session_dir, resuming=True).prepare( + invocation_fingerprint=invocation, + source_fingerprint=source, + source_paths=["app.py"], + ) + is None + ) + + +def test_checkpoint_rejects_changed_options_or_source(tmp_path): + repo, target, invocation, source, store = _prepared_store(tmp_path) + with pytest.raises(SourceHuntResumeError, match="options"): + SourceHuntCheckpoint(store.session_dir, resuming=True).prepare( + invocation_fingerprint=fingerprint_invocation({"depth": "deep"}), + source_fingerprint=source, + source_paths=["app.py"], + ) + + (repo / "app.py").write_text("print('changed')\n", encoding="utf-8") + with pytest.raises(SourceHuntResumeError, match="source inputs changed"): + SourceHuntCheckpoint(store.session_dir, resuming=True).prepare( + invocation_fingerprint=invocation, + source_fingerprint=fingerprint_source(repo, [target]), + source_paths=["app.py"], + ) + + +def test_completed_work_round_trips_and_corruption_is_a_cache_miss(tmp_path): + _repo, _target_value, _invocation, _source, store = _prepared_store(tmp_path) + work_id = "work-" + "a" * 16 + result = TargetResult( + target="app.py", + status="completed", + findings=[ + Finding( + id="finding-1", + file="app.py", + line_number=1, + severity="high", + description="unsafe input", + ) + ], + cost_usd=0.25, + tokens_used=20, + tier="A", + band="fast", + ) + store.save(work_id, result) + + restored = store.load(work_id) + assert restored is not None + assert restored.cost_usd == 0.25 + assert isinstance(restored.findings[0], Finding) + + zero_id = "work-" + "b" * 16 + store.save( + zero_id, + TargetResult( + target="app.py", status="completed", findings=[], tier="A", band="fast" + ), + ) + assert store.load(zero_id) is not None + + (store.work_dir / f"{work_id}.json").write_text('{"result":', encoding="utf-8") + assert store.load(work_id) is None + + +def test_session_ids_and_concurrent_writers_are_rejected(tmp_path): + with pytest.raises(SourceHuntResumeError, match="Invalid"): + session_directory(tmp_path, "../escape") + + session = tmp_path / "sh-test" + first = SourceHuntSessionLock(session) + second = SourceHuntSessionLock(session) + first.acquire() + try: + with pytest.raises(SourceHuntResumeError, match="already running"): + second.acquire() + finally: + first.release() + + +class _MemoryCheckpoint: + def __init__(self, values=None): + self.values = dict(values or {}) + self.loaded = [] + self.saved = [] + + def load(self, work_id): + self.loaded.append(work_id) + return self.values.get(work_id) + + def save(self, work_id, result): + self.saved.append((work_id, result)) + self.values[work_id] = result + + +@pytest.mark.asyncio +async def test_cached_findings_and_promotions_follow_the_normal_pool_path(tmp_path): + repo = tmp_path / "repo" + repo.mkdir() + target = _target(repo) + finding = Finding( + id="cached-finding", + file="app.py", + line_number=1, + severity="high", + description="cached bug", + ) + base_result = TargetResult( + target="app.py", + status="completed", + findings=[finding], + cost_usd=1.0, + tier="A", + band="fast", + stop_reason="completed", + ) + base = WorkItem(target, "fast") + promoted = WorkItem( + target, + "standard", + seed_transcript=_extract_transcript(base_result), + ) + checkpoint = _MemoryCheckpoint( + { + base.stable_identifier("sh-test", "A"): base_result, + promoted.stable_identifier("sh-test", "A"): TargetResult( + target="app.py", + status="completed", + findings=[], + tier="A", + band="standard", + ), + } + ) + findings_pool = FindingsPool() + pool = HunterPool( + HuntPoolConfig( + files=[target], + repo_path=str(repo), + checkpoint=checkpoint, + findings_pool=findings_pool, + session_id_prefix="sh-test", + redundancy_override=1, + max_parallel=1, + starting_band="fast", + max_band="standard", + ) + ) + pool._run_file_task = AsyncMock(side_effect=AssertionError("cached work reran")) + + findings = await pool.arun() + + pool._run_file_task.assert_not_awaited() + assert [item.id for item in findings] == ["cached-finding"] + assert pool.promotion_counts == {"fast→standard": 1, "standard→deep": 0} + assert pool.total_spent == 1.0 + + +@pytest.mark.asyncio +async def test_findings_pool_replay_is_idempotent(tmp_path): + path = tmp_path / "findings-pool.jsonl" + original = Finding( + id="cached-finding", + file="app.py", + line_number=1, + finding_type="command_injection", + severity="critical", + description="cached bug", + ) + pool = FindingsPool(checkpoint_path=path) + await pool.add(original) + cluster_id = original.cluster_id + + resumed = FindingsPool.from_checkpoint(path) + restored = await resumed.add( + Finding( + id="cached-finding", + file="app.py", + line_number=1, + finding_type="command_injection", + severity="critical", + description="cached bug", + ) + ) + + assert resumed.count == 1 + assert restored.cluster_id == cluster_id + assert len(path.read_text(encoding="utf-8").splitlines()) == 1 + + +@pytest.mark.asyncio +async def test_missing_work_runs_and_is_saved_even_with_zero_findings(tmp_path): + repo = tmp_path / "repo" + repo.mkdir() + target = _target(repo) + checkpoint = _MemoryCheckpoint() + pool = HunterPool( + HuntPoolConfig( + files=[target], + repo_path=str(repo), + checkpoint=checkpoint, + session_id_prefix="sh-test", + redundancy_override=1, + max_parallel=1, + starting_band="fast", + max_band="fast", + ) + ) + pool._run_file_task = AsyncMock( + return_value=TargetResult( + target="app.py", + status="completed", + findings=[], + cost_usd=0.2, + tier="A", + band="fast", + ) + ) + + assert await pool.arun() == [] + pool._run_file_task.assert_awaited_once() + assert len(checkpoint.saved) == 1 + assert checkpoint.saved[0][1].findings == [] + + +@pytest.mark.asyncio +async def test_mismatched_cached_work_is_a_cache_miss(tmp_path): + target = _target(tmp_path) + item = WorkItem(target, "fast") + checkpoint = _MemoryCheckpoint( + { + item.stable_identifier("sh-test", "A"): TargetResult( + target="other.py", + status="completed", + findings=[], + tier="A", + band="fast", + ) + } + ) + pool = HunterPool( + HuntPoolConfig( + files=[target], + repo_path=str(tmp_path), + checkpoint=checkpoint, + session_id_prefix="sh-test", + redundancy_override=1, + max_parallel=1, + starting_band="fast", + max_band="fast", + ) + ) + pool._run_file_task = AsyncMock( + return_value=TargetResult( + target="app.py", status="completed", findings=[], tier="A", band="fast" + ) + ) + + await pool.arun() + + pool._run_file_task.assert_awaited_once() + + +def _ledger(tmp_path: Path, session_id: str, *, resume: bool = False) -> SpendLedger: + return SpendLedger( + limit_usd=10.0, + session_id=session_id, + repo_url="repo", + output_dir=tmp_path, + input_price_per_million=0.0, + output_price_per_million=1_000_000.0, + resume=resume, + ) + + +def test_spend_resume_restores_settlements_and_ignores_a_truncated_tail(tmp_path): + ledger = _ledger(tmp_path, "settled") + reservation = ledger.reserve_call( + model="test", + provider="test", + stage="hunt", + input_token_upper_bound=0, + requested_max_output_tokens=2, + supports_output_limit=True, + ) + ledger.settle_call(reservation, input_tokens=2, output_tokens=1) + ledger.finalize("failed") + with ledger.ledger_path.open("a", encoding="utf-8") as stream: + stream.write('{"event":') + + resumed = _ledger(tmp_path, "settled", resume=True) + assert resumed.spent_usd == pytest.approx(1.0) + assert resumed.snapshot()["total_tokens"] == 3 + + +def test_spend_resume_charges_an_orphaned_reservation_only_once(tmp_path): + ledger = _ledger(tmp_path, "orphan") + reservation = ledger.reserve_call( + model="test", + provider="test", + stage="hunt", + input_token_upper_bound=0, + requested_max_output_tokens=2, + supports_output_limit=True, + ) + first_resume = _ledger(tmp_path, "orphan", resume=True) + assert first_resume.spent_usd == pytest.approx(reservation.reserved_usd) + first_resume.finalize("failed") + + second_resume = _ledger(tmp_path, "orphan", resume=True) + assert second_resume.spent_usd == pytest.approx(reservation.reserved_usd) + + +def test_spend_resume_refuses_a_missing_or_corrupt_ledger(tmp_path): + with pytest.raises(BudgetConfigurationError, match="ledger is missing"): + _ledger(tmp_path, "missing", resume=True) + + ledger = _ledger(tmp_path, "corrupt") + ledger.ledger_path.write_text("not-json\n{}\n", encoding="utf-8") + with pytest.raises(BudgetConfigurationError, match="ledger is corrupt"): + _ledger(tmp_path, "corrupt", resume=True) + + +def test_work_ids_distinguish_overloaded_entry_points(tmp_path): + target = _target(tmp_path) + first = WorkItem( + target, + "fast", + entry_point=EntryPoint("app.py", "parse", 1, 10, "parser", "first"), + ) + second = WorkItem( + target, + "fast", + entry_point=EntryPoint("app.py", "parse", 20, 30, "parser", "second"), + ) + + assert first.stable_identifier("sh-test", "A") != second.stable_identifier( + "sh-test", "A" + ) + + +def test_runner_reruns_preprocessing_and_restores_rank_plan(tmp_path, monkeypatch): + repo = tmp_path / "repo" + repo.mkdir() + (repo / "app.py").write_text("print('safe')\n", encoding="utf-8") + output = tmp_path / "out" + preprocess_calls = 0 + rank_calls = 0 + + def preprocess(): + nonlocal preprocess_calls + preprocess_calls += 1 + return PreprocessResult( + repo_path=str(repo), + file_targets=[_target(repo, surface=0, influence=0, priority=0.0)], + static_findings=[], + ) + + async def rank(_self, files): + nonlocal rank_calls + rank_calls += 1 + files[0].update( + surface=5, + influence=4, + priority=4.2, + surface_rationale="paid rank", + influence_rationale="paid rank", + ) + return files + + monkeypatch.setattr("clearwing.sourcehunt.runner.Ranker.arank", rank) + common = { + "repo_url": str(repo), + "local_path": str(repo), + "depth": "quick", + "output_dir": str(output), + "ranker_llm": AsyncMock(provider_name="test"), + "enable_mechanism_memory": False, + "enable_knowledge_graph": False, + } + first = SourceHuntRunner(**common) + first._preprocess = preprocess + first.run() + assert rank_calls == 1 + + resumed = SourceHuntRunner(**common, resume_session_id=first.session_id) + resumed._preprocess = preprocess + resumed.run() + assert preprocess_calls == 2 + assert rank_calls == 1 + + (repo / "app.py").write_text("print('changed')\n", encoding="utf-8") + incompatible = SourceHuntRunner(**common, resume_session_id=first.session_id) + incompatible._preprocess = preprocess + with pytest.raises(SourceHuntResumeError, match="source inputs changed"): + incompatible.run() + + +def test_invocation_compatibility_excludes_models_and_replayed_later_stages(tmp_path): + common = { + "repo_url": "repo", + "output_dir": str(tmp_path), + "enable_calibration": False, + "enable_mechanism_memory": False, + } + original = SourceHuntRunner(**common, model_override="model-a") + replacement = SourceHuntRunner( + **common, + model_override="model-b", + no_verify=True, + no_exploit=True, + ) + changed_hunt = SourceHuntRunner(**common, prompt_mode="specialist") + changed_parallelism = SourceHuntRunner(**common, max_parallel=2) + + assert original._invocation_fingerprint() == replacement._invocation_fingerprint() + assert original._invocation_fingerprint() != changed_hunt._invocation_fingerprint() + assert original._invocation_fingerprint() != changed_parallelism._invocation_fingerprint() + + +def test_cli_keeps_repo_required_and_accepts_normal_options_with_resume(): + parser = argparse.ArgumentParser() + subparsers = parser.add_subparsers(dest="command") + sourcehunt_command.add_parser(subparsers) + args = parser.parse_args( + ["sourcehunt", "repo", "--resume", "sh-test", "--no-verify", "--depth", "deep"] + ) + assert args.repo == "repo" + assert args.resume == "sh-test" + assert args.no_verify is True + assert args.depth == "deep" + + missing_repo = parser.parse_args(["sourcehunt", "--resume", "sh-test"]) + with pytest.raises(SystemExit): + sourcehunt_command.handle(None, missing_repo)