diff --git a/CHANGELOG.md b/CHANGELOG.md index b0c031fc..31902144 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -58,6 +58,26 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 upstream's `StrictImprovementAcceptance`. ### Fixed +- Backend token usage is no longer discarded when the backend's output fails + to parse. `budget.charge_llm_usage` was reachable only from the success + paths of `mutate` / `merge`, so a `MutationError` raised while parsing + threw away the attempt's accounting along with its candidate. Usage is now + recovered from raw stdout by a parse that cannot raise, carried out on + `HelixError.usage`, and charged in the apply phase with + `source="mutation_failed"` / `"merge_failed"`. Measured on one 20-generation + run, a single parse failure left the budget 7.2% low on input tokens and + 5.0% low on output tokens, silently. Failed invocations also now record the + recovered usage in `.helix_backend_result.json` instead of zeros, and the + error report prints it. +- Newline-delimited backend and evaluator output is now split on U+000A only. + `str.splitlines()` also breaks on U+0085, U+2028 and U+2029, which JSON + leaves unescaped inside string literals and which agent backends therefore + emit verbatim, fragmenting one valid JSONL record into two invalid ones. + This was the root cause of the parse failure above. Applied via the new + `helix.lines.split_lf_lines` to the JSONL parsers, the per-backend + transcript tool-event counters, the `HELIX_RESULT=` scans, and the JSONL + dataset readers; human-authored and git-generated text still uses + `splitlines()`. - `ParetoFrontier.select_parent()` and `ParetoFrontier.get_non_dominated()` are now reproducible across processes for a given seed. Candidate ids are `str` and the per-key fronts are `set` objects, so set iteration order — diff --git a/src/helix/asi.py b/src/helix/asi.py index 123e00d9..8443a471 100644 --- a/src/helix/asi.py +++ b/src/helix/asi.py @@ -71,6 +71,11 @@ def _render_field_value(value: Any) -> str: def read_text(raw: str) -> str: """Render raw HELIX ASI log text.""" lines: list[str] = [] + # ``splitlines()`` (not ``helix.lines.split_lf_lines``) is safe here and + # only here among HELIX's JSONL readers: ``log`` above writes every record + # with ``json.dumps``'s default ``ensure_ascii=True``, so this file is pure + # ASCII by construction and cannot contain a U+0085 / U+2028 / U+2029 that + # would fragment a record. for raw_line in raw.splitlines(): line = raw_line.strip() if not line: diff --git a/src/helix/config.py b/src/helix/config.py index 057e70cb..ddebe807 100644 --- a/src/helix/config.py +++ b/src/helix/config.py @@ -18,6 +18,7 @@ EFFORT_VALID_VALUES, backend_display_name, ) +from helix.lines import split_lf_lines def _load_dotenv_file(path: Path) -> None: @@ -272,7 +273,7 @@ def load_dataset_examples(train_path: Path, max_examples: int = 3) -> list[str]: ) else: # Treat as JSONL — one JSON object per non-blank line. - for line in raw.splitlines(): + for line in split_lf_lines(raw): line = line.strip() if line: items.append(json.loads(line)) diff --git a/src/helix/evolution.py b/src/helix/evolution.py index 147db9ae..a7926e21 100644 --- a/src/helix/evolution.py +++ b/src/helix/evolution.py @@ -56,6 +56,7 @@ ) from helix.executor import run_evaluator from helix.lineage import LineageEntry, find_merge_triplet, load_lineage, record_entry +from helix.lines import split_lf_lines from helix.merger import merge, select_eval_subsample_for_merged_program from helix.mutator import mutate, build_seed_generation_prompt, generate_seed from helix.proposals import ( @@ -697,7 +698,7 @@ def _load_dataset_ids(path: Path) -> list[str]: return [str(i) for i in range(len(data))] # JSONL count = 0 - for line in raw.splitlines(): + for line in split_lf_lines(raw): if line.strip(): count += 1 return [str(i) for i in range(count)] @@ -1543,6 +1544,11 @@ def _run_proposal_worker( ) # ---- Step W4: LLM mutation ---- + # ``mutate`` reports what the backend spent through this sink whether or + # not it returns a candidate, so a failed slot still carries its usage + # into the sequential apply phase for charging. The list is worker-local + # — no shared state is touched here, per the thread-safety contract above. + _spent_usage: list[UsageStats] = [] try: _child = mutate( parent=_parent, @@ -1556,6 +1562,7 @@ def _run_proposal_worker( cand, config, project_root ) ), + record_usage=_spent_usage.append, ) except Exception as _mu_exc: # Re-raise PromptArtifactCollisionError (fatal for the whole run) @@ -1588,6 +1595,7 @@ def _run_proposal_worker( presample_ctx=pre_ctx, parent_eval_result=_parent_eval, parent_n_uncached=_parent_n_uncached, + child_usage=_spent_usage[-1] if _spent_usage else None, ) if _child is None: @@ -1595,6 +1603,7 @@ def _run_proposal_worker( presample_ctx=pre_ctx, parent_eval_result=_parent_eval, parent_n_uncached=_parent_n_uncached, + child_usage=_spent_usage[-1] if _spent_usage else None, ) # ---- Step W5: Tamper check ---- @@ -2380,6 +2389,7 @@ def _has_val_support_overlap(i: str, j: str) -> bool: f"diff form for this merge." ) + merge_usage: list[UsageStats] = [] merged = merge( candidate_a=a, candidate_b=b, @@ -2395,11 +2405,21 @@ def _has_val_support_overlap(i: str, j: str) -> bool: ) ), ancestor=ancestor_candidate, + record_usage=merge_usage.append, ) if merged is None: # GEPA parity (M2/B3): merge operator failed before # any eval; no attempt, fall through to mutation. + # The failed merge still spent tokens — charge them. + if merge_usage: + live.update(usage=merge_usage[-1]) + budget_api.charge_llm_usage( + state, + merge_usage[-1], + candidate_id=merge_id, + source="merge_failed", + ) print_error( f"Merge {merge_id} failed " f"(candidates: {a.id} + {b.id}, gen {gen}). " @@ -2803,6 +2823,17 @@ def _gate_proposal( split="train", source="parent_minibatch", ) + # Charge LLM usage: the mutation failed, but its tokens + # were still spent. Skipping this is how a run's budget + # silently under-reports by a whole generation. + if wr.child_usage: + live.update(usage=wr.child_usage) + budget_api.charge_llm_usage( + state, + wr.child_usage, + candidate_id=_new_id, + source="mutation_failed", + ) print_warning(f"Mutation {_new_id} failed -- skipping.") return None diff --git a/src/helix/exceptions.py b/src/helix/exceptions.py index 8f733094..9e026a37 100644 --- a/src/helix/exceptions.py +++ b/src/helix/exceptions.py @@ -6,10 +6,14 @@ from __future__ import annotations import logging +from typing import TYPE_CHECKING from rich.console import Console from rich.panel import Panel +if TYPE_CHECKING: + from helix.display import UsageStats + # --------------------------------------------------------------------------- # Module-level Rich console for error output @@ -30,6 +34,14 @@ class HelixError(Exception): Carries structured context (operation, phase, command, cwd, stdout, stderr, exit_code, suggestion) so callers can emit rich diagnostics without parsing exception messages. + + ``usage`` carries the token usage an LLM backend had already spent when + the failure happened, recovered from the raw output independently of + whatever parse failed. Tokens are spent whether or not a usable + candidate came out, so an error that discards a candidate must not also + discard the accounting: callers charge ``usage`` to the budget on the + failure path. ``None`` means "no backend ran, or nothing was + recoverable" and is not the same as a zero-token ``UsageStats``. """ def __init__( @@ -44,6 +56,7 @@ def __init__( stderr: str = "", exit_code: int | None = None, suggestion: str = "", + usage: UsageStats | None = None, ) -> None: self.operation = operation self.phase = phase @@ -53,6 +66,7 @@ def __init__( self.stderr = stderr self.exit_code = exit_code self.suggestion = suggestion + self.usage = usage super().__init__(message) def format_full(self) -> str: @@ -68,6 +82,8 @@ def format_full(self) -> str: lines.append(f"[HELIX ERROR] Exit code: {self.exit_code}") if self.cwd: lines.append(f"[HELIX ERROR] Working dir: {self.cwd}") + if self.usage is not None: + lines.append(f"[HELIX ERROR] Backend usage: {self._usage_summary()}") if self.stdout: lines.append(f"[HELIX ERROR] Stdout:\n{self.stdout}") if self.stderr: @@ -76,6 +92,24 @@ def format_full(self) -> str: lines.append(f"[HELIX ERROR] Suggestion: {self.suggestion}") return "\n".join(lines) + def _usage_summary(self) -> str: + """One-line rendering of ``usage`` for the error report. + + Spelled out rather than dumped as a dict so that a failure whose + tokens were nonetheless charged says so plainly in the log — the + under-reporting this field exists to prevent was invisible precisely + because nothing printed it. + """ + if self.usage is None: # pragma: no cover - guarded by the caller + return "unavailable" + return ( + f"input={self.usage.input_tokens} " + f"output={self.usage.output_tokens} " + f"cached_input={self.usage.cached_input_tokens} " + f"reasoning={self.usage.reasoning_tokens} " + "(charged to the budget despite this failure)" + ) + class GitError(HelixError): """Raised when a git subprocess fails.""" @@ -144,6 +178,8 @@ def print_helix_error(exc: HelixError) -> None: lines.append(f"[red]Exit code:[/red] {exc.exit_code}") if exc.cwd: lines.append(f"[red]Working dir:[/red] {exc.cwd}") + if exc.usage is not None: + lines.append(f"[red]Backend usage:[/red] {exc._usage_summary()}") if exc.stdout: lines.append(f"[red]Stdout (full):[/red]\n{exc.stdout}") if exc.stderr: diff --git a/src/helix/executor.py b/src/helix/executor.py index e4a378f9..4f532e04 100644 --- a/src/helix/executor.py +++ b/src/helix/executor.py @@ -18,6 +18,7 @@ from helix.population import Candidate, EvalResult from helix.config import HelixConfig from helix.exceptions import EvaluatorError, format_error_context +from helix.lines import split_lf_lines from helix.parsers.helix_result import parse as parse_helix_result from helix.sandbox import ( current_evaluator_sidecar_runtime, @@ -367,7 +368,7 @@ def run_evaluator( # missing batch file, etc.). The ``helix_result`` parser does its # own reverse-scan before the parser runs. result_line_count = 0 - for line in reversed(stdout.splitlines()): + for line in reversed(split_lf_lines(stdout)): if line.startswith("HELIX_RESULT="): result_line_count += 1 if result_line_count > 1: diff --git a/src/helix/lines.py b/src/helix/lines.py new file mode 100644 index 00000000..c452854b --- /dev/null +++ b/src/helix/lines.py @@ -0,0 +1,58 @@ +"""Line splitting for newline-delimited machine formats. + +Why this exists +--------------- +``str.splitlines()`` is the wrong tool for every format whose grammar says +"records are separated by a line feed". Python splits on eight boundaries a +line feed is not:: + + \\v U+000B LINE TABULATION \\f U+000C FORM FEED + \\x1c U+001C FILE SEPARATOR \\x1d U+001D GROUP SEPARATOR + \\x1e U+001E RECORD SEPARATOR \\x85 U+0085 NEXT LINE (NEL) + \\u2028 LINE SEPARATOR \\u2029 PARAGRAPH SEPARATOR + +JSON permits U+0085, U+2028 and U+2029 unescaped inside a string literal — +they are not C0 control characters, so RFC 8259 does not require escaping — +and the serializers the agent backends use do not escape them (Rust's +``serde_json`` for Codex, ``JSON.stringify`` for the Node CLIs). The +remaining five are escaped as ``\\uXXXX`` by any conforming serializer, but +arrive verbatim in transcripts and evaluator streams that are not JSON. + +The consequence for a JSONL stream is that one agent message, or one blob of +captured command output, containing a single NEL byte turns a valid record +into two invalid fragments. Downstream that reads either as a hard parse +failure or as a silently wrong record count. + +Use :func:`split_lf_lines` wherever the line boundary is defined by the +format (JSONL, a ``KEY=`` protocol line on stdout, a transcript file). Keep +``str.splitlines()`` for human-authored text where breaking on any Unicode +line boundary is the intent. +""" + +from __future__ import annotations + + +__all__ = ["split_lf_lines"] + + +def split_lf_lines(text: str) -> list[str]: + """Split *text* on U+000A only, the way a newline-delimited format defines it. + + Unlike ``str.splitlines()`` this never breaks a record on an embedded + Unicode line boundary such as U+0085 / U+2028 / U+2029, and it never + silently rewrites one of those characters into a newline when the parts + are rejoined. + + Callers that also need to tolerate CRLF should ``strip()`` each line, as + the JSONL parsers here do; ``json.loads`` ignores trailing whitespace in + any case. + + Args: + text: The raw stream or file contents to split. + + Returns: + The line feed separated pieces of *text*. A trailing line feed + yields a final empty string, which callers skip along with any other + blank line. + """ + return text.split("\n") diff --git a/src/helix/merger.py b/src/helix/merger.py index 2d7847b2..2745b39b 100644 --- a/src/helix/merger.py +++ b/src/helix/merger.py @@ -7,6 +7,7 @@ from pathlib import Path from typing import Callable, Mapping +from helix.display import UsageStats from helix.population import Candidate, EvalResult from helix.config import HelixConfig from helix.worktree import clone_candidate, snapshot_candidate, remove_worktree, get_diff # noqa: F401 @@ -233,6 +234,7 @@ def merge( eval_result_b: EvalResult | None = None, prepare_worktree: Callable[[Candidate], None] | None = None, ancestor: Candidate | None = None, + record_usage: Callable[[UsageStats], None] | None = None, ) -> Candidate | None: """Merge *candidate_a* and *candidate_b* using Claude Code. @@ -297,6 +299,10 @@ def merge( from :func:`helix.lineage.find_merge_triplet`). When supplied, the prompt uses the two-diff (ancestor-relative) form; when ``None``, falls back to the single A↔B diff. + record_usage: + Optional sink called exactly once with the backend's token usage, + whether or not the merge produced a usable candidate — the same + contract as :func:`helix.mutator.mutate`'s parameter of that name. Returns ------- @@ -346,7 +352,13 @@ def merge( sandbox=config.sandbox, ) child.usage = usage + if record_usage is not None: + record_usage(usage) except MutationError as exc: + # Tokens spent before the failure are still spent; hand them to the + # caller before the candidate and its worktree go away. + if record_usage is not None and exc.usage is not None: + record_usage(exc.usage) exc.operation = f"merge {new_id} ({candidate_a.id} + {candidate_b.id})" print_helix_error(exc) try: @@ -354,8 +366,10 @@ def merge( except Exception: pass return None - except RateLimitError: + except RateLimitError as exc: # Rate limit — clean up orphaned worktree, then re-raise. + if record_usage is not None and exc.usage is not None: + record_usage(exc.usage) try: remove_worktree(child) except Exception: diff --git a/src/helix/mutator.py b/src/helix/mutator.py index 4f53c202..ba5d3566 100644 --- a/src/helix/mutator.py +++ b/src/helix/mutator.py @@ -15,12 +15,14 @@ from helix.population import Candidate, EvalResult from helix.config import AgentConfig, HelixConfig, SandboxConfig from helix.exceptions import ( + HelixError, MutationError, PromptArtifactCollisionError, RateLimitError, print_helix_error, ) from helix.executor import _scrub_environment +from helix.lines import split_lf_lines from helix.sandbox import resolve_sandbox_image, run_sandboxed_command from helix.worktree import clone_candidate, snapshot_candidate, remove_worktree # noqa: F401 @@ -133,7 +135,7 @@ def _strip_machine_protocol_from_evaluator_stream(text: str) -> str: return "" kept: list[str] = [] - for line in text.splitlines(): + for line in split_lf_lines(text): if line.strip().startswith("HELIX_RESULT="): continue kept.append(line) @@ -889,7 +891,7 @@ def _parse_jsonl_output( ) -> dict[str, Any]: events: list[dict[str, Any]] = [] unparsable: list[str] = [] - for raw_line in stdout.splitlines(): + for raw_line in split_lf_lines(stdout): line = raw_line.strip() if not line: continue @@ -956,6 +958,63 @@ def _parse_backend_output( raise ValueError(f"Unsupported backend: {backend}") +def _salvage_backend_usage( + backend: str, result: subprocess.CompletedProcess[str] +) -> UsageStats: + """Recover token usage from raw backend output without ever raising. + + Token usage is a fact about work the backend has already done. It must + therefore be recoverable independently of whether the output *also* + yields a usable candidate — the strict parsers above raise + :class:`MutationError` on a single malformed line, and until this + existed that error discarded a whole invocation's accounting along with + the candidate. + + The recovery is deliberately lenient and total: + + * JSONL backends reuse :func:`_parse_jsonl_output` with ``strict=False``, + which collects the records that *did* decode and sets the rest aside. + The usage record is one line of its own, so a malformed line elsewhere + in the stream does not hide it. + * Claude's single-object mode tries the object first, then falls back to + the same lenient line scan for a stream that was truncated mid-object. + + Returns a zero-token :class:`UsageStats` when nothing is recoverable, + which is the honest reading of "the backend reported no usage". + """ + stdout = result.stdout or "" + if not stdout.strip(): + return _normalise_usage_stats({}) + + parsed: dict[str, Any] | None = None + if backend == "claude": + try: + loaded = json.loads(stdout) + except (json.JSONDecodeError, ValueError, RecursionError): + loaded = None + if isinstance(loaded, dict): + parsed = loaded + + if parsed is None: + try: + parsed = _parse_jsonl_output( + stdout, + backend=backend, + cmd_str="", + worktree_path="", + stderr="", + exit_code=0, + strict=False, + ) + except (ValueError, RecursionError): # pragma: no cover - defensive + return _normalise_usage_stats({}) + + try: + return _normalise_usage_stats(parsed) + except (ValueError, TypeError, RecursionError): # pragma: no cover + return _normalise_usage_stats({}) + + def _walk_json(obj: Any) -> list[dict[str, Any]]: found: list[dict[str, Any]] = [] if isinstance(obj, dict): @@ -1183,7 +1242,7 @@ def _count_claude_transcript_tool_events(path: Path) -> tuple[int, list[str]]: count = 0 names: list[str] = [] try: - for raw in path.read_text(encoding="utf-8").splitlines(): + for raw in split_lf_lines(path.read_text(encoding="utf-8")): raw = raw.strip() if not raw: continue @@ -1218,7 +1277,7 @@ def _count_codex_stdout_tool_events(path: Path) -> tuple[int, list[str]]: count = 0 names: list[str] = [] try: - for raw in path.read_text(encoding="utf-8").splitlines(): + for raw in split_lf_lines(path.read_text(encoding="utf-8")): raw = raw.strip() if not raw: continue @@ -1264,7 +1323,7 @@ def _count_cursor_stdout_tool_events(path: Path) -> tuple[int, list[str]]: "grepToolCall": "grep", } try: - for raw in path.read_text(encoding="utf-8").splitlines(): + for raw in split_lf_lines(path.read_text(encoding="utf-8")): raw = raw.strip() if not raw: continue @@ -1301,7 +1360,7 @@ def _count_gemini_stdout_tool_events(path: Path) -> tuple[int, list[str]]: count = 0 names: list[str] = [] try: - for raw in path.read_text(encoding="utf-8").splitlines(): + for raw in split_lf_lines(path.read_text(encoding="utf-8")): raw = raw.strip() if not raw: continue @@ -1330,7 +1389,7 @@ def _count_opencode_stdout_tool_events(path: Path) -> tuple[int, list[str]]: count = 0 names: list[str] = [] try: - for raw in path.read_text(encoding="utf-8").splitlines(): + for raw in split_lf_lines(path.read_text(encoding="utf-8")): raw = raw.strip() if not raw: continue @@ -1502,13 +1561,22 @@ def _write_backend_artifacts( result: subprocess.CompletedProcess[str], parsed: dict[str, Any] | None, sandbox: SandboxConfig | None = None, + fallback_usage: UsageStats | None = None, ) -> None: try: wt = Path(worktree_path) _ignore_helix_artifacts(wt) (wt / BACKEND_STDOUT_ARTIFACT_NAME).write_text(result.stdout or "") (wt / BACKEND_STDERR_ARTIFACT_NAME).write_text(result.stderr or "") - usage = _normalise_usage_stats(parsed or {}) + # ``parsed is None`` means the strict parse failed. Record the + # leniently recovered usage rather than zeros, so the on-disk + # artifact stays a faithful account of what the invocation spent. + if parsed is not None: + usage = _normalise_usage_stats(parsed) + elif fallback_usage is not None: + usage = fallback_usage + else: + usage = _normalise_usage_stats({}) # For non-Claude backends the stdout JSONL IS the transcript; patch # ``usage`` with backend-specific tool-event counts now that the # stdout artifact is on disk. Claude is handled separately inside @@ -1649,6 +1717,13 @@ def invoke_claude_code( env=backend_env, ) + # Recover the usage record from the raw stream FIRST, with a parse that + # cannot raise. Everything below this line can fail — and when it does, + # the tokens have still been spent. Charging must not be conditional on + # the candidate being usable, so every error raised from here carries + # this on ``HelixError.usage`` for the caller to charge. + spent_usage = _salvage_backend_usage(backend, result) + parsed: dict[str, Any] | None = None try: if result.returncode == 0: @@ -1743,6 +1818,12 @@ def invoke_claude_code( exit_code=result.returncode, suggestion="Check stderr for rate limits, permission errors, or model availability.", ) + except HelixError as exc: + # Attach only when the raiser did not already supply a more precise + # record; never overwrite one. + if exc.usage is None: + exc.usage = spent_usage + raise finally: _write_backend_artifacts( worktree_path, @@ -1751,6 +1832,7 @@ def invoke_claude_code( result=result, parsed=parsed, sandbox=sandbox, + fallback_usage=spent_usage, ) @@ -1767,6 +1849,7 @@ def mutate( base_dir: Path, background: str | None = None, prepare_worktree: Callable[[Candidate], None] | None = None, + record_usage: Callable[[UsageStats], None] | None = None, ) -> Candidate | None: """Mutate *parent* using the configured backend and return the new candidate. @@ -1787,6 +1870,12 @@ def mutate( Base directory for worktrees. background: Optional background/context text injected into the prompt. + record_usage: + Optional sink called exactly once with the backend's token usage, + whether or not the mutation produced a usable candidate. The tokens + are spent either way, so this is how a caller charges the budget for + an attempt that ends in ``None``. Not called when no backend + invocation happened (e.g. the worktree clone raised). Returns ------- @@ -1825,7 +1914,13 @@ def mutate( prompt_artifact_name=prompt_artifact_name, ) child.usage = usage + if record_usage is not None: + record_usage(usage) except MutationError as exc: + # The worktree is about to be removed and the candidate dropped, but + # the tokens were spent. Hand them to the caller before both go. + if record_usage is not None and exc.usage is not None: + record_usage(exc.usage) exc.operation = f"mutate {new_id} (parent: {parent.id})" print_helix_error(exc) try: @@ -1833,10 +1928,13 @@ def mutate( except Exception: pass return None - except RateLimitError: + except RateLimitError as exc: # Rate limit — clean up orphaned worktree, then re-raise so the parallel # futures handler in evolution.py can log it and continue with a smaller - # proposal set. + # proposal set. A rate-limited invocation can still have burned tokens + # before the limit hit, so the same handoff applies. + if record_usage is not None and exc.usage is not None: + record_usage(exc.usage) try: remove_worktree(child) except Exception: diff --git a/src/helix/parsers/helix_result.py b/src/helix/parsers/helix_result.py index 20df943a..0d9e865a 100644 --- a/src/helix/parsers/helix_result.py +++ b/src/helix/parsers/helix_result.py @@ -84,6 +84,7 @@ from typing import Any from helix.exceptions import EvaluatorError +from helix.lines import split_lf_lines def _read_helix_batch(worktree_path: str | Path) -> list[str]: @@ -130,7 +131,7 @@ def _extract_helix_result_line(stdout: str) -> str | None: so the two agree on "which line wins" when (buggy) evaluators emit more than one. """ - for line in reversed(stdout.splitlines()): + for line in reversed(split_lf_lines(stdout)): if line.startswith("HELIX_RESULT="): return line return None diff --git a/src/helix/proposals.py b/src/helix/proposals.py index 9ee3e3ce..44443d00 100644 --- a/src/helix/proposals.py +++ b/src/helix/proposals.py @@ -42,11 +42,21 @@ class SkippedProposal: @dataclass class MutationFailedProposal: - """``mutate()`` raised or returned None; ``parent_eval_result`` may be None.""" + """``mutate()`` raised or returned None; ``parent_eval_result`` may be None. + + ``child_usage`` is the token usage the backend spent before it failed, + recovered from the raw output independently of the parse that failed. + It is charged in the apply phase exactly like a successful proposal's: + the tokens are gone either way, and leaving them uncharged silently + under-reports the run's cost by however much the failed attempt burned. + ``None`` means no backend invocation happened — a parent eval that + raised, or a worker that died before the LLM call. + """ presample_ctx: ProposalContext parent_eval_result: EvalResult | None parent_n_uncached: int = 0 + child_usage: UsageStats | None = None @dataclass diff --git a/src/helix/sandbox.py b/src/helix/sandbox.py index 58a388f7..e91eba0a 100644 --- a/src/helix/sandbox.py +++ b/src/helix/sandbox.py @@ -21,6 +21,7 @@ from helix.backends import BACKEND_AUTH_COMMANDS, DEFAULT_BACKEND_IMAGES from helix.config import EvaluatorSidecarConfig, SandboxConfig +from helix.lines import split_lf_lines logger = logging.getLogger(__name__) @@ -147,7 +148,7 @@ def _extract_session_id_from_json_output(stdout: str) -> str | None: try: payloads.append(json.loads(stdout)) except json.JSONDecodeError: - for raw_line in stdout.splitlines(): + for raw_line in split_lf_lines(stdout): line = raw_line.strip() if not line: continue diff --git a/tests/unit/test_budget.py b/tests/unit/test_budget.py index dfe5e445..b00d2663 100644 --- a/tests/unit/test_budget.py +++ b/tests/unit/test_budget.py @@ -19,7 +19,8 @@ SandboxConfig, ) from helix.display import UsageStats -from helix.mutator import invoke_claude_code +from helix.exceptions import MutationError +from helix.mutator import BACKEND_RESULT_ARTIFACT_NAME, invoke_claude_code from helix.state import BudgetState, EvolutionState from helix.trace import TRACE, EventType @@ -729,3 +730,188 @@ def test_evolution_counter_mutations_route_through_budget_api() -> None: ] for pattern in forbidden_patterns: assert re.search(pattern, source) is None + + +# --------------------------------------------------------------------------- +# Usage accounting must survive a backend-output parse failure +# --------------------------------------------------------------------------- +# +# Regression cover for a real under-report: a run lost one generation's +# entire usage because the backend's JSONL stream had one unparsable line +# and ``_parse_jsonl_output`` raised before anything charged the budget. +# The generation in question was the most expensive of its run, and nothing +# in the run output said its tokens had gone missing. + + +# Per-backend (malformed line, usage line) streams. In each the usage +# record is intact and a *different* line is unparsable, which is what the +# real failure looked like: the tokens were fully reported, the stream as a +# whole just would not parse. +_MALFORMED_STREAMS = { + "codex": "\n".join( + [ + '{"type":"thread.started","thread_id":"t1"}', + "{this line is not JSON", + ( + '{"type":"turn.completed","usage":{"input_tokens":2500000,' + '"cached_input_tokens":2400000,"output_tokens":8000,' + '"reasoning_output_tokens":3500}}' + ), + ] + ), + "cursor": "\n".join( + [ + "{truncated", + '{"type":"assistant","usage":{"inputTokens":13,"outputTokens":9,' + '"costUsd":0.33}}', + ] + ), + "opencode": "\n".join( + [ + "not json at all", + '{"type":"step_finish","part":{"tokens":{"input":15,"output":11},' + '"cost":0.35}}', + ] + ), +} + + +@pytest.mark.parametrize( + ("backend", "expected_input", "expected_output"), + [ + pytest.param("codex", 2500000, 8000, id="codex"), + pytest.param("cursor", 13, 9, id="cursor"), + pytest.param("opencode", 15, 11, id="opencode"), + ], +) +def test_malformed_backend_output_still_charges_llm_budget( + backend: str, + expected_input: int, + expected_output: int, + tmp_path: Path, + mocker, +) -> None: + """A parse failure must not take the token accounting down with it.""" + assert backend in BACKENDS + worktree = tmp_path / backend + worktree.mkdir() + mock_run = mocker.patch("helix.mutator.subprocess.run") + # Exit code 0 is the case that bites: it puts ``_parse_jsonl_output`` in + # strict mode, so one bad line raises instead of being set aside. + mock_run.return_value = subprocess.CompletedProcess( + args=[_BACKEND_EXECUTABLE[backend]], + returncode=0, + stdout=_MALFORMED_STREAMS[backend], + stderr="", + ) + state = make_state() + + with pytest.raises(MutationError) as excinfo: + invoke_claude_code( + str(worktree), + "read the prompt artifact", + AgentConfig(backend=backend), + ) + + # The error carries the usage recovered from the raw stream. + recovered = excinfo.value.usage + assert recovered is not None + assert recovered.input_tokens == expected_input + assert recovered.output_tokens == expected_output + + budget.charge_llm_usage( + state, + recovered, + candidate_id=f"{backend}-candidate", + source="mutation_failed", + ) + + assert state.budget.input_tokens == expected_input + assert state.budget.output_tokens == expected_output + + +def test_recovered_usage_matches_the_successful_parse_of_the_same_stream( + tmp_path: Path, mocker +) -> None: + """Recovery is lossless for the records that did decode. + + Charging a partial number would be its own quiet under-report, so the + salvage path is pinned against the strict path on an identical stream + that differs only by one appended junk line. + """ + worktree = tmp_path / "codex" + worktree.mkdir() + good = _MALFORMED_STREAMS["codex"].replace("{this line is not JSON\n", "") + mock_run = mocker.patch("helix.mutator.subprocess.run") + + mock_run.return_value = subprocess.CompletedProcess( + args=["codex"], returncode=0, stdout=good, stderr="" + ) + _parsed, clean_usage = invoke_claude_code( + str(worktree), "prompt", AgentConfig(backend="codex") + ) + + mock_run.return_value = subprocess.CompletedProcess( + args=["codex"], + returncode=0, + stdout=_MALFORMED_STREAMS["codex"], + stderr="", + ) + with pytest.raises(MutationError) as excinfo: + invoke_claude_code(str(worktree), "prompt", AgentConfig(backend="codex")) + + assert excinfo.value.usage == clean_usage + assert clean_usage.input_tokens == 2500000 + assert clean_usage.cached_input_tokens == 2400000 + assert clean_usage.reasoning_tokens == 3500 + + +def test_failed_invocation_records_recovered_usage_in_the_backend_artifact( + tmp_path: Path, mocker +) -> None: + """``.helix_backend_result.json`` must not report zeros for spent tokens.""" + worktree = tmp_path / "codex" + worktree.mkdir() + mocker.patch("helix.mutator.subprocess.run").return_value = ( + subprocess.CompletedProcess( + args=["codex"], + returncode=0, + stdout=_MALFORMED_STREAMS["codex"], + stderr="", + ) + ) + + with pytest.raises(MutationError): + invoke_claude_code(str(worktree), "prompt", AgentConfig(backend="codex")) + + artifact = json.loads( + (worktree / BACKEND_RESULT_ARTIFACT_NAME).read_text(encoding="utf-8") + ) + assert artifact["parsed"] is None + assert artifact["usage"]["input_tokens"] == 2500000 + assert artifact["usage"]["output_tokens"] == 8000 + + +def test_usage_is_none_when_no_backend_output_is_recoverable( + tmp_path: Path, mocker +) -> None: + """A zero-token charge and "nothing ran" must stay distinguishable. + + ``HelixError.usage`` is ``None`` only where no invocation produced a + usable record; a backend that ran and reported nothing yields a + zero-token ``UsageStats``, which ``charge_llm_usage`` still records as a + real (empty) backend response. + """ + worktree = tmp_path / "codex" + worktree.mkdir() + mocker.patch("helix.mutator.subprocess.run").return_value = ( + subprocess.CompletedProcess( + args=["codex"], returncode=0, stdout="{bad", stderr="" + ) + ) + + with pytest.raises(MutationError) as excinfo: + invoke_claude_code(str(worktree), "prompt", AgentConfig(backend="codex")) + + assert excinfo.value.usage == UsageStats() + assert MutationError("no backend ran").usage is None diff --git a/tests/unit/test_evolution.py b/tests/unit/test_evolution.py index 00dd80e7..5330e220 100644 --- a/tests/unit/test_evolution.py +++ b/tests/unit/test_evolution.py @@ -1082,6 +1082,82 @@ def run_eval(candidate, config, split=None, instances=None, **kwargs): assert state.budget.output_tokens == 13 assert state.budget.cost_usd == pytest.approx(0.42) + def test_failed_mutation_usage_charged_through_budget_api( + self, mocker, tmp_path, all_mocks + ): + """A mutation that produces no candidate must still charge its tokens. + + This is the shape of the measured under-report: the backend ran for + 47 minutes, its JSONL output failed to parse, ``mutate`` returned + ``None``, and the whole generation's usage never reached the budget. + """ + seed = make_candidate("g0-s0") + usage = UsageStats(input_tokens=2500000, output_tokens=8000, cost_usd=0.44) + all_mocks["create_seed_worktree"].return_value = seed + + def failing_mutate(*, record_usage=None, **kwargs): + # What ``mutate`` does when ``invoke_claude_code`` raises + # ``MutationError``: report the usage, return no candidate. + if record_usage is not None: + record_usage(usage) + return None + + all_mocks["mutate"].side_effect = failing_mutate + all_mocks["run_evaluator"].return_value = make_eval_result( + "g0-s0", {"i1": 0.3} + ) + spy = mocker.patch( + "helix.evolution.budget_api.charge_llm_usage", + wraps=budget_api.charge_llm_usage, + ) + + run_evolution( + make_config(max_generations=1, perfect_score_threshold=None), + tmp_path, + tmp_path / ".helix", + ) + + failed_calls = [ + call + for call in spy.call_args_list + if call.kwargs.get("source") == "mutation_failed" + ] + assert len(failed_calls) == 1, ( + "expected exactly one mutation_failed charge, " + f"got {len(failed_calls)}: {failed_calls!r}" + ) + state = failed_calls[0].args[0] + assert state.budget.input_tokens == 2500000 + assert state.budget.output_tokens == 8000 + assert state.budget.cost_usd == pytest.approx(0.44) + + def test_failed_mutation_without_usage_charges_nothing( + self, mocker, tmp_path, all_mocks + ): + """No backend invocation means no charge — not a zero-token charge.""" + seed = make_candidate("g0-s0") + all_mocks["create_seed_worktree"].return_value = seed + all_mocks["mutate"].return_value = None + all_mocks["run_evaluator"].return_value = make_eval_result( + "g0-s0", {"i1": 0.3} + ) + spy = mocker.patch( + "helix.evolution.budget_api.charge_llm_usage", + wraps=budget_api.charge_llm_usage, + ) + + run_evolution( + make_config(max_generations=1, perfect_score_threshold=None), + tmp_path, + tmp_path / ".helix", + ) + + assert [ + call + for call in spy.call_args_list + if call.kwargs.get("source") == "mutation_failed" + ] == [] + def test_merge_usage_charged_through_budget_api(self, mocker, tmp_path, all_mocks): seed = make_candidate("g0-s0") child = make_candidate("g1-s1", generation=1) diff --git a/tests/unit/test_lines.py b/tests/unit/test_lines.py new file mode 100644 index 00000000..7fe6e95f --- /dev/null +++ b/tests/unit/test_lines.py @@ -0,0 +1,166 @@ +"""Line splitting for newline-delimited machine formats. + +Regression cover for the U+0085 fragmentation that cost one production run a +whole generation's token accounting: Codex emitted a JSONL record whose +``command_execution`` output carried a raw NEL byte, ``str.splitlines()`` +broke the record into two invalid fragments, and the strict JSONL parse +raised on the first fragment. + +Every separator below is written as an escape on purpose — these characters +are invisible in an editor, and a test that depends on one has to say so. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +from helix.lines import split_lf_lines +from helix.mutator import ( + _count_codex_stdout_tool_events, + _parse_jsonl_output, + _strip_machine_protocol_from_evaluator_stream, +) +from helix.parsers.helix_result import _extract_helix_result_line + + +# Every boundary ``str.splitlines()`` treats as a line break but a +# newline-delimited format does not. +NON_LF_BREAKS = [ + "\v", # U+000B LINE TABULATION + "\f", # U+000C FORM FEED + "\x1c", # U+001C FILE SEPARATOR + "\x1d", # U+001D GROUP SEPARATOR + "\x1e", # U+001E RECORD SEPARATOR + "\x85", # U+0085 NEXT LINE (NEL) + "\u2028", # U+2028 LINE SEPARATOR + "\u2029", # U+2029 PARAGRAPH SEPARATOR +] + +# The subset a conforming JSON serializer leaves unescaped inside a string +# literal, and therefore the subset that actually reaches a JSONL parser. +JSON_UNESCAPED_BREAKS = ["\x85", "\u2028", "\u2029"] + +LINE_SEPARATOR = "\u2028" + + +def _nonblank(pieces: list[str]) -> list[str]: + """The pieces a JSONL parser would actually try to decode.""" + return [piece for piece in pieces if piece.strip()] + + +class TestSplitLfLines: + def test_splits_on_line_feed_only(self) -> None: + assert split_lf_lines("a\nb\nc") == ["a", "b", "c"] + + def test_trailing_line_feed_yields_trailing_empty_piece(self) -> None: + assert split_lf_lines("a\n") == ["a", ""] + + def test_no_break_on_any_non_lf_unicode_boundary(self) -> None: + for ch in NON_LF_BREAKS: + text = f"left{ch}right" + assert len(text.splitlines()) == 2, f"precondition failed for {ch!r}" + assert split_lf_lines(text) == [text] + + def test_carriage_return_is_left_for_the_caller_to_strip(self) -> None: + assert split_lf_lines("a\r\nb") == ["a\r", "b"] + assert [piece.strip() for piece in split_lf_lines("a\r\nb")] == ["a", "b"] + + +def _codex_stdout(ch: str) -> str: + """A two-record Codex stream whose first record embeds *ch* verbatim. + + Shaped like the real thing: a ``command_execution`` record carrying + captured output, then the ``turn.completed`` record that reports usage. + """ + return ( + json.dumps( + { + "type": "item.completed", + "item": { + "id": "item_1", + "type": "command_execution", + "aggregated_output": f"before{ch}after", + }, + }, + ensure_ascii=False, + ) + + "\n" + + json.dumps( + { + "type": "turn.completed", + "usage": {"input_tokens": 2500000, "output_tokens": 8000}, + } + ) + + "\n" + ) + + +class TestJsonlParseSurvivesEmbeddedUnicodeBreaks: + """``_parse_jsonl_output`` must key on the line feed, not on Unicode.""" + + def test_strict_parse_keeps_every_record(self) -> None: + for ch in JSON_UNESCAPED_BREAKS: + stdout = _codex_stdout(ch) + # Precondition: a splitlines()-based scan really does fragment this + # stream into more pieces than there are records. + assert len(_nonblank(stdout.splitlines())) == 3, repr(ch) + assert len(_nonblank(split_lf_lines(stdout))) == 2, repr(ch) + + parsed = _parse_jsonl_output( + stdout, + backend="codex", + cmd_str="codex exec --json", + worktree_path="/wt", + stderr="", + exit_code=0, + strict=True, + ) + + assert parsed["unparsable_lines"] == [] + assert len(parsed["events"]) == 2 + assert ( + parsed["events"][0]["item"]["aggregated_output"] == f"before{ch}after" + ) + assert parsed["events"][1]["usage"]["input_tokens"] == 2500000 + + def test_tool_event_counter_reads_one_record_not_two_fragments( + self, tmp_path: Path + ) -> None: + path = tmp_path / "stdout.jsonl" + path.write_text(_codex_stdout("\x85"), encoding="utf-8") + + count, names = _count_codex_stdout_tool_events(path) + + assert count == 1 + assert names == ["exec_command"] + + +class TestHelixResultLineSurvivesEmbeddedUnicodeBreaks: + """A ``HELIX_RESULT=`` payload may legitimately carry U+2028 in side info.""" + + def test_result_line_is_not_truncated_at_a_unicode_break(self) -> None: + payload = json.dumps([[1.0, {"note": "a\u2028b"}]], ensure_ascii=False) + stdout = f"noise\nHELIX_RESULT={payload}\ntrailing\n" + + line = _extract_helix_result_line(stdout) + + assert line is not None + assert json.loads(line.removeprefix("HELIX_RESULT=")) == [ + [1.0, {"note": "a\u2028b"}] + ] + + def test_machine_protocol_fragment_does_not_leak_into_the_mutation_prompt( + self, + ) -> None: + payload = json.dumps([[1.0, {"note": "a\u2028b"}]], ensure_ascii=False) + stdout = f"real evaluator output\nHELIX_RESULT={payload}\n" + + kept = _strip_machine_protocol_from_evaluator_stream(stdout) + + # A splitlines() scan dropped only the fragment up to U+2028; the tail + # no longer started with the prefix, so it leaked into the prompt. + assert kept == "real evaluator output" + assert "HELIX_RESULT" not in kept + leaked_tail = payload.split(LINE_SEPARATOR)[1] + assert leaked_tail not in kept diff --git a/tests/unit/test_mutator.py b/tests/unit/test_mutator.py index ff985953..c8d4f371 100644 --- a/tests/unit/test_mutator.py +++ b/tests/unit/test_mutator.py @@ -3,12 +3,15 @@ from __future__ import annotations import json +import subprocess from pathlib import Path from typing import Any from unittest.mock import MagicMock import pytest +from helix.display import UsageStats +from helix.exceptions import RateLimitError from helix.population import Candidate, EvalResult from helix.config import AgentConfig, HelixConfig, EvaluatorConfig, SandboxConfig from helix.mutator import ( @@ -1994,3 +1997,155 @@ def test_opencode_isolation_dir_gitignored(self, tmp_path: Path, mocker): assert ".helix_opencode_state/" in gitignore_text, ( ".helix_opencode_state/ must be gitignored to keep it out of candidate history" ) + + +# --------------------------------------------------------------------------- +# Tests: mutate hands back usage even when it hands back no candidate +# --------------------------------------------------------------------------- + + +class TestMutateRecordsUsageOnFailure: + """``record_usage`` is the only way a failed slot's tokens reach the budget. + + ``mutate`` returns ``None`` on :class:`MutationError` by contract, so + without this channel the caller has nothing to charge and the run's + budget silently under-reports by whatever the attempt spent. + """ + + def _mutate(self, tmp_path: Path, mocker, side_effect) -> tuple: + parent = make_candidate("g0-s0") + er = make_eval_result() + config = make_config() + child_path = tmp_path / "g1-s0" + child_path.mkdir() + child = make_candidate("g1-s0", str(child_path)) + mocker.patch("helix.mutator.clone_candidate", return_value=child) + mocker.patch("helix.mutator.invoke_claude_code", side_effect=side_effect) + mocker.patch("helix.mutator.remove_worktree") + mocker.patch("helix.mutator.snapshot_candidate") + + spent: list[UsageStats] = [] + result = mutate( + parent, er, "g1-s0", config, Path("/tmp"), record_usage=spent.append + ) + return result, spent + + def test_usage_is_reported_when_the_mutation_fails(self, tmp_path: Path, mocker): + usage = UsageStats(input_tokens=2500000, output_tokens=8000) + result, spent = self._mutate( + tmp_path, + mocker, + MutationError("Failed to parse Codex CLI JSONL output line", usage=usage), + ) + + assert result is None, "the None-on-failure contract still holds" + assert spent == [usage] + + def test_usage_is_reported_when_the_mutation_is_rate_limited( + self, tmp_path: Path, mocker + ): + usage = UsageStats(input_tokens=7, output_tokens=3) + parent = make_candidate("g0-s0") + er = make_eval_result() + config = make_config() + child_path = tmp_path / "g1-s0" + child_path.mkdir() + child = make_candidate("g1-s0", str(child_path)) + mocker.patch("helix.mutator.clone_candidate", return_value=child) + mocker.patch( + "helix.mutator.invoke_claude_code", + side_effect=RateLimitError("429", usage=usage), + ) + mocker.patch("helix.mutator.remove_worktree") + mocker.patch("helix.mutator.snapshot_candidate") + + spent: list[UsageStats] = [] + with pytest.raises(RateLimitError): + mutate( + parent, er, "g1-s0", config, Path("/tmp"), record_usage=spent.append + ) + + assert spent == [usage] + + def test_usage_is_reported_exactly_once_on_success(self, tmp_path: Path, mocker): + """The success path reports through the same channel, and only once.""" + usage = UsageStats(input_tokens=5, output_tokens=2) + parent = make_candidate("g0-s0") + er = make_eval_result() + config = make_config() + child_path = tmp_path / "g1-s0" + child_path.mkdir() + child = make_candidate("g1-s0", str(child_path)) + mocker.patch("helix.mutator.clone_candidate", return_value=child) + mocker.patch( + "helix.mutator.invoke_claude_code", return_value=({"result": "ok"}, usage) + ) + mocker.patch("helix.mutator.remove_worktree") + mocker.patch("helix.mutator.snapshot_candidate") + + spent: list[UsageStats] = [] + result = mutate( + parent, er, "g1-s0", config, Path("/tmp"), record_usage=spent.append + ) + + assert result is child + assert spent == [usage] + + def test_nothing_is_reported_when_the_error_carries_no_usage( + self, tmp_path: Path, mocker + ): + """No invocation happened — a zero charge would be a fabricated one.""" + result, spent = self._mutate(tmp_path, mocker, MutationError("timeout")) + + assert result is None + assert spent == [] + + +class TestSalvageBackendUsage: + """The lenient recovery used before the strict parse can fail.""" + + def test_recovers_usage_from_a_stream_with_an_unparsable_line(self): + from helix.mutator import _salvage_backend_usage + + stdout = "\n".join( + [ + "{not json", + '{"type":"turn.completed","usage":{"input_tokens":11,' + '"output_tokens":7}}', + ] + ) + usage = _salvage_backend_usage( + "codex", + subprocess.CompletedProcess( + args=["codex"], returncode=0, stdout=stdout, stderr="" + ), + ) + + assert usage.input_tokens == 11 + assert usage.output_tokens == 7 + + def test_falls_back_to_a_line_scan_for_a_truncated_claude_object(self): + from helix.mutator import _salvage_backend_usage + + stdout = '{"type":"result","usage":{"input_tokens":3,"output_tokens":1}}\n{"trunc' + usage = _salvage_backend_usage( + "claude", + subprocess.CompletedProcess( + args=["claude"], returncode=1, stdout=stdout, stderr="" + ), + ) + + assert usage.input_tokens == 3 + assert usage.output_tokens == 1 + + def test_returns_zeros_rather_than_raising_on_junk(self): + from helix.mutator import _salvage_backend_usage + + for stdout in ("", " ", "total garbage", "[]"): + usage = _salvage_backend_usage( + "codex", + subprocess.CompletedProcess( + args=["codex"], returncode=0, stdout=stdout, stderr="" + ), + ) + assert usage == UsageStats(), stdout