fix: stop losing a failed mutation's entire token accounting

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 —

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:

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))

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,13 +1595,15 @@ 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:
         return MutationFailedProposal(
             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
 

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:

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:

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")

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,16 +352,24 @@ 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:
             remove_worktree(child)
         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: