dirigence/reliquary
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
g4borg
2026-10-16 18:10:44 +02:00
parent 9564c22b34
commit d2d51049d2
2 changed files with 482 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/spire/skills/forks/forks.py
+214 -67
View File
@@ -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 (<cwd>/.claude/agents/<name>.md)")
i.add_argument("--cwd", help="working dir of base and forks (default: git toplevel)")
i.add_argument("--binary", help="claude binary (default: $CLAUDE_CODE_EXECPATH, else PATH)")
i.add_argument("--parallel-cap", type=_pos_int, default=CONFIG_DEFAULTS["parallel_cap"])
i.add_argument("--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 (<cwd>/.claude/agents/<name>.md)")
g.add_argument("--cwd", help="working dir of the base and its forks")
g.add_argument("--binary", help="claude binary")
t.add_argument("--supersedes", action="append", default=None, metavar="BASE",
help="a base this one replaces (repeatable): when the training ends ok, each "
"named base that is active becomes superseded")
def launch_overrides(sp):
# Accepted only so that they can be refused with a clear message (plan 4.3).
@@ -1923,6 +1938,9 @@ def build_parser() -> _Parser:
f.add_argument("--max-minutes", type=_pos_float, help="wall-time limit (default: init's max_minutes)")
f.add_argument("--max-budget-usd", type=_pos_float)
f.add_argument("--force", action="store_true", help="fork a base that is not active")
f.add_argument("--delta-note", action="append", default=None, metavar="PATH", dest="delta_notes",
help="a note the fork must learn (repeatable): appended to the prompt after the "
"delta header; counts toward delta_tokens when newer than the base's training")
launch_overrides(f)
c = cmd("continue", "One more turn in a fork (resume without --fork-session).")
@@ -1959,6 +1977,8 @@ def build_parser() -> _Parser:
r.add_argument("--phase", help="the new base's phase (the phase whose forks it serves)")
r.add_argument("--max-minutes", type=_pos_float, help="wall-time limit (default: init's max_minutes)")
r.add_argument("--max-budget-usd", type=_pos_float)
r.add_argument("--delta-note", action="append", default=None, metavar="PATH", dest="delta_notes",
help="a note the rebase must absorb (repeatable): appended after the delta header")
launch_overrides(r)
u = cmd("usage", "Aggregate the ledger: tokens and cost per kind, cold starts, cache-hit ratio.")
@@ -2098,39 +2118,73 @@ def build_launch(state: str, *, model, effort, permission_mode, tools, allowed,
return launch, adm["via"]
def build_init_config(args, state: str) -> dict:
"""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)
disallowed = list(ol["disallowed_tools"]) if "disallowed_tools" in ol else list(DEFAULT_DISALLOWED)
if args.cwd is None and ol.get("cwd"):
cwd = ol["cwd"]
if not os.path.isdir(cwd):
raise ForksError("--cwd is not a directory", cwd=cwd)
else:
cwd = _resolve_cwd(args.cwd)
if not MODEL_RE.match(args.model or ""): # before the binary lookup, as in Phase 1
raise ForksError("model must be a full id (aliases are refused)", model=args.model,
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)
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
for raw in paths or []:
p = norm_path(raw)
key = os.path.normcase(p)
if os.path.isfile(p):
if key not in seen:
if key in seen:
continue
seen.add(key)
out.append(p)
break
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/<job>.json` when they differ from its ledger line;
tokens and cost stay the ledger's. Returns (lines, the jobs whose status came from a result)."""
out, changed = [], []
for ln in lines:
try:
res = read_json(result_file(state, ln["job"]))
except ForksError:
res = None
if isinstance(res, dict) and any(res.get(k) != ln.get(k) for k in RESULT_STATUS_KEYS):
ln = dict(ln, status=res.get("status"), flags=list(res.get("flags") or []),
reason=res.get("reason"))
changed.append(ln["job"])
out.append(ln)
return out, changed
def cmd_usage(args) -> tuple[dict | str, int]:
"""`usage --run R [--block ID | --phase N] [--format json|md]`. R1 first: reconcile the ledger
against `results/` under the lock. Allowed inside a fork (N3)."""
@@ -3573,12 +3718,14 @@ def cmd_usage(args) -> tuple[dict | str, int]:
elif args.phase is not None:
scope = {"phase": _phase(args.phase)}
lines = [x for x in lines if x.get("phase") == scope["phase"]]
lines, from_result = status_from_results(state, lines)
if args.format == "md":
if args.block is not None:
return usage_md_block(lines, reg.get("jobs") or {}), EXIT_OK
return usage_md_phase(usage_aggregate(lines)), EXIT_OK
agg = usage_aggregate(lines)
return {"state": state, "scope": scope, "reconciled": added, **agg, "rows": lines}, EXIT_OK
return {"state": state, "scope": scope, "reconciled": added, "status_from_result": from_result,
**agg, "rows": lines}, EXIT_OK
HANDLERS = {
plugins/spire/skills/forks/tests/test_forks.py
+264 -12
View File
@@ -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`