🐛 forks: Phase 2 review fixes (I1, I2, minors 1, 2, 8-11)

plugins/spire/skills/forks/forks.py

@@ -1350,10 +1350,11 @@ def break_even(ttl) -> int:
 
 def base_warmth(base: dict, cfg: dict, now: float | None = None) -> dict:
     """Design 5: ttl = the base's inferred TTL, else the configured ttl_s; warm_until = last_hit +
-    ttl; margin = init's warm_margin_s if set, else max(120, ttl/10)."""
+    ttl; margin = init's warm_margin_s if set, capped at ttl/2 (Minor 2: a margin tuned for 1 h
+    must not make a 5-minute base `expiring` right after every hit), else max(120, ttl/10)."""
     ttl = float((base or {}).get("ttl_s") or cfg.get("ttl_s") or CONFIG_DEFAULTS["ttl_s"])
     wm = cfg.get("warm_margin_s")
-    margin = float(wm) if wm is not None else max(120.0, ttl / 10.0)
+    margin = min(float(wm), ttl / 2.0) if wm is not None else max(120.0, ttl / 10.0)
     out = {"ttl_s": int(ttl) if ttl.is_integer() else ttl,
            "margin_s": int(margin) if margin.is_integer() else margin}
     last = _ts((base or {}).get("last_hit"))
@@ -1594,6 +1595,13 @@ def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint:
                              "cost_cum_usd": u["total"] if u else None,
                              "model_usage_cum": u["model_usage"] if u else {},
                              "runtime_fingerprint": rt, "trained_at": ended})
+                # Minor 1 (`train --supersedes`): the bases this one replaces stop being active,
+                # so they get no more pings and need --force to fork. Only on `ok`.
+                for old_name in base.get("supersedes") or []:
+                    ob = reg["bases"].get(old_name)
+                    if ob is not None and ob.get("status") == "active" and old_name != base.get("name"):
+                        ob.update({"status": "superseded", "superseded_by": base.get("name"),
+                                   "superseded_at": ended})
             elif base.get("status") == "training":
                 base["status"] = "failed"
         if kind == "rebase" and status == "ok":
@@ -1861,11 +1869,14 @@ def build_parser() -> _Parser:
     w = cmd("where", "Preflight report: script, python, uv, binary, git, spawn method, env scrub.")
     w.add_argument("--binary", help="claude binary to report instead of the default resolution")
 
-    i = cmd("init", "Create the state dir and the launch config (idempotent).")
+    # Minor 8: every init flag defaults to None. On the first init an omitted flag takes its
-    i.add_argument("--model", required=True, help="full model id, e.g. claude-opus-5-5[1m]")
+    # default; on a re-run it keeps the current config value (build_init_config).
-    i.add_argument("--effort", default="high", choices=EFFORTS)
+    i = cmd("init", "Create the state dir and the launch config (idempotent). On a re-run, omitted "
-    i.add_argument("--permission-mode", default="dontAsk")
+                    "flags keep their current values.")
-    i.add_argument("--tools", default=DEFAULT_TOOLS, help="comma-separated tool names")
+    i.add_argument("--model", help="full model id, e.g. claude-opus-5-5[1m] (required on the first init)")
+    i.add_argument("--effort", choices=EFFORTS, help="default high")
+    i.add_argument("--permission-mode", help="default dontAsk")
+    i.add_argument("--tools", help="comma-separated tool names (default %s)" % DEFAULT_TOOLS)
     i.add_argument("--allowed-tools", nargs="+", action="extend", default=None,
                    help="allow rules (default: the tools)")
     i.add_argument("--disallowed-tools", nargs="+", action="extend", default=None,
@@ -1873,17 +1884,18 @@ def build_parser() -> _Parser:
     i.add_argument("--agent", help="project agent name (<cwd>/.claude/agents/<name>.md)")
     i.add_argument("--cwd", help="working dir of base and forks (default: git toplevel)")
     i.add_argument("--binary", help="claude binary (default: $CLAUDE_CODE_EXECPATH, else PATH)")
-    i.add_argument("--parallel-cap", type=_pos_int, default=CONFIG_DEFAULTS["parallel_cap"])
+    i.add_argument("--parallel-cap", type=_pos_int, help="default %d" % CONFIG_DEFAULTS["parallel_cap"])
-    i.add_argument("--orphan-after", type=_pos_int, default=CONFIG_DEFAULTS["orphan_after"])
+    i.add_argument("--orphan-after", type=_pos_int, help="default %d" % CONFIG_DEFAULTS["orphan_after"])
-    i.add_argument("--pings", choices=("on", "off"), default=CONFIG_DEFAULTS["pings"])
+    i.add_argument("--pings", choices=("on", "off"), help="default %s" % CONFIG_DEFAULTS["pings"])
-    i.add_argument("--ttl-s", type=_pos_int, default=CONFIG_DEFAULTS["ttl_s"],
+    i.add_argument("--ttl-s", type=_pos_int,
                    help="TTL assumed until a base's calls show one (default 3600)")
-    i.add_argument("--warm-margin-s", type=_pos_int, default=None,
+    i.add_argument("--warm-margin-s", type=_pos_int,
-                   help="seconds before warm_until a base turns `expiring` (default max(120, ttl/10))")
+                   help="seconds before warm_until a base turns `expiring` (default max(120, ttl/10); "
-    i.add_argument("--max-minutes", type=_pos_int, default=CONFIG_DEFAULTS["max_minutes"])
+                        "must be below ttl_s; at most half the base's TTL at runtime)")
-    i.add_argument("--stall-after-s", type=_pos_int, default=CONFIG_DEFAULTS["stall_after_s"])
+    i.add_argument("--max-minutes", type=_pos_int, help="default %d" % CONFIG_DEFAULTS["max_minutes"])
+    i.add_argument("--stall-after-s", type=_pos_int, help="default %d" % CONFIG_DEFAULTS["stall_after_s"])
     for k, v in THRESHOLD_DEFAULTS.items():
-        i.add_argument("--" + k.replace("_", "-"), type=_pos_int, default=v)
+        i.add_argument("--" + k.replace("_", "-"), type=_pos_int, help="default %d" % v)
 
     t = cmd("train", "Train a new base (headless session).")
     t.add_argument("--base", required=True)
@@ -1902,6 +1914,9 @@ def build_parser() -> _Parser:
     g.add_argument("--agent", help="project agent name (<cwd>/.claude/agents/<name>.md)")
     g.add_argument("--cwd", help="working dir of the base and its forks")
     g.add_argument("--binary", help="claude binary")
+    t.add_argument("--supersedes", action="append", default=None, metavar="BASE",
+                   help="a base this one replaces (repeatable): when the training ends ok, each "
+                        "named base that is active becomes superseded")
 
     def launch_overrides(sp):
         # Accepted only so that they can be refused with a clear message (plan 4.3).
@@ -1923,6 +1938,9 @@ def build_parser() -> _Parser:
     f.add_argument("--max-minutes", type=_pos_float, help="wall-time limit (default: init's max_minutes)")
     f.add_argument("--max-budget-usd", type=_pos_float)
     f.add_argument("--force", action="store_true", help="fork a base that is not active")
+    f.add_argument("--delta-note", action="append", default=None, metavar="PATH", dest="delta_notes",
+                   help="a note the fork must learn (repeatable): appended to the prompt after the "
+                        "delta header; counts toward delta_tokens when newer than the base's training")
     launch_overrides(f)
 
     c = cmd("continue", "One more turn in a fork (resume without --fork-session).")
@@ -1959,6 +1977,8 @@ def build_parser() -> _Parser:
     r.add_argument("--phase", help="the new base's phase (the phase whose forks it serves)")
     r.add_argument("--max-minutes", type=_pos_float, help="wall-time limit (default: init's max_minutes)")
     r.add_argument("--max-budget-usd", type=_pos_float)
+    r.add_argument("--delta-note", action="append", default=None, metavar="PATH", dest="delta_notes",
+                   help="a note the rebase must absorb (repeatable): appended after the delta header")
     launch_overrides(r)
 
     u = cmd("usage", "Aggregate the ledger: tokens and cost per kind, cold starts, cache-hit ratio.")
@@ -2098,39 +2118,73 @@ def build_launch(state: str, *, model, effort, permission_mode, tools, allowed,
     return launch, adm["via"]
 
 
-def build_init_config(args, state: str) -> dict:
+def build_init_config(args, state: str, old: dict | None = None) -> dict:
-    """Validate the launch fields and build the config block. Writes nothing."""
+    """Validate the launch fields and build the config block. Writes nothing.
-    tools = _tool_list([args.tools])
+    Minor 8: `old` is the current config (None on the first init). A flag that is not given keeps
+    its current value (launch fields and run-wide settings alike); defaults apply only on the
+    first init. What is read from disk (the binary's version, size and mtime, the preamble and
+    agent hashes) is re-read either way."""
+    ol = (old or {}).get("launch") or {}
+
+    def keep(given, key, default):
+        if given is not None:
+            return given
+        if old is not None and key in old and old[key] is not None:
+            return old[key]
+        return default
+    model = args.model if args.model is not None else ol.get("model")
+    if model is None:     # what argparse said when --model was required: a usage error
+        raise ForksError("usage", detail="the following arguments are required on the first init: --model")
+    tools = _tool_list([args.tools]) if args.tools is not None else \
+        (list(ol["tools"]) if ol.get("tools") else _tool_list([DEFAULT_TOOLS]))
     allowed = _tool_list(args.allowed_tools)
     if allowed is None:
-        allowed = list(tools)
+        allowed = list(ol["allowed_tools"]) if "allowed_tools" in ol else list(tools)
     disallowed = _tool_list(args.disallowed_tools)
     if disallowed is None:
-        disallowed = list(DEFAULT_DISALLOWED)
+        disallowed = list(ol["disallowed_tools"]) if "disallowed_tools" in ol else list(DEFAULT_DISALLOWED)
+    if args.cwd is None and ol.get("cwd"):
+        cwd = ol["cwd"]
+        if not os.path.isdir(cwd):
+            raise ForksError("--cwd is not a directory", cwd=cwd)
+    else:
         cwd = _resolve_cwd(args.cwd)
-    if not MODEL_RE.match(args.model or ""):     # before the binary lookup, as in Phase 1
+    if not MODEL_RE.match(model or ""):          # before the binary lookup, as in Phase 1
-        raise ForksError("model must be a full id (aliases are refused)", model=args.model,
+        raise ForksError("model must be a full id (aliases are refused)", model=model,
                          pattern=MODEL_RE.pattern)
+    if args.binary is None and (ol.get("binary") or {}).get("path"):
+        bpath, how = ol["binary"]["path"], (old or {}).get("binary_resolved_by")
+    else:
         bpath, how = resolve_binary(args.binary)
     if bpath is None:
         raise ForksError("no claude binary (no --binary, no $CLAUDE_CODE_EXECPATH, none on PATH)")
-    launch, via = build_launch(state, model=args.model, effort=args.effort,
+    agent = args.agent if args.agent is not None else (ol.get("agent") or {}).get("name")
-                               permission_mode=args.permission_mode, tools=tools, allowed=allowed,
+    launch, via = build_launch(state, model=model,
-                               disallowed=disallowed, agent=args.agent, cwd=cwd, binary_path=bpath)
+                               effort=args.effort if args.effort is not None else ol.get("effort") or "high",
+                               permission_mode=(args.permission_mode if args.permission_mode is not None
+                                                else ol.get("permission_mode") or "dontAsk"),
+                               tools=tools, allowed=allowed, disallowed=disallowed, agent=agent, cwd=cwd,
+                               binary_path=bpath)
+    old_th = (old or {}).get("thresholds") or {}
     cfg = {
         "launch": launch,
         "fingerprint": fingerprint_hash(launch),
         "binary_resolved_by": how,
-        "parallel_cap": args.parallel_cap,
+        "parallel_cap": keep(args.parallel_cap, "parallel_cap", CONFIG_DEFAULTS["parallel_cap"]),
-        "orphan_after": args.orphan_after,
+        "orphan_after": keep(args.orphan_after, "orphan_after", CONFIG_DEFAULTS["orphan_after"]),
-        "pings": args.pings,
+        "pings": keep(args.pings, "pings", CONFIG_DEFAULTS["pings"]),
-        "ttl_s": args.ttl_s,
+        "ttl_s": keep(args.ttl_s, "ttl_s", CONFIG_DEFAULTS["ttl_s"]),
-        "warm_margin_s": args.warm_margin_s,
+        "warm_margin_s": keep(args.warm_margin_s, "warm_margin_s", None),
-        "max_minutes": args.max_minutes,
+        "max_minutes": keep(args.max_minutes, "max_minutes", CONFIG_DEFAULTS["max_minutes"]),
-        "stall_after_s": args.stall_after_s,
+        "stall_after_s": keep(args.stall_after_s, "stall_after_s", CONFIG_DEFAULTS["stall_after_s"]),
-        "thresholds": {k: getattr(args, k) for k in THRESHOLD_DEFAULTS},
+        "thresholds": {k: (getattr(args, k) if getattr(args, k) is not None
+                           else old_th.get(k, THRESHOLD_DEFAULTS[k])) for k in THRESHOLD_DEFAULTS},
         "model_admission": via,
     }
+    # Minor 2: a margin at or above the TTL makes a base `expiring` right after every hit.
+    if cfg["warm_margin_s"] is not None and int(cfg["warm_margin_s"]) >= int(cfg["ttl_s"]):
+        raise ForksError("--warm-margin-s must be below ttl_s (a base would read `expiring` right "
+                         "after every hit)", warm_margin_s=cfg["warm_margin_s"], ttl_s=cfg["ttl_s"])
     return cfg
 
 
@@ -2145,10 +2199,14 @@ def _diff_fields(a, b, prefix="") -> list[str]:
 
 def cmd_init(args) -> tuple[dict, int]:
     state = state_dir_for(args.run)
-    cfg = build_init_config(args, state)          # all validation before any write
+    snap = read_registry(state) if os.path.isdir(state) else None
+    old_cfg = (snap or {}).get("config")
+    cfg = build_init_config(args, state, old_cfg)     # all validation before any write
     ensure_state(state)
     with RegistryLock(state):
         reg = read_registry(state)
+        if (reg or {}).get("config") != old_cfg:
+            raise ForksError("the config changed while init ran; run init again", state=state)
         created = reg is None
         if created:
             reg = new_registry()
@@ -2299,33 +2357,61 @@ def git_delta_counts(cwd: str, base_commit: str) -> tuple[int | None, int | None
     return commits, lines
 
 
-_NOTE_REF = re.compile(r"[^\s`'\"()<>\[\]{},;|*]+\.md\b")
+# I1 (Phase 2 review): "delta" is what a fork must learn that its base does not already know.
+# delta_tokens = chars / 4 of the delta header (commits and shortstat since the base's
+# base_commit) + the notes attached with `--delta-note` whose mtime is after the base's
+# `trained_at` (what the base was trained or rebased on it already knows) + the text of a
+# `## Changes since base` section of the brief. Other files a brief merely names (build plan,
+# run log, files to edit) and the rest of the brief never count.
+_CHANGES_HEADING = re.compile(r"^##[ \t]+Changes since base[ \t]*#*[ \t]*$", re.I | re.M)
+_HEADING_1_2 = re.compile(r"^#{1,2}[ \t]", re.M)
 
 
-def listed_notes(brief: str | None, cwd: str, state: str) -> list[str]:
+def changes_section(brief: str | None) -> str:
-    """The `.md` files a brief names that exist: absolute paths as given, relative ones against
+    """The body of the brief's `## Changes since base` section (up to the next level-1 or level-2
-    the base's cwd, then against the state dir (`notes/1.1.md`). Each file once."""
+    heading, or the end), or "" when the brief has none."""
+    m = _CHANGES_HEADING.search(brief or "")
+    if not m:
+        return ""
+    rest = brief[m.end():]
+    nxt = _HEADING_1_2.search(rest)
+    return (rest[:nxt.start()] if nxt else rest).strip()
+
+
+def read_delta_notes(paths) -> list[dict]:
+    """The notes given with `--delta-note`: [{"path", "text", "mtime"}], each file once.
+    A note that cannot be read is refused (exit 2) before anything is written."""
     out, seen = [], set()
-    for ref in _NOTE_REF.findall(brief or ""):
+    for raw in paths or []:
-        for root in (cwd, state):
+        p = norm_path(raw)
-            try:
-                p = norm_path(ref, cwd=root)
-            except ForksError:
-                break                       # e.g. a rooted Git Bash path cygpath cannot convert
         key = os.path.normcase(p)
-            if os.path.isfile(p):
+        if key in seen:
-                if key not in seen:
+            continue
         seen.add(key)
-                    out.append(p)
+        try:
-                break
+            text = read_bytes_retry(p).decode("utf-8", "replace")
+            mtime = os.path.getmtime(p)
+        except OSError as e:
+            raise ForksError("cannot read delta note", note=p, detail=str(e))
+        out.append({"path": p, "text": text, "mtime": mtime})
     return out
 
 
-def _text_chars(path: str) -> int:
+def new_to_base(notes: list[dict], base: dict) -> list[dict]:
-    try:
+    """The attached notes the base does not know yet: mtime after its `trained_at` (a training's
-        return len(read_bytes_retry(path).decode("utf-8", "replace"))
+    or a rebase's end). With no `trained_at` every note counts."""
-    except OSError:
+    t = _ts((base or {}).get("trained_at"))
-        return 0
+    return [n for n in notes or [] if t is None or n["mtime"] > t]
+
+
+def render_delta(header: str, notes: list[dict]) -> str:
+    """The prompt's delta section: the delta header, then each attached note with its path."""
+    if not notes:
+        return header
+    parts = [header, "### Attached delta notes"]
+    for n in notes:
+        parts.append("#### `%s`\n\n%s" % (n["path"], n["text"].rstrip("\n")))
+    return "\n\n".join(parts)
 
 
 def projects_dir() -> str:
@@ -2461,10 +2547,13 @@ def binary_state(base: dict, launch: dict, cache: dict | None = None) -> str:
 
 
 def due_flags(state: str, reg: dict, name: str, *, phase=None, brief: str | None = None,
-              delta: str | None = None, binary: str | None = None, bcache: dict | None = None) -> dict:
+              delta: str | None = None, notes: list[dict] | None = None, binary: str | None = None,
+              bcache: dict | None = None) -> dict:
     """Design 5's rebase and re-train flags for base `name`, with their reasons (advisory: only
-    `binary_missing` stops a launch). `phase`/`brief` are the fork's (none in `list`); `delta` the
+    `binary_missing` stops a launch). `phase`/`brief`/`notes` are the fork's (none in `list`, which
-    fork's delta header (computed here when not given); `binary` a `binary_state` value."""
+    counts the delta header alone); `delta` the fork's delta header (computed here when not
+    given); `notes` its `--delta-note` files (`read_delta_notes`); `binary` a `binary_state`
+    value."""
     cfg = reg["config"]
     base = reg["bases"][name]
     la = launch_record(base, cfg)
@@ -2481,7 +2570,8 @@ def due_flags(state: str, reg: dict, name: str, *, phase=None, brief: str | None
             rebase.append("lines:%d" % lines)
     if delta is None:
         delta = delta_header(la["cwd"], bc)
-    chars = len(delta) + sum(_text_chars(p) for p in listed_notes(brief, la["cwd"], state))
+    chars = (len(delta) + sum(len(n["text"]) for n in new_to_base(notes, base))
+             + len(changes_section(brief)))
     if chars // 4 > int(th["rebase_delta_tokens"]):
         rebase.append("delta_tokens:%d" % (chars // 4))
     if binary is None:
@@ -2591,6 +2681,15 @@ def cmd_train(args) -> tuple[dict, int]:
     if not re.fullmatch(r"[A-Za-z0-9_.-]+", name or ""):
         raise ForksError("bad base name", base=name)
     cfg = reg["config"]
+    supersedes = _uniq(args.supersedes or [])
+
+    def check_supersedes(r):
+        if name in supersedes:
+            raise ForksError("a base cannot supersede itself", base=name)
+        unknown = [n for n in supersedes if n not in r.get("bases", {})]
+        if unknown:
+            raise ForksError("unknown base", bases=unknown, option="--supersedes")
+    check_supersedes(reg)
     la, overrides = train_launch(args, state, cfg)
     binfo = la["binary"]
     drift: list[str] = []          # the record was built from the files just now
@@ -2607,13 +2706,14 @@ def cmd_train(args) -> tuple[dict, int]:
         pending = _pending_rebase_to(reg, name)
         if pending:
             raise ForksError("a running rebase creates this base", base=name, job=pending)
+        check_supersedes(reg)
         job = _next_job_id(reg)
         reg["bases"][name] = {"name": name, "status": "training", "session": session, "cwd": la["cwd"],
                               "launch": la, "fingerprint": fingerprint_hash(la), "overrides": overrides,
                               "binary": binfo, "created_at": _iso(), "train_job": job,
                               "phase": _phase(args.phase), "last_hit": None, "parent": None,
                               "lineage_rebases": 0, "base_commit": None, "context_tokens": None,
-                              "forks": []}
+                              "forks": [], "supersedes": supersedes}
         reg["jobs"][job] = _new_entry(
             job, "train", base=name, block=args.block, phase=_phase(args.phase), session=session,
             model_requested=la["model"], cwd=la["cwd"], argv=argv, max_s=float(cfg["max_minutes"]) * 60,
@@ -2643,10 +2743,11 @@ def cmd_fork(args) -> tuple[dict, int]:
     if not base.get("session"):
         raise ForksError("base has no session", base=args.base, status=base.get("status"))
     brief_path, brief = _read_brief(args.brief)
+    notes = read_delta_notes(args.delta_notes)
     binfo, drift = _check_binary(base.get("binary") or la["binary"], la)
     delta = delta_header(la["cwd"], base.get("base_commit"))
     due = due_flags(state, reg, args.base, phase=_phase(args.phase), brief=brief, delta=delta,
-                    binary="drift" if "binary" in drift else "ok")
+                    notes=notes, binary="drift" if "binary" in drift else "ok")
     head = _git_head(la["cwd"])
     session = str(uuid.uuid4())
     argv = claude_argv(la) + ["--resume", base["session"], "--fork-session", "--session-id", session]
@@ -2676,7 +2777,8 @@ def cmd_fork(args) -> tuple[dict, int]:
             base_context=base.get("context_tokens"), base_state_at_launch=warmth["base_state"],
             warm_until_at_launch=warmth["warm_until"], launch_drift=drift,
             binary_version=binfo["version"], head_at_launch=head, brief_source=brief_path,
-            forced=bool(args.force and base.get("status") != "active"))
+            forced=bool(args.force and base.get("status") != "active"),
+            delta_notes=[n["path"] for n in notes])
         base.setdefault("forks", []).append(job)
         write_registry(state, reg)
     if args.mode == "parallel":
@@ -2690,8 +2792,8 @@ def cmd_fork(args) -> tuple[dict, int]:
             commit_rule += " Expected paths: %s." % ", ".join("`%s`" % p for p in paths)
     prompt = render_template("fork.md", {
         "base": args.base, "run": norm_path(args.run), "state_dir": state, "block": args.block,
-        "phase": "n/a" if args.phase is None else str(args.phase), "mode": args.mode, "delta": delta,
+        "phase": "n/a" if args.phase is None else str(args.phase), "mode": args.mode,
-        "commit_rule": commit_rule, "note_path": os.path.join(state, *note_rel(args.block).split("/")),
+        "delta": render_delta(delta, notes), "commit_rule": commit_rule, "note_path": os.path.join(state, *note_rel(args.block).split("/")),
         "logs_dir": os.path.join(state, "logs"), "brief": brief})
     _start_job(state, norm_path(args.run), job, prompt, la)
     return {"job": job, "base": args.base, "base_state": warmth["base_state"],
@@ -2772,6 +2874,7 @@ def cmd_rebase(args) -> tuple[dict, int]:
         raise ForksError("base is not active", base=args.base, status=base.get("status"))
     la = launch_record(base, cfg)
     brief_path, brief = _read_brief(args.brief)
+    notes = read_delta_notes(args.delta_notes)
     binfo, drift = _check_binary(base.get("binary") or la["binary"], la)
     # The new base's record is the old one's; a binary drift means the rebase runs under the
     # current binary, which the new base then records (design 5).
@@ -2813,12 +2916,13 @@ def cmd_rebase(args) -> tuple[dict, int]:
             base_context=base.get("context_tokens"), base_state_at_launch=warmth["base_state"],
             warm_until_at_launch=warmth["warm_until"], launch_drift=drift,
             binary_version=binfo["version"], head_at_launch=head, brief_source=brief_path,
-            rebase_as=new, new_launch=new_launch)
+            rebase_as=new, new_launch=new_launch, delta_notes=[n["path"] for n in notes])
         write_registry(state, reg)
     prompt = render_template("rebase.md", {
         "base": args.base, "new_base": new, "run": norm_path(args.run), "state_dir": state,
         "notes_dir": os.path.join(state, "notes"), "logs_dir": os.path.join(state, "logs"),
-        "phase": "n/a" if args.phase is None else str(args.phase), "delta": delta, "brief": brief})
+        "phase": "n/a" if args.phase is None else str(args.phase), "delta": render_delta(delta, notes),
+        "brief": brief})
     _start_job(state, norm_path(args.run), job, prompt, la)
     return {"job": job, "base": new, "parent": args.base, "session_id": session}, EXIT_OK
 
@@ -2852,6 +2956,20 @@ def pings_since_real_call(reg: dict, name: str) -> int:
                if e.get("base") == name and e.get("kind") == "ping" and _job_num(j) > last_real)
 
 
+def ping_phase(reg: dict, name: str):
+    """I2: the phase a ping on base `name` belongs to: that of the latest (by job number) train,
+    fork, continue or rebase job on the base that has a phase; else the base's own phase; else
+    None. So `usage --phase N` carries the keep-warm cost of the phase's bases."""
+    best, ph = -1, None
+    for j, e in reg.get("jobs", {}).items():
+        if e.get("base") == name and e.get("kind") in REAL_KINDS and e.get("phase") is not None \
+                and _job_num(j) > best:
+            best, ph = _job_num(j), e["phase"]
+    if ph is None:
+        ph = (reg.get("bases", {}).get(name) or {}).get("phase")
+    return ph
+
+
 def ping_in_flight(reg: dict, base: dict | None) -> str | None:
     """The base's in-flight ping job (C3d), or None. A mark whose job is no longer running (or
     gone) does not count: it clears itself on the next marking."""
@@ -2949,7 +3067,8 @@ def tick_pass(state: str) -> dict:
                 argv = claude_argv(la) + ["--resume", base["session"], "--fork-session",
                                           "--session-id", session, "--max-turns", "1"]
                 reg["jobs"][job] = _new_entry(
-                    job, "ping", base=n, session=session, parent_session=base["session"],
+                    job, "ping", base=n, phase=ping_phase(reg, n), session=session,
+                    parent_session=base["session"],
                     model_requested=la["model"], cwd=la["cwd"], argv=argv, max_s=PING_MAX_S,
                     cost_before=float(base.get("cost_cum_usd") or 0.0),
                     model_usage_before=base.get("model_usage_cum") or {},
@@ -3179,6 +3298,9 @@ class Supervisor:
             _diag("swept %d leftover process(es) of claude: %s" % (
                 len(sweep["snapshot"]), [(x["pid"], x["name"]) for x in sweep["snapshot"]]))
         reader.join(READER_JOIN_S)
+        if not reader.is_alive():                   # the reader is done with the pipe (Minor 10)
+            with contextlib.suppress(OSError):
+                p.stdout.close()
         with contextlib.suppress(OSError):
             raw.close()
         with contextlib.suppress(OSError):
@@ -3558,6 +3680,29 @@ def usage_md_phase(agg: dict) -> str:
         "- Priced total: %s" % _cost(agg["total"]["cost_usd"])]) + "\n"
 
 
+RESULT_STATUS_KEYS = ("status", "flags", "reason")
+
+
+def status_from_results(state: str, lines: list[dict]) -> tuple[list[dict], list[str]]:
+    """Minor 11: the ledger is append-only, so a job finalised again after its line was written
+    (R1 reconciled a result whose entry still read `running`, then `reap_lost` or `kill` rewrote
+    the result as `lost`/`killed`) keeps its first status there. `usage` takes each job's
+    status, flags and reason from `results/<job>.json` when they differ from its ledger line;
+    tokens and cost stay the ledger's. Returns (lines, the jobs whose status came from a result)."""
+    out, changed = [], []
+    for ln in lines:
+        try:
+            res = read_json(result_file(state, ln["job"]))
+        except ForksError:
+            res = None
+        if isinstance(res, dict) and any(res.get(k) != ln.get(k) for k in RESULT_STATUS_KEYS):
+            ln = dict(ln, status=res.get("status"), flags=list(res.get("flags") or []),
+                      reason=res.get("reason"))
+            changed.append(ln["job"])
+        out.append(ln)
+    return out, changed
+
+
 def cmd_usage(args) -> tuple[dict | str, int]:
     """`usage --run R [--block ID | --phase N] [--format json|md]`. R1 first: reconcile the ledger
     against `results/` under the lock. Allowed inside a fork (N3)."""
@@ -3573,12 +3718,14 @@ def cmd_usage(args) -> tuple[dict | str, int]:
     elif args.phase is not None:
         scope = {"phase": _phase(args.phase)}
         lines = [x for x in lines if x.get("phase") == scope["phase"]]
+    lines, from_result = status_from_results(state, lines)
     if args.format == "md":
         if args.block is not None:
             return usage_md_block(lines, reg.get("jobs") or {}), EXIT_OK
         return usage_md_phase(usage_aggregate(lines)), EXIT_OK
     agg = usage_aggregate(lines)
-    return {"state": state, "scope": scope, "reconciled": added, **agg, "rows": lines}, EXIT_OK
+    return {"state": state, "scope": scope, "reconciled": added, "status_from_result": from_result,
+            **agg, "rows": lines}, EXIT_OK
 
 
 HANDLERS = {

plugins/spire/skills/forks/tests/test_forks.py

@@ -721,6 +721,62 @@ class InitTests(Base):
         self.assertFalse(out["changed"])
         self.assertEqual(self.reg_bytes(), before)
 
+    def test_rerun_keeps_omitted_settings(self):
+        """Minor 8: on a re-run, flags that are not given keep their current values (launch fields
+        and run-wide settings); defaults apply only on the first init; `changed_fields` lists only
+        what the given flags changed."""
+        code, first, _ = self.init("--pings", "off", "--parallel-cap", "3", "--rebase-commits", "9",
+                                   "--orphan-after", "1500", "--effort", "low", "--tools", "Read,Bash",
+                                   model="claude-sonnet-5-5")
+        self.assertEqual(code, 0, first)
+        self.add_fake_base()
+        before = forks.read_registry(self.state)["config"]
+        code, out, raw = call(["init", "--run", self.run_log, "--warm-margin-s", "100"])
+        self.assertEqual(code, 0, out)
+        self.assert_one_json_line(raw)
+        self.assertTrue(out["changed"])
+        self.assertEqual(out["changed_fields"], ["warm_margin_s"])
+        cfg = forks.read_registry(self.state)["config"]
+        self.assertEqual(cfg["warm_margin_s"], 100)
+        self.assertEqual((cfg["pings"], cfg["parallel_cap"], cfg["orphan_after"]), ("off", 3, 1500))
+        self.assertEqual(cfg["thresholds"], dict(forks.THRESHOLD_DEFAULTS, rebase_commits=9))
+        self.assertEqual(cfg["launch"], before["launch"])
+        self.assertEqual((cfg["launch"]["model"], cfg["launch"]["effort"], cfg["launch"]["tools"]),
+                         ("claude-sonnet-5-5", "low", ["Read", "Bash"]))
+        self.assertEqual(cfg["fingerprint"], before["fingerprint"])
+        # a given flag still changes its value, and only it is listed
+        code, out, _ = call(["init", "--run", self.run_log, "--pings", "on"])
+        self.assertEqual(code, 0, out)
+        self.assertEqual(out["changed_fields"], ["pings"])
+        cfg = forks.read_registry(self.state)["config"]
+        self.assertEqual((cfg["pings"], cfg["warm_margin_s"], cfg["parallel_cap"]), ("on", 100, 3))
+        # the first init still needs --model (a usage error, nothing written)
+        other = os.path.join(self.tmp, "reports", "2026-10-16-09-u-run.md")
+        code, out, raw = call(["init", "--run", other, "--cwd", self.cwd, "--binary", self.stub])
+        self.assertEqual(code, 2, out)
+        self.assert_one_json_line(raw)
+        self.assertEqual(out["error"], "usage")
+        self.assertIn("--model", out["detail"])
+        self.assertFalse(os.path.exists(forks.state_dir_for(other)))
+
+    def test_warm_margin_must_be_below_ttl(self):
+        """Minor 2: `init` refuses a warm_margin_s at or above the configured ttl_s (exit 2), on the
+        first init and on a re-run that lowers ttl_s under the current margin."""
+        out = self.assert_refused_no_state("--warm-margin-s", "3600")
+        self.assertEqual((out["warm_margin_s"], out["ttl_s"]), (3600, 3600))
+        self.assert_refused_no_state("--ttl-s", "300", "--warm-margin-s", "300")
+        code, out, _ = self.init("--warm-margin-s", "3599")
+        self.assertEqual(code, 0, out)
+        before = self.reg_bytes()
+        code, out, raw = self.init("--ttl-s", "3000")
+        self.assertEqual(code, 2, out)
+        self.assert_one_json_line(raw)
+        self.assertIn("below ttl_s", out["error"])
+        self.assertEqual(self.reg_bytes(), before)
+        code, out, _ = self.init("--ttl-s", "300", "--warm-margin-s", "200")
+        self.assertEqual(code, 0, out)
+        self.assertEqual(sorted(out["changed_fields"]), ["ttl_s", "warm_margin_s"])
+
     def test_fingerprint_hash_stable_across_runs(self):
         """M7: the hash covers the mcp config's content, not its per-run path: two runs with the
         same launch fields hash the same; the path stays in the config, outside the hash."""
@@ -1635,11 +1691,28 @@ class WarmthTests(Base):
             with self.subTest(margin=900, t=t - T0):
                 self.assertEqual(st(b, m, t), want)
         self.assertEqual(forks.base_warmth(b, m, T0)["margin_s"], 900)
-        for t, want in ((T0, "expiring"), (T0 + 299, "expiring"), (T0 + 300, "cold")):
+        # Minor 2: at a 5-minute TTL the 900 s margin is capped at ttl/2 = 150 s
+        for t, want in ((T0, "warm"), (T0 + 149, "warm"), (T0 + 150, "expiring"), (T0 + 299, "expiring"),
+                        (T0 + 300, "cold")):
             with self.subTest(margin=900, base_ttl=300, t=t - T0):
                 self.assertEqual(st({**b, "ttl_s": 300}, m, t), want)
+        self.assertEqual(forks.base_warmth({**b, "ttl_s": 300}, m, T0)["margin_s"], 150)
         self.assertEqual(st({}, cfg, T0), "cold")
 
+    def test_margin_capped_at_half_the_ttl(self):
+        """Minor 2: a configured margin above ttl/2 is capped there, so a 5-minute downgrade does not
+        leave a base `expiring` (and pinged every tick) right after every hit."""
+        T0 = self.T0
+        b = {"last_hit": forks._iso(T0), "ttl_s": 300}
+        w = forks.base_warmth(b, {"ttl_s": 3600, "warm_margin_s": 600}, T0 + 1)
+        self.assertEqual((w["base_state"], w["margin_s"], w["ttl_s"]), ("warm", 150, 300))
+        reg = self.make_reg(T0 + 1, margin=600, base_ttl=300)
+        self.assertEqual(self.decide(reg, T0 + 1)["why"], "state:warm")
+        self.assertTrue(self.decide(reg, T0 + 150)["ping"])
+        # a margin below ttl/2 is used as given; the formula applies when none is set
+        self.assertEqual(forks.base_warmth(b, {"ttl_s": 3600, "warm_margin_s": 100}, T0)["margin_s"], 100)
+        self.assertEqual(forks.base_warmth(b, {"ttl_s": 3600, "warm_margin_s": None}, T0)["margin_s"], 120)
+
     def test_infer_ttl_and_downgrade(self):
         def t(h=0, m=0, r=0):
             return {"cache_creation_1h": h, "cache_creation_5m": m, "cache_read": r}
@@ -1684,8 +1757,11 @@ class WarmthTests(Base):
         # 5 min TTL inferred on the base, margin from init
         d = self.decide(self.make_reg(self.E5M, base_ttl=300), self.E5M)
         self.assertEqual((d["ping"], d["ttl_s"], d["break_even"]), (True, 300, 11))
-        t = self.T0 + 100
+        # init's margin, capped at ttl/2 (Minor 2): 3600 acts as 1800 at a 1 h TTL
+        t = self.T0 + 1800
         self.assertTrue(self.decide(self.make_reg(t, margin=3600), t)["ping"])
+        t = self.T0 + 1799
+        self.assertEqual(self.decide(self.make_reg(t, margin=3600), t)["why"], "state:warm")
 
     def test_ping_rule_activity_window(self):
         now = self.E1H
@@ -1847,6 +1923,53 @@ class TickTests(Base):
         self.assertEqual(len(out["pinged"]), 1)
         self.assertEqual(out["bases"][0]["pings_since_real_call"], 1)    # counted before this one
 
+    def test_ping_carries_the_phase_and_counts_in_usage_phase(self):
+        """I2: a ping records the phase of the latest train/fork/continue/rebase job on its base
+        (the base's own phase if none has one), so `usage --phase N` (md and json) lists it and its
+        cost is in the priced total."""
+        reg = self.reg()
+        self.assertIsNone(forks.ping_phase(reg, "B1"))                   # nothing has a phase
+        reg["bases"]["B1"]["phase"] = 1
+        self.assertEqual(forks.ping_phase(reg, "B1"), 1)                 # the base's own
+        reg["jobs"]["j0002"] = forks._new_entry("j0002", "fork", base="B1", phase=2, status="ok",
+                                                ended_at=forks._iso(time.time() - 4000))
+        reg["jobs"]["j0003"] = forks._new_entry("j0003", "fork", base="B9", phase=7, status="ok")
+        reg["jobs"]["j0004"] = forks._new_entry("j0004", "continue", base="B1", phase=None, status="ok")
+        reg["jobs"]["j0005"] = forks._new_entry("j0005", "ping", base="B1", phase=5, status="ok")
+        self.assertEqual(forks.ping_phase(reg, "B1"), 2)                 # latest real job with a phase
+        with forks.RegistryLock(self.state):
+            forks.write_registry(self.state, reg)
+        out = forks.tick_pass(self.state)
+        self.assertEqual(len(out["pinged"]), 1, out)
+        job = out["pinged"][0]
+        self.assertEqual(self.reg()["jobs"][job]["phase"], 2)
+        # the ping ends ok with a known cost (cost basis: the base's 1.25)
+        ss = forks.StreamState()
+        t = time.time()
+        u = {"input_tokens": 1, "cache_read_input_tokens": 1000, "cache_creation_input_tokens": 0}
+        ss.feed({"type": "assistant", "message": {"id": "m1", "usage": u}}, t)
+        ss.feed({"type": "result", "is_error": False, "num_turns": 1, "total_cost_usd": 1.2623,
+                 "usage": dict(u, output_tokens=2, cache_creation={}),
+                 "modelUsage": {TEST_MODEL: {"inputTokens": 2, "outputTokens": 2,
+                                             "cacheReadInputTokens": 1000, "cacheCreationInputTokens": 0}}},
+                t)
+        res = forks.finalize_job(self.state, job, ss=ss, hint="exited", exit_code=0, wall_s=1.0)
+        self.assertEqual((res["status"], res["phase"], res["cost_delta_usd"]), ("ok", 2, 0.0123), res)
+        line = [x for x in forks.read_ledger(self.state) if x["job"] == job][0]
+        self.assertEqual(line["phase"], 2)
+        code, js, _ = call(["usage", "--run", self.run_log, "--phase", "2"])
+        self.assertEqual(code, 0, js)
+        self.assertEqual([x["job"] for x in js["rows"]], [job])
+        self.assertEqual(js["kinds"]["ping"]["calls"], 1)
+        self.assertAlmostEqual(js["kinds"]["ping"]["cost_usd"], 0.0123, places=6)
+        self.assertAlmostEqual(js["total"]["cost_usd"], 0.0123, places=6)
+        code, md = call_raw(["usage", "--run", self.run_log, "--phase", "2", "--format", "md"])
+        self.assertEqual(code, 0, md)
+        self.assertIn("| ping | 1 | 1 | 2 | 0 | 1,000 | $0.0123 |\n", md)
+        self.assertIn("| total | 1 | 1 | 2 | 0 | 1,000 | $0.0123 |\n", md)
+        self.assertTrue(md.endswith("- Priced total: $0.0123\n"), md)
+        self.assertEqual(call(["usage", "--run", self.run_log, "--phase", "1"])[1]["rows"], [])
+
     def test_tick_command_and_n3(self):
         code, out, raw = call(["tick", "--run", self.run_log])
         self.assertEqual(code, 0, out)
@@ -2185,21 +2308,48 @@ class UsageTests(Base):
         self.assertEqual(self.usage("--phase", "3")["reconciled"], [])
 
     def test_r1_waits_for_the_registry_lock(self):
-        """With the lock held and a line missing, `usage` blocks until it is released, then
+        """With the lock held and a line missing, the reconciliation itself blocks until the lock is
-        reconciles."""
+        released, then appends. Minor 9: `reconcile_ledger` is called directly (`usage` would wait
+        on its orchestrator stamp first, so a reconciliation without the lock would pass too)."""
         self.result_for("j0010", block="4.1", phase=4)
         done = []
         lock = forks.RegistryLock(self.state)
         lock.acquire()
         try:
-            th = threading.Thread(target=lambda: done.append(self.usage("--phase", "4")))
+            th = threading.Thread(target=lambda: done.append(forks.reconcile_ledger(self.state)))
             th.start()
             time.sleep(1.0)
             self.assertEqual(done, [])            # blocked on the lock
+            self.assertNotIn("j0010", [x["job"] for x in self.ledger_lines()])
         finally:
             lock.release()
         th.join(30)
-        self.assertEqual(done[0]["reconciled"], ["j0010"])
+        self.assertEqual(done, [["j0010"]])
+        self.assertEqual(len([x for x in self.ledger_lines() if x["job"] == "j0010"]), 1)
+
+    def test_status_from_result_when_it_changed_after_the_ledger_line(self):
+        """Minor 11: the ledger stays append-only; a job finalised again after its line was written
+        (R1 line, then `reap_lost`/`kill` rewrote the result) reads its status, flags and reason
+        from the result in `usage`; tokens and cost stay the ledger's."""
+        self.result_for("j0002", status="lost", flags=["cold_start"], reason="supervisor_dead",
+                        block="1.1", phase=1, tokens=dict(zip(forks.TOKEN_KEYS, (9, 9, 9, 9, 0, 9))),
+                        cost_delta_usd=9.0)
+        self.result_for("j0004", block="1.2", phase=1, flags=[])       # agrees with its line
+        n0 = len(self.ledger_lines())
+        out = self.usage("--phase", "1")
+        self.assertEqual(out["status_from_result"], ["j0002"])
+        row = {x["job"]: x for x in out["rows"]}["j0002"]
+        self.assertEqual((row["status"], row["flags"], row["reason"]), ("lost", ["cold_start"], "supervisor_dead"))
+        self.assertEqual(row["tokens"]["cache_read"], 1234567)          # the ledger's, not the result's
+        self.assertAlmostEqual(row["cost_delta_usd"], 1.2345, places=6)
+        self.assertAlmostEqual(out["total"]["cost_usd"], 2.0845, places=6)
+        self.assertEqual(out["cold_starts"], 2)                          # the result's flags count
+        self.assertEqual({x["job"]: x["status"] for x in out["rows"]}["j0004"], "ok")
+        # the ledger itself is unchanged (append-only)
+        led = self.ledger_lines()
+        self.assertEqual(len(led), n0)
+        self.assertEqual([x["status"] for x in led if x["job"] == "j0002"], ["ok"])
+        self.assertEqual(self.usage("--phase", "2")["status_from_result"], [])
 
     def test_r1_nothing_missing_takes_no_lock(self):
         self.result_for("j0002")                       # a result that has its ledger line
@@ -2862,11 +3012,12 @@ class LaunchTests(LaunchBase):
         """A stubbed ping: launched by `wait`'s tick, returned once through `wait`'s `pings` (never
         `done`); a `partial` ping refreshes last_hit, a missed or failed one does not and carries
         `ping_missed`; the in-flight mark clears; each ping has its ledger line."""
-        self.setup_run("--warm-margin-s", "3600")      # expiring from the last hit on
+        # expiring from 1800 s after the last hit on (a margin must be below ttl_s; Minor 2)
+        self.setup_run("--warm-margin-s", "1800")
         self.train("B1")
         self.finish("j0001")
         base_session = self.reg()["bases"]["B1"]["session"]
-        old = forks._iso(time.time() - 100)
+        old = forks._iso(time.time() - 2000)
         self.set_last_hit(old)
         # 1. partial, launched by wait's tick (the wait itself is the activity)
         with mock.patch.object(forks, "PING_PROMPT", "Reply OK\n@stub cache=partial"):
@@ -3035,12 +3186,14 @@ class LaunchTests(LaunchBase):
         self.clone_base("Bl", base_commit=c7)
         self.assert_due(self.fake_fork("Bl"), "lines:1600")
         # every clone from here on starts at HEAD
-        # delta_tokens: chars / 4 of the delta header plus the notes the brief lists > 12000
+        # delta_tokens: chars / 4 of the delta header plus the attached delta notes > 12000 (I1)
-        with open(os.path.join(self.state, "notes", "big-note.md"), "w") as f:
+        big_note = os.path.join(self.state, "notes", "big-note.md")
+        with open(big_note, "w") as f:
             f.write("n" * 60000)
         self.clone_base("Bd")
-        self.assert_due(self.fake_fork("Bd"))                                    # brief lists no note
+        self.assert_due(self.fake_fork("Bd"))                                    # no note attached
-        out = self.fake_fork("Bd", brief=self.brief(text="Read notes/big-note.md first."))
+        self.assert_due(self.fake_fork("Bd", brief=self.brief(text="Read notes/big-note.md first.")))
+        out = self.fake_fork("Bd", "--delta-note", big_note)
         self.assertEqual(len(out["reasons"]), 1, out)
         self.assertRegex(out["reasons"][0], r"^delta_tokens:\d+$")
         self.assertGreater(int(out["reasons"][0].split(":")[1]), 15000)
@@ -3102,6 +3255,105 @@ class LaunchTests(LaunchBase):
             with self.subTest(base=name):
                 self.assert_due(rows[name], *want)
 
+    def test_delta_tokens_counts_only_what_the_base_does_not_know(self):
+        """I1: delta_tokens = the delta header + `--delta-note` files newer than the base's training
+        + a `## Changes since base` section of the brief. Large reference docs a brief merely names
+        never count; an attached note is in the fork's prompt with its path; a base trained or
+        rebased after the note's mtime does not count it, so a rebase clears the reason."""
+        self.setup_run(cap=4)
+        self.train("B1")
+        self.finish("j0001")
+        os.makedirs(os.path.join(self.cwd, "docs"))
+        for rel in ("docs/build-plan.md", "docs/run-log.md"):
+            with open(os.path.join(self.cwd, *rel.split("/")), "w") as f:
+                f.write("r" * 80000)
+        with open(os.path.join(self.state, "notes", "1.1.md"), "w") as f:
+            f.write("h" * 80000)
+        named = self.brief(text="Read docs/build-plan.md, docs/run-log.md and notes/1.1.md; edit "
+                                "docs/run-log.md and %s." % os.path.join(self.cwd, "docs", "build-plan.md"))
+        self.assert_due(self.fake_fork("B1", brief=named))                       # named only: nothing
+        # a large attached note trips it, and the fork's prompt carries it with its path
+        note = os.path.join(self.tmp, "delta-1.md")
+        with open(note, "w", newline="\n") as f:
+            f.write("NOTE-MARKER\n" + "d" * 60000 + "\n")
+        out = self.fake_fork("B1", "--delta-note", note, brief=named)
+        self.assertEqual(len(out["reasons"]), 1, out)
+        self.assertRegex(out["reasons"][0], r"^delta_tokens:\d+$")
+        self.assertGreater(int(out["reasons"][0].split(":")[1]), 15000)
+        with open(os.path.join(self.state, "briefs", out["job"] + ".md"), encoding="utf-8") as fh:
+            prompt = fh.read()
+        delta_at = prompt.index("## Delta since training")
+        self.assertLess(delta_at, prompt.index("`%s`" % forks.norm_path(note)))
+        self.assertLess(prompt.index("NOTE-MARKER"), prompt.index("## Rules for this fork"))
+        self.assertEqual(self.reg()["jobs"][out["job"]]["delta_notes"], [forks.norm_path(note)])
+        # a small attached note does not; a missing one is refused
+        small = os.path.join(self.tmp, "delta-small.md")
+        with open(small, "w") as f:
+            f.write("small\n")
+        self.assert_due(self.fake_fork("B1", "--delta-note", small))
+        self.assertEqual(self.fake_fork("B1", "--delta-note", os.path.join(self.tmp, "nope.md"),
+                                        code=2)["error"], "cannot read delta note")
+        # a base trained after the note's mtime already knows it
+        mtime = os.path.getmtime(note)
+        self.clone_base("Bk", trained_at=forks._iso(mtime + 60))
+        self.assert_due(self.fake_fork("Bk", "--delta-note", note))
+        # a large `## Changes since base` section trips it; the same text under another heading not
+        body = "c" * 60000
+        out = self.fake_fork("Bk", brief=self.brief(text="## Changes since base\n\n%s\n\n## Gate\n\nrun it" % body))
+        self.assertEqual(len(out["reasons"]), 1, out)
+        self.assertRegex(out["reasons"][0], r"^delta_tokens:\d+$")
+        self.assert_due(self.fake_fork("Bk", brief=self.brief(text="## Background\n\n%s\n" % body)))
+        # a rebase after the note clears the reason; the rebase's prompt carries the note too
+        r = self.cmd("rebase", "--base", "B1", "--as", "B2", "--delta-note", note,
+                     "--brief", self.brief("turns=1", text="Absorb."))
+        with open(os.path.join(self.state, "briefs", r["job"] + ".md"), encoding="utf-8") as fh:
+            rprompt = fh.read()
+        self.assertIn("NOTE-MARKER", rprompt)
+        self.assertLess(rprompt.index("## Delta since the base commit"), rprompt.index("NOTE-MARKER"))
+        self.assertLess(rprompt.index("NOTE-MARKER"), rprompt.index("## Brief"))
+        self.assertEqual(self.finish(r["job"])["status"], "ok")
+        self.assertGreater(forks._ts(self.reg()["bases"]["B2"]["trained_at"]), mtime)
+        self.assert_due(self.fake_fork("B2", "--delta-note", note))
+        # list counts the delta header only
+        rows = {b["name"]: b for b in self.cmd("list")["bases"]}
+        self.assert_due(rows["B2"])
+
+    def test_train_supersedes(self):
+        """Minor 1: `train --supersedes B` (repeatable) marks each named active base superseded by
+        the new one when its training ends ok: no more pings, and a fork needs --force. Unknown
+        names are refused at launch; a failed training changes nothing."""
+        self.setup_run(cap=4)
+        self.train("B1")
+        self.finish("j0001")
+        self.train("B0")
+        self.finish("j0002")
+        out = self.cmd("train", "--base", "B2", "--brief", self.brief(), "--supersedes", "nope", code=2)
+        self.assertEqual((out["error"], out["bases"]), ("unknown base", ["nope"]))
+        out = self.cmd("train", "--base", "B2", "--brief", self.brief(), "--supersedes", "B2", code=2)
+        self.assertIn("itself", out["error"])
+        self.assertNotIn("B2", self.reg()["bases"])
+        # a failed training: nothing changes
+        t = self.cmd("train", "--base", "B2", "--brief", self.brief("error"), "--supersedes", "B1")
+        self.assertEqual(self.finish(t["job"])["status"], "failed")
+        reg = self.reg()
+        self.assertEqual((reg["bases"]["B2"]["status"], reg["bases"]["B1"]["status"]), ("failed", "active"))
+        self.assertNotIn("superseded_by", reg["bases"]["B1"])
+        # an ok training supersedes every named active base
+        t = self.cmd("train", "--base", "B3", "--brief", self.brief(), "--supersedes", "B1",
+                     "--supersedes", "B0", "--supersedes", "B2")
+        self.assertEqual(self.finish(t["job"])["status"], "ok")
+        reg = self.reg()
+        for name in ("B1", "B0"):
+            with self.subTest(base=name):
+                b = reg["bases"][name]
+                self.assertEqual((b["status"], b["superseded_by"]), ("superseded", "B3"))
+                self.assertIsNotNone(b["superseded_at"])
+        self.assertEqual(reg["bases"]["B2"]["status"], "failed")                 # not active: untouched
+        self.assertEqual(reg["bases"]["B3"]["status"], "active")
+        self.assertEqual(forks.ping_decision(reg, "B1")["why"], "status:superseded")
+        self.assertIn("not active", self.fake_fork("B1", code=2)["error"])
+        self.fake_fork("B1", "--force")
+
     def test_train_overrides_and_init_model_change(self):
         """M10 (plan 4.3): `train` takes init's launch flags as overrides; the base records its
         launch fields and their hash, and every later launch on it uses that record. `init`