From e291bacb53e6d1d6ed677e8359988710522ee7d5 Mon Sep 17 00:00:00 2001 From: Gadi Evron Date: Thu, 10 Sep 2026 22:49:23 +0300 Subject: [PATCH] =?UTF-8?q?feat(llm-reach):=20the=20in-pass=20split-and-re?= =?UTF-8?q?try=20for=20dropped=20batches=20=E2=80=94=20a=20malformed=20bat?= =?UTF-8?q?ch=20re-issued=20once=20as=20two=20halves,=20the=20recovery=20c?= =?UTF-8?q?arrying=20its=20own=20provenance=20(#558)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The post-lift measurement (the issue's step 1, run 2026-09-09): 4 malformed batches in ~95 at the DEFAULT_MAX_TOKENS cap — ALL the end_turn broken-JSON finish class (0 max_tokens; the cap lift eliminated the truncation-at-cap class). The close-out rule failed: the residual is the model's own structurally-broken finishes, a fresh-roll class — re-generation usually recovers. The implementation (the issue's direction 3 — split-and-retry, NEVER the JSON corrector: it cannot recover signals the model never emitted, and on the truncation class it would freeze partial batches as reviewed, defeating absence-as-retry): - a dropped or truncated batch of >= 2 units re-issues ONCE as two halves — smaller outputs (less likely to exhaust a cap or break mid-structure) AND a fresh roll. Bounded: one split level, no recursion; a still-dropped half stays dropped (its units re-run on the next resume via absence-as-retry); - the ORIGINAL drop's counters are REVISED (subtracted) so the halves count their own outcomes — the coverage truth never double-counted; - the recovery's own provenance counters (batches_split_recovered / batches_split_lost, direction 4 — a recovered batch never silently overwrites the units_not_reviewed count); - the recovered halves' units persist checkpoint records (they were reviewed this pass); the still-dropped leave none. The batch loop is restructured into a _attempt() helper (call+parse+ counters per sub-batch) — the same behavior for the non-split paths, with the split at the loop level. --- libs/openant-core/core/llm_reachability.py | 197 +++++++++++------- libs/openant-core/core/scanner.py | 6 + .../test_issue294_reach_batch_diagnostics.py | 11 +- .../tests/test_issue532_llr_resume.py | 137 +++++++++++- 4 files changed, 272 insertions(+), 79 deletions(-) diff --git a/libs/openant-core/core/llm_reachability.py b/libs/openant-core/core/llm_reachability.py index 7ece4350..66826718 100644 --- a/libs/openant-core/core/llm_reachability.py +++ b/libs/openant-core/core/llm_reachability.py @@ -461,6 +461,10 @@ def analyze_reachability( units_not_reviewed = 0 batches_truncated = 0 batches_failed = 0 + # #558: the split-and-retry provenance counters (direction 4 — the + # recovery never silently overwrites the coverage counts). + batches_split_recovered = 0 + batches_split_lost = 0 # ------------------------------------------------------------------ # #532: resume/adopt machinery — the checkpoint family's own pattern @@ -597,15 +601,22 @@ def _projection_sha(unit: dict) -> str: batches = _chunk(units_to_run, batch_size) persisted = 0 - for i, batch in enumerate(batches): + + # #558: one call+parse attempt for a (sub-)batch. Returns + # (signals, outcome) where outcome ∈ {"ok", "failed", "dropped", + # "truncated"}; the counters are mutated here so the split-and-retry + # below can revise them (a recovered original subtracts what its + # drop counted; the halves count their own outcomes). + def _attempt(sub_batch, label) -> tuple: + """Returns (signals, outcome, deltas) — deltas are THIS attempt's + applied counter changes ({dropped, units, truncated}), so the + split-and-retry subtracts exactly what was applied (never an + inferred shape — the refutation's negative-counter catch).""" + nonlocal dropped_batches, units_not_reviewed, batches_truncated, \ + batches_failed prompt = build_prompt( - batch, app_context=app_context, max_code_bytes=max_code_bytes + sub_batch, app_context=app_context, max_code_bytes=max_code_bytes ) - if tracker is not None: - try: - tracker.start_unit_tracking() - except Exception: # noqa: BLE001 - pass try: result = simple_completion(binding, prompt, max_tokens=DEFAULT_MAX_TOKENS, @@ -620,87 +631,132 @@ def _projection_sha(unit: dict) -> str: raise except Exception as exc: # noqa: BLE001 — advisory stage; never crash pipeline # #541: a provider-exception batch is counted in the coverage - # truth — the #386 counters covered the parse path only, so a - # step report could read success/0/0 for a pass that reviewed - # nothing (4 empty-completion batches on the receipt run were - # invisible to every surviving counter). A distinct counter - # (different failure class, different remediation) + the same - # units_not_reviewed so the coverage gap is the visible sum. - msg = f"batch {i + 1}/{len(batches)} failed: {exc}" + # truth. A distinct counter (different failure class, different + # remediation) + the same units_not_reviewed. + msg = f"{label} failed: {exc}" if on_error: on_error(msg) else: print(f"[LLMReach] {msg}", file=sys.stderr) batches_failed += 1 - units_not_reviewed += len(batch) - continue + units_not_reviewed += len(sub_batch) + return [], "failed", {"dropped": 0, "units": 0, "truncated": 0} - # #532 (disclosed behavior change): valid ids are PER-BATCH, so a - # signal for a unit outside the producing batch is ungrounded by - # construction — under persistence, adopted state must equal - # applied state (an out-of-batch signal would apply this run but - # never persist). - batch_ids = {u.get("id") for u in batch if u.get("id")} - first = batch[0].get("id", "?") if batch else "?" - last = batch[-1].get("id", "?") if batch else "?" - dropped_this_batch = False - - def _count_drop_and_flag(batch=batch, truncated=False): - nonlocal dropped_batches, units_not_reviewed, \ - dropped_this_batch, batches_truncated + batch_ids = {u.get("id") for u in sub_batch if u.get("id")} + first = sub_batch[0].get("id", "?") if sub_batch else "?" + last = sub_batch[-1].get("id", "?") if sub_batch else "?" + dropped = [] + _delta = {"dropped": 0, "units": 0, "truncated": 0} + + def _count_drop(sb=sub_batch, truncated=False): + nonlocal dropped_batches, units_not_reviewed, batches_truncated dropped_batches += 1 - units_not_reviewed += len(batch) - dropped_this_batch = True + units_not_reviewed += len(sb) + _delta["dropped"] += 1 + _delta["units"] += len(sb) if truncated: batches_truncated += 1 + _delta["truncated"] += 1 + dropped.append(True) parsed = parse_response( text, valid_unit_ids=batch_ids, on_error=on_error, - batch_label=f"batch {i + 1}/{len(batches)}, units {first}..{last}", - on_batch_drop=_count_drop_and_flag, + batch_label=f"{label}, units {first}..{last}", + on_batch_drop=_count_drop, stop_reason=result.stop_reason, ) - # #538 gate fold (fable+astra — the salvage-success gap): a - # max_tokens reply drops the batch WHOLE — the parsed prefix - # signals are NOT applied (the applied==adopted invariant: the - # signals gated the re-filter this run but were absent from the - # checkpoint, so a resume silently re-filtered on un-replayed - # state), and batches_truncated counts the truncation - # INDEPENDENTLY of the parse outcome (the salvage parse may - # succeed without the on_batch_drop callback ever firing — the - # counter was blind to exactly the case the gate acts on). - # Deduplicated: when the salvage parse DID drop (the callback - # fired), its counters already stand — the fold only adds the - # counts the callback never made. - _truncated = (result.stop_reason == "max_tokens") - if _truncated: - if not dropped_this_batch: + # #538 gate fold: a max_tokens reply drops the batch WHOLE (the + # applied==adopted invariant; the salvage prefix discarded) — + # independently of the parse outcome (the callback may not fire). + if result.stop_reason == "max_tokens": + if not dropped: batches_truncated += 1 dropped_batches += 1 - units_not_reviewed += len(batch) - print(f"[LLMReach] batch {i + 1}/{len(batches)} truncated at " - f"max_tokens — dropping whole (salvage prefix discarded); " - f"{len(batch)} units re-run on the next pass", - file=sys.stderr) + units_not_reviewed += len(sub_batch) + _delta["truncated"] += 1 + _delta["dropped"] += 1 + _delta["units"] += len(sub_batch) + return [], "truncated", _delta + if dropped: + return [], "dropped", _delta + return parsed, "ok", _delta + + for i, batch in enumerate(batches): + # #558 (the refutation's usage fix): the tracking window spans the + # ORIGINAL batch AND its split halves — started once here, never + # inside _attempt, so the per-unit records carry the whole + # recovery's true cost (a lost half's spend shared over the + # recovered units — the conservative choice, else it vanishes). + if tracker is not None: + try: + tracker.start_unit_tracking() + except Exception: # noqa: BLE001 + pass + parsed, outcome, deltas = _attempt( + batch, f"batch {i + 1}/{len(batches)}") + record_units: list = batch if outcome == "ok" else [] + + # #558: SPLIT-AND-RETRY (never the JSON corrector — it cannot + # recover signals the model never emitted, and on the truncation + # class it would freeze partial batches as reviewed). A dropped + # or truncated batch of >= 2 units is re-issued once as two + # halves: the halves are smaller outputs (less likely to exhaust + # a cap or break mid-structure) AND a fresh roll (the model's + # own broken-JSON finishes — the measured residual class at the + # lifted cap — recover on re-generation; the #292 rationale). + # Bounded: ONE split level, no recursion — a half that still + # drops stays dropped (its units re-run on the next resume via + # absence-as-retry). The ORIGINAL drop's counters are revised + # (subtracted) so the halves count their own outcomes: the + # coverage truth is never double-counted, and the recovery has + # its own provenance counters (#558's direction 4). + if outcome in ("dropped", "truncated") and len(batch) >= 2: + # Subtract EXACTLY this attempt's applied deltas (the + # refutation's negative-counter catch: a max_tokens reply + # with a no-brace shape incremented dropped but NOT truncated + # — inferring the shape drove batches_truncated to -1). + dropped_batches -= deltas["dropped"] + units_not_reviewed -= deltas["units"] + batches_truncated -= deltas["truncated"] + halves = [] + mid = (len(batch) + 1) // 2 + for j, half in enumerate((batch[:mid], batch[mid:])): + if not half: + continue + p2, o2, _d2 = _attempt( + half, f"batch {i + 1}/{len(batches)} half {j + 1}/2") + if o2 == "ok": + halves.extend(p2) + record_units.extend(half) + if halves or record_units: + batches_split_recovered += 1 + else: + batches_split_lost += 1 + print(f"[LLMReach] batch {i + 1}/{len(batches)} split-retry: " + f"{sum(1 for u in record_units)} units recovered via " + f"halves", file=sys.stderr) + parsed = halves + outcome = "recovered" if record_units else outcome + + # NOTE: a "failed" (provider-exception) batch is deliberately NOT + # split — the exception class (empty completions, transport) is + # not output-size-shaped; the #541 counters + the resume own it. + if outcome == "failed" or (outcome in ("dropped", "truncated") + and not record_units): continue signals.extend(parsed) - # Persist per-unit records ONLY for a batch that completed without a - # drop — dropped batches leave no records (absence = the retry - # marker on the next resume). Save failures cost persistence, not - # the pass (the stage's own advisory doctrine). - # #538 (4)-primitive: a max_tokens reply NEVER reaches this block — - # the fold's `continue` above drops the whole batch BEFORE the - # persist (the applied==adopted invariant; the salvage prefix - # discarded). dropped_this_batch stays the parse-drop marker. - if (checkpoint is not None and not dropped_this_batch): + # Persist per-unit records for the OK units — dropped halves leave + # no records (absence = the retry marker on the next resume). Save + # failures cost persistence, not the pass (the advisory doctrine). + if checkpoint is not None and record_units: batch_usage = {} if tracker is not None: try: batch_usage = tracker.get_unit_usage() or {} except Exception: # noqa: BLE001 batch_usage = {} - n_units = max(len(batch), 1) + n_units = max(len(record_units), 1) def _share(units_in_batch: int) -> dict: share = { @@ -708,8 +764,6 @@ def _share(units_in_batch: int) -> dict: "output_tokens": int(batch_usage.get("output_tokens", 0) or 0) // units_in_batch, "cost_usd": round(float(batch_usage.get("cost_usd", 0.0) or 0.0) / units_in_batch, 6), } - # #216: the incomplete-cost marker travels with the record so - # a resume restores it (the family's analyzer contract). if batch_usage.get("unpriced_models"): share["cost_incomplete"] = True share["unpriced_models"] = sorted( @@ -717,12 +771,12 @@ def _share(units_in_batch: int) -> dict: return share sig_by_unit: Dict[str, List[dict]] = {} - for s in parsed: - sig_by_unit.setdefault(s.unit_id, []).append({ - "unit_id": s.unit_id, "kind": s.kind, - "confidence": s.confidence, "reason": s.reason, + for sig in parsed: + sig_by_unit.setdefault(sig.unit_id, []).append({ + "unit_id": sig.unit_id, "kind": sig.kind, + "confidence": sig.confidence, "reason": sig.reason, }) - for u in batch: + for u in record_units: uid = u.get("id") if not uid: continue @@ -770,6 +824,9 @@ def _share(units_in_batch: int) -> dict: # #541: the provider-exception class — distinct from the parse # drops, same coverage-truth denominator. stats["batches_failed"] = batches_failed + # #558: the split-and-retry provenance. + stats["batches_split_recovered"] = batches_split_recovered + stats["batches_split_lost"] = batches_split_lost return signals diff --git a/libs/openant-core/core/scanner.py b/libs/openant-core/core/scanner.py index 43642232..ad60564a 100644 --- a/libs/openant-core/core/scanner.py +++ b/libs/openant-core/core/scanner.py @@ -901,6 +901,12 @@ def _step_label(name: str) -> str: # #541: the provider-exception class — the coverage # truth the parse-path-only counters missed. "batches_failed": reach_stats.get("batches_failed", 0), + # #558: the split-and-retry provenance — the recovery + # never silently overwrites the coverage counts. + "batches_split_recovered": reach_stats.get( + "batches_split_recovered", 0), + "batches_split_lost": reach_stats.get( + "batches_split_lost", 0), "units_not_reviewed": reach_stats.get("units_not_reviewed", 0), # #541 (the refute round): the #285/#376 partial- # status contract — dropped + failed batches make diff --git a/libs/openant-core/tests/test_issue294_reach_batch_diagnostics.py b/libs/openant-core/tests/test_issue294_reach_batch_diagnostics.py index 60f5ae2a..a577e73d 100644 --- a/libs/openant-core/tests/test_issue294_reach_batch_diagnostics.py +++ b/libs/openant-core/tests/test_issue294_reach_batch_diagnostics.py @@ -167,7 +167,10 @@ def test_analyze_reachability_counts_dropped_units(tmp_path, monkeypatch): # batch 1: valid ('{"signals": [{"unit_id": "f0.py:fn", "kind": "entry_point", ' '"confidence": "high", "reason": "ok"}]}'), - # batch 2: malformed (prose refusal) + # batch 2: malformed (prose refusal) — and its #558 split halves + # also malformed (the batch must STAY dropped for this pin) + "I can't help with analyzing this code for security purposes.", + "I can't help with analyzing this code for security purposes.", "I can't help with analyzing this code for security purposes.", ]) errs: list[str] = [] @@ -180,10 +183,14 @@ def test_analyze_reachability_counts_dropped_units(tmp_path, monkeypatch): # batch 1's signal survived the drop of batch 2 assert [s.unit_id for s in signals] == ["f0.py:fn"] - assert stats["batches_dropped"] == 1 + # #558: the original drop is REVISED (subtracted) and the two halves + # count their own outcomes — both dropped: 2 half-drops, 2 units. + assert stats["batches_dropped"] == 2 assert stats["units_not_reviewed"] == 2, "exact: batch membership known" + assert stats["batches_split_lost"] == 1 # the drop message is attributable assert any("batch 2/2" in e and "prose/refusal" in e for e in errs), errs + assert any("half" in e and "prose/refusal" in e for e in errs), errs def test_analyze_reachability_stats_absent_means_uncounted(): diff --git a/libs/openant-core/tests/test_issue532_llr_resume.py b/libs/openant-core/tests/test_issue532_llr_resume.py index 936bf7a4..19cae516 100644 --- a/libs/openant-core/tests/test_issue532_llr_resume.py +++ b/libs/openant-core/tests/test_issue532_llr_resume.py @@ -189,13 +189,17 @@ def test_removed_unit_record_not_adopted(self, tmp_path): class TestRetryMarkers: def test_dropped_batch_writes_no_records(self, tmp_path): - """A malformed batch must leave no records: absence IS the retry - marker (the #386 counter family + #532's 'must not freeze drops').""" + """A malformed batch (and its split-halves STILL failing) must + leave no records: absence IS the retry marker (#538 + #558's + split-retry — the original drop is revised, the halves count + their own outcomes).""" cp = str(tmp_path / "llm_reach_checkpoints") dataset = {"units": [_make_unit("a:f1"), _make_unit("b:f2")]} - # Malformed JSON for the whole (single) batch. + # The batch drops AND both halves drop (3 malformed responses: + # the original + the 2 split halves). analyze_reachability( - dataset, binding=_binding(FakeAdapter(["not json {"])), + dataset, binding=_binding(FakeAdapter( + ["not json {", "not json {", "not json {"])), checkpoint_path=cp, tracker=FakeTracker(), ) adapter = FakeAdapter([_canned(_sig("a:f1"), _sig("b:f2"))]) @@ -203,7 +207,7 @@ def test_dropped_batch_writes_no_records(self, tmp_path): dataset, binding=_binding(adapter), checkpoint_path=cp, tracker=FakeTracker(), ) - # Both units had no records -> both re-ran in one batch. + # No records from the dropped pass -> both re-ran in one batch. assert len(adapter.calls) == 1 def test_exception_batch_writes_no_records(self, tmp_path): @@ -381,9 +385,11 @@ def test_summary_incomplete_when_batches_dropped(self, tmp_path): cp = str(tmp_path / "llm_reach_checkpoints") dataset = {"units": [_make_unit("a:f1"), _make_unit("b:f2")]} - # Malformed response drops the whole (single) batch. + # Malformed response drops the whole (single) batch; the split + # halves ALSO drop (3 malformed responses total). analyze_reachability( - dataset, binding=_binding(FakeAdapter(["not json {"])), + dataset, binding=_binding(FakeAdapter( + ["not json {", "not json {", "not json {"])), checkpoint_path=cp, tracker=FakeTracker(), ) s = StepCheckpoint.read_summary(cp) @@ -730,3 +736,120 @@ def test_step_status_partial_when_batches_fail(self, tmp_path): with open(_os.path.join(str(tmp_path), "llm-reachability.report.json")) as fh: rep = _json.load(fh) assert rep["status"] == "partial" + + +class TestIssue558SplitRetry: + """#558: the split-and-retry — a dropped batch re-issued once as two + halves; never the JSON corrector; the recovery carries provenance.""" + + def test_dropped_batch_recovers_via_halves(self): + """The measurement's class (an end_turn broken-JSON finish): the + batch drops, the halves re-roll OK — the units are reviewed, the + original drop REVISED away, the recovery counted.""" + stats: dict = {} + # resp 1: the batch drops (malformed); resp 2-3: the halves OK. + adapter = FakeAdapter([ + "not json {", + _canned(_sig("a:f1")), + _canned(_sig("b:f2")), + ]) + signals = analyze_reachability( + {"units": [_make_unit("a:f1"), _make_unit("b:f2")]}, + binding=_binding(adapter), batch_size=2, + tracker=FakeTracker(), stats=stats) + assert {s.unit_id for s in signals} == {"a:f1", "b:f2"} + assert stats["batches_split_recovered"] == 1 + assert stats["batches_dropped"] == 0 # the original revised + assert stats["units_not_reviewed"] == 0 # the halves recovered + + def test_recovered_units_persist(self, tmp_path): + """The recovered halves' units get checkpoint records (they were + reviewed this pass — absence-as-retry only for the STILL-dropped).""" + cp = str(tmp_path / "llm_reach_checkpoints") + adapter = FakeAdapter(["not json {", _canned(_sig("a:f1")), + _canned(_sig("b:f2"))]) + analyze_reachability( + {"units": [_make_unit("a:f1"), _make_unit("b:f2")]}, + binding=_binding(adapter), batch_size=2, + checkpoint_path=cp, tracker=FakeTracker()) + import os + recs = sorted(f for f in os.listdir(cp) + if f.endswith(".json") and not f.startswith("_")) + assert recs == ["a_f1.json", "b_f2.json"] + + def test_one_unit_batch_never_splits(self): + """A 1-unit dropped batch cannot split — it stays dropped (the + bounded design: no recursion, the resume owns it).""" + stats: dict = {} + analyze_reachability( + {"units": [_make_unit("a:f1")]}, + binding=_binding(FakeAdapter(["not json {"])), + stats=stats) + assert stats["batches_dropped"] == 1 + assert stats["units_not_reviewed"] == 1 + assert stats["batches_split_recovered"] == 0 + assert stats["batches_split_lost"] == 0 + + def test_half_recovered_half_lost(self): + """A mixed split: one half OK, one half still drops — the + recovered units reviewed + persisted-eligible; the lost half's + units counted unreviewed (the coverage truth exact).""" + stats: dict = {} + adapter = FakeAdapter([ + "not json {", # the batch drops + _canned(_sig("a:f1")), # half 1 OK + "not json {", # half 2 drops (1 unit: no split) + ]) + signals = analyze_reachability( + {"units": [_make_unit("a:f1"), _make_unit("b:f2")]}, + binding=_binding(adapter), batch_size=2, + tracker=FakeTracker(), stats=stats) + assert [s.unit_id for s in signals] == ["a:f1"] + assert stats["units_not_reviewed"] == 1 # b:f2 only + assert stats["batches_split_recovered"] == 1 # partial recovery + assert stats["batches_dropped"] == 1 # the lost half + + def test_truncated_no_brace_never_negative(self): + """The refutation's negative-counter catch: a max_tokens reply + with a NO-BRACE shape (prose preamble hit the cap) incremented + dropped but NOT truncated — the old inferred subtraction drove + batches_truncated to -1. The deltas fix keeps it exact.""" + from utilities.llm import CompletionResult, TextBlock + + class TruncProse(FakeAdapter): + def complete(self, **kw): + return CompletionResult( + content=[TextBlock("prose preamble that never opens " + "the JSON before the cap")], + input_tokens=5, output_tokens=5, + stop_reason="max_tokens") + + stats: dict = {} + analyze_reachability( + {"units": [_make_unit("a:f1"), _make_unit("b:f2")]}, + binding=_binding(TruncProse()), batch_size=2, + tracker=FakeTracker(), stats=stats) + # The deltas fix: the prose+max_tokens shape counts in the + # malformed class (the #538 dedup: the parse's own drop stands), + # NOT truncated — and NO counter ever goes negative (the old + # inferred subtraction drove batches_truncated to -1 here). + assert stats["batches_dropped"] == 2 # the 2 lost halves + assert stats["units_not_reviewed"] == 2 + assert stats["batches_truncated"] == 0 # non-negative, exact + assert stats["batches_split_lost"] == 1 + + def test_usage_spans_the_recovery(self): + """The refutation's usage fix: the tracking window starts ONCE + per original batch — the records' usage carries the whole + recovery's cost (original + halves), not the last half's only.""" + from utilities.llm_client import TokenTracker + tracker = TokenTracker() + adapter = FakeAdapter(["not json {", _canned(_sig("a:f1")), + _canned(_sig("b:f2"))]) + analyze_reachability( + {"units": [_make_unit("a:f1"), _make_unit("b:f2")]}, + binding=_binding(adapter), batch_size=2, + tracker=tracker) + # All three calls' tokens are in the tracker totals (the records' + # shares derive from the window spanning them). + assert tracker.total_input_tokens >= 3 # 3 calls happened