From d2d51049d22350416880491f616a395ac88a83a9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gabor=20K=C3=B6rber?= Date: Fri, 16 Oct 2026 18:10:44 +0200 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B=20forks:=20Phase=202=20review=20fi?= =?UTF-8?q?xes=20(I1,=20I2,=20minors=201,=202,=208-11)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- plugins/spire/skills/forks/forks.py | 289 +++++++++++++----- .../spire/skills/forks/tests/test_forks.py | 276 ++++++++++++++++- 2 files changed, 482 insertions(+), 83 deletions(-) diff --git a/plugins/spire/skills/forks/forks.py b/plugins/spire/skills/forks/forks.py index 5836f6a..d954cba 100644 --- a/plugins/spire/skills/forks/forks.py +++ b/plugins/spire/skills/forks/forks.py @@ -1350,10 +1350,11 @@ def break_even(ttl) -> int: def base_warmth(base: dict, cfg: dict, now: float | None = None) -> dict: """Design 5: ttl = the base's inferred TTL, else the configured ttl_s; warm_until = last_hit + - ttl; margin = init's warm_margin_s if set, else max(120, ttl/10).""" + ttl; margin = init's warm_margin_s if set, capped at ttl/2 (Minor 2: a margin tuned for 1 h + must not make a 5-minute base `expiring` right after every hit), else max(120, ttl/10).""" ttl = float((base or {}).get("ttl_s") or cfg.get("ttl_s") or CONFIG_DEFAULTS["ttl_s"]) wm = cfg.get("warm_margin_s") - margin = float(wm) if wm is not None else max(120.0, ttl / 10.0) + margin = min(float(wm), ttl / 2.0) if wm is not None else max(120.0, ttl / 10.0) out = {"ttl_s": int(ttl) if ttl.is_integer() else ttl, "margin_s": int(margin) if margin.is_integer() else margin} last = _ts((base or {}).get("last_hit")) @@ -1594,6 +1595,13 @@ def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint: "cost_cum_usd": u["total"] if u else None, "model_usage_cum": u["model_usage"] if u else {}, "runtime_fingerprint": rt, "trained_at": ended}) + # Minor 1 (`train --supersedes`): the bases this one replaces stop being active, + # so they get no more pings and need --force to fork. Only on `ok`. + for old_name in base.get("supersedes") or []: + ob = reg["bases"].get(old_name) + if ob is not None and ob.get("status") == "active" and old_name != base.get("name"): + ob.update({"status": "superseded", "superseded_by": base.get("name"), + "superseded_at": ended}) elif base.get("status") == "training": base["status"] = "failed" if kind == "rebase" and status == "ok": @@ -1861,11 +1869,14 @@ def build_parser() -> _Parser: w = cmd("where", "Preflight report: script, python, uv, binary, git, spawn method, env scrub.") w.add_argument("--binary", help="claude binary to report instead of the default resolution") - i = cmd("init", "Create the state dir and the launch config (idempotent).") - i.add_argument("--model", required=True, help="full model id, e.g. claude-opus-5-5[1m]") - i.add_argument("--effort", default="high", choices=EFFORTS) - i.add_argument("--permission-mode", default="dontAsk") - i.add_argument("--tools", default=DEFAULT_TOOLS, help="comma-separated tool names") + # Minor 8: every init flag defaults to None. On the first init an omitted flag takes its + # default; on a re-run it keeps the current config value (build_init_config). + i = cmd("init", "Create the state dir and the launch config (idempotent). On a re-run, omitted " + "flags keep their current values.") + i.add_argument("--model", help="full model id, e.g. claude-opus-5-5[1m] (required on the first init)") + i.add_argument("--effort", choices=EFFORTS, help="default high") + i.add_argument("--permission-mode", help="default dontAsk") + i.add_argument("--tools", help="comma-separated tool names (default %s)" % DEFAULT_TOOLS) i.add_argument("--allowed-tools", nargs="+", action="extend", default=None, help="allow rules (default: the tools)") i.add_argument("--disallowed-tools", nargs="+", action="extend", default=None, @@ -1873,17 +1884,18 @@ def build_parser() -> _Parser: i.add_argument("--agent", help="project agent name (/.claude/agents/.md)") i.add_argument("--cwd", help="working dir of base and forks (default: git toplevel)") i.add_argument("--binary", help="claude binary (default: $CLAUDE_CODE_EXECPATH, else PATH)") - i.add_argument("--parallel-cap", type=_pos_int, default=CONFIG_DEFAULTS["parallel_cap"]) - i.add_argument("--orphan-after", type=_pos_int, default=CONFIG_DEFAULTS["orphan_after"]) - i.add_argument("--pings", choices=("on", "off"), default=CONFIG_DEFAULTS["pings"]) - i.add_argument("--ttl-s", type=_pos_int, default=CONFIG_DEFAULTS["ttl_s"], + i.add_argument("--parallel-cap", type=_pos_int, help="default %d" % CONFIG_DEFAULTS["parallel_cap"]) + i.add_argument("--orphan-after", type=_pos_int, help="default %d" % CONFIG_DEFAULTS["orphan_after"]) + i.add_argument("--pings", choices=("on", "off"), help="default %s" % CONFIG_DEFAULTS["pings"]) + i.add_argument("--ttl-s", type=_pos_int, help="TTL assumed until a base's calls show one (default 3600)") - i.add_argument("--warm-margin-s", type=_pos_int, default=None, - help="seconds before warm_until a base turns `expiring` (default max(120, ttl/10))") - i.add_argument("--max-minutes", type=_pos_int, default=CONFIG_DEFAULTS["max_minutes"]) - i.add_argument("--stall-after-s", type=_pos_int, default=CONFIG_DEFAULTS["stall_after_s"]) + i.add_argument("--warm-margin-s", type=_pos_int, + help="seconds before warm_until a base turns `expiring` (default max(120, ttl/10); " + "must be below ttl_s; at most half the base's TTL at runtime)") + i.add_argument("--max-minutes", type=_pos_int, help="default %d" % CONFIG_DEFAULTS["max_minutes"]) + i.add_argument("--stall-after-s", type=_pos_int, help="default %d" % CONFIG_DEFAULTS["stall_after_s"]) for k, v in THRESHOLD_DEFAULTS.items(): - i.add_argument("--" + k.replace("_", "-"), type=_pos_int, default=v) + i.add_argument("--" + k.replace("_", "-"), type=_pos_int, help="default %d" % v) t = cmd("train", "Train a new base (headless session).") t.add_argument("--base", required=True) @@ -1902,6 +1914,9 @@ def build_parser() -> _Parser: g.add_argument("--agent", help="project agent name (/.claude/agents/.md)") g.add_argument("--cwd", help="working dir of the base and its forks") g.add_argument("--binary", help="claude binary") + t.add_argument("--supersedes", action="append", default=None, metavar="BASE", + help="a base this one replaces (repeatable): when the training ends ok, each " + "named base that is active becomes superseded") def launch_overrides(sp): # Accepted only so that they can be refused with a clear message (plan 4.3). @@ -1923,6 +1938,9 @@ def build_parser() -> _Parser: f.add_argument("--max-minutes", type=_pos_float, help="wall-time limit (default: init's max_minutes)") f.add_argument("--max-budget-usd", type=_pos_float) f.add_argument("--force", action="store_true", help="fork a base that is not active") + f.add_argument("--delta-note", action="append", default=None, metavar="PATH", dest="delta_notes", + help="a note the fork must learn (repeatable): appended to the prompt after the " + "delta header; counts toward delta_tokens when newer than the base's training") launch_overrides(f) c = cmd("continue", "One more turn in a fork (resume without --fork-session).") @@ -1959,6 +1977,8 @@ def build_parser() -> _Parser: r.add_argument("--phase", help="the new base's phase (the phase whose forks it serves)") r.add_argument("--max-minutes", type=_pos_float, help="wall-time limit (default: init's max_minutes)") r.add_argument("--max-budget-usd", type=_pos_float) + r.add_argument("--delta-note", action="append", default=None, metavar="PATH", dest="delta_notes", + help="a note the rebase must absorb (repeatable): appended after the delta header") launch_overrides(r) u = cmd("usage", "Aggregate the ledger: tokens and cost per kind, cold starts, cache-hit ratio.") @@ -2098,39 +2118,73 @@ def build_launch(state: str, *, model, effort, permission_mode, tools, allowed, return launch, adm["via"] -def build_init_config(args, state: str) -> dict: - """Validate the launch fields and build the config block. Writes nothing.""" - tools = _tool_list([args.tools]) +def build_init_config(args, state: str, old: dict | 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.""" + ol = (old or {}).get("launch") or {} + + def keep(given, key, default): + if given is not None: + return given + if old is not None and key in old and old[key] is not None: + return old[key] + return default + model = args.model if args.model is not None else ol.get("model") + if model is None: # what argparse said when --model was required: a usage error + raise ForksError("usage", detail="the following arguments are required on the first init: --model") + tools = _tool_list([args.tools]) if args.tools is not None else \ + (list(ol["tools"]) if ol.get("tools") else _tool_list([DEFAULT_TOOLS])) allowed = _tool_list(args.allowed_tools) if allowed is None: - allowed = list(tools) + allowed = list(ol["allowed_tools"]) if "allowed_tools" in ol else list(tools) disallowed = _tool_list(args.disallowed_tools) if disallowed is None: - disallowed = list(DEFAULT_DISALLOWED) - 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, + disallowed = list(ol["disallowed_tools"]) if "disallowed_tools" in ol else list(DEFAULT_DISALLOWED) + if args.cwd is None and ol.get("cwd"): + cwd = ol["cwd"] + if not os.path.isdir(cwd): + raise ForksError("--cwd is not a directory", cwd=cwd) + else: + cwd = _resolve_cwd(args.cwd) + if not MODEL_RE.match(model or ""): # before the binary lookup, as in Phase 1 + raise ForksError("model must be a full id (aliases are refused)", model=model, pattern=MODEL_RE.pattern) - bpath, how = resolve_binary(args.binary) + if args.binary is None and (ol.get("binary") or {}).get("path"): + bpath, how = ol["binary"]["path"], (old or {}).get("binary_resolved_by") + else: + bpath, how = resolve_binary(args.binary) if bpath is None: raise ForksError("no claude binary (no --binary, no $CLAUDE_CODE_EXECPATH, none on PATH)") - launch, via = build_launch(state, model=args.model, effort=args.effort, - permission_mode=args.permission_mode, tools=tools, allowed=allowed, - disallowed=disallowed, agent=args.agent, cwd=cwd, binary_path=bpath) + agent = args.agent if args.agent is not None else (ol.get("agent") or {}).get("name") + launch, via = build_launch(state, model=model, + effort=args.effort if args.effort is not None else ol.get("effort") or "high", + permission_mode=(args.permission_mode if args.permission_mode is not None + else ol.get("permission_mode") or "dontAsk"), + tools=tools, allowed=allowed, disallowed=disallowed, agent=agent, cwd=cwd, + binary_path=bpath) + old_th = (old or {}).get("thresholds") or {} cfg = { "launch": launch, "fingerprint": fingerprint_hash(launch), "binary_resolved_by": how, - "parallel_cap": args.parallel_cap, - "orphan_after": args.orphan_after, - "pings": args.pings, - "ttl_s": args.ttl_s, - "warm_margin_s": args.warm_margin_s, - "max_minutes": args.max_minutes, - "stall_after_s": args.stall_after_s, - "thresholds": {k: getattr(args, k) for k in THRESHOLD_DEFAULTS}, + "parallel_cap": keep(args.parallel_cap, "parallel_cap", CONFIG_DEFAULTS["parallel_cap"]), + "orphan_after": keep(args.orphan_after, "orphan_after", CONFIG_DEFAULTS["orphan_after"]), + "pings": keep(args.pings, "pings", CONFIG_DEFAULTS["pings"]), + "ttl_s": keep(args.ttl_s, "ttl_s", CONFIG_DEFAULTS["ttl_s"]), + "warm_margin_s": keep(args.warm_margin_s, "warm_margin_s", None), + "max_minutes": keep(args.max_minutes, "max_minutes", CONFIG_DEFAULTS["max_minutes"]), + "stall_after_s": keep(args.stall_after_s, "stall_after_s", CONFIG_DEFAULTS["stall_after_s"]), + "thresholds": {k: (getattr(args, k) if getattr(args, k) is not None + else old_th.get(k, THRESHOLD_DEFAULTS[k])) for k in THRESHOLD_DEFAULTS}, "model_admission": via, } + # Minor 2: a margin at or above the TTL makes a base `expiring` right after every hit. + if cfg["warm_margin_s"] is not None and int(cfg["warm_margin_s"]) >= int(cfg["ttl_s"]): + raise ForksError("--warm-margin-s must be below ttl_s (a base would read `expiring` right " + "after every hit)", warm_margin_s=cfg["warm_margin_s"], ttl_s=cfg["ttl_s"]) return cfg @@ -2145,10 +2199,14 @@ def _diff_fields(a, b, prefix="") -> list[str]: def cmd_init(args) -> tuple[dict, int]: state = state_dir_for(args.run) - cfg = build_init_config(args, state) # all validation before any write + snap = read_registry(state) if os.path.isdir(state) else None + old_cfg = (snap or {}).get("config") + cfg = build_init_config(args, state, old_cfg) # all validation before any write ensure_state(state) with RegistryLock(state): reg = read_registry(state) + if (reg or {}).get("config") != old_cfg: + raise ForksError("the config changed while init ran; run init again", state=state) created = reg is None if created: reg = new_registry() @@ -2299,33 +2357,61 @@ def git_delta_counts(cwd: str, base_commit: str) -> tuple[int | None, int | None return commits, lines -_NOTE_REF = re.compile(r"[^\s`'\"()<>\[\]{},;|*]+\.md\b") +# I1 (Phase 2 review): "delta" is what a fork must learn that its base does not already know. +# delta_tokens = chars / 4 of the delta header (commits and shortstat since the base's +# base_commit) + the notes attached with `--delta-note` whose mtime is after the base's +# `trained_at` (what the base was trained or rebased on it already knows) + the text of a +# `## Changes since base` section of the brief. Other files a brief merely names (build plan, +# run log, files to edit) and the rest of the brief never count. +_CHANGES_HEADING = re.compile(r"^##[ \t]+Changes since base[ \t]*#*[ \t]*$", re.I | re.M) +_HEADING_1_2 = re.compile(r"^#{1,2}[ \t]", re.M) -def listed_notes(brief: str | None, cwd: str, state: str) -> list[str]: - """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.""" +def changes_section(brief: str | None) -> str: + """The body of the brief's `## Changes since base` section (up to the next level-1 or level-2 + heading, or the end), or "" when the brief has none.""" + m = _CHANGES_HEADING.search(brief or "") + if not m: + return "" + rest = brief[m.end():] + nxt = _HEADING_1_2.search(rest) + return (rest[:nxt.start()] if nxt else rest).strip() + + +def read_delta_notes(paths) -> list[dict]: + """The notes given with `--delta-note`: [{"path", "text", "mtime"}], each file once. + A note that cannot be read is refused (exit 2) before anything is written.""" out, seen = [], set() - for ref in _NOTE_REF.findall(brief or ""): - for 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 + for raw in paths or []: + p = norm_path(raw) + key = os.path.normcase(p) + if key in seen: + continue + seen.add(key) + try: + text = read_bytes_retry(p).decode("utf-8", "replace") + mtime = os.path.getmtime(p) + except OSError as e: + raise ForksError("cannot read delta note", note=p, detail=str(e)) + out.append({"path": p, "text": text, "mtime": mtime}) return out -def _text_chars(path: str) -> int: - try: - return len(read_bytes_retry(path).decode("utf-8", "replace")) - except OSError: - return 0 +def new_to_base(notes: list[dict], base: dict) -> list[dict]: + """The attached notes the base does not know yet: mtime after its `trained_at` (a training's + or a rebase's end). With no `trained_at` every note counts.""" + t = _ts((base or {}).get("trained_at")) + return [n for n in notes or [] if t is None or n["mtime"] > t] + + +def render_delta(header: str, notes: list[dict]) -> str: + """The prompt's delta section: the delta header, then each attached note with its path.""" + if not notes: + return header + parts = [header, "### Attached delta notes"] + for n in notes: + parts.append("#### `%s`\n\n%s" % (n["path"], n["text"].rstrip("\n"))) + return "\n\n".join(parts) def projects_dir() -> str: @@ -2461,10 +2547,13 @@ def binary_state(base: dict, launch: dict, cache: dict | None = None) -> str: def due_flags(state: str, reg: dict, name: str, *, phase=None, brief: str | None = None, - delta: str | None = None, binary: str | None = None, bcache: dict | None = None) -> dict: + delta: str | None = None, notes: list[dict] | None = None, binary: str | None = None, + bcache: dict | None = None) -> dict: """Design 5's rebase and re-train flags for base `name`, with their reasons (advisory: only - `binary_missing` stops a launch). `phase`/`brief` are the fork's (none in `list`); `delta` the - fork's delta header (computed here when not given); `binary` a `binary_state` value.""" + `binary_missing` stops a launch). `phase`/`brief`/`notes` are the fork's (none in `list`, which + counts the delta header alone); `delta` the fork's delta header (computed here when not + given); `notes` its `--delta-note` files (`read_delta_notes`); `binary` a `binary_state` + value.""" cfg = reg["config"] base = reg["bases"][name] la = launch_record(base, cfg) @@ -2481,7 +2570,8 @@ def due_flags(state: str, reg: dict, name: str, *, phase=None, brief: str | None rebase.append("lines:%d" % lines) if delta is None: delta = delta_header(la["cwd"], bc) - chars = len(delta) + sum(_text_chars(p) for p in listed_notes(brief, la["cwd"], state)) + chars = (len(delta) + sum(len(n["text"]) for n in new_to_base(notes, base)) + + len(changes_section(brief))) if chars // 4 > int(th["rebase_delta_tokens"]): rebase.append("delta_tokens:%d" % (chars // 4)) if binary is None: @@ -2591,6 +2681,15 @@ def cmd_train(args) -> tuple[dict, int]: if not re.fullmatch(r"[A-Za-z0-9_.-]+", name or ""): raise ForksError("bad base name", base=name) cfg = reg["config"] + supersedes = _uniq(args.supersedes or []) + + def check_supersedes(r): + if name in supersedes: + raise ForksError("a base cannot supersede itself", base=name) + unknown = [n for n in supersedes if n not in r.get("bases", {})] + if unknown: + raise ForksError("unknown base", bases=unknown, option="--supersedes") + check_supersedes(reg) la, overrides = train_launch(args, state, cfg) binfo = la["binary"] drift: list[str] = [] # the record was built from the files just now @@ -2607,13 +2706,14 @@ def cmd_train(args) -> tuple[dict, int]: pending = _pending_rebase_to(reg, name) if pending: raise ForksError("a running rebase creates this base", base=name, job=pending) + check_supersedes(reg) job = _next_job_id(reg) reg["bases"][name] = {"name": name, "status": "training", "session": session, "cwd": la["cwd"], "launch": la, "fingerprint": fingerprint_hash(la), "overrides": overrides, "binary": binfo, "created_at": _iso(), "train_job": job, "phase": _phase(args.phase), "last_hit": None, "parent": None, "lineage_rebases": 0, "base_commit": None, "context_tokens": None, - "forks": []} + "forks": [], "supersedes": supersedes} reg["jobs"][job] = _new_entry( job, "train", base=name, block=args.block, phase=_phase(args.phase), session=session, model_requested=la["model"], cwd=la["cwd"], argv=argv, max_s=float(cfg["max_minutes"]) * 60, @@ -2643,10 +2743,11 @@ def cmd_fork(args) -> tuple[dict, int]: if not base.get("session"): raise ForksError("base has no session", base=args.base, status=base.get("status")) brief_path, brief = _read_brief(args.brief) + notes = read_delta_notes(args.delta_notes) binfo, drift = _check_binary(base.get("binary") or la["binary"], la) delta = delta_header(la["cwd"], base.get("base_commit")) due = due_flags(state, reg, args.base, phase=_phase(args.phase), brief=brief, delta=delta, - binary="drift" if "binary" in drift else "ok") + notes=notes, binary="drift" if "binary" in drift else "ok") head = _git_head(la["cwd"]) session = str(uuid.uuid4()) argv = claude_argv(la) + ["--resume", base["session"], "--fork-session", "--session-id", session] @@ -2676,7 +2777,8 @@ def cmd_fork(args) -> tuple[dict, int]: base_context=base.get("context_tokens"), base_state_at_launch=warmth["base_state"], warm_until_at_launch=warmth["warm_until"], launch_drift=drift, binary_version=binfo["version"], head_at_launch=head, brief_source=brief_path, - forced=bool(args.force and base.get("status") != "active")) + forced=bool(args.force and base.get("status") != "active"), + delta_notes=[n["path"] for n in notes]) base.setdefault("forks", []).append(job) write_registry(state, reg) if args.mode == "parallel": @@ -2690,8 +2792,8 @@ def cmd_fork(args) -> tuple[dict, int]: commit_rule += " Expected paths: %s." % ", ".join("`%s`" % p for p in paths) prompt = render_template("fork.md", { "base": args.base, "run": norm_path(args.run), "state_dir": state, "block": args.block, - "phase": "n/a" if args.phase is None else str(args.phase), "mode": args.mode, "delta": delta, - "commit_rule": commit_rule, "note_path": os.path.join(state, *note_rel(args.block).split("/")), + "phase": "n/a" if args.phase is None else str(args.phase), "mode": args.mode, + "delta": render_delta(delta, notes), "commit_rule": commit_rule, "note_path": os.path.join(state, *note_rel(args.block).split("/")), "logs_dir": os.path.join(state, "logs"), "brief": brief}) _start_job(state, norm_path(args.run), job, prompt, la) return {"job": job, "base": args.base, "base_state": warmth["base_state"], @@ -2772,6 +2874,7 @@ def cmd_rebase(args) -> tuple[dict, int]: raise ForksError("base is not active", base=args.base, status=base.get("status")) la = launch_record(base, cfg) brief_path, brief = _read_brief(args.brief) + notes = read_delta_notes(args.delta_notes) binfo, drift = _check_binary(base.get("binary") or la["binary"], la) # The new base's record is the old one's; a binary drift means the rebase runs under the # current binary, which the new base then records (design 5). @@ -2813,12 +2916,13 @@ def cmd_rebase(args) -> tuple[dict, int]: base_context=base.get("context_tokens"), base_state_at_launch=warmth["base_state"], warm_until_at_launch=warmth["warm_until"], launch_drift=drift, binary_version=binfo["version"], head_at_launch=head, brief_source=brief_path, - rebase_as=new, new_launch=new_launch) + rebase_as=new, new_launch=new_launch, delta_notes=[n["path"] for n in notes]) write_registry(state, reg) prompt = render_template("rebase.md", { "base": args.base, "new_base": new, "run": norm_path(args.run), "state_dir": state, "notes_dir": os.path.join(state, "notes"), "logs_dir": os.path.join(state, "logs"), - "phase": "n/a" if args.phase is None else str(args.phase), "delta": delta, "brief": brief}) + "phase": "n/a" if args.phase is None else str(args.phase), "delta": render_delta(delta, notes), + "brief": brief}) _start_job(state, norm_path(args.run), job, prompt, la) return {"job": job, "base": new, "parent": args.base, "session_id": session}, EXIT_OK @@ -2852,6 +2956,20 @@ def pings_since_real_call(reg: dict, name: str) -> int: if e.get("base") == name and e.get("kind") == "ping" and _job_num(j) > last_real) +def ping_phase(reg: dict, name: str): + """I2: the phase a ping on base `name` belongs to: that of the latest (by job number) train, + fork, continue or rebase job on the base that has a phase; else the base's own phase; else + None. So `usage --phase N` carries the keep-warm cost of the phase's bases.""" + best, ph = -1, None + for j, e in reg.get("jobs", {}).items(): + if e.get("base") == name and e.get("kind") in REAL_KINDS and e.get("phase") is not None \ + and _job_num(j) > best: + best, ph = _job_num(j), e["phase"] + if ph is None: + ph = (reg.get("bases", {}).get(name) or {}).get("phase") + return ph + + def ping_in_flight(reg: dict, base: dict | None) -> str | None: """The base's in-flight ping job (C3d), or None. A mark whose job is no longer running (or gone) does not count: it clears itself on the next marking.""" @@ -2949,7 +3067,8 @@ def tick_pass(state: str) -> dict: argv = claude_argv(la) + ["--resume", base["session"], "--fork-session", "--session-id", session, "--max-turns", "1"] reg["jobs"][job] = _new_entry( - job, "ping", base=n, session=session, parent_session=base["session"], + job, "ping", base=n, phase=ping_phase(reg, n), session=session, + parent_session=base["session"], model_requested=la["model"], cwd=la["cwd"], argv=argv, max_s=PING_MAX_S, cost_before=float(base.get("cost_cum_usd") or 0.0), model_usage_before=base.get("model_usage_cum") or {}, @@ -3179,6 +3298,9 @@ class Supervisor: _diag("swept %d leftover process(es) of claude: %s" % ( len(sweep["snapshot"]), [(x["pid"], x["name"]) for x in sweep["snapshot"]])) reader.join(READER_JOIN_S) + if not reader.is_alive(): # the reader is done with the pipe (Minor 10) + with contextlib.suppress(OSError): + p.stdout.close() with contextlib.suppress(OSError): raw.close() with contextlib.suppress(OSError): @@ -3558,6 +3680,29 @@ def usage_md_phase(agg: dict) -> str: "- Priced total: %s" % _cost(agg["total"]["cost_usd"])]) + "\n" +RESULT_STATUS_KEYS = ("status", "flags", "reason") + + +def status_from_results(state: str, lines: list[dict]) -> tuple[list[dict], list[str]]: + """Minor 11: the ledger is append-only, so a job finalised again after its line was written + (R1 reconciled a result whose entry still read `running`, then `reap_lost` or `kill` rewrote + the result as `lost`/`killed`) keeps its first status there. `usage` takes each job's + status, flags and reason from `results/.json` when they differ from its ledger line; + tokens and cost stay the ledger's. Returns (lines, the jobs whose status came from a result).""" + out, changed = [], [] + for ln in lines: + try: + res = read_json(result_file(state, ln["job"])) + except ForksError: + res = None + if isinstance(res, dict) and any(res.get(k) != ln.get(k) for k in RESULT_STATUS_KEYS): + ln = dict(ln, status=res.get("status"), flags=list(res.get("flags") or []), + reason=res.get("reason")) + changed.append(ln["job"]) + out.append(ln) + return out, changed + + def cmd_usage(args) -> tuple[dict | str, int]: """`usage --run R [--block ID | --phase N] [--format json|md]`. R1 first: reconcile the ledger against `results/` under the lock. Allowed inside a fork (N3).""" @@ -3573,12 +3718,14 @@ def cmd_usage(args) -> tuple[dict | str, int]: elif args.phase is not None: scope = {"phase": _phase(args.phase)} lines = [x for x in lines if x.get("phase") == scope["phase"]] + lines, from_result = status_from_results(state, lines) if args.format == "md": if args.block is not None: return usage_md_block(lines, reg.get("jobs") or {}), EXIT_OK return usage_md_phase(usage_aggregate(lines)), EXIT_OK agg = usage_aggregate(lines) - return {"state": state, "scope": scope, "reconciled": added, **agg, "rows": lines}, EXIT_OK + return {"state": state, "scope": scope, "reconciled": added, "status_from_result": from_result, + **agg, "rows": lines}, EXIT_OK HANDLERS = { diff --git a/plugins/spire/skills/forks/tests/test_forks.py b/plugins/spire/skills/forks/tests/test_forks.py index 7bd70ec..43e54a6 100644 --- a/plugins/spire/skills/forks/tests/test_forks.py +++ b/plugins/spire/skills/forks/tests/test_forks.py @@ -721,6 +721,62 @@ class InitTests(Base): self.assertFalse(out["changed"]) self.assertEqual(self.reg_bytes(), before) + def test_rerun_keeps_omitted_settings(self): + """Minor 8: on a re-run, flags that are not given keep their current values (launch fields + and run-wide settings); defaults apply only on the first init; `changed_fields` lists only + what the given flags changed.""" + code, first, _ = self.init("--pings", "off", "--parallel-cap", "3", "--rebase-commits", "9", + "--orphan-after", "1500", "--effort", "low", "--tools", "Read,Bash", + model="claude-sonnet-5-5") + self.assertEqual(code, 0, first) + self.add_fake_base() + before = forks.read_registry(self.state)["config"] + code, out, raw = call(["init", "--run", self.run_log, "--warm-margin-s", "100"]) + self.assertEqual(code, 0, out) + self.assert_one_json_line(raw) + self.assertTrue(out["changed"]) + self.assertEqual(out["changed_fields"], ["warm_margin_s"]) + cfg = forks.read_registry(self.state)["config"] + self.assertEqual(cfg["warm_margin_s"], 100) + self.assertEqual((cfg["pings"], cfg["parallel_cap"], cfg["orphan_after"]), ("off", 3, 1500)) + self.assertEqual(cfg["thresholds"], dict(forks.THRESHOLD_DEFAULTS, rebase_commits=9)) + self.assertEqual(cfg["launch"], before["launch"]) + self.assertEqual((cfg["launch"]["model"], cfg["launch"]["effort"], cfg["launch"]["tools"]), + ("claude-sonnet-5-5", "low", ["Read", "Bash"])) + self.assertEqual(cfg["fingerprint"], before["fingerprint"]) + # a given flag still changes its value, and only it is listed + code, out, _ = call(["init", "--run", self.run_log, "--pings", "on"]) + self.assertEqual(code, 0, out) + self.assertEqual(out["changed_fields"], ["pings"]) + cfg = forks.read_registry(self.state)["config"] + self.assertEqual((cfg["pings"], cfg["warm_margin_s"], cfg["parallel_cap"]), ("on", 100, 3)) + # the first init still needs --model (a usage error, nothing written) + other = os.path.join(self.tmp, "reports", "2026-10-16-09-u-run.md") + code, out, raw = call(["init", "--run", other, "--cwd", self.cwd, "--binary", self.stub]) + self.assertEqual(code, 2, out) + self.assert_one_json_line(raw) + self.assertEqual(out["error"], "usage") + self.assertIn("--model", out["detail"]) + self.assertFalse(os.path.exists(forks.state_dir_for(other))) + + def test_warm_margin_must_be_below_ttl(self): + """Minor 2: `init` refuses a warm_margin_s at or above the configured ttl_s (exit 2), on the + first init and on a re-run that lowers ttl_s under the current margin.""" + out = self.assert_refused_no_state("--warm-margin-s", "3600") + self.assertEqual((out["warm_margin_s"], out["ttl_s"]), (3600, 3600)) + self.assert_refused_no_state("--ttl-s", "300", "--warm-margin-s", "300") + code, out, _ = self.init("--warm-margin-s", "3599") + self.assertEqual(code, 0, out) + before = self.reg_bytes() + code, out, raw = self.init("--ttl-s", "3000") + self.assertEqual(code, 2, out) + self.assert_one_json_line(raw) + self.assertIn("below ttl_s", out["error"]) + self.assertEqual(self.reg_bytes(), before) + code, out, _ = self.init("--ttl-s", "300", "--warm-margin-s", "200") + self.assertEqual(code, 0, out) + self.assertEqual(sorted(out["changed_fields"]), ["ttl_s", "warm_margin_s"]) + def test_fingerprint_hash_stable_across_runs(self): """M7: the hash covers the mcp config's content, not its per-run path: two runs with the same launch fields hash the same; the path stays in the config, outside the hash.""" @@ -1635,11 +1691,28 @@ class WarmthTests(Base): with self.subTest(margin=900, t=t - T0): self.assertEqual(st(b, m, t), want) self.assertEqual(forks.base_warmth(b, m, T0)["margin_s"], 900) - for t, want in ((T0, "expiring"), (T0 + 299, "expiring"), (T0 + 300, "cold")): + # Minor 2: at a 5-minute TTL the 900 s margin is capped at ttl/2 = 150 s + for t, want in ((T0, "warm"), (T0 + 149, "warm"), (T0 + 150, "expiring"), (T0 + 299, "expiring"), + (T0 + 300, "cold")): with self.subTest(margin=900, base_ttl=300, t=t - T0): self.assertEqual(st({**b, "ttl_s": 300}, m, t), want) + self.assertEqual(forks.base_warmth({**b, "ttl_s": 300}, m, T0)["margin_s"], 150) self.assertEqual(st({}, cfg, T0), "cold") + def test_margin_capped_at_half_the_ttl(self): + """Minor 2: a configured margin above ttl/2 is capped there, so a 5-minute downgrade does not + leave a base `expiring` (and pinged every tick) right after every hit.""" + T0 = self.T0 + b = {"last_hit": forks._iso(T0), "ttl_s": 300} + w = forks.base_warmth(b, {"ttl_s": 3600, "warm_margin_s": 600}, T0 + 1) + self.assertEqual((w["base_state"], w["margin_s"], w["ttl_s"]), ("warm", 150, 300)) + reg = self.make_reg(T0 + 1, margin=600, base_ttl=300) + self.assertEqual(self.decide(reg, T0 + 1)["why"], "state:warm") + self.assertTrue(self.decide(reg, T0 + 150)["ping"]) + # a margin below ttl/2 is used as given; the formula applies when none is set + self.assertEqual(forks.base_warmth(b, {"ttl_s": 3600, "warm_margin_s": 100}, T0)["margin_s"], 100) + self.assertEqual(forks.base_warmth(b, {"ttl_s": 3600, "warm_margin_s": None}, T0)["margin_s"], 120) + def test_infer_ttl_and_downgrade(self): def t(h=0, m=0, r=0): return {"cache_creation_1h": h, "cache_creation_5m": m, "cache_read": r} @@ -1684,8 +1757,11 @@ class WarmthTests(Base): # 5 min TTL inferred on the base, margin from init d = self.decide(self.make_reg(self.E5M, base_ttl=300), self.E5M) self.assertEqual((d["ping"], d["ttl_s"], d["break_even"]), (True, 300, 11)) - t = self.T0 + 100 + # init's margin, capped at ttl/2 (Minor 2): 3600 acts as 1800 at a 1 h TTL + t = self.T0 + 1800 self.assertTrue(self.decide(self.make_reg(t, margin=3600), t)["ping"]) + t = self.T0 + 1799 + self.assertEqual(self.decide(self.make_reg(t, margin=3600), t)["why"], "state:warm") def test_ping_rule_activity_window(self): now = self.E1H @@ -1847,6 +1923,53 @@ class TickTests(Base): self.assertEqual(len(out["pinged"]), 1) self.assertEqual(out["bases"][0]["pings_since_real_call"], 1) # counted before this one + def test_ping_carries_the_phase_and_counts_in_usage_phase(self): + """I2: a ping records the phase of the latest train/fork/continue/rebase job on its base + (the base's own phase if none has one), so `usage --phase N` (md and json) lists it and its + cost is in the priced total.""" + reg = self.reg() + self.assertIsNone(forks.ping_phase(reg, "B1")) # nothing has a phase + reg["bases"]["B1"]["phase"] = 1 + self.assertEqual(forks.ping_phase(reg, "B1"), 1) # the base's own + reg["jobs"]["j0002"] = forks._new_entry("j0002", "fork", base="B1", phase=2, status="ok", + ended_at=forks._iso(time.time() - 4000)) + reg["jobs"]["j0003"] = forks._new_entry("j0003", "fork", base="B9", phase=7, status="ok") + reg["jobs"]["j0004"] = forks._new_entry("j0004", "continue", base="B1", phase=None, status="ok") + reg["jobs"]["j0005"] = forks._new_entry("j0005", "ping", base="B1", phase=5, status="ok") + self.assertEqual(forks.ping_phase(reg, "B1"), 2) # latest real job with a phase + with forks.RegistryLock(self.state): + forks.write_registry(self.state, reg) + out = forks.tick_pass(self.state) + self.assertEqual(len(out["pinged"]), 1, out) + job = out["pinged"][0] + self.assertEqual(self.reg()["jobs"][job]["phase"], 2) + # the ping ends ok with a known cost (cost basis: the base's 1.25) + ss = forks.StreamState() + t = time.time() + u = {"input_tokens": 1, "cache_read_input_tokens": 1000, "cache_creation_input_tokens": 0} + ss.feed({"type": "assistant", "message": {"id": "m1", "usage": u}}, t) + ss.feed({"type": "result", "is_error": False, "num_turns": 1, "total_cost_usd": 1.2623, + "usage": dict(u, output_tokens=2, cache_creation={}), + "modelUsage": {TEST_MODEL: {"inputTokens": 2, "outputTokens": 2, + "cacheReadInputTokens": 1000, "cacheCreationInputTokens": 0}}}, + t) + res = forks.finalize_job(self.state, job, ss=ss, hint="exited", exit_code=0, wall_s=1.0) + self.assertEqual((res["status"], res["phase"], res["cost_delta_usd"]), ("ok", 2, 0.0123), res) + line = [x for x in forks.read_ledger(self.state) if x["job"] == job][0] + self.assertEqual(line["phase"], 2) + code, js, _ = call(["usage", "--run", self.run_log, "--phase", "2"]) + self.assertEqual(code, 0, js) + self.assertEqual([x["job"] for x in js["rows"]], [job]) + self.assertEqual(js["kinds"]["ping"]["calls"], 1) + self.assertAlmostEqual(js["kinds"]["ping"]["cost_usd"], 0.0123, places=6) + self.assertAlmostEqual(js["total"]["cost_usd"], 0.0123, places=6) + code, md = call_raw(["usage", "--run", self.run_log, "--phase", "2", "--format", "md"]) + self.assertEqual(code, 0, md) + self.assertIn("| ping | 1 | 1 | 2 | 0 | 1,000 | $0.0123 |\n", md) + self.assertIn("| total | 1 | 1 | 2 | 0 | 1,000 | $0.0123 |\n", md) + self.assertTrue(md.endswith("- Priced total: $0.0123\n"), md) + self.assertEqual(call(["usage", "--run", self.run_log, "--phase", "1"])[1]["rows"], []) + def test_tick_command_and_n3(self): code, out, raw = call(["tick", "--run", self.run_log]) self.assertEqual(code, 0, out) @@ -2185,21 +2308,48 @@ class UsageTests(Base): self.assertEqual(self.usage("--phase", "3")["reconciled"], []) def test_r1_waits_for_the_registry_lock(self): - """With the lock held and a line missing, `usage` blocks until it is released, then - reconciles.""" + """With the lock held and a line missing, the reconciliation itself blocks until the lock is + released, then appends. Minor 9: `reconcile_ledger` is called directly (`usage` would wait + on its orchestrator stamp first, so a reconciliation without the lock would pass too).""" self.result_for("j0010", block="4.1", phase=4) done = [] lock = forks.RegistryLock(self.state) lock.acquire() try: - th = threading.Thread(target=lambda: done.append(self.usage("--phase", "4"))) + th = threading.Thread(target=lambda: done.append(forks.reconcile_ledger(self.state))) th.start() time.sleep(1.0) self.assertEqual(done, []) # blocked on the lock + self.assertNotIn("j0010", [x["job"] for x in self.ledger_lines()]) finally: lock.release() th.join(30) - self.assertEqual(done[0]["reconciled"], ["j0010"]) + self.assertEqual(done, [["j0010"]]) + self.assertEqual(len([x for x in self.ledger_lines() if x["job"] == "j0010"]), 1) + + def test_status_from_result_when_it_changed_after_the_ledger_line(self): + """Minor 11: the ledger stays append-only; a job finalised again after its line was written + (R1 line, then `reap_lost`/`kill` rewrote the result) reads its status, flags and reason + from the result in `usage`; tokens and cost stay the ledger's.""" + self.result_for("j0002", status="lost", flags=["cold_start"], reason="supervisor_dead", + block="1.1", phase=1, tokens=dict(zip(forks.TOKEN_KEYS, (9, 9, 9, 9, 0, 9))), + cost_delta_usd=9.0) + self.result_for("j0004", block="1.2", phase=1, flags=[]) # agrees with its line + n0 = len(self.ledger_lines()) + out = self.usage("--phase", "1") + self.assertEqual(out["status_from_result"], ["j0002"]) + row = {x["job"]: x for x in out["rows"]}["j0002"] + self.assertEqual((row["status"], row["flags"], row["reason"]), ("lost", ["cold_start"], "supervisor_dead")) + self.assertEqual(row["tokens"]["cache_read"], 1234567) # the ledger's, not the result's + self.assertAlmostEqual(row["cost_delta_usd"], 1.2345, places=6) + self.assertAlmostEqual(out["total"]["cost_usd"], 2.0845, places=6) + self.assertEqual(out["cold_starts"], 2) # the result's flags count + self.assertEqual({x["job"]: x["status"] for x in out["rows"]}["j0004"], "ok") + # the ledger itself is unchanged (append-only) + led = self.ledger_lines() + self.assertEqual(len(led), n0) + self.assertEqual([x["status"] for x in led if x["job"] == "j0002"], ["ok"]) + self.assertEqual(self.usage("--phase", "2")["status_from_result"], []) def test_r1_nothing_missing_takes_no_lock(self): self.result_for("j0002") # a result that has its ledger line @@ -2862,11 +3012,12 @@ class LaunchTests(LaunchBase): """A stubbed ping: launched by `wait`'s tick, returned once through `wait`'s `pings` (never `done`); a `partial` ping refreshes last_hit, a missed or failed one does not and carries `ping_missed`; the in-flight mark clears; each ping has its ledger line.""" - self.setup_run("--warm-margin-s", "3600") # expiring from the last hit on + # expiring from 1800 s after the last hit on (a margin must be below ttl_s; Minor 2) + self.setup_run("--warm-margin-s", "1800") self.train("B1") self.finish("j0001") base_session = self.reg()["bases"]["B1"]["session"] - old = forks._iso(time.time() - 100) + old = forks._iso(time.time() - 2000) self.set_last_hit(old) # 1. partial, launched by wait's tick (the wait itself is the activity) with mock.patch.object(forks, "PING_PROMPT", "Reply OK\n@stub cache=partial"): @@ -3035,12 +3186,14 @@ class LaunchTests(LaunchBase): self.clone_base("Bl", base_commit=c7) self.assert_due(self.fake_fork("Bl"), "lines:1600") # every clone from here on starts at HEAD - # delta_tokens: chars / 4 of the delta header plus the notes the brief lists > 12000 - with open(os.path.join(self.state, "notes", "big-note.md"), "w") as f: + # delta_tokens: chars / 4 of the delta header plus the attached delta notes > 12000 (I1) + big_note = os.path.join(self.state, "notes", "big-note.md") + with open(big_note, "w") as f: f.write("n" * 60000) self.clone_base("Bd") - self.assert_due(self.fake_fork("Bd")) # brief lists no note - out = self.fake_fork("Bd", brief=self.brief(text="Read notes/big-note.md first.")) + self.assert_due(self.fake_fork("Bd")) # no note attached + self.assert_due(self.fake_fork("Bd", brief=self.brief(text="Read notes/big-note.md first."))) + out = self.fake_fork("Bd", "--delta-note", big_note) self.assertEqual(len(out["reasons"]), 1, out) self.assertRegex(out["reasons"][0], r"^delta_tokens:\d+$") self.assertGreater(int(out["reasons"][0].split(":")[1]), 15000) @@ -3102,6 +3255,105 @@ class LaunchTests(LaunchBase): with self.subTest(base=name): self.assert_due(rows[name], *want) + def test_delta_tokens_counts_only_what_the_base_does_not_know(self): + """I1: delta_tokens = the delta header + `--delta-note` files newer than the base's training + + a `## Changes since base` section of the brief. Large reference docs a brief merely names + never count; an attached note is in the fork's prompt with its path; a base trained or + rebased after the note's mtime does not count it, so a rebase clears the reason.""" + self.setup_run(cap=4) + self.train("B1") + self.finish("j0001") + os.makedirs(os.path.join(self.cwd, "docs")) + for rel in ("docs/build-plan.md", "docs/run-log.md"): + with open(os.path.join(self.cwd, *rel.split("/")), "w") as f: + f.write("r" * 80000) + with open(os.path.join(self.state, "notes", "1.1.md"), "w") as f: + f.write("h" * 80000) + named = self.brief(text="Read docs/build-plan.md, docs/run-log.md and notes/1.1.md; edit " + "docs/run-log.md and %s." % os.path.join(self.cwd, "docs", "build-plan.md")) + self.assert_due(self.fake_fork("B1", brief=named)) # named only: nothing + # a large attached note trips it, and the fork's prompt carries it with its path + note = os.path.join(self.tmp, "delta-1.md") + with open(note, "w", newline="\n") as f: + f.write("NOTE-MARKER\n" + "d" * 60000 + "\n") + out = self.fake_fork("B1", "--delta-note", note, brief=named) + self.assertEqual(len(out["reasons"]), 1, out) + self.assertRegex(out["reasons"][0], r"^delta_tokens:\d+$") + self.assertGreater(int(out["reasons"][0].split(":")[1]), 15000) + with open(os.path.join(self.state, "briefs", out["job"] + ".md"), encoding="utf-8") as fh: + prompt = fh.read() + delta_at = prompt.index("## Delta since training") + self.assertLess(delta_at, prompt.index("`%s`" % forks.norm_path(note))) + self.assertLess(prompt.index("NOTE-MARKER"), prompt.index("## Rules for this fork")) + self.assertEqual(self.reg()["jobs"][out["job"]]["delta_notes"], [forks.norm_path(note)]) + # a small attached note does not; a missing one is refused + small = os.path.join(self.tmp, "delta-small.md") + with open(small, "w") as f: + f.write("small\n") + self.assert_due(self.fake_fork("B1", "--delta-note", small)) + self.assertEqual(self.fake_fork("B1", "--delta-note", os.path.join(self.tmp, "nope.md"), + code=2)["error"], "cannot read delta note") + # a base trained after the note's mtime already knows it + mtime = os.path.getmtime(note) + self.clone_base("Bk", trained_at=forks._iso(mtime + 60)) + self.assert_due(self.fake_fork("Bk", "--delta-note", note)) + # a large `## Changes since base` section trips it; the same text under another heading not + body = "c" * 60000 + out = self.fake_fork("Bk", brief=self.brief(text="## Changes since base\n\n%s\n\n## Gate\n\nrun it" % body)) + self.assertEqual(len(out["reasons"]), 1, out) + self.assertRegex(out["reasons"][0], r"^delta_tokens:\d+$") + self.assert_due(self.fake_fork("Bk", brief=self.brief(text="## Background\n\n%s\n" % body))) + # a rebase after the note clears the reason; the rebase's prompt carries the note too + r = self.cmd("rebase", "--base", "B1", "--as", "B2", "--delta-note", note, + "--brief", self.brief("turns=1", text="Absorb.")) + with open(os.path.join(self.state, "briefs", r["job"] + ".md"), encoding="utf-8") as fh: + rprompt = fh.read() + self.assertIn("NOTE-MARKER", rprompt) + self.assertLess(rprompt.index("## Delta since the base commit"), rprompt.index("NOTE-MARKER")) + self.assertLess(rprompt.index("NOTE-MARKER"), rprompt.index("## Brief")) + self.assertEqual(self.finish(r["job"])["status"], "ok") + self.assertGreater(forks._ts(self.reg()["bases"]["B2"]["trained_at"]), mtime) + self.assert_due(self.fake_fork("B2", "--delta-note", note)) + # list counts the delta header only + rows = {b["name"]: b for b in self.cmd("list")["bases"]} + self.assert_due(rows["B2"]) + + def test_train_supersedes(self): + """Minor 1: `train --supersedes B` (repeatable) marks each named active base superseded by + the new one when its training ends ok: no more pings, and a fork needs --force. Unknown + names are refused at launch; a failed training changes nothing.""" + self.setup_run(cap=4) + self.train("B1") + self.finish("j0001") + self.train("B0") + self.finish("j0002") + out = self.cmd("train", "--base", "B2", "--brief", self.brief(), "--supersedes", "nope", code=2) + self.assertEqual((out["error"], out["bases"]), ("unknown base", ["nope"])) + out = self.cmd("train", "--base", "B2", "--brief", self.brief(), "--supersedes", "B2", code=2) + self.assertIn("itself", out["error"]) + self.assertNotIn("B2", self.reg()["bases"]) + # a failed training: nothing changes + t = self.cmd("train", "--base", "B2", "--brief", self.brief("error"), "--supersedes", "B1") + self.assertEqual(self.finish(t["job"])["status"], "failed") + reg = self.reg() + self.assertEqual((reg["bases"]["B2"]["status"], reg["bases"]["B1"]["status"]), ("failed", "active")) + self.assertNotIn("superseded_by", reg["bases"]["B1"]) + # an ok training supersedes every named active base + t = self.cmd("train", "--base", "B3", "--brief", self.brief(), "--supersedes", "B1", + "--supersedes", "B0", "--supersedes", "B2") + self.assertEqual(self.finish(t["job"])["status"], "ok") + reg = self.reg() + for name in ("B1", "B0"): + with self.subTest(base=name): + b = reg["bases"][name] + self.assertEqual((b["status"], b["superseded_by"]), ("superseded", "B3")) + self.assertIsNotNone(b["superseded_at"]) + self.assertEqual(reg["bases"]["B2"]["status"], "failed") # not active: untouched + self.assertEqual(reg["bases"]["B3"]["status"], "active") + self.assertEqual(forks.ping_decision(reg, "B1")["why"], "status:superseded") + self.assertIn("not active", self.fake_fork("B1", code=2)["error"]) + self.fake_fork("B1", "--force") + def test_train_overrides_and_init_model_change(self): """M10 (plan 4.3): `train` takes init's launch flags as overrides; the base records its launch fields and their hash, and every later launch on it uses that record. `init`