Wave 21b: skip zero-signal SDPO on empty-recovery error turns + real-trace validation

Found by validating the Wave 21 pipeline against REAL Claude Code session logs
(738 local sessions, 66 with is_error:true) rather than only the synthetic
fixture. Two outcomes: a collator fix and a permanent validation example.

## Collator fix: empty-recovery error turns

On real traces, ~67% of error sites have EMPTY recovery content when
strip_thinking=True — because the error-RECOVERY turn is frequently pure
[THINKING] (the model reasons about the failure, then silently retries a tool),
and stripping thinking empties it. The old code set any_errors=True and injected
a hint the moment hint_text existed, BEFORE checking recovery content, so an
empty-recovery turn produced an all-ignore_index sdpo_loss_mask: a zero-signal
SDPO row that wastes a forward pass and dilutes the channel.

Fix: only treat a turn as an SDPO error site when BOTH a hint was produced AND
the recovery turn has content (`hint_text and turn.get("content")`). Applied
symmetrically to `_build_hint_injected_trace` (teacher) and
`_build_aligned_student_one` (student) so the student/teacher message lists stay
in lockstep and the SDPO shape-match gate never breaks. Empty-recovery turns
fall through to the (also-skipped) empty passthrough.

## The strip_thinking x SDPO lesson

SDPO hint-distillation on real agent traces REQUIRES strip_thinking=False — the
recovery reasoning IS the thinking. With it kept: 0% empty-recovery, real signal.
With it stripped: ~67% empty, channel goes mostly dark. Documented in the new
example's README.

## Real-trace validation result (10 sessions, strip_thinking=False)
- 10/10 processed, 0 crashes
- 141 error sites, 170 structural-flagged users, 0 string-tag-only
- 0% empty-recovery, SDPO alignment 832/832 = 100.0%
Confirms the Wave 21 _build_chat_aligned_mask fix holds at population scale.

## Added
- examples/validate_real_trace_alignment/ (run.py + README): auto-discovers
error-bearing ~/.claude sessions, runs ingestion->adapter->collator->SDPO,
reports alignment ratio + empty-recovery rate. Exit 0 PASS / 1 FAIL / 2 no-data.
- 3 stub-based empty-recovery tests in trainer/tests (always run, no model):
empty -> no SDPO; mixed -> fires on non-empty; shapes stay matched.

## Tests
Full package + spike collator: 164 passed, 16 skipped, 0 failed.

Note: empty-recovery tests live in composer_replication/trainer/tests/ (the
PACKAGE collator), not the spike — spikes/005 imports the legacy trl_path copy.

composer_replication/trainer/data_collator.py

@@ -296,25 +296,35 @@ class ComposerDataCollator:
                     turn.get("tool_error", "unknown"),
                     turn.get("error_meta", {}),
                 )
-                if hint_text:
+                # Only treat this as an SDPO error site when BOTH a hint was
+                # produced AND the recovery turn has content to distill against.
+                # Real Claude Code traces frequently have empty recovery content
+                # — e.g. when strip_thinking=True nukes a recovery turn that was
+                # pure [THINKING] reasoning (observed ~67% of real error sites).
+                # Injecting a hint with no recovery content produces an
+                # all-ignore_index mask: a zero-signal SDPO row that wastes a
+                # forward pass and silently dilutes the channel. Skip it; the
+                # turn then falls through to the (also-skipped) empty passthrough.
+                if hint_text and turn.get("content"):
                     any_errors = True
+                    recovery_content = turn.get("content") or ""
                     # Inject hint as a system-style addendum BEFORE the assistant's response
                     teacher_messages.append({"role": "system", "content": hint_text})
                     teacher_loss_segments.append((False, hint_text))
-                    if turn.get("content"):
-                        teacher_messages.append({
-                            "role": turn.get("role", "assistant"),
-                            "content": turn["content"],
-                        })
-                        teacher_loss_segments.append((True, turn["content"]))  # post-hint tokens = loss
+                    teacher_messages.append({
+                        "role": turn.get("role", "assistant"),
+                        "content": recovery_content,
+                    })
+                    teacher_loss_segments.append((True, recovery_content))  # post-hint tokens = loss
                     continue
-            # Non-error turn (or hint generator returned None) — passthrough
-            if turn.get("content"):
+            # Non-error turn (or hint generator returned None / empty recovery) — passthrough
+            content = turn.get("content")
+            if content:
                 teacher_messages.append({
                     "role": turn.get("role", "assistant"),
-                    "content": turn["content"],
+                    "content": content,
                 })
-                teacher_loss_segments.append((False, turn["content"]))
+                teacher_loss_segments.append((False, content))
 
         # Tokenize the full teacher conversation
         teacher_ids = self._tokenize_messages(teacher_messages)
@@ -449,28 +459,33 @@ class ComposerDataCollator:
                     turn.get("tool_error", "unknown"),
                     turn.get("error_meta", {}),
                 )
-                if hint_text:
+                # MUST mirror the teacher path's condition exactly (hint AND
+                # recovery content) or the student/teacher message lists diverge
+                # and the SDPO shape-match gate breaks. Empty-recovery error
+                # turns are skipped on both sides — see _build_hint_injected_trace.
+                if hint_text and turn.get("content"):
                     any_errors = True
+                    recovery_content = turn.get("content") or ""
                     placeholder = self._make_placeholder_for_hint_length(hint_text)
                     # Student gets a placeholder system-msg at the SAME slot
                     # the teacher gets the hint system-msg.
                     student_messages.append({"role": "system", "content": placeholder})
                     student_loss_segments.append((False, placeholder))
-                    if turn.get("content"):
-                        student_messages.append({
-                            "role": turn.get("role", "assistant"),
-                            "content": turn["content"],
-                        })
-                        is_assistant = turn.get("role") == "assistant"
-                        student_loss_segments.append((is_assistant, turn["content"]))
+                    student_messages.append({
+                        "role": turn.get("role", "assistant"),
+                        "content": recovery_content,
+                    })
+                    is_assistant = turn.get("role") == "assistant"
+                    student_loss_segments.append((is_assistant, recovery_content))
                     continue
-            if turn.get("content"):
+            content = turn.get("content")
+            if content:
                 student_messages.append({
                     "role": turn.get("role", "assistant"),
-                    "content": turn["content"],
+                    "content": content,
                 })
                 is_assistant = turn.get("role") == "assistant"
-                student_loss_segments.append((is_assistant, turn["content"]))
+                student_loss_segments.append((is_assistant, content))
 
         # Tokenize the full student conversation via apply_chat_template
         # (mirrors teacher's path so chat-template markers are identical).

composer_replication/trainer/tests/test_chat_template_alignment.py

@@ -144,3 +144,102 @@ def test_real_chat_template_student_teacher_shapes_match(real_chat_tok, multitur
     collator = ComposerDataCollator(tokenizer=real_chat_tok, config=cfg)
     batch = collator([multiturn_error_trace])
     assert batch["input_ids"].shape == batch["ctx_teacher_input_ids"].shape
+
+
+# ----------------------------------------------------------------------------
+# Empty-recovery guard (Wave 21 — discovered on real Claude Code traces)
+# ----------------------------------------------------------------------------
+#
+# ~67% of real error sites have EMPTY recovery content: when strip_thinking=True
+# the recovery turn (which was pure [THINKING] reasoning) becomes empty. Injecting
+# an SDPO hint with no recovery content yields an all-ignore_index mask — a
+# zero-signal row that wastes a forward pass and dilutes the channel. The collator
+# must treat empty-recovery error turns as non-error sites. These use a stub
+# tokenizer (pure logic, no model needed) so they always run.
+
+
+class _StubTok:
+    """Word-level deterministic tokenizer; apply_chat_template space-joins."""
+
+    pad_token_id = 0
+
+    def __init__(self) -> None:
+        self._v: dict[str, int] = {"<pad>": 0, "<bos>": 1, "<eos>": 2}
+
+    def _id(self, w: str) -> int:
+        if w not in self._v:
+            self._v[w] = len(self._v)
+        return self._v[w]
+
+    def __call__(self, text, **_k):
+        return {"input_ids": [self._id(w) for w in text.split()] if text else []}
+
+    def apply_chat_template(self, messages, tokenize=True, **_k):  # noqa: ARG002
+        return [self._id(w) for w in " ".join(m.get("content", "") for m in messages).split()]
+
+
+def _hint_for_tnf(kind, _meta):
+    return "HINT use a real tool" if kind == "tool_not_found" else None
+
+
+def test_empty_recovery_does_not_fire_sdpo():
+    """An error turn with empty recovery content must NOT emit an SDPO mask."""
+    tok = _StubTok()
+    trace = {
+        "trace_id": "empty-recovery",
+        "turns": [
+            {"role": "user", "content": "do the thing"},
+            {"role": "assistant", "content": "", "tool_error": "tool_not_found", "error_meta": {}},
+            {"role": "user", "content": "tool not found"},
+        ],
+        "final_reward": 0.0,
+    }
+    cfg = CollatorConfig(hint_generator=_hint_for_tnf)
+    collator = ComposerDataCollator(tokenizer=tok, config=cfg)
+    batch = collator([trace])
+    assert "sdpo_loss_mask" not in batch, (
+        "Empty-recovery error turn fired a zero-signal SDPO mask; it must be skipped."
+    )
+
+
+def test_mixed_recovery_fires_on_nonempty_only():
+    """A trace mixing empty + non-empty recovery turns fires SDPO from the
+    non-empty one and has loss-active positions."""
+    tok = _StubTok()
+    trace = {
+        "trace_id": "mixed-recovery",
+        "turns": [
+            {"role": "user", "content": "first task"},
+            {"role": "assistant", "content": "", "tool_error": "tool_not_found", "error_meta": {}},
+            {"role": "user", "content": "tool not found"},
+            {"role": "assistant", "content": "let me use a real tool instead",
+             "tool_error": "tool_not_found", "error_meta": {}},
+        ],
+        "final_reward": 0.0,
+    }
+    cfg = CollatorConfig(hint_generator=_hint_for_tnf)
+    collator = ComposerDataCollator(tokenizer=tok, config=cfg)
+    batch = collator([trace])
+    assert "sdpo_loss_mask" in batch
+    assert int((batch["sdpo_loss_mask"] == 1).sum()) > 0
+
+
+def test_empty_recovery_keeps_student_teacher_shapes_matched():
+    """Even with a skipped empty-recovery turn, when SDPO DOES fire elsewhere
+    the student/teacher shapes must still match (lockstep skip on both sides)."""
+    tok = _StubTok()
+    trace = {
+        "trace_id": "mixed-shape",
+        "turns": [
+            {"role": "user", "content": "task"},
+            {"role": "assistant", "content": "", "tool_error": "tool_not_found", "error_meta": {}},
+            {"role": "user", "content": "tool not found"},
+            {"role": "assistant", "content": "recover now with a real tool",
+             "tool_error": "tool_not_found", "error_meta": {}},
+        ],
+        "final_reward": 0.0,
+    }
+    cfg = CollatorConfig(hint_generator=_hint_for_tnf)
+    collator = ComposerDataCollator(tokenizer=tok, config=cfg)
+    batch = collator([trace])
+    assert batch["input_ids"].shape == batch["ctx_teacher_input_ids"].shape

examples/validate_real_trace_alignment/README.md

@@ -0,0 +1,63 @@
+# Real-trace SDPO alignment validation
+
+Runs the full **ingestion → adapter → collator → SDPO** data path against your
+own local Claude Code session logs (`~/.claude/projects/**/*.jsonl`) and reports
+the live SDPO mask alignment ratio. This is the population-level proof that
+Wave 21's `_build_chat_aligned_mask` fix holds on real-world data, not just the
+synthetic fixture.
+
+## Run
+
+```bash
+python examples/validate_real_trace_alignment/run.py
+# options:
+#   --projects-dir ~/.claude/projects   where to discover sessions
+#   --max-sessions 8                    how many error-bearing sessions to sample
+#   --model Qwen/Qwen2.5-0.5B-Instruct  a real chat-template tokenizer
+#   --pass-threshold 0.95               min alignment ratio to PASS
+#   --strip-thinking                    (default OFF — see below)
+```
+
+Exit code: `0` PASS (alignment ≥ threshold, no crashes), `1` FAIL, `2` no
+error-bearing sessions found / no chat template.
+
+## What it measures
+
+- **ingestion yield** — states emitted, error sites detected
+- **structural vs string-only flagging** — the Wave 21 `is_error` fix. The
+  ingester sets a structural `tool_error: True` boolean; `string-tag-only`
+  should be ~0 (the brittle `[TOOL_RESULT (ERROR)]` grep is fallback-only).
+- **empty-recovery rate** — see below.
+- **SDPO alignment** — fraction of in-loss `sdpo_loss_mask` positions where
+  student token id == teacher token id. ~100% means the mask lands exactly on
+  content tokens; <95% means chat-template drift has regressed.
+
+## The `--strip-thinking` gotcha (important for SDPO)
+
+`ClaudeCodeIngester(strip_thinking=...)` controls whether `[THINKING]` blocks
+survive. For most ingestion you strip them. **For SDPO hint-distillation you
+must NOT** — on real Claude Code traces the error-*recovery* turn is very often
+**pure thinking** (the model reasons about the failure, then silently retries a
+tool). Strip it and that turn's content goes empty, so ~67% of error sites carry
+no recovery content to distill against and produce a zero-signal SDPO row.
+
+This script therefore defaults to `strip_thinking=False`. The collator also
+guards against the empty case (an empty-recovery error turn is treated as a
+non-error site rather than firing an all-`ignore_index` mask), but the *signal*
+only exists if you keep the thinking. Pass `--strip-thinking` to see the
+empty-recovery warning fire.
+
+## Representative result (Codeseys' machine, 2026-05-28)
+
+```
+sessions processed:       10/10
+total error sites:        141
+structural-flagged users: 170
+string-tag-only users:    0
+empty-recovery sites:     0/141 (0%)     # strip_thinking=False
+SDPO alignment (REAL):    832/832 = 100.0%
+RESULT: PASS ✅
+```
+
+With `--strip-thinking` the same sessions report ~67% empty-recovery and the
+measurable in-loss positions collapse accordingly — the lever is visible.

examples/validate_real_trace_alignment/run.py

@@ -0,0 +1,203 @@
+"""Validate the full ingestion -> adapter -> collator -> SDPO data path against
+REAL Claude Code session logs, and report the live SDPO alignment ratio.
+
+Why this exists
+---------------
+The synthetic fixture in `spikes/007-real-trace-ingestion/fixtures/` proves the
+pipeline works on hand-built data. This script proves it on REAL traces — long
+tool outputs, multi-block content, thinking blocks, genuinely weird tool errors
+— which is where the Wave 19 chat-template drift bug (residual ~33%
+misalignment) actually bit. Wave 21's `_build_chat_aligned_mask` fix is verified
+here at the population level.
+
+What it measures
+----------------
+  * ingestion yield (states emitted, error sites detected)
+  * structural vs string-only error flagging (the Wave 21 TOOL_ERROR_TAG fix —
+    structural should dominate; string-only should be ~0)
+  * SDPO alignment ratio: fraction of in-loss `sdpo_loss_mask` positions where
+    student token id == teacher token id. ~100% means the mask lands exactly on
+    content tokens; <95% means chat-template drift has regressed.
+
+Usage
+-----
+    python examples/validate_real_trace_alignment/run.py \
+        [--projects-dir ~/.claude/projects] \
+        [--max-sessions 8] [--model Qwen/Qwen2.5-0.5B-Instruct]
+
+Requires a real chat-template tokenizer (transformers + a cached/instruct model)
+and at least one local Claude Code session containing `is_error: true`. Exits 0
+on PASS (>=95% alignment), 1 on FAIL, 2 if no error-bearing sessions were found.
+"""
+from __future__ import annotations
+
+import argparse
+import os
+import sys
+import traceback
+from pathlib import Path
+
+
+def _discover_error_sessions(projects_dir: Path, limit: int) -> list[Path]:
+    """Find session JSONLs that contain at least one is_error:true tool_result,
+    skipping subagent (`agent-*`) files. Returns up to `limit`, smallest first
+    (faster to process, still representative)."""
+    hits: list[tuple[int, Path]] = []
+    for p in projects_dir.rglob("*.jsonl"):
+        if p.name.startswith("agent-"):
+            continue
+        try:
+            text = p.read_text(encoding="utf-8", errors="ignore")
+        except OSError:
+            continue
+        if '"is_error":true' in text or '"is_error": true' in text:
+            hits.append((p.stat().st_size, p))
+    hits.sort(key=lambda t: t[0])
+    return [p for _, p in hits[:limit]]
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--projects-dir", default=str(Path.home() / ".claude" / "projects"))
+    ap.add_argument("--max-sessions", type=int, default=8)
+    ap.add_argument("--model", default="Qwen/Qwen2.5-0.5B-Instruct")
+    ap.add_argument("--pass-threshold", type=float, default=0.95)
+    ap.add_argument(
+        "--strip-thinking",
+        action="store_true",
+        help="Strip [THINKING] blocks. DEFAULT IS FALSE for SDPO: on real "
+        "Claude Code traces the error-recovery turn is frequently pure "
+        "thinking, so stripping it empties ~67%% of error sites and the SDPO "
+        "channel sees no signal. Keep thinking for hint-distillation.",
+    )
+    args = ap.parse_args()
+
+    os.environ.setdefault("HF_HUB_OFFLINE", "1")
+    os.environ.setdefault("TRANSFORMERS_OFFLINE", "1")
+
+    from transformers import AutoTokenizer
+
+    from composer_replication.ingestion import ClaudeCodeIngester
+    from composer_replication.ingestion.trace_examples import (
+        TOOL_ERROR_TAG,
+        claude_states_to_trace_examples,
+    )
+    from composer_replication.trainer.data_collator import (
+        CollatorConfig,
+        ComposerDataCollator,
+    )
+
+    projects_dir = Path(args.projects_dir).expanduser()
+    if not projects_dir.exists():
+        print(f"projects dir not found: {projects_dir}")
+        return 2
+
+    sessions = _discover_error_sessions(projects_dir, args.max_sessions)
+    if not sessions:
+        print(f"no error-bearing sessions under {projects_dir}")
+        return 2
+
+    tok = AutoTokenizer.from_pretrained(args.model)
+    if not getattr(tok, "chat_template", None):
+        print(f"{args.model} has no chat template; pick an -Instruct model")
+        return 2
+
+    def hint_gen(kind, _meta):
+        return f"Recover from the {kind}: re-check the path/args before retrying."
+
+    cfg = CollatorConfig(hint_generator=hint_gen, enable_replay_dpo=False, max_seq_len=8192)
+    collator = ComposerDataCollator(tokenizer=tok, config=cfg)
+
+    tot_states = tot_err_sites = 0
+    tot_aligned = tot_inloss = 0
+    n_struct = n_string_only = 0
+    n_empty_recovery = n_nonempty_recovery = 0
+    sessions_with_sdpo = 0
+    crashes: list[tuple[str, str]] = []
+
+    for path in sessions:
+        label = path.name[:18]
+        try:
+            ing = ClaudeCodeIngester(skip_sidechain=True, strip_thinking=args.strip_thinking)
+            states = list(ing.ingest(path))
+            for s in states:
+                for m in s["messages"]:
+                    if m.get("role") != "user":
+                        continue
+                    if m.get("tool_error") is True:
+                        n_struct += 1
+                    elif isinstance(m.get("content"), str) and TOOL_ERROR_TAG in m["content"]:
+                        n_string_only += 1
+            examples = claude_states_to_trace_examples(states)
+            # Count empty vs non-empty recovery content among detected error turns.
+            for ex in examples:
+                for t in ex["turns"]:
+                    if t.get("tool_error"):
+                        if (t.get("content") or "").strip():
+                            n_nonempty_recovery += 1
+                        else:
+                            n_empty_recovery += 1
+            err_examples = [
+                ex for ex in examples if any(t.get("tool_error") for t in ex["turns"])
+            ]
+            tot_states += len(states)
+            tot_err_sites += sum(
+                sum(1 for t in ex["turns"] if t.get("tool_error")) for ex in examples
+            )
+
+            if err_examples:
+                batch = collator(err_examples[:4])
+                if "sdpo_loss_mask" in batch:
+                    sessions_with_sdpo += 1
+                    s_in = batch["input_ids"]
+                    t_in = batch["ctx_teacher_input_ids"]
+                    m_in = batch["sdpo_loss_mask"]
+                    for row in range(s_in.shape[0]):
+                        il = m_in[row] == 1
+                        if int(il.sum()) == 0:
+                            continue
+                        tot_aligned += int((s_in[row][il] == t_in[row][il]).sum().item())
+                        tot_inloss += int(il.sum().item())
+            print(f"  OK    {label}: {len(states):4d} states, {len(err_examples):3d} err-examples")
+        except Exception as e:  # noqa: BLE001 — report-and-continue is the point
+            crashes.append((path.name, repr(e)))
+            print(f"  CRASH {label}: {e!r}")
+            traceback.print_exc()
+
+    print("\n" + "=" * 64)
+    print("REAL-TRACE PIPELINE VALIDATION")
+    print("=" * 64)
+    print(f"  sessions processed:       {len(sessions) - len(crashes)}/{len(sessions)}")
+    print(f"  total states emitted:     {tot_states}")
+    print(f"  total error sites:        {tot_err_sites}")
+    print(f"  structural-flagged users: {n_struct}")
+    print(f"  string-tag-only users:    {n_string_only}  (Wave 21: should be ~0)")
+    _tot_recovery = n_empty_recovery + n_nonempty_recovery
+    if _tot_recovery:
+        pct_empty = 100 * n_empty_recovery / _tot_recovery
+        print(
+            f"  empty-recovery sites:     {n_empty_recovery}/{_tot_recovery} "
+            f"({pct_empty:.0f}%) — these fire NO SDPO signal"
+        )
+        if args.strip_thinking and pct_empty > 30:
+            print(
+                "    ⚠ high empty-recovery rate with --strip-thinking: the recovery "
+                "turns are pure [THINKING]. Re-run WITHOUT --strip-thinking to "
+                "recover SDPO signal on these sites."
+            )
+    print(f"  sessions firing SDPO:     {sessions_with_sdpo}")
+
+    if not tot_inloss:
+        print("  no in-loss positions measured — cannot assess alignment")
+        return 2
+    ratio = tot_aligned / tot_inloss
+    print(f"  SDPO alignment (REAL):    {tot_aligned}/{tot_inloss} = {100 * ratio:.1f}%")
+    ok = ratio >= args.pass_threshold and not crashes
+    print(f"  RESULT: {'PASS ✅' if ok else 'FAIL ❌'} (threshold {100*args.pass_threshold:.0f}%)")
+    if crashes:
+        print(f"  {len(crashes)} crash(es): {[c[0] for c in crashes]}")
+    return 0 if ok else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())