🐛 gremlins: init re-resolves the binary, late rebase of an inactive base, kill verification

plugins/spire/skills/gremlins/gremlins.py

@@ -1179,7 +1179,8 @@ def kill_tree(rec: dict | None, extra=(), alive_until: float | None = None) -> d
         break   # one tree kill of the first live root; the sweep covers the rest
     deadline = time.monotonic() + KILL_VERIFY_S
     while True:
-        survivors = [p for p in snap if record_alive(p)]
+        table = {pid: name for pid, _, name in process_table()}     # one table per sweep
+        survivors = [p for p in snap if snapshot_survivor(p, table)]
         if not survivors or time.monotonic() >= deadline:
             break
         for p in survivors:
@@ -1188,6 +1189,22 @@ def kill_tree(rec: dict | None, extra=(), alive_until: float | None = None) -> d
     return {"snapshot": snap, "tree_kill_rc": rc, "survivors": survivors}
 
 
+def snapshot_survivor(p: dict, table: dict) -> bool:
+    """Is snapshot entry `p` ({pid, ctime, name}) still alive? (C16 as narrowed by A2.)
+    `table` is {pid: name} from one `process_table()`. The pid is dead when the table is
+    non-empty and lacks it, or when its snapshot name is known, the table now gives the pid another
+    name and its creation time cannot be read (`proc_ctime` None): pid reuse by a process that
+    refuses `OpenProcess`, which `pid_alive` reports alive (access denied) without being able to
+    compare creation times (class (a)). In every other case `record_alive` decides."""
+    pid = int(p["pid"])
+    if table and pid not in table:
+        return False
+    name = p.get("name")
+    if name is not None and pid in table and table[pid] != name and proc_ctime(pid) is None:
+        return False
+    return record_alive(p)
+
+
 def _run(args, cwd=None, timeout=60, env=None) -> subprocess.CompletedProcess:
     """Run a short helper process: stdout captured (never leaks to ours), no window."""
     return subprocess.run(args, cwd=cwd, env=env, stdin=subprocess.DEVNULL, capture_output=True,
@@ -1295,15 +1312,29 @@ def resolve_binary(explicit: str | None = None, environ=None) -> tuple[str | Non
         environ = os.environ
     if explicit:
         return norm_path(explicit), "--binary"
-    ep = environ.get("CLAUDE_CODE_EXECPATH")
+    for how in BINARY_SOURCES:
-    if ep:
+        path = resolve_binary_via(how, environ)
-        return norm_path(ep), "CLAUDE_CODE_EXECPATH"
+        if path:
-    w = shutil.which("claude")
+            return path, how
-    if w:
-        return norm_path(w), "PATH"
     return None, None
 
 
+BINARY_SOURCES = ("CLAUDE_CODE_EXECPATH", "PATH")   # the lookup order without --binary
+
+
+def resolve_binary_via(how: str, environ=None) -> str | None:
+    """The binary through one source only (N3, A4): `CLAUDE_CODE_EXECPATH` or a `PATH` lookup."""
+    if environ is None:
+        environ = os.environ
+    if how == "CLAUDE_CODE_EXECPATH":
+        ep = environ.get("CLAUDE_CODE_EXECPATH")
+        return norm_path(ep) if ep else None
+    if how == "PATH":
+        w = shutil.which("claude")
+        return norm_path(w) if w else None
+    return None
+
+
 def _binary_version(path: str) -> str | None:
     try:
         r = _run([path, "--version"], timeout=60)
@@ -1785,6 +1816,12 @@ def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint:
                       and bool(ss and ss.first_at))
     if kind == "ping" and not ping_refreshed:
         flags.append("ping_missed")
+    # N4 (C15): a rebase that ends ok after its base stopped being active (a `train --supersedes`
+    # ended first, or an `expire` raced the kill) still creates its new base, with the old base's
+    # status. Flagged only when the new base is actually created (not on a `rebase_conflict`).
+    if (kind == "rebase" and status == "ok" and base is not None and base.get("status") != "active"
+            and _rebase_target_free(reg, entry)):
+        flags.append("parent_inactive")
     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
@@ -1867,20 +1904,35 @@ def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint:
     return result, ledger
 
 
+def _rebase_target_free(reg: dict, entry: dict) -> bool:
+    """True if a successful rebase `entry` may create its new base: it names one, and no base of
+    that name exists other than a failed one (which it replaces)."""
+    name = entry.get("rebase_as")
+    if not name:
+        return False
+    existing = reg["bases"].get(name)
+    return not (existing and existing.get("status") != "failed")
+
+
 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."""
+    base stays active and the new one is never created.
+    N4 (C15): if the old base is no longer active when the rebase ends (superseded by a training
+    that ended first, or retired by an `expire` that raced the kill), the new base takes the old
+    base's status (`superseded` by the same base, or `retired`, as of this end time) and the old
+    base is left untouched; the job records `rebase_inactive_parent`."""
     name = entry.get("rebase_as")
     if not name:
         return
-    existing = reg["bases"].get(name)
+    if not _rebase_target_free(reg, entry):
-    if existing and existing.get("status") != "failed":
+        existing = reg["bases"][name]
         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"])
+    old_status = old.get("status")
     reg["bases"][name] = {
         "name": name, "status": "active", "session": entry.get("session"), "cwd": launch["cwd"],
         "launch": launch, "fingerprint": fingerprint_hash(launch),
@@ -1896,6 +1948,16 @@ def _rebase_succeeded(reg: dict, entry: dict, old: dict, ss, u, rt, base_commit,
         "ttl_s": infer_ttl(u["tokens"] if u else None, old.get("ttl_s")), "ttl_from": entry["job"],
         "ping_in_flight": None,
     }
+    if old_status != "active":
+        new = reg["bases"][name]
+        new["status"] = old_status
+        if old_status == "superseded":
+            new.update({"superseded_by": old.get("superseded_by"), "superseded_at": ended})
+        elif old_status == "retired":
+            new["retired_at"] = ended
+        entry["rebase_inactive_parent"] = "%s was %s when the rebase ended" % (
+            old.get("name") or entry.get("base"), old_status)
+        return
     old["status"] = "superseded"
     old["superseded_by"] = name
     old["superseded_at"] = ended
@@ -2393,12 +2455,34 @@ def build_launch(state: str, *, model, effort, permission_mode, tools, allowed,
     return launch, adm["via"]
 
 
-def build_init_config(args, state: str, old: dict | None = None) -> dict:
+def rerun_binary(stored: str, how: str | None, warnings: list) -> tuple[str, str | None]:
+    """N3 (C14, A4): the binary of an `init` re-run without `--binary`. A `--binary` pin keeps the
+    stored path; otherwise the path is looked up again through the source it came from
+    (`CLAUDE_CODE_EXECPATH` or `PATH`, never the other one). Nothing found there keeps the stored
+    path. A change and a failed lookup each add one entry to `warnings`."""
+    if how not in BINARY_SOURCES:      # "--binary" (an explicit pin), or no recorded source
+        return stored, how
+    found = resolve_binary_via(how)
+    if found is None:
+        warnings.append("could not re-resolve the claude binary through %s; kept the stored path %s"
+                        % (how, stored))
+        return stored, how
+    if os.path.normcase(found) == os.path.normcase(stored):
+        return stored, how
+    warnings.append("binary changed: %s → %s (resolved by %s); existing bases keep their own binary "
+                    "record, new trainings use the new one" % (stored, found, how))
+    return found, how
+
+
+def build_init_config(args, state: str, old: dict | None = None, warnings: list | None = None) -> dict:
     """Validate the launch fields and build the config block. Writes nothing.
     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."""
+    agent hashes) is re-read either way. N3: the binary is the exception (`rerun_binary`); its
+    warnings go to `warnings`."""
+    if warnings is None:
+        warnings = []
     ol = (old or {}).get("launch") or {}
 
     def keep(given, key, default):
@@ -2428,7 +2512,7 @@ def build_init_config(args, state: str, old: dict | None = None) -> dict:
         raise GremlinsError("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")
+        bpath, how = rerun_binary(ol["binary"]["path"], (old or {}).get("binary_resolved_by"), warnings)
     else:
         bpath, how = resolve_binary(args.binary)
     if bpath is None:
@@ -2477,7 +2561,8 @@ def cmd_init(args) -> tuple[dict, int]:
     state = state_dir_for(args.run)
     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
+    warnings: list[str] = []
+    cfg = build_init_config(args, state, old_cfg, warnings)     # all validation before any write
     ensure_state(state)
     with RegistryLock(state):
         reg = read_registry(state)
@@ -2508,7 +2593,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,
-            "bases": sorted(reg.get("bases", {})), "config": cfg}, EXIT_OK
+            "bases": sorted(reg.get("bases", {})), "config": cfg, "warnings": warnings}, EXIT_OK
 
 
 # --- launching: train, spawn, continue ------------------------------------------------

plugins/spire/skills/gremlins/tests/test_gremlins.py

@@ -687,6 +687,67 @@ class ProcessTreeTests(Base):
             self.assertEqual(gremlins.reap_lost(st), ["j0001"])
         self.assertEqual(seen, [(self.rec(), gremlins._ts(gremlins._iso(old)))])
 
+    # --- 3.3 / A2: kill verification and pid reuse (fake table; nothing real is killed) ---
+    SNAP = 900051
+
+    def verify(self, *, table, snap_name="sleep.exe", ctime_readable=True):
+        """`kill_tree` of a dead root whose snapshot holds one pid, SNAP, that `pid_alive` reports
+        alive the way the Windows platform does on access denied (running; the creation time is
+        compared only when it can be read). Returns the survivors' pids and the terminate calls."""
+        ct = self.ctimes[self.C]
+        snap = [{"pid": self.SNAP, "ctime": ct, "ppid": self.ROOT, "name": snap_name}]
+        self.table = list(table)
+        self.live = {self.SNAP}
+        self.ctimes = {self.SNAP: ct if ctime_readable else None}
+
+        def alive(pid, ctime=None):
+            pid = int(pid)
+            if pid not in self.live:
+                return False
+            now = self.ctimes.get(pid)
+            return not (ctime is not None and now is not None and now != ctime)
+        killed = []
+        with mock.patch.object(gremlins, "pid_alive", alive), \
+                mock.patch.object(gremlins, "process_tree", lambda rec, alive_until=None: list(snap)), \
+                mock.patch.object(gremlins, "terminate", lambda rec: killed.append(rec["pid"]) or False), \
+                mock.patch.object(gremlins.PLATFORM, "tree_kill", lambda pid: self.fail("root is dead")), \
+                mock.patch.object(gremlins, "KILL_VERIFY_S", 0.3):
+            k = gremlins.kill_tree(self.rec(), alive_until=self.hb)
+        self.assertEqual(k["snapshot"], snap)
+        self.assertIsNone(k["tree_kill_rc"])
+        return [p["pid"] for p in k["survivors"]], killed
+
+    def test_kill_verify_pid_reused_by_unopenable_process_is_dead(self):
+        """Class (a): the pid now carries another name and its creation time cannot be read."""
+        survivors, killed = self.verify(table=[(self.SNAP, 4, "svchost.exe"), (self.OTHER, 4, "other.exe")],
+                                        ctime_readable=False)
+        self.assertEqual((survivors, killed), ([], []))
+
+    def test_kill_verify_pid_gone_from_a_nonempty_table_is_dead(self):
+        survivors, killed = self.verify(table=[(self.OTHER, 4, "other.exe")], ctime_readable=False)
+        self.assertEqual((survivors, killed), ([], []))
+
+    def test_kill_verify_same_name_unverifiable_survives(self):
+        survivors, killed = self.verify(table=[(self.SNAP, self.ROOT, "sleep.exe")], ctime_readable=False)
+        self.assertEqual(survivors, [self.SNAP])
+        self.assertIn(self.SNAP, killed)
+
+    def test_kill_verify_empty_table_keeps_the_survivor(self):
+        survivors, killed = self.verify(table=[], ctime_readable=False)
+        self.assertEqual(survivors, [self.SNAP])
+        self.assertIn(self.SNAP, killed)
+
+    def test_kill_verify_unknown_snapshot_name_keeps_the_survivor(self):
+        survivors, _ = self.verify(table=[(self.SNAP, 4, "svchost.exe")], snap_name=None,
+                                   ctime_readable=False)
+        self.assertEqual(survivors, [self.SNAP])
+
+    def test_kill_verify_other_name_with_readable_ctime_is_record_alive(self):
+        """A differing name with a readable creation time leaves the decision to `record_alive`:
+        the same creation time is still the same process (a survivor)."""
+        survivors, _ = self.verify(table=[(self.SNAP, 4, "svchost.exe")], ctime_readable=True)
+        self.assertEqual(survivors, [self.SNAP])
+
 
 # --------------------------------------------------------------------------------------
 class InitTests(Base):
@@ -770,6 +831,7 @@ class InitTests(Base):
         self.assertEqual(cfg["stall_after_s"], 900)
         self.assertEqual(cfg["thresholds"], gremlins.THRESHOLD_DEFAULTS)
         self.assertEqual(out["config"], cfg)
+        self.assertEqual(out["warnings"], [])                 # N3: always present
 
     def test_fingerprint_canonical(self):
         a = {"b": [1, 2], "a": {"y": 1, "x": None}}
@@ -871,6 +933,140 @@ class InitTests(Base):
         self.assertIn("--model", out["detail"])
         self.assertFalse(os.path.exists(gremlins.state_dir_for(other)))
 
+    # --- N3: a re-run and the binary (3.3; digest §5.1, amendment A3, A4) ---
+    def named_stub(self, sub):
+        """A stub named `claude` (`claude.cmd` on Windows) in its own dir, found by a PATH lookup."""
+        d = os.path.join(self.tmp, sub)
+        os.makedirs(d)
+        dst = os.path.join(d, "claude.cmd" if IS_WINDOWS else "claude")
+        os.replace(make_stub_binary(d), dst)
+        return dst
+
+    def bare_init(self, *extra):
+        """A first init without --binary: the binary comes from $CLAUDE_CODE_EXECPATH or PATH."""
+        return call(["init", "--run", self.run_log, "--model", "claude-opus-5-5[1m]", "--cwd", self.cwd,
+                     *extra])
+
+    def rerun(self, *extra):
+        return call(["init", "--run", self.run_log, *extra])
+
+    def path_only(self, d):
+        """PATH is exactly `d` (an empty dir when it should hold no claude)."""
+        os.makedirs(d, exist_ok=True)
+        os.environ["PATH"] = d
+
+    def stored_binary(self):
+        cfg = gremlins.read_registry(self.state)["config"]
+        return cfg["launch"]["binary"]["path"], cfg["binary_resolved_by"]
+
+    def test_rerun_reresolves_execpath(self):
+        """C14: a re-run without --binary follows $CLAUDE_CODE_EXECPATH to a new binary, warns, and
+        lists the change; an existing base keeps its own binary record."""
+        a, b = self.named_stub("a"), self.named_stub("b")
+        os.environ["CLAUDE_CODE_EXECPATH"] = a
+        code, first, _ = self.bare_init()
+        self.assertEqual(code, 0, first)
+        self.assertEqual(self.stored_binary(), (gremlins.norm_path(a), "CLAUDE_CODE_EXECPATH"))
+        self.assertEqual(first["warnings"], [])
+        self.add_fake_base()
+        os.environ["CLAUDE_CODE_EXECPATH"] = b
+        code, out, raw = self.rerun("--pings", "off")
+        self.assertEqual(code, 0, out)
+        self.assert_one_json_line(raw)
+        self.assertEqual(self.stored_binary(), (gremlins.norm_path(b), "CLAUDE_CODE_EXECPATH"))
+        self.assertEqual(out["config"]["launch"]["binary"]["path"], gremlins.norm_path(b))
+        self.assertIn("launch.binary.path", out["changed_fields"])
+        self.assertIn("pings", out["changed_fields"])
+        self.assertEqual(len(out["warnings"]), 1, out["warnings"])
+        w = out["warnings"][0]
+        self.assertTrue(w.startswith("binary changed: "), w)
+        for s in (gremlins.norm_path(a), gremlins.norm_path(b), "(resolved by CLAUDE_CODE_EXECPATH)",
+                  "existing bases keep their own binary record, new trainings use the new one"):
+            self.assertIn(s, w)
+        reg = gremlins.read_registry(self.state)
+        self.assertEqual(reg["bases"]["B1"]["launch"]["binary"]["path"], gremlins.norm_path(a))
+        # the same source again: nothing changes, no warning
+        code, out, _ = self.rerun()
+        self.assertEqual((code, out["changed"], out["changed_fields"], out["warnings"]), (0, False, [], []))
+
+    def test_rerun_keeps_a_binary_pin(self):
+        """A `--binary` pin is kept on a re-run without --binary, whatever $CLAUDE_CODE_EXECPATH says."""
+        os.environ["CLAUDE_CODE_EXECPATH"] = self.named_stub("b")
+        code, first, _ = self.init()
+        self.assertEqual(code, 0, first)
+        self.assertEqual(self.stored_binary(), (gremlins.norm_path(self.stub), "--binary"))
+        code, out, _ = self.rerun("--pings", "off")
+        self.assertEqual(code, 0, out)
+        self.assertEqual(self.stored_binary(), (gremlins.norm_path(self.stub), "--binary"))
+        self.assertEqual(out["changed_fields"], ["pings"])
+        self.assertEqual(out["warnings"], [])
+
+    def test_rerun_without_a_source_keeps_the_stored_path(self):
+        """A3 (first case): stored file present, $CLAUDE_CODE_EXECPATH unset, no claude on PATH: the
+        stored path is kept with the "could not re-resolve" warning, exit 0."""
+        a = self.named_stub("a")
+        os.environ["CLAUDE_CODE_EXECPATH"] = a
+        self.assertEqual(self.bare_init()[0], 0)
+        os.environ.pop("CLAUDE_CODE_EXECPATH")
+        self.path_only(os.path.join(self.tmp, "empty-path"))
+        code, out, raw = self.rerun("--pings", "off")
+        self.assertEqual(code, 0, out)
+        self.assert_one_json_line(raw)
+        self.assertEqual(self.stored_binary(), (gremlins.norm_path(a), "CLAUDE_CODE_EXECPATH"))
+        self.assertEqual(out["changed_fields"], ["pings"])
+        self.assertEqual(out["warnings"], [
+            "could not re-resolve the claude binary through CLAUDE_CODE_EXECPATH; kept the stored path %s"
+            % gremlins.norm_path(a)])
+
+    def test_rerun_with_the_stored_file_gone_fails_loud(self):
+        """A3 (second case): the stored file is deleted and nothing re-resolves it: exit 2, `claude
+        binary not found` (today's loud failure), nothing written."""
+        a = self.named_stub("a")
+        os.environ["CLAUDE_CODE_EXECPATH"] = a
+        self.assertEqual(self.bare_init()[0], 0)
+        before = self.reg_bytes()
+        os.remove(a)
+        os.environ.pop("CLAUDE_CODE_EXECPATH")
+        self.path_only(os.path.join(self.tmp, "empty-path"))
+        code, out, raw = self.rerun("--pings", "off")
+        self.assertEqual(code, 2, out)
+        self.assert_one_json_line(raw)
+        self.assertEqual((out["error"], out["binary"]), ("claude binary not found", gremlins.norm_path(a)))
+        self.assertEqual(self.reg_bytes(), before)
+
+    def test_rerun_never_crosses_sources(self):
+        """A4: a path stored from $CLAUDE_CODE_EXECPATH is not replaced by a PATH lookup (kept, with
+        the warning), and a path stored from PATH is re-resolved by PATH, not by the env."""
+        a, p1, p2 = self.named_stub("a"), self.named_stub("p1"), self.named_stub("p2")
+        os.environ["CLAUDE_CODE_EXECPATH"] = a
+        self.assertEqual(self.bare_init()[0], 0)
+        os.environ.pop("CLAUDE_CODE_EXECPATH")
+        self.path_only(os.path.dirname(p1))
+        code, out, _ = self.rerun("--pings", "off")
+        self.assertEqual(code, 0, out)
+        self.assertEqual(self.stored_binary(), (gremlins.norm_path(a), "CLAUDE_CODE_EXECPATH"))
+        self.assertEqual(len(out["warnings"]), 1, out["warnings"])
+        self.assertTrue(out["warnings"][0].startswith(
+            "could not re-resolve the claude binary through CLAUDE_CODE_EXECPATH"), out["warnings"])
+        # stored from PATH: the env does not win on a re-run
+        other = os.path.join(self.tmp, "reports", "2026-10-16-09-p-run.md")
+        # (a PATH lookup on Windows spells the extension from PATHEXT, e.g. `claude.CMD`: compare
+        # the paths case-folded, as the file system does)
+        nc = os.path.normcase
+        code, out, _ = call(["init", "--run", other, "--model", "claude-opus-5-5[1m]", "--cwd", self.cwd])
+        self.assertEqual(code, 0, out)
+        self.assertEqual((nc(out["config"]["launch"]["binary"]["path"]), out["config"]["binary_resolved_by"]),
+                         (nc(gremlins.norm_path(p1)), "PATH"))
+        os.environ["CLAUDE_CODE_EXECPATH"] = a
+        self.path_only(os.path.dirname(p2))
+        code, out, _ = call(["init", "--run", other, "--pings", "off"])
+        self.assertEqual(code, 0, out)
+        self.assertEqual((nc(out["config"]["launch"]["binary"]["path"]), out["config"]["binary_resolved_by"]),
+                         (nc(gremlins.norm_path(p2)), "PATH"))
+        self.assertIn("launch.binary.path", out["changed_fields"])
+        self.assertEqual(len(out["warnings"]), 1, out["warnings"])
+        self.assertIn("(resolved by PATH)", out["warnings"][0])
+
     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."""
@@ -1527,6 +1723,10 @@ def main():
         prev = inp + cw + cr
         if i == 0 and "hang" in d:
             time.sleep(600 if d["hang"] == "1" else float(d["hang"]))
+        if i == 0 and "gate" in d:             # gate=NAME: wait (at most 120 s) for HERE/NAME to exist
+            t_gate = time.time() + 120
+            while not os.path.exists(os.path.join(HERE, d["gate"])) and time.time() < t_gate:
+                time.sleep(0.05)
         if i == 0 and "die" in d:
             os._exit(int(d.get("code", "3")))
         if "sleep" in d:
@@ -1940,6 +2140,28 @@ class RuleTests(Base):
         os.utime(p, (start - 60, start - 60))
         self.assertFalse(gremlins.note_written_by(p, {"created_at": gremlins._iso(start)}))   # ISO fallback
 
+    def test_late_rebase_of_a_retired_base(self):
+        """N4 (C15), the `expire` race: the old base was retired when the rebase ended ok. The new
+        base is created `retired` as of the end time; the old base is left as it was."""
+        ended = gremlins._iso(time.time())
+        launch = {"cwd": self.tmp, "binary": {"path": "claude", "version": STUB_VERSION}}
+        old = {"name": "B1", "status": "retired", "retired_at": gremlins._iso(time.time() - 60),
+               "session": SESS_TRAIN, "ttl_s": 3600}
+        reg = {"config": {"launch": launch}, "bases": {"B1": old}, "jobs": {}}
+        entry = {"job": "j0002", "kind": "rebase", "base": "B1", "rebase_as": "B1r", "session": SESS_SPAWN,
+                 "new_launch": launch, "phase": 2}
+        before = dict(old)
+        self.assertTrue(gremlins._rebase_target_free(reg, entry))
+        gremlins._rebase_succeeded(reg, entry, old, None, None, None, "abc1234", ended)
+        new = reg["bases"]["B1r"]
+        self.assertEqual((new["status"], new["retired_at"], new["parent"], new["session"]),
+                         ("retired", ended, "B1", SESS_SPAWN))
+        self.assertNotIn("superseded_by", new)
+        self.assertEqual(old, before)
+        self.assertEqual(entry["rebase_inactive_parent"], "B1 was retired when the rebase ended")
+        # the name is now taken: a second success would be a conflict, not a replacement
+        self.assertFalse(gremlins._rebase_target_free(reg, dict(entry, job="j0003")))
+
 
 class WarmthTests(Base):
     """Design 5 under an injected clock (block 2.1): warmth boundaries, TTL inference, the ping rule."""
@@ -3181,7 +3403,8 @@ class LaunchTests(LaunchBase):
             os.environ.pop(gremlins.CHILD_MARKER, None)
         for j in (j1, j2):
             self.until_running(j)
-            self.assertTrue(self.cmd("kill", "--job", j)["ok"])
+            out = self.cmd("kill", "--job", j)
+            self.assertTrue(out["ok"], out)          # 3.3: a failure prints the survivors
         code, out = self.wait(j1, j2)
         self.assertEqual(code, 4)
         self.assertEqual({x["job"]: x["paths"] for x in out["done"]}, {j1: ["a.py"], j2: ["b.py", "c.py"]})
@@ -3885,6 +4108,61 @@ class LaunchTests(LaunchBase):
         self.assertEqual(reg["bases"]["B1"]["status"], "active")
         self.assertNotIn("B2", reg["bases"])
 
+    def test_late_rebase_of_a_superseded_base(self):
+        """N4 (C15): `rebase --base B1 --as B1r` is still running when `train --base B3 --supersedes
+        B1` ends ok. When the rebase then ends ok, its result is `ok` with `parent_inactive`; B1r is
+        created superseded by B3 (no pings, a spawn needs --force); B1 keeps `superseded_by == B3`.
+        A second rebase whose new name is taken by then (`rebase_conflict`) creates nothing and is
+        not flagged."""
+        self.setup_run(cap=4)
+        self.train("B1")
+        self.finish("j0001")
+        self.train("B2")
+        self.finish("j0002")
+        gates = os.path.dirname(self.stub)            # the stub's `gate=NAME` waits for this file
+        r1 = self.cmd("rebase", "--base", "B1", "--as", "B1r", "--phase", "2",
+                      "--brief", self.brief("gate=g1", "turns=2"))
+        r2 = self.cmd("rebase", "--base", "B2", "--as", "B2r", "--brief", self.brief("gate=g2", "turns=2"))
+        t = self.cmd("train", "--base", "B3", "--brief", self.brief("turns=1"), "--supersedes", "B1",
+                     "--supersedes", "B2")
+        self.assertEqual(self.finish(t["job"])["status"], "ok")
+        reg = self.reg()
+        self.assertEqual([reg["jobs"][j]["status"] in gremlins.RUNNING for j in (r1["job"], r2["job"])],
+                         [True, True], "the rebases must still run when the training ends")
+        b1_before = dict(reg["bases"]["B1"])
+        self.assertEqual((b1_before["status"], b1_before["superseded_by"]), ("superseded", "B3"))
+        self.clone_base("B2r", src="B3")                # B2r's name is taken before its rebase ends
+        b2r_before = self.reg()["bases"]["B2r"]
+        for g in ("g1", "g2"):
+            with open(os.path.join(gates, g), "w"):
+                pass
+        rr1, rr2 = self.finish(r1["job"]), self.finish(r2["job"])
+        # the late rebase: ok, flagged, the new base superseded by the same base as its parent
+        self.assertEqual((rr1["kind"], rr1["status"], rr1["base"]), ("rebase", "ok", "B1"), rr1)
+        self.assertIn("parent_inactive", rr1["flags"])
+        reg = self.reg()
+        new, b1 = reg["bases"]["B1r"], reg["bases"]["B1"]
+        e1 = reg["jobs"][r1["job"]]
+        self.assertEqual((new["status"], new["superseded_by"], new["superseded_at"]),
+                         ("superseded", "B3", e1["ended_at"]))
+        self.assertEqual((new["parent"], new["session"], new["lineage_rebases"]), ("B1", r1["session_id"], 1))
+        lineage = ("status", "superseded_by", "superseded_at")                 # B1's lineage untouched
+        self.assertEqual([b1[k] for k in lineage], [b1_before[k] for k in lineage])
+        self.assertEqual(e1["rebase_inactive_parent"], "B1 was superseded when the rebase ended")
+        self.assertIn("parent_inactive", e1["flags"])
+        self.assertIn("parent_inactive", {x["job"]: x for x in self.ledger()}[r1["job"]]["flags"])
+        self.assertEqual(gremlins.ping_decision(reg, "B1r")["why"], "status:superseded")
+        self.assertIn("not active", self.spawn(base="B1r", code=2)["error"])     # needs --force
+        self.fake_spawn("B1r", "--force")
+        # the conflict: ok, not flagged, nothing created or replaced
+        self.assertEqual(rr2["status"], "ok", rr2)
+        self.assertNotIn("parent_inactive", rr2["flags"])
+        reg = self.reg()
+        self.assertIn("rebase_conflict", reg["jobs"][r2["job"]])
+        self.assertNotIn("rebase_inactive_parent", reg["jobs"][r2["job"]])
+        self.assertEqual(reg["bases"]["B2r"], b2r_before)
+        self.assertEqual((reg["bases"]["B2"]["status"], reg["bases"]["B2"]["superseded_by"]), ("superseded", "B3"))
+
     def test_head_at_end_and_usage_commit_column(self):
         """2.3: the supervisor records `head_at_end`; `usage --block` lists the commits made during a
         sequential spawn, and the ledger/usage figures agree with the result."""