feat(tools): segment the transition stream at every replan

What a strategy change bought is a question the transition stream already
answers, and answering it in a scratchpad after the first corpus arrives is
how the analysis logic ends up outside the tree again. --replan-episodes
writes one object per replan: where it happened, what resolved it, and what
was spent in between.

The resolution is read from the resolving transition's own kind, so terminal
is not folded into advance the way a progress revision folds them — "the plan
finished" and "the plan moved one step" are different answers. A rewrite does
not resolve anything, so restating the steps after changing them stays inside
the interval, which is precisely the interval worth measuring. An episode
still open when the sample ends is written as "end" rather than dropped: a
replan nothing ever followed is the case a survivor-only view loses, and it is
likely the most informative one.

Semantic samples only. The proxy source carries no replan verdict, so there is
nothing to segment on and inferring one would put a made-up classification
into a corpus meant to calibrate against real ones.

This is a segmentation, not a statistic. No rate is aggregated and no rate is
called healthy — and before any of them mean anything, a corpus has to settle
whether the text-identity fallback reports renames as replans. The README says
which ratio to look at first.

tools/trajectory/README.md

@@ -69,11 +69,54 @@ The two bounds that survive the proxy caveat: at least 348.81M tokens and at
 least 739 minutes passed between two host-observed advances. That is the claim
 liveness work rests on, and it is stated in resources rather than rounds.
 
+## Replan episodes
+
+`--replan-episodes out.jsonl` segments the transition stream at every replan and
+records what resolved it, one object per episode:
+
+```json
+{"replan_at": 812.4, "replan_tokens": 12400000, "outcome": "advance",
+ "outcome_at": 851.2, "outcome_tokens": 13100000,
+ "tokens_to_outcome": 700000, "wall_to_outcome": 38.8}
+```
+
+`outcome` is one of four, read from the resolving transition's own kind:
+
+| outcome | the replan was followed by |
+| --- | --- |
+| `advance` | a step completing and execution moving on |
+| `terminal` | the last step completing |
+| `replan` | another change of plan |
+| `end` | nothing — the sample ended first |
+
+Three properties are deliberate. `terminal` is not folded into `advance`, even
+though a progress revision counts both: "the plan finished" and "the plan moved
+one step" are different answers to what a strategy change bought. A `rewrite`
+does not end an episode, so restating the steps after changing them stays
+inside the interval rather than resolving it. And an episode still open when
+the sample ends is written as `end` rather than dropped, because a replan that
+nothing ever followed is the case a survivor-only view would lose.
+
+Semantic samples only: the proxy source carries no replan verdict, so there is
+nothing to segment on and the tool refuses rather than inferring one.
+
+This is a segmentation, not a statistic. Nothing here aggregates episodes or
+names a healthy rate — what fraction of replans lead anywhere is a question for
+a corpus, and the first thing that corpus has to settle is whether the text
+identity fallback (below) is reporting renames as replans.
+
 ## Known fragility
 
 Semantic transitions are read back from the rendered trajectory line
 (`todo advance · content 4 · plan 1 · progress 2`), because that is what an
-export carries. A renderer change breaks the parse — deliberately loudly: a
+export carries. Step identity falls back to normalized text when a list carries
+no `step_id`, so a step renamed mid-run reads as one identity leaving and
+another arriving — a replan by this tool's definition, and possibly a rewrite by
+a person's. The `replan / all transitions` ratio is the diagnostic; a corpus
+full of `replan → replan` at small token gaps is a reason to check identity
+before concluding anything about strategy changes.
+
+A renderer change breaks the parse — deliberately loudly: a
 line that starts with `todo ` and does not parse raises rather than reading as
 "no semantic progress", which would quietly turn every later sample into a
 proxy. If the export ever carries the frame's fields structurally, read those

tools/trajectory/progress_corpus.py

@@ -174,6 +174,42 @@ def _summarise(transitions, cumulative, span, source, fidelity):
     return out, transitions
 
 
+# The transitions that end a replan episode. rewrite is deliberately absent: a
+# turn that keeps restating its steps after changing them has not yet answered
+# whether the change led anywhere, which is the question an episode asks.
+_EPISODE_OUTCOMES = ("advance", "terminal", "replan")
+
+
+def replan_episodes(transitions, total_tokens, span):
+    """Segment the transition stream at every replan.
+
+    An episode runs from a replan to the first transition that resolves it, and
+    the resolution is read from the transition's own kind — terminal is not
+    folded into advance, because "the plan finished" and "the plan moved one
+    step" are different answers to what a strategy change bought. An episode
+    still open when the sample ends is censored, not dropped: a replan nothing
+    ever followed is the case a survivor-only view would lose.
+    """
+    episodes = []
+    for index, start in enumerate(transitions):
+        if start["kind"] != "replan":
+            continue
+        resolution = next(
+            (t for t in transitions[index + 1:] if t["kind"] in _EPISODE_OUTCOMES), None)
+        if resolution is None:
+            outcome, at, tokens = "end", span, total_tokens
+        else:
+            outcome = resolution["kind"]
+            at, tokens = resolution["at"], resolution["cumulative_tokens"]
+        episodes.append({
+            "replan_at": start["at"], "replan_tokens": start["cumulative_tokens"],
+            "outcome": outcome, "outcome_at": at, "outcome_tokens": tokens,
+            "tokens_to_outcome": tokens - start["cumulative_tokens"],
+            "wall_to_outcome": at - start["at"],
+        })
+    return episodes
+
+
 def _million(value):
     return "-" if value is None else f"{value / 1e6:.2f}M"
 
@@ -209,6 +245,8 @@ def main(argv=None):
                         help="write one JSON object per transition (no tool payloads, no source)")
     parser.add_argument("--require-semantic", action="store_true",
                         help="fail instead of falling back to the complete_step proxy")
+    parser.add_argument("--replan-episodes", metavar="PATH",
+                        help="write one JSON object per replan episode; semantic samples only")
     args = parser.parse_args(argv)
 
     try:
@@ -222,6 +260,17 @@ def main(argv=None):
               "fallback is available only in proxy mode", file=sys.stderr)
         return 2
 
+    if args.replan_episodes and summary["progress_fidelity"] != "semantic":
+        print("error: replan episodes need semantic transitions; the complete_step "
+              "proxy carries no replan verdict to segment on", file=sys.stderr)
+        return 2
+
+    if args.replan_episodes:
+        episodes = replan_episodes(transitions, summary["total_tokens"], summary["wall_total_s"])
+        with open(args.replan_episodes, "w", encoding="utf-8") as fh:
+            for episode in episodes:
+                fh.write(json.dumps(episode, sort_keys=True) + "\n")
+
     if args.transitions:
         with open(args.transitions, "w", encoding="utf-8") as fh:
             for transition in transitions:

tools/trajectory/test_progress_corpus.py

@@ -138,3 +138,85 @@ class CommandLine(unittest.TestCase):
 
 if __name__ == "__main__":
     unittest.main()
+
+
+def _transition(at, tokens, kind, progress=0):
+    return {"at": at, "cumulative_tokens": tokens, "kind": kind,
+            "content_rev": 0, "plan_rev": 0, "progress_rev": progress}
+
+
+class ReplanEpisodes(unittest.TestCase):
+    """An episode asks what a strategy change bought, so it ends only on a
+    transition that answers: the plan moved, finished, or was changed again."""
+
+    def test_rewrites_do_not_end_an_episode(self):
+        episodes = pc.replan_episodes([
+            _transition(0, 0, "replan"),
+            _transition(1, 1e5, "rewrite"),
+            _transition(2, 2e5, "rewrite"),
+            _transition(3, 3e5, "rewrite"),
+            _transition(4, 4e5, "advance", 1),
+        ], total_tokens=4e5, span=10.0)
+        self.assertEqual(len(episodes), 1)
+        self.assertEqual(episodes[0]["outcome"], "advance")
+        self.assertEqual(episodes[0]["tokens_to_outcome"], 4e5)
+        self.assertEqual(episodes[0]["wall_to_outcome"], 4.0)
+
+    def test_a_replan_after_rewrites_resolves_the_episode(self):
+        episodes = pc.replan_episodes([
+            _transition(0, 0, "replan"),
+            _transition(1, 1e5, "rewrite"),
+            _transition(2, 2e5, "replan"),
+        ], total_tokens=2e5, span=10.0)
+        self.assertEqual(episodes[0]["outcome"], "replan")
+        self.assertEqual(episodes[0]["tokens_to_outcome"], 2e5)
+        # The second replan opens an episode of its own, censored here.
+        self.assertEqual(len(episodes), 2)
+        self.assertEqual(episodes[1]["outcome"], "end")
+
+    def test_terminal_is_not_folded_into_advance(self):
+        """Advances() covers both for a progress revision; an episode must not,
+        because "the plan finished" and "it moved one step" are different
+        answers to what the replan bought."""
+        episodes = pc.replan_episodes([
+            _transition(0, 0, "replan"),
+            _transition(5, 1e6, "terminal", 1),
+        ], total_tokens=1e6, span=10.0)
+        self.assertEqual(episodes[0]["outcome"], "terminal")
+
+    def test_an_unresolved_replan_is_censored_not_dropped(self):
+        episodes = pc.replan_episodes([
+            _transition(10, 2e6, "replan"),
+            _transition(20, 3e6, "rewrite"),
+        ], total_tokens=5e6, span=100.0)
+        self.assertEqual(len(episodes), 1)
+        self.assertEqual(episodes[0]["outcome"], "end")
+        self.assertEqual(episodes[0]["outcome_tokens"], 5e6)
+        self.assertEqual(episodes[0]["tokens_to_outcome"], 3e6)
+        self.assertEqual(episodes[0]["wall_to_outcome"], 90.0)
+
+    def test_proxy_samples_produce_no_episodes(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            out = os.path.join(tmp, "e.jsonl")
+            code = pc.main([fixture("legacy_proxy.json"), "--replan-episodes", out])
+            self.assertEqual(code, 2)
+            self.assertFalse(os.path.exists(out))
+
+    def test_semantic_samples_write_the_file(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            out = os.path.join(tmp, "e.jsonl")
+            code = pc.main([fixture("semantic.json"), "--replan-episodes", out, "--json"])
+            self.assertEqual(code, 0)
+            with open(out, encoding="utf-8") as fh:
+                episodes = [json.loads(line) for line in fh]
+        self.assertEqual(len(episodes), 1)
+        self.assertEqual(episodes[0]["outcome"], "advance")
+        self.assertEqual(
+            set(episodes[0]),
+            {"replan_at", "replan_tokens", "outcome", "outcome_at", "outcome_tokens",
+             "tokens_to_outcome", "wall_to_outcome"})
+
+    def test_episodes_stay_out_of_the_default_summary(self):
+        summary, _ = pc.extract(fixture("semantic.json"))
+        for key in summary:
+            self.assertNotIn("replan_", key)