🤖 forks: rebase and re-train flags (block 2.2)

plugins/spire/skills/forks/forks.py

@@ -122,7 +122,7 @@ SCRUB_WATCH_PREFIXES = ("CLAUDE_", "ANTHROPIC_")   # `where` lists unknown vars
 # C8: the user settings file. Module-level so tests can point it at a temp file.
 USER_SETTINGS_PATH: str | None = None
 
-PHASE2_COMMANDS = ("rebase", "usage")
+PHASE2_COMMANDS = ("usage",)
 # N3: a fork may only run where, list, usage and wait (`tick` launches pings).
 CHILD_REFUSED = ("train", "fork", "continue", "rebase", "kill", "expire", "init", "tick")
 
@@ -156,6 +156,21 @@ LAUNCH_FIELDS = ("model", "effort", "permission_mode", "tools", "allowed_tools",
 MU_FIELDS = ("inputTokens", "outputTokens", "cacheReadInputTokens", "cacheCreationInputTokens")
 TOKEN_KEYS = ("input", "output", "cache_creation", "cache_creation_1h", "cache_creation_5m", "cache_read")
 TEMPLATES_DIR = "templates"
+CAPPED_KINDS = ("fork", "continue", "rebase")   # parallel_cap counts these (pings never)
+
+# Rebase and re-train flags (design 5, block 2.2).
+# Runtime fields of a `system/init` event (compared with the base's; M6 adds `skills`).
+RUNTIME_FIELDS = ("tools", "permissionMode", "plugins", "agents", "apiKeySource", "skills")
+# Fingerprint-defining launch fields: a difference means a new base (retrain_due `defining:`).
+DEFINING_FIELDS = ("model", "cwd", "tools", "permission_mode", "agent", "append_system_prompt")
+# M11: a note counts for a job when its mtime is at or after the job's creation. File times on
+# Windows can lag the precise clock by a timer tick; this slack absorbs that (a stale note from an
+# earlier job of the same block is minutes old).
+NOTE_MTIME_SLACK_S = 2.0
+# P0.7: transcripts live in <projects dir>/<cwd, every non-alphanumeric -> '-'>/<session>.jsonl.
+# Module-level so tests can point it at a temp dir.
+CLAUDE_PROJECTS_DIR: str | None = None
+TRANSCRIPT_TAIL_BYTES = 256 * 1024
 
 # CIM hidden spawn (P0.1 rung 3 + ShowWindow=0). Command line and cwd arrive through env vars.
 POWERSHELL_CIM = (
@@ -237,8 +252,18 @@ def canonical_json(obj) -> str:
     return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=True)
 
 
+def fingerprint_view(launch: dict) -> dict:
+    """What the fingerprint hash covers (M7): the launch fields, with the mcp config's **content**
+    in place of its per-run path, so the same launch hashes the same in every run."""
+    view = dict(launch)
+    m = view.get("mcp")
+    if isinstance(m, dict):
+        view["mcp"] = {"strict": bool(m.get("strict")), "servers": dict(MCP_EMPTY["mcpServers"])}
+    return view
+
+
 def fingerprint_hash(launch: dict) -> str:
-    return _sha256_bytes(canonical_json(launch).encode("utf-8"))
+    return _sha256_bytes(canonical_json(fingerprint_view(launch)).encode("utf-8"))
 
 
 # --------------------------------------------------------------------------------------
@@ -248,10 +273,36 @@ def fingerprint_hash(launch: dict) -> str:
 _GITBASH_DRIVE = re.compile(r"^/([A-Za-z])(?=/|$)")
 
 
+def find_cygpath() -> str | None:
+    """`cygpath` on PATH (Git Bash ships it), or None. Tests monkeypatch this lookup."""
+    return shutil.which("cygpath")
+
+
+def cygpath_w(p: str) -> str:
+    """M13: a Git Bash rooted path that is not a drive form (`/tmp/x`, `/usr/...`) converted with
+    `cygpath -w`. Refused (exit 2) when cygpath is not on PATH or fails; never mapped onto the
+    cwd's drive."""
+    exe = find_cygpath()
+    if not exe:
+        raise ForksError("Git Bash rooted path is not a drive form and cygpath is not on PATH to "
+                         "convert it; pass a drive path (G:\\... or /g/...)", path=p)
+    try:
+        r = _run([exe, "-w", p], timeout=30)
+    except (OSError, subprocess.SubprocessError) as e:
+        raise ForksError("cygpath failed", path=p, detail=str(e))
+    out = r.stdout.decode("utf-8", "replace").strip()
+    if r.returncode != 0 or not out:
+        raise ForksError("cygpath failed", path=p,
+                         detail=r.stderr.decode("utf-8", "replace").strip()[:200] or "no output")
+    return out
+
+
 def norm_path(p: str, cwd: str | None = None, windows: bool | None = None) -> str:
     """Normalise relative, `G:\\…`, `G:/…` and Git Bash `/g/…` forms.
 
-    Windows: absolute path with backslashes and an upper-case drive letter.
+    Windows: absolute path with backslashes and an upper-case drive letter. A Git Bash rooted
+    path that is not a drive form (`/tmp/x`) goes through `cygpath -w` (M13); a native rooted
+    path (`\\x`) keeps its Windows meaning (the cwd's drive).
     POSIX: `os.path.abspath` semantics (relative to `cwd`).
     """
     if windows is None:
@@ -265,6 +316,8 @@ def norm_path(p: str, cwd: str | None = None, windows: bool | None = None) -> st
     m = _GITBASH_DRIVE.match(p)
     if m:
         p = m.group(1).upper() + ":" + (p[2:] or "/")
+    elif p.startswith("/") and not p.startswith("//"):
+        p = cygpath_w(p)
     p = p.replace("/", "\\")
     drive, rest = ntpath.splitdrive(p)
     if not (drive and rest.startswith("\\")):
@@ -1153,25 +1206,41 @@ def context_tokens(usage: dict | None) -> int:
         + int(usage.get("cache_read_input_tokens") or 0)
 
 
+def plugin_id(p: dict) -> str:
+    """`name@version`; a plugin without a version is identified by its path's last segment (M6),
+    so an in-place update of a versionless plugin (a new cache dir) shows as drift."""
+    ver = p.get("version")
+    if ver:
+        return "%s@%s" % (p.get("name"), ver)
+    path = str(p.get("path") or "").rstrip("/\\")
+    seg = re.split(r"[/\\]", path)[-1] if path else ""
+    return "%s@path:%s" % (p.get("name"), seg) if seg else "%s@" % p.get("name")
+
+
+def _names(items) -> list[str]:
+    return sorted(str(a) if not isinstance(a, dict) else str(a.get("name")) for a in (items or []))
+
+
 def runtime_fingerprint(init: dict | None) -> dict | None:
-    """The runtime fields of a `system/init` event that must match the base's."""
+    """The runtime fields of a `system/init` event that must match the base's (RUNTIME_FIELDS)."""
     if not init:
         return None
     tools = sorted(str(x) for x in (init.get("tools") or []))
-    plugins = sorted("%s@%s" % (p.get("name"), p.get("version") or "")
+    plugins = sorted(plugin_id(p) for p in (init.get("plugins") or []) if isinstance(p, dict))
-                     for p in (init.get("plugins") or []) if isinstance(p, dict))
-    agents = sorted(str(a) if not isinstance(a, dict) else str(a.get("name")) for a in (init.get("agents") or []))
     return {"tools": _sha256_bytes(canonical_json(tools).encode("utf-8")),
-            "permissionMode": init.get("permissionMode"), "plugins": plugins, "agents": agents,
+            "permissionMode": init.get("permissionMode"), "plugins": plugins,
-            "apiKeySource": init.get("apiKeySource")}
+            "agents": _names(init.get("agents")), "apiKeySource": init.get("apiKeySource"),
+            "skills": _names(init.get("skills"))}
 
 
 def runtime_drift(rt: dict | None, ref: dict | None, launch: dict) -> list[str]:
+    """Runtime fields of `rt` that differ from the base's `ref`. A field the reference does not
+    carry (a base recorded before that field was fingerprinted) is not compared."""
     if not rt:
         return []
     out = []
     if ref:
-        out += [k for k in sorted(rt) if rt.get(k) != ref.get(k)]
+        out += [k for k in sorted(rt) if k in ref and rt.get(k) != ref.get(k)]
     elif rt.get("permissionMode") not in (None, launch.get("permission_mode")):
         out.append("permissionMode")
     return out
@@ -1321,6 +1390,20 @@ def note_rel(block) -> str | None:
     return None if block in (None, "") else "notes/%s.md" % _safe_name(block)
 
 
+def note_written_by(path: str, entry: dict) -> bool:
+    """M11: the note file exists and was written during this job: its mtime is at or after the
+    job's creation (less NOTE_MTIME_SLACK_S for coarse file times). A note left by an earlier job
+    of the same block (a re-run, or the fork a continue resumes) does not count."""
+    try:
+        mtime = os.path.getmtime(path)
+    except OSError:
+        return False
+    start = entry.get("created_ts")
+    if start is None:
+        start = _ts(entry.get("created_at"))
+    return start is None or mtime >= float(start) - NOTE_MTIME_SLACK_S
+
+
 def _phase(v):
     if v is None:
         return None
@@ -1381,9 +1464,9 @@ def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint:
     """Build the result, the ledger line, and apply the registry updates for one finished job.
     `hint`: "exited" (decide ok/failed from the stream), or a final status given by the caller."""
     cfg = reg["config"]
-    launch = cfg["launch"]
     kind = entry["kind"]
     base = reg["bases"].get(entry.get("base")) if entry.get("base") else None
+    launch = launch_record(base, cfg)
     res = ss.result if ss else None
     u = usage_from_result(res, entry.get("model_usage_before") or {}, entry.get("cost_before") or 0.0) \
         if res else None
@@ -1423,16 +1506,17 @@ def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint:
     if denials:
         flags.append("needs_attention")
     cache = cache_verdict(ss.first_usage if ss else None, entry.get("base_context")) \
-        if kind in ("fork", "continue", "ping") else None
+        if kind in ("fork", "continue", "ping", "rebase") else None
     rt = runtime_fingerprint(ss.init) if ss else None
     drift = list(entry.get("launch_drift") or [])
     ref = (base or {}).get("runtime_fingerprint") if kind != "train" else None
-    drift += runtime_drift(rt, ref, launch)
+    rt_drift = runtime_drift(rt, ref, launch)
+    drift += rt_drift
     drift = sorted(set(drift))
     fingerprint = "ok" if not drift else "drift:" + ",".join(drift)
     base_state = entry.get("base_state_at_launch")
-    if kind in ("fork", "continue") and (base_state == "cold" or (cache not in (None, "hit")
+    if kind in ("fork", "continue", "rebase") and (base_state == "cold" or (
-                                                                   and fingerprint == "ok")):
+            cache not in (None, "hit") and fingerprint == "ok")):
         flags.append("cold_start")
     if u and wrote_only_5m(u["tokens"]):
         flags.append("ttl_downgrade")
@@ -1445,7 +1529,7 @@ def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint:
     flags = _uniq(flags)
     turns = int(res.get("num_turns") or 0) if res else (ss.turns if ss else 0)
     note = note_rel(entry.get("block")) if kind != "train" else None
-    if note and not os.path.isfile(os.path.join(state, note)):
+    if note and not note_written_by(os.path.join(state, note), entry):
         note = None
     session = (ss.session_id if ss and ss.session_id else None) or entry.get("session")
     tokens = u["tokens"] if u else {k: None for k in TOKEN_KEYS}
@@ -1472,7 +1556,10 @@ def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint:
     # registry updates
     entry.update({"status": status, "flags": flags, "reason": reason, "reason_detail": detail,
                   "ended_at": ended, "session": session, "turns": turns, "exit_code": exit_code,
-                  "fingerprint": fingerprint, "cache": cache})
+                  "fingerprint": fingerprint, "cache": cache,
+                  # runtime fields that drifted, None when the call reported no init event:
+                  # `rebase_due: runtime_drift:<fields>` reads the base's last call with one.
+                  "runtime_drift": rt_drift if rt else None})
     if ss and ss.first_at:
         entry["first_event_at"] = _iso(ss.first_at)
     if u:
@@ -1488,7 +1575,7 @@ def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint:
         hit_t = None
         if ss and kind == "train":
             hit_t = ss.last_assistant_at
-        elif ss and kind == "fork":
+        elif ss and kind in ("fork", "rebase"):    # a rebase's first request reads the base prefix
             hit_t = ss.first_at
         elif ping_refreshed:
             hit_t = ss.first_at
@@ -1509,9 +1596,45 @@ def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint:
                              "runtime_fingerprint": rt, "trained_at": ended})
             elif base.get("status") == "training":
                 base["status"] = "failed"
+        if kind == "rebase" and status == "ok":
+            _rebase_succeeded(reg, entry, base, ss, u, rt, base_commit, ended)
     return result, ledger
 
 
+def _rebase_succeeded(reg: dict, entry: dict, old: dict, ss, u, rt, base_commit, ended: str) -> None:
+    """A rebase ended `ok`: its session becomes base `entry.rebase_as` (active, parent = the old
+    base, one more rebase in the lineage, HEAD at the end, context from its last assistant
+    message, the launch record computed at launch: the old base's, with the binary it ran
+    under); the old base becomes `superseded`. On any other status nothing here runs: the old
+    base stays active and the new one is never created."""
+    name = entry.get("rebase_as")
+    if not name:
+        return
+    existing = reg["bases"].get(name)
+    if existing and existing.get("status") != "failed":
+        entry["rebase_conflict"] = "base %s exists (%s); not replaced" % (name, existing.get("status"))
+        return
+    launch = entry.get("new_launch") or launch_record(old, reg["config"])
+    reg["bases"][name] = {
+        "name": name, "status": "active", "session": entry.get("session"), "cwd": launch["cwd"],
+        "launch": launch, "fingerprint": fingerprint_hash(launch),
+        "overrides": list(old.get("overrides") or []), "binary": launch["binary"],
+        "created_at": entry.get("created_at"), "train_job": None, "rebase_job": entry["job"],
+        "parent": old.get("name") or entry.get("base"),
+        "lineage_rebases": int(old.get("lineage_rebases") or 0) + 1,
+        "phase": entry.get("phase"), "last_hit": _iso(ss.last_assistant_at) if ss and ss.last_assistant_at else None,
+        "base_commit": base_commit, "context_tokens": context_tokens(ss.last_usage if ss else None),
+        "training_cost_usd": u["cost_delta"] if u else None,
+        "cost_cum_usd": u["total"] if u else None, "model_usage_cum": u["model_usage"] if u else {},
+        "runtime_fingerprint": rt, "trained_at": ended, "forks": [],
+        "ttl_s": infer_ttl(u["tokens"] if u else None, old.get("ttl_s")), "ttl_from": entry["job"],
+        "ping_in_flight": None,
+    }
+    old["status"] = "superseded"
+    old["superseded_by"] = name
+    old["superseded_at"] = ended
+
+
 def finalize_job(state: str, job: str, *, ss: StreamState | None, hint: str, flags=(), reason=None,
                  detail=None, exit_code=None, wall_s=None, base_commit=None) -> dict | None:
     """Write `results/<job>.json`, the registry update and the ledger line, exactly once, under the
@@ -1759,6 +1882,18 @@ def build_parser() -> _Parser:
     t.add_argument("--brief", required=True)
     t.add_argument("--block")
     t.add_argument("--phase")
+    # M10 (plan 4.3): train takes launch-field overrides, the same flags as init; a field not
+    # given comes from the run config. The base records what it was trained with.
+    g = t.add_argument_group("launch overrides (default: the init config)")
+    g.add_argument("--model", help="full model id")
+    g.add_argument("--effort", choices=EFFORTS)
+    g.add_argument("--permission-mode")
+    g.add_argument("--tools", help="comma-separated tool names")
+    g.add_argument("--allowed-tools", nargs="+", action="extend", default=None)
+    g.add_argument("--disallowed-tools", nargs="+", action="extend", default=None)
+    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")
 
     def launch_overrides(sp):
         # Accepted only so that they can be refused with a clear message (plan 4.3).
@@ -1809,9 +1944,16 @@ def build_parser() -> _Parser:
 
     cmd("tick", "One keep-warm pass: ping every base the rule picks (design 5).")
 
-    for name, text in (("rebase", "Rebase a base onto the current state (Phase 2)."),
+    r = cmd("rebase", "Fork a base to absorb what changed; its session becomes a new base.")
-                       ("usage", "Aggregate the ledger (Phase 2).")):
+    r.add_argument("--base", required=True)
-        cmd(name, text)   # arguments arrive with Phase 2; until then extra args are tolerated
+    r.add_argument("--as", dest="as_name", required=True, help="name of the new base")
+    r.add_argument("--brief", required=True)
+    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)
+    launch_overrides(r)
+
+    cmd("usage", "Aggregate the ledger (Phase 2).")   # arguments arrive with 2.3; extra args tolerated
     return p
 
 
@@ -1875,30 +2017,31 @@ def _resolve_cwd(arg: str | None) -> str:
     return cwd
 
 
-def build_init_config(args, state: str) -> dict:
+def agent_path(cwd: str, name: str) -> str:
-    """Validate the launch fields and build the config block. Writes nothing."""
+    return os.path.join(cwd, ".claude", "agents", name + ".md")
-    model = args.model
+
+
+def build_launch(state: str, *, model, effort, permission_mode, tools, allowed, disallowed, agent,
+                 cwd: str, binary_path: str) -> tuple[dict, str]:
+    """Validate launch fields and build a launch record (init's config and train's base record;
+    design 3.3). `tools`/`allowed`/`disallowed` are lists, `cwd` and `binary_path` normalised.
+    Returns (launch, model admission `via`). Writes nothing."""
     if not MODEL_RE.match(model or ""):
         raise ForksError("model must be a full id (aliases are refused)", model=model,
                          pattern=MODEL_RE.pattern)
-    mode = args.permission_mode
+    mode = permission_mode
     if mode in REFUSED_PERMISSION_MODES:
         raise ForksError("permission mode refused", permission_mode=mode,
                          detail="plan mode swaps models; use dontAsk")
     if mode not in PERMISSION_MODES:
         raise ForksError("unknown permission mode", permission_mode=mode,
                          allowed=[m for m in PERMISSION_MODES if m not in REFUSED_PERMISSION_MODES])
-    tools = _tool_list([args.tools])
+    if effort not in EFFORTS:
+        raise ForksError("unknown effort", effort=effort, allowed=list(EFFORTS))
     if not tools:
         raise ForksError("--tools is empty")
-    allowed = _tool_list(args.allowed_tools)
+    if not os.path.isdir(cwd):
-    if allowed is None:
+        raise ForksError("--cwd is not a directory", cwd=cwd)
-        allowed = list(tools)
-    disallowed = _tool_list(args.disallowed_tools)
-    if disallowed is None:
-        disallowed = list(DEFAULT_DISALLOWED)
-
-    cwd = _resolve_cwd(args.cwd)
 
     adm = model_admission(model, cwd)
     if not adm["admitted"]:
@@ -1909,39 +2052,58 @@ def build_init_config(args, state: str) -> dict:
                      "add the id (or its family alias) to one of these files; forks never writes "
                      "settings." % (model, ", ".join(files))))
 
-    bpath, how = resolve_binary(args.binary)
+    binfo = binary_info(binary_path)
-    if bpath is None:
-        raise ForksError("no claude binary (no --binary, no $CLAUDE_CODE_EXECPATH, none on PATH)")
-    binfo = binary_info(bpath)
 
     pre = preamble_path()
     if not os.path.isfile(pre):
         raise ForksError("preamble.md missing next to forks.py", path=pre)
 
-    agent = None
+    agent_rec = None
-    if args.agent:
+    if agent:
-        name = args.agent
+        name = agent
         if not re.match(r"^[A-Za-z0-9_.:-]+$", name) or name in (".", ".."):
             raise ForksError("bad agent name", agent=name)
-        apath = os.path.join(cwd, ".claude", "agents", name + ".md")
+        apath = agent_path(cwd, name)
         if not os.path.isfile(apath):
             raise ForksError("agent file not found", agent=name, path=apath)
-        agent = {"name": name, "path": apath, "sha256": _sha256_file(apath)}
+        agent_rec = {"name": name, "path": apath, "sha256": _sha256_file(apath)}
 
     launch = {
         "model": model,
-        "effort": args.effort,
+        "effort": effort,
         "permission_mode": mode,
-        "tools": tools,
+        "tools": list(tools),
-        "allowed_tools": allowed,
+        "allowed_tools": list(allowed),
-        "disallowed_tools": disallowed,
+        "disallowed_tools": list(disallowed),
         "mcp": {"strict": True, "config": mcp_empty_path(state)},
         "append_system_prompt": {"path": pre, "sha256": _sha256_file(pre)},
-        "agent": agent,
+        "agent": agent_rec,
         "cwd": cwd,
         "binary": binfo,
         "env": {"pinned": dict(ENV_PINNED_LAUNCH)},
     }
+    return launch, adm["via"]
+
+
+def build_init_config(args, state: str) -> dict:
+    """Validate the launch fields and build the config block. Writes nothing."""
+    tools = _tool_list([args.tools])
+    allowed = _tool_list(args.allowed_tools)
+    if allowed is None:
+        allowed = list(tools)
+    disallowed = _tool_list(args.disallowed_tools)
+    if disallowed is None:
+        disallowed = list(DEFAULT_DISALLOWED)
+    cwd = _resolve_cwd(args.cwd)
+    if not MODEL_RE.match(args.model or ""):     # before the binary lookup, as in Phase 1
+        raise ForksError("model must be a full id (aliases are refused)", model=args.model,
+                         pattern=MODEL_RE.pattern)
+    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,
+                               permission_mode=args.permission_mode, tools=tools, allowed=allowed,
+                               disallowed=disallowed, agent=args.agent, cwd=cwd, binary_path=bpath)
     cfg = {
         "launch": launch,
         "fingerprint": fingerprint_hash(launch),
@@ -1954,7 +2116,7 @@ def build_init_config(args, state: str) -> dict:
         "max_minutes": args.max_minutes,
         "stall_after_s": args.stall_after_s,
         "thresholds": {k: getattr(args, k) for k in THRESHOLD_DEFAULTS},
-        "model_admission": adm["via"],
+        "model_admission": via,
     }
     return cfg
 
@@ -1982,10 +2144,16 @@ def cmd_init(args) -> tuple[dict, int]:
         # `model_admission` and `binary_resolved_by` describe how, not what: not a change.
         changed_fields = [f for f in changed_fields
                           if f not in ("model_admission", "binary_resolved_by")]
-        if old is not None and changed_fields and reg.get("bases"):
+        # Block 2.2: values may change after bases exist (Phase 1 refused it). Every base keeps
-            raise ForksError("init values differ and a base exists; train a new run instead",
+        # launching with its own record; a base that did not override a changed defining field
-                             fields=changed_fields, bases=sorted(reg["bases"]), state=state)
+        # reports `retrain_due: defining:<fields>`. A base from before per-base records gets the
+        # config it was trained with as its record first.
         if old is None or changed_fields:
+            if old is not None:
+                for b in reg.get("bases", {}).values():
+                    if not b.get("launch") and old.get("launch"):
+                        b["launch"] = old["launch"]
+                        b.setdefault("overrides", [])
             reg["config"] = cfg
             reg.setdefault("run", {})["init_at"] = _iso()
             write_registry(state, reg)
@@ -1993,7 +2161,7 @@ def cmd_init(args) -> tuple[dict, int]:
             cfg = old
     return {"ok": True, "state": state, "registry": registry_path(state), "created": created,
             "changed": bool(old is not None and changed_fields), "changed_fields": changed_fields,
-            "config": cfg}, EXIT_OK
+            "bases": sorted(reg.get("bases", {})), "config": cfg}, EXIT_OK
 
 
 def _not_implemented(block=None, phase=None):
@@ -2040,7 +2208,8 @@ def _check_binary(ref: dict, launch: dict) -> tuple[dict, list[str]]:
     path = launch["binary"]["path"]
     if not os.path.isfile(path):
         raise ForksError("claude binary missing (the stored path is gone); the base cannot be "
-                         "launched with another binary", binary=path, reason="binary_missing")
+                         "launched with another binary", binary=path, reason="binary_missing",
+                         retrain_due=True, reasons=["binary_missing"])
     info = binary_info(path)
     drift = [] if all(info.get(k) == (ref or {}).get(k) for k in ("version", "size", "mtime")) else ["binary"]
     asp = launch["append_system_prompt"]
@@ -2107,6 +2276,233 @@ def delta_header(cwd: str, base_commit: str | None) -> str:
     return "\n\n".join(parts)
 
 
+# --- rebase and re-train flags (design 5, block 2.2) ------------------------------------
+
+
+def git_delta_counts(cwd: str, base_commit: str) -> tuple[int | None, int | None]:
+    """(commits since `base_commit`, insertions + deletions of `git diff --shortstat <base_commit>`);
+    None where git fails."""
+    commits = lines = None
+    with contextlib.suppress(OSError, subprocess.SubprocessError, ValueError):
+        r = git(["rev-list", "--count", "%s..HEAD" % base_commit], cwd=cwd)
+        if r.returncode == 0:
+            commits = int(r.stdout.decode("utf-8", "replace").strip() or 0)
+    with contextlib.suppress(OSError, subprocess.SubprocessError, ValueError):
+        s = git(["diff", "--shortstat", base_commit], cwd=cwd)
+        if s.returncode == 0:
+            txt = s.stdout.decode("utf-8", "replace")
+            ins = re.search(r"(\d+) insertion", txt)
+            dels = re.search(r"(\d+) deletion", txt)
+            lines = (int(ins.group(1)) if ins else 0) + (int(dels.group(1)) if dels else 0)
+    return commits, lines
+
+
+_NOTE_REF = re.compile(r"[^\s`'\"()<>\[\]{},;|*]+\.md\b")
+
+
+def listed_notes(brief: str | None, cwd: str, state: str) -> list[str]:
+    """The `.md` files a brief names that exist: absolute paths as given, relative ones against
+    the base's cwd, then against the state dir (`notes/1.1.md`). Each file once."""
+    out, seen = [], set()
+    for ref in _NOTE_REF.findall(brief or ""):
+        for root in (cwd, state):
+            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 not in seen:
+                    seen.add(key)
+                    out.append(p)
+                break
+    return out
+
+
+def _text_chars(path: str) -> int:
+    try:
+        return len(read_bytes_retry(path).decode("utf-8", "replace"))
+    except OSError:
+        return 0
+
+
+def projects_dir() -> str:
+    return CLAUDE_PROJECTS_DIR or os.path.join(os.path.expanduser("~"), ".claude", "projects")
+
+
+def transcript_path(cwd: str, session: str) -> str:
+    """P0.7: `~/.claude/projects/<cwd with every non-alphanumeric -> '-'>/<session>.jsonl`."""
+    return os.path.join(projects_dir(), re.sub(r"[^A-Za-z0-9]", "-", cwd), session + ".jsonl")
+
+
+def transcript_context(path: str) -> int | None:
+    """Context tokens of the last assistant line (with `message.usage`) of a transcript, read from
+    the tail; None when the file or such a line is missing."""
+    try:
+        size = os.path.getsize(path)
+    except OSError:
+        return None
+    n = TRANSCRIPT_TAIL_BYTES
+    while True:
+        start = max(0, size - n)
+        try:
+            with open(path, "rb") as f:
+                f.seek(start)
+                data = f.read()
+        except OSError:
+            return None
+        lines = data.splitlines()
+        if start > 0:
+            lines = lines[1:]                # the first line of a tail may be cut
+        for ln in reversed(lines):
+            if b'"assistant"' not in ln:
+                continue
+            try:
+                obj = json.loads(ln.decode("utf-8", "replace"))
+            except ValueError:
+                continue
+            if isinstance(obj, dict) and obj.get("type") == "assistant":
+                u = (obj.get("message") or {}).get("usage")
+                if u:
+                    return context_tokens(u)
+        if start == 0:
+            return None
+        n *= 4
+
+
+def base_context_now(base: dict, launch: dict) -> int | None:
+    """The base's context tokens: from its transcript's last assistant line when the file exists
+    (P0.7), else the context stored at training / rebase."""
+    ctx = None
+    if base.get("session"):
+        ctx = transcript_context(transcript_path(launch["cwd"], base["session"]))
+    return ctx if ctx is not None else base.get("context_tokens")
+
+
+def _int_phase(p) -> bool:
+    return isinstance(p, int) and not isinstance(p, bool)
+
+
+def phase_boundary(reg: dict, name: str, phase) -> bool:
+    """The fork's `--phase` is greater than the highest phase of any fork from this base, or than
+    the base's own phase (integers only; a side that is not known does not count)."""
+    if not _int_phase(phase):
+        return False
+    base = reg["bases"][name]
+    forks_ = [e.get("phase") for e in reg.get("jobs", {}).values()
+              if e.get("base") == name and e.get("kind") == "fork" and _int_phase(e.get("phase"))]
+    own = base.get("phase")
+    return bool(forks_ and phase > max(forks_)) or bool(_int_phase(own) and phase > own)
+
+
+def last_runtime_drift(reg: dict, name: str) -> list[str]:
+    """Runtime fields (M6 included) the last call on base `name` that reported an init event
+    found drifted against the base (any kind but train)."""
+    best, fields = -1, []
+    for j, e in reg.get("jobs", {}).items():
+        if e.get("base") != name or e.get("kind") == "train" or e.get("runtime_drift") is None:
+            continue
+        if _job_num(j) > best:
+            best, fields = _job_num(j), list(e["runtime_drift"])
+    return [f for f in fields if f in RUNTIME_FIELDS]
+
+
+def defining_drift(base: dict, cfg: dict) -> list[str]:
+    """Fingerprint-defining fields of the base's launch record that differ from what a launch
+    would use now: for a field the base overrode at `train`, its own value; otherwise the run
+    config's. The agent file and the preamble are re-hashed from disk."""
+    rec = base.get("launch")
+    if not rec:
+        return []
+    over = set(base.get("overrides") or [])
+    now = cfg.get("launch") or {}
+
+    def want(f):
+        return rec.get(f) if f in over else now.get(f)
+    out = [f for f in ("model", "cwd", "tools", "permission_mode") if want(f) != rec.get(f)]
+    ra = rec.get("agent") or {}
+    name = ra.get("name") if "agent" in over else (now.get("agent") or {}).get("name")
+    sha = None
+    if name and want("cwd"):
+        try:
+            sha = _sha256_file(agent_path(want("cwd"), name))
+        except OSError:
+            sha = None
+    if (name, sha) != (ra.get("name"), ra.get("sha256")):
+        out.append("agent")
+    try:
+        pre = _sha256_file(preamble_path())
+    except OSError:
+        pre = None
+    if pre != (rec.get("append_system_prompt") or {}).get("sha256"):
+        out.append("append_system_prompt")
+    return [f for f in DEFINING_FIELDS if f in out]
+
+
+def binary_state(base: dict, launch: dict, cache: dict | None = None) -> str:
+    """"missing" (the stored path is gone), "drift" (version, size or mtime differ from the base's
+    record) or "ok". `cache` maps a path to its binary_info within one call."""
+    path = launch["binary"]["path"]
+    if not os.path.isfile(path):
+        return "missing"
+    cache = {} if cache is None else cache
+    if path not in cache:
+        try:
+            cache[path] = binary_info(path)
+        except ForksError:
+            cache[path] = None
+    info = cache[path]
+    ref = base.get("binary") or launch["binary"]
+    if info is None or any(info.get(k) != ref.get(k) for k in ("version", "size", "mtime")):
+        return "drift"
+    return "ok"
+
+
+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:
+    """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
+    fork's delta header (computed here when not given); `binary` a `binary_state` value."""
+    cfg = reg["config"]
+    base = reg["bases"][name]
+    la = launch_record(base, cfg)
+    th = dict(THRESHOLD_DEFAULTS, **(cfg.get("thresholds") or {}))
+    rebase, retrain = [], []
+    if phase_boundary(reg, name, phase):
+        rebase.append("phase_boundary")
+    bc = base.get("base_commit")
+    if bc:
+        commits, lines = git_delta_counts(la["cwd"], bc)
+        if commits is not None and commits > int(th["rebase_commits"]):
+            rebase.append("commits:%d" % commits)
+        if lines is not None and lines > int(th["rebase_lines"]):
+            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))
+    if chars // 4 > int(th["rebase_delta_tokens"]):
+        rebase.append("delta_tokens:%d" % (chars // 4))
+    if binary is None:
+        binary = binary_state(base, la, bcache)
+    if binary == "drift":
+        rebase.append("binary_drift")
+    rd = last_runtime_drift(reg, name)
+    if rd:
+        rebase.append("runtime_drift:" + ",".join(rd))
+    dd = defining_drift(base, cfg)
+    if dd:
+        retrain.append("defining:" + ",".join(dd))
+    ctx = base_context_now(base, la)
+    if ctx is not None and ctx > int(th["max_base_tokens"]):
+        retrain.append("max_base_tokens:%d" % ctx)
+    lr = int(base.get("lineage_rebases") or 0)
+    if lr >= int(th["max_rebases"]):
+        retrain.append("rebases:%d" % lr)
+    if binary == "missing":
+        retrain.append("binary_missing")
+    return {"rebase_due": bool(rebase), "retrain_due": bool(retrain), "reasons": rebase + retrain}
+
+
 def _next_job_id(reg: dict) -> str:
     n = 0
     for j in reg.get("jobs", {}):
@@ -2118,13 +2514,22 @@ def _next_job_id(reg: dict) -> str:
 
 def _running_capped(reg: dict) -> list[str]:
     return sorted(j for j, e in reg.get("jobs", {}).items()
-                  if e.get("kind") in ("fork", "continue") and e.get("status") in RUNNING)
+                  if e.get("kind") in CAPPED_KINDS and e.get("status") in RUNNING)
+
+
+def _pending_rebase_to(reg: dict, name: str) -> str | None:
+    """The running rebase job that will create base `name`, if any."""
+    for j, e in sorted(reg.get("jobs", {}).items()):
+        if e.get("kind") == "rebase" and e.get("rebase_as") == name and e.get("status") in RUNNING:
+            return j
+    return None
 
 
 def _new_entry(job: str, kind: str, **kw) -> dict:
     e = {"job": job, "kind": kind, "base": None, "block": None, "phase": None, "mode": None,
          "paths": None, "status": "created", "flags": [], "reason": None, "session": None,
-         "parent_session": None, "model_requested": None, "created_at": _iso(), "launched_at": None,
+         "parent_session": None, "model_requested": None, "created_at": _iso(),
+         "created_ts": _now(), "launched_at": None,
          "ended_at": None, "supervisor": None, "cwd": None, "argv": None,
          "brief": "briefs/%s.md" % job, "raw": "raw/%s.jsonl" % job, "max_s": None,
          "cost_before": 0.0, "model_usage_before": {}, "base_context": None,
@@ -2154,15 +2559,40 @@ def _start_job(state: str, run: str, job: str, prompt: str, launch: dict) -> dic
     return sp
 
 
+def train_launch(args, state: str, cfg: dict) -> tuple[dict, list[str]]:
+    """M10 (plan 4.3): the base's launch record = the run config's launch fields with the
+    overrides given on `train` (the same flags as `init`), validated as `init` validates them.
+    Returns (launch, sorted overridden field names)."""
+    la = cfg["launch"]
+    over = sorted(n for n in LAUNCH_FIELDS if getattr(args, n, None) is not None)
+    tools = _tool_list([args.tools]) if args.tools is not None else la["tools"]
+    allowed = _tool_list(args.allowed_tools) if args.allowed_tools is not None else la["allowed_tools"]
+    disallowed = (_tool_list(args.disallowed_tools) if args.disallowed_tools is not None
+                  else la["disallowed_tools"])
+    agent = args.agent if args.agent is not None else (la.get("agent") or {}).get("name")
+    cwd = norm_path(args.cwd) if args.cwd is not None else la["cwd"]
+    bpath = norm_path(args.binary) if args.binary is not None else la["binary"]["path"]
+    if not os.path.isfile(bpath):
+        raise ForksError("claude binary missing (the stored path is gone); train with --binary or "
+                         "re-run init with another binary", binary=bpath, reason="binary_missing")
+    launch, _ = build_launch(
+        state, model=args.model if args.model is not None else la["model"],
+        effort=args.effort if args.effort is not None else la["effort"],
+        permission_mode=args.permission_mode if args.permission_mode is not None else la["permission_mode"],
+        tools=tools, allowed=allowed, disallowed=disallowed, agent=agent, cwd=cwd, binary_path=bpath)
+    return launch, over
+
+
 def cmd_train(args) -> tuple[dict, int]:
     state, reg = _load_run(args)
     name = args.base
     if not re.fullmatch(r"[A-Za-z0-9_.-]+", name or ""):
         raise ForksError("bad base name", base=name)
     cfg = reg["config"]
-    la = cfg["launch"]
+    la, overrides = train_launch(args, state, cfg)
+    binfo = la["binary"]
+    drift: list[str] = []          # the record was built from the files just now
     brief_path, brief = _read_brief(args.brief)
-    binfo, drift = _check_binary(la["binary"], la)
     session = str(uuid.uuid4())
     argv = claude_argv(la) + ["--session-id", session]
     head = _git_head(la["cwd"])
@@ -2172,11 +2602,16 @@ def cmd_train(args) -> tuple[dict, int]:
         old = reg["bases"].get(name)
         if old and old.get("status") != "failed":
             raise ForksError("base exists", base=name, status=old.get("status"))
+        pending = _pending_rebase_to(reg, name)
+        if pending:
+            raise ForksError("a running rebase creates this base", base=name, job=pending)
         job = _next_job_id(reg)
         reg["bases"][name] = {"name": name, "status": "training", "session": session, "cwd": la["cwd"],
-                              "fingerprint": cfg["fingerprint"], "binary": binfo, "created_at": _iso(),
+                              "launch": la, "fingerprint": fingerprint_hash(la), "overrides": overrides,
-                              "train_job": job, "phase": _phase(args.phase), "last_hit": None,
+                              "binary": binfo, "created_at": _iso(), "train_job": job,
-                              "base_commit": None, "context_tokens": None, "forks": []}
+                              "phase": _phase(args.phase), "last_hit": None, "parent": None,
+                              "lineage_rebases": 0, "base_commit": None, "context_tokens": None,
+                              "forks": []}
         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,
@@ -2196,10 +2631,10 @@ def cmd_fork(args) -> tuple[dict, int]:
         raise ForksError("--mode parallel needs --paths (the files this fork may touch)", mode="parallel")
     state, reg = _load_run(args)
     cfg = reg["config"]
-    la = cfg["launch"]
     base = reg["bases"].get(args.base)
     if base is None:
         raise ForksError("unknown base", base=args.base, bases=sorted(reg["bases"]))
+    la = launch_record(base, cfg)          # M10: every launch on a base uses the base's record
     if base.get("status") != "active" and not args.force:
         raise ForksError("base is not active (use --force to fork it anyway)", base=args.base,
                          status=base.get("status"))
@@ -2208,6 +2643,8 @@ def cmd_fork(args) -> tuple[dict, int]:
     brief_path, brief = _read_brief(args.brief)
     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")
     head = _git_head(la["cwd"])
     session = str(uuid.uuid4())
     argv = claude_argv(la) + ["--resume", base["session"], "--fork-session", "--session-id", session]
@@ -2256,15 +2693,14 @@ def cmd_fork(args) -> tuple[dict, int]:
         "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"],
-            "warm_until": warmth["warm_until"], "rebase_due": False, "retrain_due": False,
+            "warm_until": warmth["warm_until"], "rebase_due": due["rebase_due"],
-            "reasons": []}, EXIT_OK
+            "retrain_due": due["retrain_due"], "reasons": due["reasons"]}, EXIT_OK
 
 
 def cmd_continue(args) -> tuple[dict, int]:
     _refuse_overrides(args)
     state, reg = _load_run(args)
     cfg = reg["config"]
-    la = cfg["launch"]
     fj = reg["jobs"].get(args.job)
     if fj is None:
         raise ForksError("unknown job", job=args.job)
@@ -2279,6 +2715,7 @@ def cmd_continue(args) -> tuple[dict, int]:
     if not fj.get("session"):
         raise ForksError("the fork has no session", job=args.job)
     base = reg["bases"].get(fj.get("base")) or {}
+    la = launch_record(base, cfg)          # M10: the fork's base's record
     brief_path, brief = _read_brief(args.brief)
     binfo, drift = _check_binary(base.get("binary") or la["binary"], la)
     head = _git_head(la["cwd"])
@@ -2313,6 +2750,77 @@ def cmd_continue(args) -> tuple[dict, int]:
     return {"job": job}, EXIT_OK
 
 
+def cmd_rebase(args) -> tuple[dict, int]:
+    """`rebase --base B --as NEW --brief FILE [--phase N]`: a fork of B that absorbs what changed
+    (`templates/rebase.md`). On `ok` its session becomes base NEW and B is superseded (finaliser,
+    `_rebase_succeeded`); otherwise B stays active and NEW is never created. Counts toward the cap;
+    N3-refused (dispatch)."""
+    _refuse_overrides(args)
+    state, reg = _load_run(args)
+    cfg = reg["config"]
+    new = args.as_name
+    if not re.fullmatch(r"[A-Za-z0-9_.-]+", new or ""):
+        raise ForksError("bad base name", base=new)
+    base = reg["bases"].get(args.base)
+    if base is None:
+        raise ForksError("unknown base", base=args.base, bases=sorted(reg["bases"]))
+    if new == args.base:
+        raise ForksError("--as must name a new base", base=new)
+    if base.get("status") != "active" or not base.get("session"):
+        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)
+    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).
+    new_launch = dict(la, binary=binfo)
+    delta = delta_header(la["cwd"], base.get("base_commit"))
+    head = _git_head(la["cwd"])
+    session = str(uuid.uuid4())
+    argv = claude_argv(la) + ["--resume", base.get("session") or "", "--fork-session", "--session-id", session]
+    if args.max_budget_usd:
+        argv += ["--max-budget-usd", "%g" % args.max_budget_usd]
+    reap_lost(state)
+    with RegistryLock(state):
+        reg = read_registry(state)
+        stamp_orchestrator(reg)
+        base = reg["bases"][args.base]
+        if base.get("status") != "active" or not base.get("session"):
+            raise ForksError("base is not active", base=args.base, status=base.get("status"))
+        old = reg["bases"].get(new)
+        if old and old.get("status") != "failed":
+            raise ForksError("base exists", base=new, status=old.get("status"))
+        pending = _pending_rebase_to(reg, new)
+        if pending:
+            raise ForksError("a running rebase creates this base", base=new, job=pending)
+        busy = sorted(j for j, e in reg["jobs"].items() if e.get("kind") == "rebase"
+                      and e.get("base") == args.base and e.get("status") in RUNNING)
+        if busy:
+            raise ForksError("the base is being rebased", base=args.base, running=busy)
+        running = _running_capped(reg)
+        if len(running) >= int(cfg["parallel_cap"]):
+            raise ForksError("parallel cap reached", parallel_cap=cfg["parallel_cap"], running=running)
+        job = _next_job_id(reg)
+        warmth = base_warmth(base, cfg)
+        reg["jobs"][job] = _new_entry(
+            job, "rebase", base=args.base, phase=_phase(args.phase), session=session,
+            parent_session=base["session"], model_requested=la["model"], cwd=la["cwd"], argv=argv,
+            max_s=float(args.max_minutes if args.max_minutes else cfg["max_minutes"]) * 60,
+            cost_before=float(base.get("cost_cum_usd") or 0.0),
+            model_usage_before=base.get("model_usage_cum") or {},
+            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)
+        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})
+    _start_job(state, norm_path(args.run), job, prompt, la)
+    return {"job": job, "base": new, "parent": args.base, "session_id": session}, EXIT_OK
+
+
 # --- keep warm: the ping rule and the tick (design 5, C3d) ----------------------------
 
 
@@ -2322,8 +2830,8 @@ def run_for_state(state: str) -> str:
 
 
 def launch_record(base: dict | None, cfg: dict) -> dict:
-    """The launch fields a launch on `base` uses: the base's own record when it has one, else the
+    """The launch fields a launch on `base` uses (M10): the base's own record (written at `train`
-    run config (every base has the init config's fields until bases carry their own record)."""
+    or `rebase`); the run config only for a base from before per-base records."""
     return (base or {}).get("launch") or cfg["launch"]
 
 
@@ -2609,7 +3117,8 @@ class Supervisor:
         stall_s = float(self.cfg.get("stall_after_s") or CONFIG_DEFAULTS["stall_after_s"])
         orphan_s = float(self.cfg.get("orphan_after") or CONFIG_DEFAULTS["orphan_after"])
         hint, reason, detail, flags_extra = "exited", None, None, []
-        hit_written = e["kind"] != "fork"            # only a fork's first request refreshes (I3)
+        # only a fork's (or a rebase's) first request reads the base prefix and refreshes it (I3)
+        hit_written = e["kind"] not in ("fork", "rebase")
         clock = SupClock(t0)
         last_orphan_check = 0.0
         last_messages = 0
@@ -2674,7 +3183,7 @@ class Supervisor:
             err.close()
         wall = _now() - t0
         base_commit = None
-        if e["kind"] == "train":
+        if e["kind"] in ("train", "rebase"):        # the new base's HEAD at the end of the call
             base_commit = _git_head(e["cwd"]) or e.get("head_at_launch")
         res = finalize_job(self.state, self.job, ss=self.ss, hint=hint, flags=self.flags + flags_extra,
                            reason=reason, detail=detail, exit_code=exit_code, wall_s=wall,
@@ -2795,7 +3304,8 @@ LIST_JOB_KEYS = ("job", "kind", "base", "block", "phase", "mode", "paths", "stat
                  "ended_at", "fingerprint", "cache", "delivered_at", "settled_at")
 LIST_BASE_KEYS = ("name", "status", "session", "cwd", "base_commit", "context_tokens",
                   "training_cost_usd", "cost_cum_usd", "last_hit", "train_job", "forks", "phase",
-                  "created_at", "trained_at", "ttl_from", "ping_in_flight")
+                  "created_at", "trained_at", "ttl_from", "ping_in_flight", "parent",
+                  "lineage_rebases", "rebase_job", "superseded_by", "overrides", "fingerprint")
 
 
 def cmd_list(args) -> tuple[dict, int]:
@@ -2806,10 +3316,16 @@ def cmd_list(args) -> tuple[dict, int]:
     now = _now()
     cfg = reg["config"]
     bases = []
+    bcache: dict = {}
     for name, b in sorted(reg["bases"].items()):
         row = {k: b.get(k) for k in LIST_BASE_KEYS}
         row["name"] = name
         row.update(base_warmth(b, cfg, now))
+        # The flags are computed for active bases only (no fork's --phase or brief here, so no
+        # phase_boundary and delta_tokens from the delta header alone).
+        if b.get("status") == "active":
+            row.update(due_flags(state, reg, name, bcache=bcache))
+        else:
             row.update({"rebase_due": False, "retrain_due": False, "reasons": []})
         bases.append(row)
     jobs, running = [], []
@@ -2866,7 +3382,7 @@ HANDLERS = {
     "expire": cmd_expire,
     "kill": cmd_kill,
     "_supervise": cmd_supervise,
-    "rebase": _not_implemented(phase=2),
+    "rebase": cmd_rebase,
     "usage": _not_implemented(phase=2),
     "tick": cmd_tick,
 }

plugins/spire/skills/forks/templates/rebase.md

@@ -0,0 +1,17 @@
+# Rebase (spire:forks)
+
+You are a fork of base `${base}` (run log `${run}`). When you finish, this session becomes the new base `${new_base}` (phase ${phase}): every later fork of this build starts from what you know at the end of it. Your job is to absorb what changed since `${base}` was trained or last rebased.
+
+Your system prompt was snapshotted when the base was trained: the git status, the CLAUDE.md files and the environment it shows are **stale**. The delta header below lists what changed in the repository since then. Trust it and the files on disk over the snapshot and over what you remember.
+
+- Read the handoff notes and the diffs the brief lists (`git --no-optional-locks diff`, `git --no-optional-locks log -p`), and re-read the files they touch. Handoff notes live under `${notes_dir}`.
+- Do not change code and do not commit. If the brief names a trunk-notes path, update that file; it is the only file you write.
+- Finish with one line `READY`, then a short summary of what changed and anything the brief asked for that you could not do.
+
+## Delta since the base commit
+
+${delta}
+
+## Brief
+
+${brief}

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

@@ -102,14 +102,79 @@ class PathTests(Base):
             "/c": "C:\\",
             "/c/": "C:\\",
             "../other/./z.md": "G:\\dev\\other\\z.md",
-            "\\rooted\\a": "G:\\rooted\\a",
+            "\\rooted\\a": "G:\\rooted\\a",   # native rooted path: the cwd's drive (Windows meaning)
-            "/tmp/x": "G:\\tmp\\x",       # not a drive form: rooted on the cwd's drive
             "  G:/x/y.md  ": "G:\\x\\y.md",
         }
         for src, want in cases.items():
             with self.subTest(src=src):
                 self.assertEqual(forks.norm_path(src, cwd=cwd, windows=True), want)
 
+    def fake_cygpath(self, code=0):
+        """A fake `cygpath` (python behind a wrapper) that maps `/a/b` to `C:\\Fake\\a\\b` and logs
+        its arguments."""
+        d = os.path.join(self.tmp, "cyg")
+        os.makedirs(d, exist_ok=True)
+        py = os.path.join(d, "cygpath.py")
+        log = os.path.join(d, "calls.txt")
+        with open(py, "w", newline="\n") as f:
+            f.write("import sys\nopen(%r, 'a').write(' '.join(sys.argv[1:]) + '\\n')\n"
+                    "print('C:\\\\Fake' + sys.argv[-1].replace('/', '\\\\'))\nsys.exit(%d)\n" % (log, code))
+        if IS_WINDOWS:
+            exe = os.path.join(d, "cygpath.cmd")
+            with open(exe, "w", newline="\r\n") as f:
+                f.write('@echo off\n"%s" -I "%s" %%*\n' % (stub_interpreter(), py))
+        else:
+            exe = os.path.join(d, "cygpath")
+            with open(exe, "w", newline="\n") as f:
+                f.write('#!/bin/sh\nexec "%s" -I "%s" "$@"\n' % (stub_interpreter(), py))
+            os.chmod(exe, 0o755)
+        return exe, log
+
+    def test_gitbash_rooted_without_cygpath(self):
+        """M13: a Git Bash rooted path that is not a drive form is refused (exit 2) when cygpath is
+        not on PATH; it is never mapped onto the cwd's drive. Drive forms need no cygpath."""
+        cwd = "G:\\dev\\proj"
+        with mock.patch.object(forks, "find_cygpath", lambda: None):
+            for src in ("/tmp/x", "/usr/share/doc", "/tmp/x-run.md"):
+                with self.subTest(src=src):
+                    with self.assertRaises(forks.ForksError) as cm:
+                        forks.norm_path(src, cwd=cwd, windows=True)
+                    self.assertEqual(cm.exception.code, 2)
+                    self.assertIn("cygpath", cm.exception.payload["error"])
+                    self.assertEqual(cm.exception.payload["path"], src)
+            self.assertEqual(forks.norm_path("/g/x/y.md", cwd=cwd, windows=True), "G:\\x\\y.md")
+            self.assertEqual(forks.norm_path("/c", cwd=cwd, windows=True), "C:\\")
+            self.assertEqual(forks.norm_path("\\rooted\\a", cwd=cwd, windows=True), "G:\\rooted\\a")
+            if IS_WINDOWS:     # end to end: --run /tmp/... is refused, nothing written on the cwd's drive
+                drive = os.path.splitdrive(os.getcwd())[0]
+                code, out, raw = call(["init", "--run", "/tmp/forks-m13-x-run.md", "--model",
+                                       "claude-opus-5-5", "--cwd", self.tmp])
+                self.assertEqual(code, 2, out)
+                self.assert_one_json_line(raw)
+                self.assertIn("cygpath", out["error"])
+                self.assertFalse(os.path.exists(drive + "\\tmp\\forks-m13-x-run-forks"))
+
+    def test_gitbash_rooted_with_cygpath(self):
+        """M13: with cygpath on PATH the path is converted by `cygpath -w`; a failing cygpath
+        refuses."""
+        exe, log = self.fake_cygpath()
+        with mock.patch.object(forks, "find_cygpath", lambda: exe):
+            self.assertEqual(forks.norm_path("/tmp/x", cwd="G:\\dev\\proj", windows=True),
+                             "C:\\Fake\\tmp\\x")
+            self.assertEqual(forks.norm_path("/g/x", cwd="G:\\dev\\proj", windows=True), "G:\\x")
+        with open(log) as f:
+            self.assertEqual(f.read().splitlines(), ["-w /tmp/x"])     # drive forms skip cygpath
+        bad, _ = self.fake_cygpath(code=1)
+        with mock.patch.object(forks, "find_cygpath", lambda: bad):
+            with self.assertRaises(forks.ForksError) as cm:
+                forks.norm_path("/tmp/x", cwd="G:\\dev\\proj", windows=True)
+            self.assertEqual(cm.exception.payload["error"], "cygpath failed")
+        real = shutil.which("cygpath")
+        if IS_WINDOWS and real:        # the real one (Git Bash): never the cwd's drive + \tmp
+            got = forks.norm_path("/tmp/x", cwd="G:\\dev\\proj", windows=True)
+            self.assertNotEqual(got, "G:\\tmp\\x")
+            self.assertRegex(got, r"^[A-Z]:\\.+\\x$")
+
     def test_windows_cwd_in_gitbash_form(self):
         self.assertEqual(forks.norm_path("a.md", cwd="/g/dev/proj", windows=True), "G:\\dev\\proj\\a.md")
 
@@ -632,22 +697,56 @@ class InitTests(Base):
         self.assertEqual(cfg["orphan_after"], 1500)
         self.assertNotEqual(cfg["fingerprint"], first["config"]["fingerprint"])
 
-    def test_changed_values_after_base_refused(self):
+    def test_changed_values_after_base_allowed(self):
+        """Block 2.2 replaces Phase 1's refusal: `init` may change values after a base exists (the
+        input path of `retrain_due: defining:`). The base keeps its own launch record; a base from
+        before per-base records gets the config it was trained with as its record."""
         self.assertEqual(self.init()[0], 0)
         self.add_fake_base()
-        before = self.reg_bytes()
         code, out, raw = self.init("--effort", "medium")
-        self.assertEqual(code, 2)
+        self.assertEqual(code, 0, out)
         self.assert_one_json_line(raw)
-        self.assertIn("base exists", out["error"])
+        self.assertTrue(out["changed"])
+        self.assertIn("launch.effort", out["changed_fields"])
         self.assertEqual(out["bases"], ["B1"])
-        self.assertIn("launch.effort", out["fields"])
+        reg = forks.read_registry(self.state)
-        self.assertEqual(self.reg_bytes(), before)
+        self.assertEqual(reg["config"]["launch"]["effort"], "medium")
+        self.assertEqual(reg["bases"]["B1"]["launch"]["effort"], "high")
+        self.assertEqual(reg["bases"]["B1"]["overrides"], [])
         self.assertFalse(os.path.exists(forks.lock_path(self.state)))
-        # identical values stay fine with a base
+        # identical values stay unchanged with a base
-        code, out, _ = self.init()
+        before = self.reg_bytes()
+        code, out, _ = self.init("--effort", "medium")
         self.assertEqual(code, 0, out)
         self.assertFalse(out["changed"])
+        self.assertEqual(self.reg_bytes(), before)
+
+    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."""
+        code, a, _ = self.init()
+        self.assertEqual(code, 0, a)
+        other = os.path.join(self.tmp, "elsewhere", "2026-10-16-09-u-run.md")
+        os.makedirs(os.path.dirname(other))
+        code, b, _ = call(["init", "--run", other, "--model", "claude-opus-5-5[1m]", "--cwd", self.cwd,
+                           "--binary", self.stub])
+        self.assertEqual(code, 0, b)
+        la, lb = a["config"]["launch"], b["config"]["launch"]
+        self.assertNotEqual(la["mcp"]["config"], lb["mcp"]["config"])
+        self.assertEqual(la["mcp"]["config"], forks.mcp_empty_path(self.state))
+        self.assertEqual(a["config"]["fingerprint"], b["config"]["fingerprint"])
+        self.assertEqual(forks.fingerprint_view(la)["mcp"], {"strict": True, "servers": {}})
+        self.assertNotIn("config", forks.fingerprint_view(la)["mcp"])
+        # a second process computes the same hash (no per-process input)
+        r = subprocess.run([sys.executable, "-I", "-c",
+                            "import importlib.util,json,sys;"
+                            "s=importlib.util.spec_from_file_location('f',sys.argv[1]);m=importlib.util.module_from_spec(s);"
+                            "s.loader.exec_module(m);print(m.fingerprint_hash(json.loads(sys.argv[2])))",
+                            FORKS_PY, json.dumps(lb)], capture_output=True, timeout=60)
+        self.assertEqual(r.stdout.decode().strip(), a["config"]["fingerprint"], r.stderr)
+        # the content does count
+        self.assertNotEqual(forks.fingerprint_hash(dict(la, mcp=dict(la["mcp"], strict=False))),
+                            a["config"]["fingerprint"])
 
     # --- refusals ---
     def assert_refused_no_state(self, *extra, model="claude-opus-5-5[1m]"):
@@ -759,7 +858,7 @@ class InitTests(Base):
 class CliTests(Base):
     def test_phase2_not_implemented(self):
         run_log = os.path.join(self.tmp, "r-run.md")
-        for argv in (("rebase",), ("usage", "--phase", "1")):   # `tick` is implemented (2.1)
+        for argv in (("usage", "--phase", "1"),):   # `tick` (2.1) and `rebase` (2.2) are implemented
             with self.subTest(cmd=argv[0]):
                 code, out, raw = call([argv[0], "--run", run_log, *argv[1:]])
                 self.assertEqual(code, 2)
@@ -777,7 +876,8 @@ class CliTests(Base):
                  ("fork", "--base", "B1", "--block", "1.1", "--brief", brief),
                  ("continue", "--job", "j0001", "--brief", brief),
                  ("wait", "--all"), ("list",), ("expire", "--all"), ("kill", "--job", "j0001"),
-                 ("_supervise", "--job", "j0001"), ("tick",))
+                 ("_supervise", "--job", "j0001"), ("tick",),
+                 ("rebase", "--base", "B1", "--as", "B2", "--brief", brief))
         for argv in cases:
             with self.subTest(cmd=argv[0]):
                 code, out, raw = call([argv[0], "--run", run_log, *argv[1:]])
@@ -1015,7 +1115,10 @@ def main():
           "permissionMode": opt(a, "--permission-mode"), "tools": tools, "mcp_servers": [],
           "apiKeySource": d.get("apikey", "none"), "claude_code_version": VERSION,
           "plugins": [{"name": "spire", "path": "x", "source": "spire@reliquary",
-                       "version": d.get("plugin", "0.2.2")}],
+                       "version": d.get("plugin", "0.2.2")},
+                      {"name": "skill-creator", "source": "skill-creator@official",
+                       "path": "C:\\plugins\\cache\\skill-creator\\" + d.get("nover", "abc123")}],
+          "skills": sorted(["implementation", "workpad"] + ([d["skill"]] if "skill" in d else [])),
           "agents": ["claude", "general-purpose"]})
     used = model
     if "restricted" in d:
@@ -1200,6 +1303,9 @@ class LaunchBase(Base):
         p = mock.patch.object(forks, "USER_SETTINGS_PATH", os.path.join(self.tmp, "home", "settings.json"))
         p.start()
         self.addCleanup(p.stop)
+        p = mock.patch.object(forks, "CLAUDE_PROJECTS_DIR", os.path.join(self.tmp, "home", "projects"))
+        p.start()
+        self.addCleanup(p.stop)
         self.cwd = os.path.join(self.tmp, "proj")
         os.makedirs(self.cwd)
         self.git("init", "-q")
@@ -1446,6 +1552,46 @@ class RuleTests(Base):
         self.assertEqual(forks.runtime_drift(other, rt, {}), ["apiKeySource", "plugins"])
         self.assertEqual(forks.runtime_drift(rt, None, {"permission_mode": "default"}), ["permissionMode"])
 
+    def test_runtime_fingerprint_skills_and_versionless_plugins(self):
+        """M6: sorted `skills` are compared; a plugin without a version is identified by the last
+        segment of its path. Both drift."""
+        init = {"tools": ["Read"], "permissionMode": "dontAsk", "apiKeySource": "none",
+                "agents": ["claude"], "skills": ["workpad", "implementation"],
+                "plugins": [{"name": "spire", "version": "0.2.2", "path": "C:\\c\\spire\\0.2.2"},
+                            {"name": "skill-creator",
+                             "path": "C:\\Users\\u\\.claude\\plugins\\cache\\skill-creator\\abc123\\"},
+                            {"name": "builtin"}]}
+        rt = forks.runtime_fingerprint(init)
+        self.assertEqual(rt["plugins"], ["builtin@", "skill-creator@path:abc123", "spire@0.2.2"])
+        self.assertEqual(rt["skills"], ["implementation", "workpad"])
+        self.assertEqual(sorted(rt), sorted(forks.RUNTIME_FIELDS))
+        upd = json.loads(json.dumps(init))
+        upd["plugins"][1]["path"] = "/home/u/.claude/plugins/cache/skill-creator/def456"
+        self.assertEqual(forks.runtime_drift(forks.runtime_fingerprint(upd), rt, {}), ["plugins"])
+        more = dict(init, skills=["workpad", "implementation", "new-skill"])
+        self.assertEqual(forks.runtime_drift(forks.runtime_fingerprint(more), rt, {}), ["skills"])
+        same = dict(init, skills=["implementation", "workpad"])                  # order does not count
+        self.assertEqual(forks.runtime_drift(forks.runtime_fingerprint(same), rt, {}), [])
+        old_ref = {k: v for k, v in rt.items() if k != "skills"}                 # a base from before M6
+        self.assertEqual(forks.runtime_drift(forks.runtime_fingerprint(more), old_ref, {}), [])
+
+    def test_note_written_by(self):
+        """M11: a note counts only when its mtime is at or after the job's creation (less the
+        file-time slack)."""
+        p = os.path.join(self.tmp, "n.md")
+        start = time.time() - 100
+        entry = {"created_ts": start, "created_at": forks._iso(start)}
+        self.assertFalse(forks.note_written_by(p, entry))                        # no file
+        with open(p, "w") as f:
+            f.write("note\n")
+        for delta, want in ((-forks.NOTE_MTIME_SLACK_S - 1, False), (-forks.NOTE_MTIME_SLACK_S + 0.5, True),
+                            (0, True), (50, True)):
+            with self.subTest(delta=delta):
+                os.utime(p, (start + delta, start + delta))
+                self.assertEqual(forks.note_written_by(p, entry), want)
+        os.utime(p, (start - 60, start - 60))
+        self.assertFalse(forks.note_written_by(p, {"created_at": forks._iso(start)}))   # ISO fallback
+
 
 class WarmthTests(Base):
     """Design 5 under an injected clock (block 2.1): warmth boundaries, TTL inference, the ping rule."""
@@ -1798,9 +1944,9 @@ class LaunchTests(LaunchBase):
         # --- fork, after a commit on the trunk ---
         self.commit_file("src.py", "x = 1\n", "add src after training")
         os.makedirs(os.path.join(self.state, "notes"), exist_ok=True)
-        with open(os.path.join(self.state, "notes", "1.1.md"), "w") as f:
+        f_out = self.fork("turns=2", "sleep=1", extra=("--phase", "1", "--paths", "src.py", "docs/a.md"))
+        with open(os.path.join(self.state, "notes", "1.1.md"), "w") as f:   # M11: written during the job
             f.write("note\n")
-        f_out = self.fork("turns=2", extra=("--phase", "1", "--paths", "src.py", "docs/a.md"))
         self.assertEqual(set(f_out), {"job", "base", "base_state", "warm_until", "rebase_due",
                                       "retrain_due", "reasons"})
         self.assertEqual(f_out["base_state"], "warm")
@@ -2213,7 +2359,8 @@ class LaunchTests(LaunchBase):
         for name in forks.LAUNCH_FIELDS:
             flag = "--" + name.replace("_", "-")
             for argv in (["fork", "--base", "B1", "--block", "1", "--brief", "b.md"],
-                         ["continue", "--job", "j0001", "--brief", "b.md"]):
+                         ["continue", "--job", "j0001", "--brief", "b.md"],
+                         ["rebase", "--base", "B1", "--as", "B2", "--brief", "b.md"]):
                 with self.subTest(field=name, cmd=argv[0]):
                     out = self.cmd(*argv, flag, values[name], code=2)
                     self.assertEqual(out["fields"], [name])
@@ -2499,6 +2646,358 @@ class LaunchTests(LaunchBase):
         self.assertIn("messages", self.runtime(job))                  # M16: the job file's counter
         self.assertNotIn("turns", self.runtime(job))
 
+    # --- block 2.2 -----------------------------------------------------------------------
+
+    REBASE_REASONS = ("phase_boundary", "commits", "lines", "delta_tokens", "binary_drift", "runtime_drift")
+
+    def dead_spawn(self) -> dict:
+        p = subprocess.Popen([sys.executable, "-c", "pass"])
+        dead = forks.pid_record(p.pid)
+        p.wait()
+        return {"ok": True, "method": "test", "pid": dead["pid"], "ctime": dead["ctime"]}
+
+    def fake_fork(self, base, *extra, brief=None, code=0):
+        """A fork whose supervisor is never started (spawn stubbed with a dead pid), finalised at once
+        as killed: only its launch return is looked at."""
+        fake = self.dead_spawn()
+        with mock.patch.object(forks, "spawn_supervisor", lambda *a: fake):
+            out = self.cmd("fork", "--base", base, "--block", "d", "--brief", brief or self.brief(),
+                           *extra, code=code)
+        if code == 0:
+            forks.finalize_job(self.state, out["job"], ss=None, hint="killed", reason="kill")
+        return out
+
+    def clone_base(self, name, src="B1", **changes):
+        """A copy of base `src` under a new name, at the current HEAD unless `base_commit` is given."""
+        head = self.git("rev-parse", "HEAD")
+
+        def f(reg):
+            b = json.loads(json.dumps(reg["bases"][src]))
+            b.update({"name": name, "forks": [], "base_commit": head})
+            b.update(changes)
+            reg["bases"][name] = b
+        self.set_registry(f)
+
+    def assert_due(self, out, *reasons):
+        self.assertEqual(out["reasons"], list(reasons), out)
+        rebase = [r for r in reasons if r.split(":")[0] in self.REBASE_REASONS]
+        self.assertEqual(out["rebase_due"], bool(rebase), out)
+        self.assertEqual(out["retrain_due"], len(rebase) < len(reasons), out)
+
+    def test_due_reasons_each_alone(self):
+        """Design 5 / digest 2.2: every rebase_due and retrain_due reason triggered on its own; the
+        fork's launch return lists exactly it (and `list` the base-level ones)."""
+        self.setup_run(cap=4)
+        self.train("B1")
+        self.finish("j0001")
+        c0 = self.git("rev-parse", "HEAD")
+        self.assert_due(self.fake_fork("B1"))                                    # nothing due
+        # phase_boundary: above the highest phase of the base's forks, or of the base itself
+        self.clone_base("Bp")
+        self.assert_due(self.fake_fork("Bp", "--phase", "1"))                    # nothing known yet
+        self.assert_due(self.fake_fork("Bp", "--phase", "1"))
+        self.assert_due(self.fake_fork("Bp", "--phase", "2"), "phase_boundary")
+        self.assert_due(self.fake_fork("Bp"))                                    # no --phase
+        self.clone_base("Bq", phase=3)
+        self.assert_due(self.fake_fork("Bq", "--phase", "3"))
+        self.assert_due(self.fake_fork("Bq", "--phase", "4"), "phase_boundary")
+        self.assert_due(self.fake_fork("Bq", "--phase", "4"), "phase_boundary")   # still above the base's 3
+        self.assert_due(self.fake_fork("Bq", "--phase", "2"))
+        # commits: 7 > rebase_commits 6 (seven one-line files)
+        for i in range(7):
+            self.commit_file("c%d.txt" % i, "x\n", "commit %d" % i)
+        self.clone_base("Bc", base_commit=c0)
+        self.assert_due(self.fake_fork("Bc"), "commits:7")
+        # lines: 1600 > rebase_lines 1500, in one commit
+        c7 = self.git("rev-parse", "HEAD")
+        self.commit_file("big.txt", "line\n" * 1600, "big change")
+        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
+        with open(os.path.join(self.state, "notes", "big-note.md"), "w") as f:
+            f.write("n" * 60000)
+        self.clone_base("Bd")
+        self.assert_due(self.fake_fork("Bd"))                                    # brief lists no note
+        out = self.fake_fork("Bd", brief=self.brief(text="Read notes/big-note.md first."))
+        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)
+        self.assertEqual((out["rebase_due"], out["retrain_due"]), (True, False))
+        # binary_drift: a clone with its own copy of the binary, changed after it was recorded
+        ext = os.path.splitext(self.stub)[1]
+        b2 = os.path.join(os.path.dirname(self.stub), "claude2" + ext)
+        shutil.copy2(self.stub, b2)
+        info = forks.binary_info(forks.norm_path(b2))
+        b1_launch = self.reg()["bases"]["B1"]["launch"]
+        self.clone_base("Bb", launch=dict(b1_launch, binary=info), binary=info)
+        self.assert_due(self.fake_fork("Bb"))
+        with open(b2, "a", newline="\r\n" if IS_WINDOWS else "\n") as f:
+            f.write("rem drift\n" if IS_WINDOWS else "# drift\n")
+        self.assert_due(self.fake_fork("Bb"), "binary_drift")
+        # runtime_drift (M6): the base's last call saw a versionless plugin and the skills change
+        self.clone_base("Br")
+        r = self.finish(self.fork("turns=1", "nover=def456", "skill=extra", base="Br")["job"])
+        self.assertEqual((r["status"], r["fingerprint"]), ("ok", "drift:plugins,skills"), r)
+        self.assert_due(self.fake_fork("Br"), "runtime_drift:plugins,skills")
+        # defining: the base's record differs from the config in a field it did not override
+        self.clone_base("Bm", launch=dict(b1_launch, model="claude-opus-4-1"))
+        self.assert_due(self.fake_fork("Bm"), "defining:model")
+        # max_base_tokens: from the stored context, and from the transcript when it exists (P0.7)
+        self.clone_base("Bx", context_tokens=300000)
+        self.assert_due(self.fake_fork("Bx"), "max_base_tokens:300000")
+        sid = "11111111-2222-3333-4444-555555555555"
+        self.clone_base("Bt", session=sid, context_tokens=1000)
+        tp = forks.transcript_path(b1_launch["cwd"], sid)
+        self.assertTrue(tp.startswith(os.path.join(self.tmp, "home", "projects")), tp)
+        os.makedirs(os.path.dirname(tp))
+        lines = [{"type": "user", "message": {"role": "user", "content": "x" * 300}},
+                 {"type": "assistant", "message": {"id": "m1", "usage": {
+                     "input_tokens": 2, "cache_creation_input_tokens": 500, "cache_read_input_tokens": 1000}}},
+                 {"type": "assistant", "message": {"id": "m2", "usage": {
+                     "input_tokens": 2, "cache_creation_input_tokens": 1000, "cache_read_input_tokens": 260000}}},
+                 {"type": "system", "subtype": "informational", "content": "y" * 300}]
+        with open(tp, "w", newline="\n") as f:
+            f.write("".join(json.dumps(x) + "\n" for x in lines))
+        with mock.patch.object(forks, "TRANSCRIPT_TAIL_BYTES", 64):              # the tail grows
+            self.assert_due(self.fake_fork("Bt"), "max_base_tokens:261002")
+        # rebases: lineage_rebases >= max_rebases 4
+        self.clone_base("Bn3", lineage_rebases=3)
+        self.assert_due(self.fake_fork("Bn3"))
+        self.clone_base("Bn", lineage_rebases=4)
+        self.assert_due(self.fake_fork("Bn"), "rebases:4")
+        # binary_missing: the stored path is gone; launches refuse (exit 2)
+        gone = dict(info, path=os.path.join(self.tmp, "gone", "claude" + ext))
+        self.clone_base("Bz", launch=dict(b1_launch, binary=gone), binary=gone)
+        out = self.fake_fork("Bz", code=2)
+        self.assertEqual((out["reason"], out["retrain_due"], out["reasons"]),
+                         ("binary_missing", True, ["binary_missing"]), out)
+        # list: the base-level reasons, per active base
+        rows = {b["name"]: b for b in self.cmd("list")["bases"]}
+        for name, want in (("Bl", ["lines:1600"]), ("Bd", []), ("Bb", ["binary_drift"]),
+                           ("Br", ["runtime_drift:plugins,skills"]), ("Bm", ["defining:model"]),
+                           ("Bx", ["max_base_tokens:300000"]), ("Bt", ["max_base_tokens:261002"]),
+                           ("Bn", ["rebases:4"]), ("Bz", ["binary_missing"])):
+            with self.subTest(base=name):
+                self.assert_due(rows[name], *want)
+
+    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`
+        changing the model afterwards: a base that did not override it reports `retrain_due:
+        defining:model`; one that did reports nothing."""
+        self.setup_run(cap=4)
+        t = self.cmd("train", "--base", "B1", "--brief", self.brief(), "--model", "claude-sonnet-5-5",
+                     "--effort", "low", "--tools", "Read,Bash")
+        r = self.finish(t["job"])
+        self.assertEqual((r["status"], r["model_requested"], r["models_used"]),
+                         ("ok", "claude-sonnet-5-5", ["claude-sonnet-5-5"]), r)
+        reg = self.reg()
+        cfg, b1 = reg["config"], reg["bases"]["B1"]
+        rec = b1["launch"]
+        self.assertEqual((rec["model"], rec["effort"], rec["tools"]), ("claude-sonnet-5-5", "low", ["Read", "Bash"]))
+        self.assertEqual(rec["allowed_tools"], cfg["launch"]["allowed_tools"])      # not overridden
+        self.assertEqual(rec["cwd"], cfg["launch"]["cwd"])
+        self.assertEqual(b1["overrides"], ["effort", "model", "tools"])
+        self.assertEqual(b1["fingerprint"], forks.fingerprint_hash(rec))
+        self.assertNotEqual(b1["fingerprint"], cfg["fingerprint"])
+        a = stub_calls(self.stub)[0]["argv"]
+        self.assertEqual((a[a.index("--model") + 1], a[a.index("--effort") + 1], a[a.index("--tools") + 1]),
+                         ("claude-sonnet-5-5", "low", "Read,Bash"))
+        # a base trained without overrides records the config's fields
+        self.train("B2")
+        self.assertEqual(self.finish("j0002")["status"], "ok")
+        b2 = self.reg()["bases"]["B2"]
+        self.assertEqual((b2["launch"], b2["overrides"], b2["fingerprint"]),
+                         (cfg["launch"], [], cfg["fingerprint"]))
+        # fork and continue of B1 build their command line from B1's record
+        f = self.fork("turns=1", base="B1")
+        self.assert_due(f)
+        rf = self.finish(f["job"])
+        self.assertEqual((rf["status"], rf["model_requested"], rf["models_used"]),
+                         ("ok", "claude-sonnet-5-5", ["claude-sonnet-5-5"]), rf)
+        c = self.cmd("continue", "--job", f["job"], "--brief", self.brief("turns=1"))
+        rc = self.finish(c["job"])
+        self.assertEqual((rc["status"], rc["model_requested"]), ("ok", "claude-sonnet-5-5"), rc)
+        for job in (f["job"], c["job"]):
+            with self.subTest(job=job):
+                argv = self.reg()["jobs"][job]["argv"]
+                self.assertEqual((argv[argv.index("--model") + 1], argv[argv.index("--effort") + 1],
+                                  argv[argv.index("--tools") + 1]), ("claude-sonnet-5-5", "low", "Read,Bash"))
+        # init changes the model after the bases exist
+        out = self.setup_run("--model", "claude-opus-4-1", cap=4)
+        self.assertTrue(out["changed"], out)
+        self.assertIn("launch.model", out["changed_fields"])
+        rows = {b["name"]: b for b in self.cmd("list")["bases"]}
+        self.assert_due(rows["B2"], "defining:model")
+        self.assert_due(rows["B1"])
+        f2 = self.fork("turns=1", base="B2")
+        self.assert_due(f2, "defining:model")                 # advisory: the fork still launches
+        r2 = self.finish(f2["job"])
+        self.assertEqual((r2["status"], r2["model_requested"]), ("ok", TEST_MODEL), r2)   # B2's record
+        f1 = self.fork("turns=1", base="B1")
+        self.assert_due(f1)
+        self.assertEqual(self.finish(f1["job"])["model_requested"], "claude-sonnet-5-5")
+        # overrides pass init's checks: aliases, plan mode, availableModels
+        self.cmd("train", "--base", "B3", "--brief", self.brief(), "--model", "sonnet", code=2)
+        self.cmd("train", "--base", "B3", "--brief", self.brief(), "--permission-mode", "plan", code=2)
+        settings = os.path.join(self.tmp, "home", "settings.json")
+        os.makedirs(os.path.dirname(settings), exist_ok=True)
+        with open(settings, "w") as fh:
+            json.dump({"availableModels": ["opus"]}, fh)
+        out = self.cmd("train", "--base", "B3", "--brief", self.brief(), "--model", "claude-sonnet-5-5", code=2)
+        self.assertEqual(out["error"], "model not admitted by availableModels")
+        self.assertNotIn("B3", self.reg()["bases"])
+
+    def test_note_only_from_this_job(self):
+        """M11: `note` is set only when the note file was written during this job; a note left by an
+        earlier fork of the block, or by the fork a continue resumes, does not count."""
+        self.setup_run()
+        self.train("B1")
+        self.finish("j0001")
+        note = os.path.join(self.state, "notes", "1.1.md")
+        with open(note, "w") as f:
+            f.write("old note\n")
+        old = time.time() - 600
+        os.utime(note, (old, old))
+        r = self.finish(self.fork("turns=1")["job"])
+        self.assertEqual((r["status"], r["note"]), ("ok", None), r)
+        j = self.fork("sleep=1", "turns=2")["job"]
+        with open(note, "w") as f:                                  # written while the job runs
+            f.write("new note\n")
+        r = self.finish(j)
+        self.assertEqual((r["status"], r["note"]), ("ok", "notes/1.1.md"), r)
+        os.utime(note, (old, old))
+        rc = self.finish(self.cmd("continue", "--job", j, "--brief", self.brief("turns=1"))["job"])
+        self.assertEqual((rc["status"], rc["note"]), ("ok", None), rc)
+        led = {x["job"]: x for x in self.ledger()}
+        self.assertEqual((led[j]["note"], led[rc["job"]]["note"]), ("notes/1.1.md", None))
+
+    def test_rebase_end_to_end(self):
+        """`rebase` with the stub: the new base is active (parent, lineage, HEAD, context, launch
+        record), the old one superseded; a fork of the new base hits the new context; a failed
+        rebase leaves the base active and creates nothing; a binary drift is recorded by the new
+        base."""
+        self.setup_run()
+        self.train("B1")
+        self.finish("j0001")
+        b1 = self.reg()["bases"]["B1"]
+        self.commit_file("src.py", "x = 1\n", "work of phase 1")
+        f1 = self.fork("turns=1", extra=("--phase", "1"))
+        self.assert_due(f1)
+        self.finish(f1["job"])
+        f2 = self.fork("turns=1", extra=("--phase", "2"))
+        self.assert_due(f2, "phase_boundary")
+        self.finish(f2["job"])
+        out = self.cmd("rebase", "--base", "B1", "--as", "B2", "--phase", "2",
+                       "--brief", self.brief("turns=2", text="Absorb phase 1."))
+        self.assertEqual(set(out), {"job", "base", "parent", "session_id"})
+        self.assertEqual((out["base"], out["parent"]), ("B2", "B1"))
+        rr = self.finish(out["job"])
+        self.assertEqual((rr["kind"], rr["status"], rr["base"], rr["cache"], rr["phase"], rr["fingerprint"]),
+                         ("rebase", "ok", "B1", "hit", 2, "ok"), rr)
+        self.assertEqual((rr["session"], rr["parent_session"]), (out["session_id"], b1["session"]))
+        reg = self.reg()
+        old, new = reg["bases"]["B1"], reg["bases"]["B2"]
+        self.assertEqual((old["status"], old["superseded_by"]), ("superseded", "B2"))
+        self.assertEqual((new["status"], new["parent"], new["lineage_rebases"], new["session"], new["phase"],
+                          new["rebase_job"]), ("active", "B1", 1, out["session_id"], 2, out["job"]))
+        self.assertEqual(new["base_commit"], self.git("rev-parse", "HEAD"))
+        self.assertEqual((new["launch"], new["fingerprint"], new["overrides"]),
+                         (old["launch"], old["fingerprint"], []))
+        self.assertGreater(new["context_tokens"], old["context_tokens"])
+        self.assertIsNotNone(new["last_hit"])
+        line = {x["job"]: x for x in self.ledger()}[out["job"]]
+        self.assertEqual((line["kind"], line["status"]), ("rebase", "ok"))
+        self.assertAlmostEqual(new["cost_cum_usd"], line["cost_cumulative_usd"], places=6)
+        a = [c for c in stub_calls(self.stub) if c["session"] == out["session_id"]][0]["argv"]
+        self.assertEqual(a[a.index("--resume") + 1], b1["session"])
+        self.assertIn("--fork-session", a)
+        self.assertEqual(a[a.index("--session-id") + 1], out["session_id"])
+        with open(os.path.join(self.state, "briefs", out["job"] + ".md"), encoding="utf-8") as fh:
+            txt = fh.read()
+        for s in ("READY", "Do not change code", "work of phase 1", "`B2`", "Absorb phase 1.", "stale"):
+            self.assertIn(s, txt)
+        # a fork of the new base: a hit against the new base's context
+        f3 = self.fork("turns=1", base="B2", extra=("--phase", "2"))
+        self.assert_due(f3)
+        r3 = self.finish(f3["job"])
+        self.assertEqual((r3["status"], r3["cache"], r3["parent_session"]), ("ok", "hit", new["session"]), r3)
+        self.assertEqual(self.reg()["jobs"][f3["job"]]["base_context"], new["context_tokens"])
+        self.assertGreaterEqual(r3["tokens"]["cache_read"], new["context_tokens"])
+        a = [c for c in stub_calls(self.stub) if c["session"] == r3["session"]][0]["argv"]
+        self.assertEqual(a[a.index("--resume") + 1], new["session"])
+        # the superseded base: no fork, no rebase
+        self.fork(base="B1", code=2)
+        self.cmd("rebase", "--base", "B1", "--as", "B9", "--brief", self.brief(), code=2)
+        # a failed rebase: the base stays active, the new base is never created
+        rf = self.finish(self.cmd("rebase", "--base", "B2", "--as", "B3", "--brief", self.brief("error"))["job"])
+        self.assertEqual((rf["status"], rf["reason"]), ("failed", "is_error"), rf)
+        reg = self.reg()
+        self.assertEqual(reg["bases"]["B2"]["status"], "active")
+        self.assertNotIn("B3", reg["bases"])
+        rows = {b["name"]: b for b in self.cmd("list")["bases"]}
+        self.assertEqual((rows["B1"]["status"], rows["B1"]["superseded_by"], rows["B2"]["parent"]),
+                         ("superseded", "B2", "B1"))
+        # refusals
+        for argv, why in ((["--base", "B2", "--as", "B2"], "new base"),
+                          (["--base", "B2", "--as", "B1"], "base exists"),
+                          (["--base", "B9", "--as", "B4"], "unknown base"),
+                          (["--base", "B2", "--as", "a/b"], "bad base name")):
+            with self.subTest(argv=argv):
+                self.assertIn(why, self.cmd("rebase", *argv, "--brief", self.brief(), code=2)["error"])
+        os.environ[forks.CHILD_MARKER] = "1"
+        try:
+            out = self.cmd("rebase", "--base", "B2", "--as", "B4", "--brief", self.brief(), code=2)
+            self.assertIn("inside a fork", out["error"])
+        finally:
+            os.environ.pop(forks.CHILD_MARKER, None)
+        # binary drift: the rebase runs under the current binary, and the new base records it
+        with open(self.stub, "a", newline="\r\n" if IS_WINDOWS else "\n") as f:
+            f.write("rem drift\n" if IS_WINDOWS else "# drift\n")
+        fd = self.fork("turns=1", base="B2")
+        self.assert_due(fd, "binary_drift")
+        self.finish(fd["job"])
+        out = self.cmd("rebase", "--base", "B2", "--as", "B4", "--brief", self.brief("turns=1"))
+        rb = self.finish(out["job"])
+        self.assertEqual((rb["status"], rb["fingerprint"]), ("ok", "drift:binary"), rb)
+        b2, b4 = self.reg()["bases"]["B2"], self.reg()["bases"]["B4"]
+        cur = forks.binary_info(forks.norm_path(self.stub))
+        for k in ("version", "size", "mtime"):
+            self.assertEqual((b4["binary"][k], b4["launch"]["binary"][k]), (cur[k], cur[k]), k)
+        self.assertEqual((b4["parent"], b4["lineage_rebases"], b2["status"]), ("B2", 2, "superseded"))
+        self.assertEqual(b4["fingerprint"], forks.fingerprint_hash(b4["launch"]))
+        self.assertNotEqual(b4["fingerprint"], b2["fingerprint"])
+        f4 = self.fork("turns=1", base="B4")
+        self.assert_due(f4)
+        r4 = self.finish(f4["job"])
+        self.assertEqual((r4["status"], r4["fingerprint"], r4["cache"]), ("ok", "ok", "hit"), r4)
+
+    def test_rebase_counts_toward_cap(self):
+        """A running rebase counts toward parallel_cap, blocks a second rebase of its base and a
+        train of its new name; killed, it leaves the base active and creates nothing."""
+        self.setup_run(cap=1)
+        self.train("B1")
+        self.finish("j0001")
+        nxt = forks._next_job_id(self.reg())
+        out = self.cmd("rebase", "--base", "B1", "--as", "B2",
+                       "--brief", self.brief("hang=1", text="RUN-%s" % nxt))
+        job = out["job"]
+        self.assertEqual(job, nxt)
+        self.until_running(job)
+        refused = self.fork(code=2)
+        self.assertEqual((refused["error"], refused["running"]), ("parallel cap reached", [job]))
+        self.assertIn("rebased", self.cmd("rebase", "--base", "B1", "--as", "B3", "--brief", self.brief(),
+                                          code=2)["error"])
+        self.assertIn("rebase creates", self.cmd("train", "--base", "B2", "--brief", self.brief(),
+                                                 code=2)["error"])
+        self.assertTrue(self.cmd("kill", "--job", job)["ok"])
+        r = self.finish(job)
+        self.assertEqual((r["status"], r["kind"]), ("killed", "rebase"))
+        reg = self.reg()
+        self.assertEqual(reg["bases"]["B1"]["status"], "active")
+        self.assertNotIn("B2", reg["bases"])
+
 
 def tearDownModule():
     """No supervisor or stub of this suite may survive it (checked by command line, never by name)."""