From 5fda2e2f3692a6081965539b16073762864aad73 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Gabor=20K=C3=B6rber?= Date: Fri, 16 Oct 2026 17:23:20 +0200 Subject: [PATCH] =?UTF-8?q?=F0=9F=A4=96=20forks:=20rebase=20and=20re-train?= =?UTF-8?q?=20flags=20(block=202.2)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- plugins/spire/skills/forks/forks.py | 660 ++++++++++++++++-- .../spire/skills/forks/templates/rebase.md | 17 + .../spire/skills/forks/tests/test_forks.py | 531 +++++++++++++- 3 files changed, 1120 insertions(+), 88 deletions(-) create mode 100644 plugins/spire/skills/forks/templates/rebase.md diff --git a/plugins/spire/skills/forks/forks.py b/plugins/spire/skills/forks/forks.py index 460639b..64c9570 100644 --- a/plugins/spire/skills/forks/forks.py +++ b/plugins/spire/skills/forks/forks.py @@ -122,7 +122,7 @@ SCRUB_WATCH_PREFIXES = ("CLAUDE_", "ANTHROPIC_") # `where` lists unknown vars # C8: the user settings file. Module-level so tests can point it at a temp file. USER_SETTINGS_PATH: str | None = None -PHASE2_COMMANDS = ("rebase", "usage") +PHASE2_COMMANDS = ("usage",) # N3: a fork may only run where, list, usage and wait (`tick` launches pings). CHILD_REFUSED = ("train", "fork", "continue", "rebase", "kill", "expire", "init", "tick") @@ -156,6 +156,21 @@ LAUNCH_FIELDS = ("model", "effort", "permission_mode", "tools", "allowed_tools", MU_FIELDS = ("inputTokens", "outputTokens", "cacheReadInputTokens", "cacheCreationInputTokens") TOKEN_KEYS = ("input", "output", "cache_creation", "cache_creation_1h", "cache_creation_5m", "cache_read") TEMPLATES_DIR = "templates" +CAPPED_KINDS = ("fork", "continue", "rebase") # parallel_cap counts these (pings never) + +# Rebase and re-train flags (design 5, block 2.2). +# Runtime fields of a `system/init` event (compared with the base's; M6 adds `skills`). +RUNTIME_FIELDS = ("tools", "permissionMode", "plugins", "agents", "apiKeySource", "skills") +# Fingerprint-defining launch fields: a difference means a new base (retrain_due `defining:`). +DEFINING_FIELDS = ("model", "cwd", "tools", "permission_mode", "agent", "append_system_prompt") +# M11: a note counts for a job when its mtime is at or after the job's creation. File times on +# Windows can lag the precise clock by a timer tick; this slack absorbs that (a stale note from an +# earlier job of the same block is minutes old). +NOTE_MTIME_SLACK_S = 2.0 +# P0.7: transcripts live in / '-'>/.jsonl. +# Module-level so tests can point it at a temp dir. +CLAUDE_PROJECTS_DIR: str | None = None +TRANSCRIPT_TAIL_BYTES = 256 * 1024 # CIM hidden spawn (P0.1 rung 3 + ShowWindow=0). Command line and cwd arrive through env vars. POWERSHELL_CIM = ( @@ -237,8 +252,18 @@ def canonical_json(obj) -> str: return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=True) +def fingerprint_view(launch: dict) -> dict: + """What the fingerprint hash covers (M7): the launch fields, with the mcp config's **content** + in place of its per-run path, so the same launch hashes the same in every run.""" + view = dict(launch) + m = view.get("mcp") + if isinstance(m, dict): + view["mcp"] = {"strict": bool(m.get("strict")), "servers": dict(MCP_EMPTY["mcpServers"])} + return view + + def fingerprint_hash(launch: dict) -> str: - return _sha256_bytes(canonical_json(launch).encode("utf-8")) + return _sha256_bytes(canonical_json(fingerprint_view(launch)).encode("utf-8")) # -------------------------------------------------------------------------------------- @@ -248,10 +273,36 @@ def fingerprint_hash(launch: dict) -> str: _GITBASH_DRIVE = re.compile(r"^/([A-Za-z])(?=/|$)") +def find_cygpath() -> str | None: + """`cygpath` on PATH (Git Bash ships it), or None. Tests monkeypatch this lookup.""" + return shutil.which("cygpath") + + +def cygpath_w(p: str) -> str: + """M13: a Git Bash rooted path that is not a drive form (`/tmp/x`, `/usr/...`) converted with + `cygpath -w`. Refused (exit 2) when cygpath is not on PATH or fails; never mapped onto the + cwd's drive.""" + exe = find_cygpath() + if not exe: + raise ForksError("Git Bash rooted path is not a drive form and cygpath is not on PATH to " + "convert it; pass a drive path (G:\\... or /g/...)", path=p) + try: + r = _run([exe, "-w", p], timeout=30) + except (OSError, subprocess.SubprocessError) as e: + raise ForksError("cygpath failed", path=p, detail=str(e)) + out = r.stdout.decode("utf-8", "replace").strip() + if r.returncode != 0 or not out: + raise ForksError("cygpath failed", path=p, + detail=r.stderr.decode("utf-8", "replace").strip()[:200] or "no output") + return out + + def norm_path(p: str, cwd: str | None = None, windows: bool | None = None) -> str: """Normalise relative, `G:\\…`, `G:/…` and Git Bash `/g/…` forms. - Windows: absolute path with backslashes and an upper-case drive letter. + Windows: absolute path with backslashes and an upper-case drive letter. A Git Bash rooted + path that is not a drive form (`/tmp/x`) goes through `cygpath -w` (M13); a native rooted + path (`\\x`) keeps its Windows meaning (the cwd's drive). POSIX: `os.path.abspath` semantics (relative to `cwd`). """ if windows is None: @@ -265,6 +316,8 @@ def norm_path(p: str, cwd: str | None = None, windows: bool | None = None) -> st m = _GITBASH_DRIVE.match(p) if m: p = m.group(1).upper() + ":" + (p[2:] or "/") + elif p.startswith("/") and not p.startswith("//"): + p = cygpath_w(p) p = p.replace("/", "\\") drive, rest = ntpath.splitdrive(p) if not (drive and rest.startswith("\\")): @@ -1153,25 +1206,41 @@ def context_tokens(usage: dict | None) -> int: + int(usage.get("cache_read_input_tokens") or 0) +def plugin_id(p: dict) -> str: + """`name@version`; a plugin without a version is identified by its path's last segment (M6), + so an in-place update of a versionless plugin (a new cache dir) shows as drift.""" + ver = p.get("version") + if ver: + return "%s@%s" % (p.get("name"), ver) + path = str(p.get("path") or "").rstrip("/\\") + seg = re.split(r"[/\\]", path)[-1] if path else "" + return "%s@path:%s" % (p.get("name"), seg) if seg else "%s@" % p.get("name") + + +def _names(items) -> list[str]: + return sorted(str(a) if not isinstance(a, dict) else str(a.get("name")) for a in (items or [])) + + def runtime_fingerprint(init: dict | None) -> dict | None: - """The runtime fields of a `system/init` event that must match the base's.""" + """The runtime fields of a `system/init` event that must match the base's (RUNTIME_FIELDS).""" if not init: return None tools = sorted(str(x) for x in (init.get("tools") or [])) - plugins = sorted("%s@%s" % (p.get("name"), p.get("version") or "") - for p in (init.get("plugins") or []) if isinstance(p, dict)) - agents = sorted(str(a) if not isinstance(a, dict) else str(a.get("name")) for a in (init.get("agents") or [])) + plugins = sorted(plugin_id(p) for p in (init.get("plugins") or []) if isinstance(p, dict)) return {"tools": _sha256_bytes(canonical_json(tools).encode("utf-8")), - "permissionMode": init.get("permissionMode"), "plugins": plugins, "agents": agents, - "apiKeySource": init.get("apiKeySource")} + "permissionMode": init.get("permissionMode"), "plugins": plugins, + "agents": _names(init.get("agents")), "apiKeySource": init.get("apiKeySource"), + "skills": _names(init.get("skills"))} def runtime_drift(rt: dict | None, ref: dict | None, launch: dict) -> list[str]: + """Runtime fields of `rt` that differ from the base's `ref`. A field the reference does not + carry (a base recorded before that field was fingerprinted) is not compared.""" if not rt: return [] out = [] if ref: - out += [k for k in sorted(rt) if rt.get(k) != ref.get(k)] + out += [k for k in sorted(rt) if k in ref and rt.get(k) != ref.get(k)] elif rt.get("permissionMode") not in (None, launch.get("permission_mode")): out.append("permissionMode") return out @@ -1321,6 +1390,20 @@ def note_rel(block) -> str | None: return None if block in (None, "") else "notes/%s.md" % _safe_name(block) +def note_written_by(path: str, entry: dict) -> bool: + """M11: the note file exists and was written during this job: its mtime is at or after the + job's creation (less NOTE_MTIME_SLACK_S for coarse file times). A note left by an earlier job + of the same block (a re-run, or the fork a continue resumes) does not count.""" + try: + mtime = os.path.getmtime(path) + except OSError: + return False + start = entry.get("created_ts") + if start is None: + start = _ts(entry.get("created_at")) + return start is None or mtime >= float(start) - NOTE_MTIME_SLACK_S + + def _phase(v): if v is None: return None @@ -1381,9 +1464,9 @@ def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint: """Build the result, the ledger line, and apply the registry updates for one finished job. `hint`: "exited" (decide ok/failed from the stream), or a final status given by the caller.""" cfg = reg["config"] - launch = cfg["launch"] kind = entry["kind"] base = reg["bases"].get(entry.get("base")) if entry.get("base") else None + launch = launch_record(base, cfg) res = ss.result if ss else None u = usage_from_result(res, entry.get("model_usage_before") or {}, entry.get("cost_before") or 0.0) \ if res else None @@ -1423,16 +1506,17 @@ def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint: if denials: flags.append("needs_attention") cache = cache_verdict(ss.first_usage if ss else None, entry.get("base_context")) \ - if kind in ("fork", "continue", "ping") else None + if kind in ("fork", "continue", "ping", "rebase") else None rt = runtime_fingerprint(ss.init) if ss else None drift = list(entry.get("launch_drift") or []) ref = (base or {}).get("runtime_fingerprint") if kind != "train" else None - drift += runtime_drift(rt, ref, launch) + rt_drift = runtime_drift(rt, ref, launch) + drift += rt_drift drift = sorted(set(drift)) fingerprint = "ok" if not drift else "drift:" + ",".join(drift) base_state = entry.get("base_state_at_launch") - if kind in ("fork", "continue") and (base_state == "cold" or (cache not in (None, "hit") - and fingerprint == "ok")): + if kind in ("fork", "continue", "rebase") and (base_state == "cold" or ( + cache not in (None, "hit") and fingerprint == "ok")): flags.append("cold_start") if u and wrote_only_5m(u["tokens"]): flags.append("ttl_downgrade") @@ -1445,7 +1529,7 @@ def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint: flags = _uniq(flags) turns = int(res.get("num_turns") or 0) if res else (ss.turns if ss else 0) note = note_rel(entry.get("block")) if kind != "train" else None - if note and not os.path.isfile(os.path.join(state, note)): + if note and not note_written_by(os.path.join(state, note), entry): note = None session = (ss.session_id if ss and ss.session_id else None) or entry.get("session") tokens = u["tokens"] if u else {k: None for k in TOKEN_KEYS} @@ -1472,7 +1556,10 @@ def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint: # registry updates entry.update({"status": status, "flags": flags, "reason": reason, "reason_detail": detail, "ended_at": ended, "session": session, "turns": turns, "exit_code": exit_code, - "fingerprint": fingerprint, "cache": cache}) + "fingerprint": fingerprint, "cache": cache, + # runtime fields that drifted, None when the call reported no init event: + # `rebase_due: runtime_drift:` reads the base's last call with one. + "runtime_drift": rt_drift if rt else None}) if ss and ss.first_at: entry["first_event_at"] = _iso(ss.first_at) if u: @@ -1488,7 +1575,7 @@ def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint: hit_t = None if ss and kind == "train": hit_t = ss.last_assistant_at - elif ss and kind == "fork": + elif ss and kind in ("fork", "rebase"): # a rebase's first request reads the base prefix hit_t = ss.first_at elif ping_refreshed: hit_t = ss.first_at @@ -1509,9 +1596,45 @@ def _conclude(state: str, reg: dict, entry: dict, ss: StreamState | None, hint: "runtime_fingerprint": rt, "trained_at": ended}) elif base.get("status") == "training": base["status"] = "failed" + if kind == "rebase" and status == "ok": + _rebase_succeeded(reg, entry, base, ss, u, rt, base_commit, ended) return result, ledger +def _rebase_succeeded(reg: dict, entry: dict, old: dict, ss, u, rt, base_commit, ended: str) -> None: + """A rebase ended `ok`: its session becomes base `entry.rebase_as` (active, parent = the old + base, one more rebase in the lineage, HEAD at the end, context from its last assistant + message, the launch record computed at launch: the old base's, with the binary it ran + under); the old base becomes `superseded`. On any other status nothing here runs: the old + base stays active and the new one is never created.""" + name = entry.get("rebase_as") + if not name: + return + existing = reg["bases"].get(name) + if existing and existing.get("status") != "failed": + entry["rebase_conflict"] = "base %s exists (%s); not replaced" % (name, existing.get("status")) + return + launch = entry.get("new_launch") or launch_record(old, reg["config"]) + reg["bases"][name] = { + "name": name, "status": "active", "session": entry.get("session"), "cwd": launch["cwd"], + "launch": launch, "fingerprint": fingerprint_hash(launch), + "overrides": list(old.get("overrides") or []), "binary": launch["binary"], + "created_at": entry.get("created_at"), "train_job": None, "rebase_job": entry["job"], + "parent": old.get("name") or entry.get("base"), + "lineage_rebases": int(old.get("lineage_rebases") or 0) + 1, + "phase": entry.get("phase"), "last_hit": _iso(ss.last_assistant_at) if ss and ss.last_assistant_at else None, + "base_commit": base_commit, "context_tokens": context_tokens(ss.last_usage if ss else None), + "training_cost_usd": u["cost_delta"] if u else None, + "cost_cum_usd": u["total"] if u else None, "model_usage_cum": u["model_usage"] if u else {}, + "runtime_fingerprint": rt, "trained_at": ended, "forks": [], + "ttl_s": infer_ttl(u["tokens"] if u else None, old.get("ttl_s")), "ttl_from": entry["job"], + "ping_in_flight": None, + } + old["status"] = "superseded" + old["superseded_by"] = name + old["superseded_at"] = ended + + def finalize_job(state: str, job: str, *, ss: StreamState | None, hint: str, flags=(), reason=None, detail=None, exit_code=None, wall_s=None, base_commit=None) -> dict | None: """Write `results/.json`, the registry update and the ledger line, exactly once, under the @@ -1759,6 +1882,18 @@ def build_parser() -> _Parser: t.add_argument("--brief", required=True) t.add_argument("--block") t.add_argument("--phase") + # M10 (plan 4.3): train takes launch-field overrides, the same flags as init; a field not + # given comes from the run config. The base records what it was trained with. + g = t.add_argument_group("launch overrides (default: the init config)") + g.add_argument("--model", help="full model id") + g.add_argument("--effort", choices=EFFORTS) + g.add_argument("--permission-mode") + g.add_argument("--tools", help="comma-separated tool names") + g.add_argument("--allowed-tools", nargs="+", action="extend", default=None) + g.add_argument("--disallowed-tools", nargs="+", action="extend", default=None) + g.add_argument("--agent", help="project agent name (/.claude/agents/.md)") + g.add_argument("--cwd", help="working dir of the base and its forks") + g.add_argument("--binary", help="claude binary") def launch_overrides(sp): # Accepted only so that they can be refused with a clear message (plan 4.3). @@ -1809,9 +1944,16 @@ def build_parser() -> _Parser: cmd("tick", "One keep-warm pass: ping every base the rule picks (design 5).") - for name, text in (("rebase", "Rebase a base onto the current state (Phase 2)."), - ("usage", "Aggregate the ledger (Phase 2).")): - cmd(name, text) # arguments arrive with Phase 2; until then extra args are tolerated + r = cmd("rebase", "Fork a base to absorb what changed; its session becomes a new base.") + r.add_argument("--base", required=True) + r.add_argument("--as", dest="as_name", required=True, help="name of the new base") + r.add_argument("--brief", required=True) + r.add_argument("--phase", help="the new base's phase (the phase whose forks it serves)") + r.add_argument("--max-minutes", type=_pos_float, help="wall-time limit (default: init's max_minutes)") + r.add_argument("--max-budget-usd", type=_pos_float) + launch_overrides(r) + + cmd("usage", "Aggregate the ledger (Phase 2).") # arguments arrive with 2.3; extra args tolerated return p @@ -1875,30 +2017,31 @@ def _resolve_cwd(arg: str | None) -> str: return cwd -def build_init_config(args, state: str) -> dict: - """Validate the launch fields and build the config block. Writes nothing.""" - model = args.model +def agent_path(cwd: str, name: str) -> str: + return os.path.join(cwd, ".claude", "agents", name + ".md") + + +def build_launch(state: str, *, model, effort, permission_mode, tools, allowed, disallowed, agent, + cwd: str, binary_path: str) -> tuple[dict, str]: + """Validate launch fields and build a launch record (init's config and train's base record; + design 3.3). `tools`/`allowed`/`disallowed` are lists, `cwd` and `binary_path` normalised. + Returns (launch, model admission `via`). Writes nothing.""" if not MODEL_RE.match(model or ""): raise ForksError("model must be a full id (aliases are refused)", model=model, pattern=MODEL_RE.pattern) - mode = args.permission_mode + mode = permission_mode if mode in REFUSED_PERMISSION_MODES: raise ForksError("permission mode refused", permission_mode=mode, detail="plan mode swaps models; use dontAsk") if mode not in PERMISSION_MODES: raise ForksError("unknown permission mode", permission_mode=mode, allowed=[m for m in PERMISSION_MODES if m not in REFUSED_PERMISSION_MODES]) - tools = _tool_list([args.tools]) + if effort not in EFFORTS: + raise ForksError("unknown effort", effort=effort, allowed=list(EFFORTS)) if not tools: raise ForksError("--tools is empty") - allowed = _tool_list(args.allowed_tools) - if allowed is None: - allowed = list(tools) - disallowed = _tool_list(args.disallowed_tools) - if disallowed is None: - disallowed = list(DEFAULT_DISALLOWED) - - cwd = _resolve_cwd(args.cwd) + if not os.path.isdir(cwd): + raise ForksError("--cwd is not a directory", cwd=cwd) adm = model_admission(model, cwd) if not adm["admitted"]: @@ -1909,39 +2052,58 @@ def build_init_config(args, state: str) -> dict: "add the id (or its family alias) to one of these files; forks never writes " "settings." % (model, ", ".join(files)))) - bpath, how = resolve_binary(args.binary) - if bpath is None: - raise ForksError("no claude binary (no --binary, no $CLAUDE_CODE_EXECPATH, none on PATH)") - binfo = binary_info(bpath) + binfo = binary_info(binary_path) pre = preamble_path() if not os.path.isfile(pre): raise ForksError("preamble.md missing next to forks.py", path=pre) - agent = None - if args.agent: - name = args.agent + agent_rec = None + if agent: + name = agent if not re.match(r"^[A-Za-z0-9_.:-]+$", name) or name in (".", ".."): raise ForksError("bad agent name", agent=name) - apath = os.path.join(cwd, ".claude", "agents", name + ".md") + apath = agent_path(cwd, name) if not os.path.isfile(apath): raise ForksError("agent file not found", agent=name, path=apath) - agent = {"name": name, "path": apath, "sha256": _sha256_file(apath)} + agent_rec = {"name": name, "path": apath, "sha256": _sha256_file(apath)} launch = { "model": model, - "effort": args.effort, + "effort": effort, "permission_mode": mode, - "tools": tools, - "allowed_tools": allowed, - "disallowed_tools": disallowed, + "tools": list(tools), + "allowed_tools": list(allowed), + "disallowed_tools": list(disallowed), "mcp": {"strict": True, "config": mcp_empty_path(state)}, "append_system_prompt": {"path": pre, "sha256": _sha256_file(pre)}, - "agent": agent, + "agent": agent_rec, "cwd": cwd, "binary": binfo, "env": {"pinned": dict(ENV_PINNED_LAUNCH)}, } + return launch, adm["via"] + + +def build_init_config(args, state: str) -> dict: + """Validate the launch fields and build the config block. Writes nothing.""" + tools = _tool_list([args.tools]) + allowed = _tool_list(args.allowed_tools) + if allowed is None: + allowed = list(tools) + disallowed = _tool_list(args.disallowed_tools) + if disallowed is None: + disallowed = list(DEFAULT_DISALLOWED) + cwd = _resolve_cwd(args.cwd) + if not MODEL_RE.match(args.model or ""): # before the binary lookup, as in Phase 1 + raise ForksError("model must be a full id (aliases are refused)", model=args.model, + pattern=MODEL_RE.pattern) + bpath, how = resolve_binary(args.binary) + if bpath is None: + raise ForksError("no claude binary (no --binary, no $CLAUDE_CODE_EXECPATH, none on PATH)") + launch, via = build_launch(state, model=args.model, effort=args.effort, + permission_mode=args.permission_mode, tools=tools, allowed=allowed, + disallowed=disallowed, agent=args.agent, cwd=cwd, binary_path=bpath) cfg = { "launch": launch, "fingerprint": fingerprint_hash(launch), @@ -1954,7 +2116,7 @@ def build_init_config(args, state: str) -> dict: "max_minutes": args.max_minutes, "stall_after_s": args.stall_after_s, "thresholds": {k: getattr(args, k) for k in THRESHOLD_DEFAULTS}, - "model_admission": adm["via"], + "model_admission": via, } return cfg @@ -1982,10 +2144,16 @@ def cmd_init(args) -> tuple[dict, int]: # `model_admission` and `binary_resolved_by` describe how, not what: not a change. changed_fields = [f for f in changed_fields if f not in ("model_admission", "binary_resolved_by")] - if old is not None and changed_fields and reg.get("bases"): - raise ForksError("init values differ and a base exists; train a new run instead", - fields=changed_fields, bases=sorted(reg["bases"]), state=state) + # Block 2.2: values may change after bases exist (Phase 1 refused it). Every base keeps + # launching with its own record; a base that did not override a changed defining field + # reports `retrain_due: defining:`. A base from before per-base records gets the + # config it was trained with as its record first. if old is None or changed_fields: + if old is not None: + for b in reg.get("bases", {}).values(): + if not b.get("launch") and old.get("launch"): + b["launch"] = old["launch"] + b.setdefault("overrides", []) reg["config"] = cfg reg.setdefault("run", {})["init_at"] = _iso() write_registry(state, reg) @@ -1993,7 +2161,7 @@ def cmd_init(args) -> tuple[dict, int]: cfg = old return {"ok": True, "state": state, "registry": registry_path(state), "created": created, "changed": bool(old is not None and changed_fields), "changed_fields": changed_fields, - "config": cfg}, EXIT_OK + "bases": sorted(reg.get("bases", {})), "config": cfg}, EXIT_OK def _not_implemented(block=None, phase=None): @@ -2040,7 +2208,8 @@ def _check_binary(ref: dict, launch: dict) -> tuple[dict, list[str]]: path = launch["binary"]["path"] if not os.path.isfile(path): raise ForksError("claude binary missing (the stored path is gone); the base cannot be " - "launched with another binary", binary=path, reason="binary_missing") + "launched with another binary", binary=path, reason="binary_missing", + retrain_due=True, reasons=["binary_missing"]) info = binary_info(path) drift = [] if all(info.get(k) == (ref or {}).get(k) for k in ("version", "size", "mtime")) else ["binary"] asp = launch["append_system_prompt"] @@ -2107,6 +2276,233 @@ def delta_header(cwd: str, base_commit: str | None) -> str: return "\n\n".join(parts) +# --- rebase and re-train flags (design 5, block 2.2) ------------------------------------ + + +def git_delta_counts(cwd: str, base_commit: str) -> tuple[int | None, int | None]: + """(commits since `base_commit`, insertions + deletions of `git diff --shortstat `); + None where git fails.""" + commits = lines = None + with contextlib.suppress(OSError, subprocess.SubprocessError, ValueError): + r = git(["rev-list", "--count", "%s..HEAD" % base_commit], cwd=cwd) + if r.returncode == 0: + commits = int(r.stdout.decode("utf-8", "replace").strip() or 0) + with contextlib.suppress(OSError, subprocess.SubprocessError, ValueError): + s = git(["diff", "--shortstat", base_commit], cwd=cwd) + if s.returncode == 0: + txt = s.stdout.decode("utf-8", "replace") + ins = re.search(r"(\d+) insertion", txt) + dels = re.search(r"(\d+) deletion", txt) + lines = (int(ins.group(1)) if ins else 0) + (int(dels.group(1)) if dels else 0) + return commits, lines + + +_NOTE_REF = re.compile(r"[^\s`'\"()<>\[\]{},;|*]+\.md\b") + + +def listed_notes(brief: str | None, cwd: str, state: str) -> list[str]: + """The `.md` files a brief names that exist: absolute paths as given, relative ones against + the base's cwd, then against the state dir (`notes/1.1.md`). Each file once.""" + out, seen = [], set() + for ref in _NOTE_REF.findall(brief or ""): + for root in (cwd, state): + try: + p = norm_path(ref, cwd=root) + except ForksError: + break # e.g. a rooted Git Bash path cygpath cannot convert + key = os.path.normcase(p) + if os.path.isfile(p): + if key not in seen: + seen.add(key) + out.append(p) + break + return out + + +def _text_chars(path: str) -> int: + try: + return len(read_bytes_retry(path).decode("utf-8", "replace")) + except OSError: + return 0 + + +def projects_dir() -> str: + return CLAUDE_PROJECTS_DIR or os.path.join(os.path.expanduser("~"), ".claude", "projects") + + +def transcript_path(cwd: str, session: str) -> str: + """P0.7: `~/.claude/projects/ '-'>/.jsonl`.""" + return os.path.join(projects_dir(), re.sub(r"[^A-Za-z0-9]", "-", cwd), session + ".jsonl") + + +def transcript_context(path: str) -> int | None: + """Context tokens of the last assistant line (with `message.usage`) of a transcript, read from + the tail; None when the file or such a line is missing.""" + try: + size = os.path.getsize(path) + except OSError: + return None + n = TRANSCRIPT_TAIL_BYTES + while True: + start = max(0, size - n) + try: + with open(path, "rb") as f: + f.seek(start) + data = f.read() + except OSError: + return None + lines = data.splitlines() + if start > 0: + lines = lines[1:] # the first line of a tail may be cut + for ln in reversed(lines): + if b'"assistant"' not in ln: + continue + try: + obj = json.loads(ln.decode("utf-8", "replace")) + except ValueError: + continue + if isinstance(obj, dict) and obj.get("type") == "assistant": + u = (obj.get("message") or {}).get("usage") + if u: + return context_tokens(u) + if start == 0: + return None + n *= 4 + + +def base_context_now(base: dict, launch: dict) -> int | None: + """The base's context tokens: from its transcript's last assistant line when the file exists + (P0.7), else the context stored at training / rebase.""" + ctx = None + if base.get("session"): + ctx = transcript_context(transcript_path(launch["cwd"], base["session"])) + return ctx if ctx is not None else base.get("context_tokens") + + +def _int_phase(p) -> bool: + return isinstance(p, int) and not isinstance(p, bool) + + +def phase_boundary(reg: dict, name: str, phase) -> bool: + """The fork's `--phase` is greater than the highest phase of any fork from this base, or than + the base's own phase (integers only; a side that is not known does not count).""" + if not _int_phase(phase): + return False + base = reg["bases"][name] + forks_ = [e.get("phase") for e in reg.get("jobs", {}).values() + if e.get("base") == name and e.get("kind") == "fork" and _int_phase(e.get("phase"))] + own = base.get("phase") + return bool(forks_ and phase > max(forks_)) or bool(_int_phase(own) and phase > own) + + +def last_runtime_drift(reg: dict, name: str) -> list[str]: + """Runtime fields (M6 included) the last call on base `name` that reported an init event + found drifted against the base (any kind but train).""" + best, fields = -1, [] + for j, e in reg.get("jobs", {}).items(): + if e.get("base") != name or e.get("kind") == "train" or e.get("runtime_drift") is None: + continue + if _job_num(j) > best: + best, fields = _job_num(j), list(e["runtime_drift"]) + return [f for f in fields if f in RUNTIME_FIELDS] + + +def defining_drift(base: dict, cfg: dict) -> list[str]: + """Fingerprint-defining fields of the base's launch record that differ from what a launch + would use now: for a field the base overrode at `train`, its own value; otherwise the run + config's. The agent file and the preamble are re-hashed from disk.""" + rec = base.get("launch") + if not rec: + return [] + over = set(base.get("overrides") or []) + now = cfg.get("launch") or {} + + def want(f): + return rec.get(f) if f in over else now.get(f) + out = [f for f in ("model", "cwd", "tools", "permission_mode") if want(f) != rec.get(f)] + ra = rec.get("agent") or {} + name = ra.get("name") if "agent" in over else (now.get("agent") or {}).get("name") + sha = None + if name and want("cwd"): + try: + sha = _sha256_file(agent_path(want("cwd"), name)) + except OSError: + sha = None + if (name, sha) != (ra.get("name"), ra.get("sha256")): + out.append("agent") + try: + pre = _sha256_file(preamble_path()) + except OSError: + pre = None + if pre != (rec.get("append_system_prompt") or {}).get("sha256"): + out.append("append_system_prompt") + return [f for f in DEFINING_FIELDS if f in out] + + +def binary_state(base: dict, launch: dict, cache: dict | None = None) -> str: + """"missing" (the stored path is gone), "drift" (version, size or mtime differ from the base's + record) or "ok". `cache` maps a path to its binary_info within one call.""" + path = launch["binary"]["path"] + if not os.path.isfile(path): + return "missing" + cache = {} if cache is None else cache + if path not in cache: + try: + cache[path] = binary_info(path) + except ForksError: + cache[path] = None + info = cache[path] + ref = base.get("binary") or launch["binary"] + if info is None or any(info.get(k) != ref.get(k) for k in ("version", "size", "mtime")): + return "drift" + return "ok" + + +def due_flags(state: str, reg: dict, name: str, *, phase=None, brief: str | None = None, + delta: str | None = None, binary: str | None = None, bcache: dict | None = None) -> dict: + """Design 5's rebase and re-train flags for base `name`, with their reasons (advisory: only + `binary_missing` stops a launch). `phase`/`brief` are the fork's (none in `list`); `delta` the + fork's delta header (computed here when not given); `binary` a `binary_state` value.""" + cfg = reg["config"] + base = reg["bases"][name] + la = launch_record(base, cfg) + th = dict(THRESHOLD_DEFAULTS, **(cfg.get("thresholds") or {})) + rebase, retrain = [], [] + if phase_boundary(reg, name, phase): + rebase.append("phase_boundary") + bc = base.get("base_commit") + if bc: + commits, lines = git_delta_counts(la["cwd"], bc) + if commits is not None and commits > int(th["rebase_commits"]): + rebase.append("commits:%d" % commits) + if lines is not None and lines > int(th["rebase_lines"]): + rebase.append("lines:%d" % lines) + if delta is None: + delta = delta_header(la["cwd"], bc) + chars = len(delta) + sum(_text_chars(p) for p in listed_notes(brief, la["cwd"], state)) + if chars // 4 > int(th["rebase_delta_tokens"]): + rebase.append("delta_tokens:%d" % (chars // 4)) + if binary is None: + binary = binary_state(base, la, bcache) + if binary == "drift": + rebase.append("binary_drift") + rd = last_runtime_drift(reg, name) + if rd: + rebase.append("runtime_drift:" + ",".join(rd)) + dd = defining_drift(base, cfg) + if dd: + retrain.append("defining:" + ",".join(dd)) + ctx = base_context_now(base, la) + if ctx is not None and ctx > int(th["max_base_tokens"]): + retrain.append("max_base_tokens:%d" % ctx) + lr = int(base.get("lineage_rebases") or 0) + if lr >= int(th["max_rebases"]): + retrain.append("rebases:%d" % lr) + if binary == "missing": + retrain.append("binary_missing") + return {"rebase_due": bool(rebase), "retrain_due": bool(retrain), "reasons": rebase + retrain} + + def _next_job_id(reg: dict) -> str: n = 0 for j in reg.get("jobs", {}): @@ -2118,13 +2514,22 @@ def _next_job_id(reg: dict) -> str: def _running_capped(reg: dict) -> list[str]: return sorted(j for j, e in reg.get("jobs", {}).items() - if e.get("kind") in ("fork", "continue") and e.get("status") in RUNNING) + if e.get("kind") in CAPPED_KINDS and e.get("status") in RUNNING) + + +def _pending_rebase_to(reg: dict, name: str) -> str | None: + """The running rebase job that will create base `name`, if any.""" + for j, e in sorted(reg.get("jobs", {}).items()): + if e.get("kind") == "rebase" and e.get("rebase_as") == name and e.get("status") in RUNNING: + return j + return None def _new_entry(job: str, kind: str, **kw) -> dict: e = {"job": job, "kind": kind, "base": None, "block": None, "phase": None, "mode": None, "paths": None, "status": "created", "flags": [], "reason": None, "session": None, - "parent_session": None, "model_requested": None, "created_at": _iso(), "launched_at": None, + "parent_session": None, "model_requested": None, "created_at": _iso(), + "created_ts": _now(), "launched_at": None, "ended_at": None, "supervisor": None, "cwd": None, "argv": None, "brief": "briefs/%s.md" % job, "raw": "raw/%s.jsonl" % job, "max_s": None, "cost_before": 0.0, "model_usage_before": {}, "base_context": None, @@ -2154,15 +2559,40 @@ def _start_job(state: str, run: str, job: str, prompt: str, launch: dict) -> dic return sp +def train_launch(args, state: str, cfg: dict) -> tuple[dict, list[str]]: + """M10 (plan 4.3): the base's launch record = the run config's launch fields with the + overrides given on `train` (the same flags as `init`), validated as `init` validates them. + Returns (launch, sorted overridden field names).""" + la = cfg["launch"] + over = sorted(n for n in LAUNCH_FIELDS if getattr(args, n, None) is not None) + tools = _tool_list([args.tools]) if args.tools is not None else la["tools"] + allowed = _tool_list(args.allowed_tools) if args.allowed_tools is not None else la["allowed_tools"] + disallowed = (_tool_list(args.disallowed_tools) if args.disallowed_tools is not None + else la["disallowed_tools"]) + agent = args.agent if args.agent is not None else (la.get("agent") or {}).get("name") + cwd = norm_path(args.cwd) if args.cwd is not None else la["cwd"] + bpath = norm_path(args.binary) if args.binary is not None else la["binary"]["path"] + if not os.path.isfile(bpath): + raise ForksError("claude binary missing (the stored path is gone); train with --binary or " + "re-run init with another binary", binary=bpath, reason="binary_missing") + launch, _ = build_launch( + state, model=args.model if args.model is not None else la["model"], + effort=args.effort if args.effort is not None else la["effort"], + permission_mode=args.permission_mode if args.permission_mode is not None else la["permission_mode"], + tools=tools, allowed=allowed, disallowed=disallowed, agent=agent, cwd=cwd, binary_path=bpath) + return launch, over + + def cmd_train(args) -> tuple[dict, int]: state, reg = _load_run(args) name = args.base if not re.fullmatch(r"[A-Za-z0-9_.-]+", name or ""): raise ForksError("bad base name", base=name) cfg = reg["config"] - la = cfg["launch"] + la, overrides = train_launch(args, state, cfg) + binfo = la["binary"] + drift: list[str] = [] # the record was built from the files just now brief_path, brief = _read_brief(args.brief) - binfo, drift = _check_binary(la["binary"], la) session = str(uuid.uuid4()) argv = claude_argv(la) + ["--session-id", session] head = _git_head(la["cwd"]) @@ -2172,11 +2602,16 @@ def cmd_train(args) -> tuple[dict, int]: old = reg["bases"].get(name) if old and old.get("status") != "failed": raise ForksError("base exists", base=name, status=old.get("status")) + pending = _pending_rebase_to(reg, name) + if pending: + raise ForksError("a running rebase creates this base", base=name, job=pending) job = _next_job_id(reg) reg["bases"][name] = {"name": name, "status": "training", "session": session, "cwd": la["cwd"], - "fingerprint": cfg["fingerprint"], "binary": binfo, "created_at": _iso(), - "train_job": job, "phase": _phase(args.phase), "last_hit": None, - "base_commit": None, "context_tokens": None, "forks": []} + "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": []} reg["jobs"][job] = _new_entry( job, "train", base=name, block=args.block, phase=_phase(args.phase), session=session, model_requested=la["model"], cwd=la["cwd"], argv=argv, max_s=float(cfg["max_minutes"]) * 60, @@ -2196,10 +2631,10 @@ def cmd_fork(args) -> tuple[dict, int]: raise ForksError("--mode parallel needs --paths (the files this fork may touch)", mode="parallel") state, reg = _load_run(args) cfg = reg["config"] - la = cfg["launch"] base = reg["bases"].get(args.base) if base is None: raise ForksError("unknown base", base=args.base, bases=sorted(reg["bases"])) + la = launch_record(base, cfg) # M10: every launch on a base uses the base's record if base.get("status") != "active" and not args.force: raise ForksError("base is not active (use --force to fork it anyway)", base=args.base, status=base.get("status")) @@ -2208,6 +2643,8 @@ def cmd_fork(args) -> tuple[dict, int]: brief_path, brief = _read_brief(args.brief) binfo, drift = _check_binary(base.get("binary") or la["binary"], la) delta = delta_header(la["cwd"], base.get("base_commit")) + due = due_flags(state, reg, args.base, phase=_phase(args.phase), brief=brief, delta=delta, + binary="drift" if "binary" in drift else "ok") head = _git_head(la["cwd"]) session = str(uuid.uuid4()) argv = claude_argv(la) + ["--resume", base["session"], "--fork-session", "--session-id", session] @@ -2256,15 +2693,14 @@ def cmd_fork(args) -> tuple[dict, int]: "logs_dir": os.path.join(state, "logs"), "brief": brief}) _start_job(state, norm_path(args.run), job, prompt, la) return {"job": job, "base": args.base, "base_state": warmth["base_state"], - "warm_until": warmth["warm_until"], "rebase_due": False, "retrain_due": False, - "reasons": []}, EXIT_OK + "warm_until": warmth["warm_until"], "rebase_due": due["rebase_due"], + "retrain_due": due["retrain_due"], "reasons": due["reasons"]}, EXIT_OK def cmd_continue(args) -> tuple[dict, int]: _refuse_overrides(args) state, reg = _load_run(args) cfg = reg["config"] - la = cfg["launch"] fj = reg["jobs"].get(args.job) if fj is None: raise ForksError("unknown job", job=args.job) @@ -2279,6 +2715,7 @@ def cmd_continue(args) -> tuple[dict, int]: if not fj.get("session"): raise ForksError("the fork has no session", job=args.job) base = reg["bases"].get(fj.get("base")) or {} + la = launch_record(base, cfg) # M10: the fork's base's record brief_path, brief = _read_brief(args.brief) binfo, drift = _check_binary(base.get("binary") or la["binary"], la) head = _git_head(la["cwd"]) @@ -2313,6 +2750,77 @@ def cmd_continue(args) -> tuple[dict, int]: return {"job": job}, EXIT_OK +def cmd_rebase(args) -> tuple[dict, int]: + """`rebase --base B --as NEW --brief FILE [--phase N]`: a fork of B that absorbs what changed + (`templates/rebase.md`). On `ok` its session becomes base NEW and B is superseded (finaliser, + `_rebase_succeeded`); otherwise B stays active and NEW is never created. Counts toward the cap; + N3-refused (dispatch).""" + _refuse_overrides(args) + state, reg = _load_run(args) + cfg = reg["config"] + new = args.as_name + if not re.fullmatch(r"[A-Za-z0-9_.-]+", new or ""): + raise ForksError("bad base name", base=new) + base = reg["bases"].get(args.base) + if base is None: + raise ForksError("unknown base", base=args.base, bases=sorted(reg["bases"])) + if new == args.base: + raise ForksError("--as must name a new base", base=new) + if base.get("status") != "active" or not base.get("session"): + raise ForksError("base is not active", base=args.base, status=base.get("status")) + la = launch_record(base, cfg) + brief_path, brief = _read_brief(args.brief) + binfo, drift = _check_binary(base.get("binary") or la["binary"], la) + # The new base's record is the old one's; a binary drift means the rebase runs under the + # current binary, which the new base then records (design 5). + new_launch = dict(la, binary=binfo) + delta = delta_header(la["cwd"], base.get("base_commit")) + head = _git_head(la["cwd"]) + session = str(uuid.uuid4()) + argv = claude_argv(la) + ["--resume", base.get("session") or "", "--fork-session", "--session-id", session] + if args.max_budget_usd: + argv += ["--max-budget-usd", "%g" % args.max_budget_usd] + reap_lost(state) + with RegistryLock(state): + reg = read_registry(state) + stamp_orchestrator(reg) + base = reg["bases"][args.base] + if base.get("status") != "active" or not base.get("session"): + raise ForksError("base is not active", base=args.base, status=base.get("status")) + old = reg["bases"].get(new) + if old and old.get("status") != "failed": + raise ForksError("base exists", base=new, status=old.get("status")) + pending = _pending_rebase_to(reg, new) + if pending: + raise ForksError("a running rebase creates this base", base=new, job=pending) + busy = sorted(j for j, e in reg["jobs"].items() if e.get("kind") == "rebase" + and e.get("base") == args.base and e.get("status") in RUNNING) + if busy: + raise ForksError("the base is being rebased", base=args.base, running=busy) + running = _running_capped(reg) + if len(running) >= int(cfg["parallel_cap"]): + raise ForksError("parallel cap reached", parallel_cap=cfg["parallel_cap"], running=running) + job = _next_job_id(reg) + warmth = base_warmth(base, cfg) + reg["jobs"][job] = _new_entry( + job, "rebase", base=args.base, phase=_phase(args.phase), session=session, + parent_session=base["session"], model_requested=la["model"], cwd=la["cwd"], argv=argv, + max_s=float(args.max_minutes if args.max_minutes else cfg["max_minutes"]) * 60, + cost_before=float(base.get("cost_cum_usd") or 0.0), + model_usage_before=base.get("model_usage_cum") or {}, + base_context=base.get("context_tokens"), base_state_at_launch=warmth["base_state"], + warm_until_at_launch=warmth["warm_until"], launch_drift=drift, + binary_version=binfo["version"], head_at_launch=head, brief_source=brief_path, + rebase_as=new, new_launch=new_launch) + write_registry(state, reg) + prompt = render_template("rebase.md", { + "base": args.base, "new_base": new, "run": norm_path(args.run), "state_dir": state, + "notes_dir": os.path.join(state, "notes"), "logs_dir": os.path.join(state, "logs"), + "phase": "n/a" if args.phase is None else str(args.phase), "delta": delta, "brief": brief}) + _start_job(state, norm_path(args.run), job, prompt, la) + return {"job": job, "base": new, "parent": args.base, "session_id": session}, EXIT_OK + + # --- keep warm: the ping rule and the tick (design 5, C3d) ---------------------------- @@ -2322,8 +2830,8 @@ def run_for_state(state: str) -> str: def launch_record(base: dict | None, cfg: dict) -> dict: - """The launch fields a launch on `base` uses: the base's own record when it has one, else the - run config (every base has the init config's fields until bases carry their own record).""" + """The launch fields a launch on `base` uses (M10): the base's own record (written at `train` + or `rebase`); the run config only for a base from before per-base records.""" return (base or {}).get("launch") or cfg["launch"] @@ -2609,7 +3117,8 @@ class Supervisor: stall_s = float(self.cfg.get("stall_after_s") or CONFIG_DEFAULTS["stall_after_s"]) orphan_s = float(self.cfg.get("orphan_after") or CONFIG_DEFAULTS["orphan_after"]) hint, reason, detail, flags_extra = "exited", None, None, [] - hit_written = e["kind"] != "fork" # only a fork's first request refreshes (I3) + # only a fork's (or a rebase's) first request reads the base prefix and refreshes it (I3) + hit_written = e["kind"] not in ("fork", "rebase") clock = SupClock(t0) last_orphan_check = 0.0 last_messages = 0 @@ -2674,7 +3183,7 @@ class Supervisor: err.close() wall = _now() - t0 base_commit = None - if e["kind"] == "train": + if e["kind"] in ("train", "rebase"): # the new base's HEAD at the end of the call base_commit = _git_head(e["cwd"]) or e.get("head_at_launch") res = finalize_job(self.state, self.job, ss=self.ss, hint=hint, flags=self.flags + flags_extra, reason=reason, detail=detail, exit_code=exit_code, wall_s=wall, @@ -2795,7 +3304,8 @@ LIST_JOB_KEYS = ("job", "kind", "base", "block", "phase", "mode", "paths", "stat "ended_at", "fingerprint", "cache", "delivered_at", "settled_at") LIST_BASE_KEYS = ("name", "status", "session", "cwd", "base_commit", "context_tokens", "training_cost_usd", "cost_cum_usd", "last_hit", "train_job", "forks", "phase", - "created_at", "trained_at", "ttl_from", "ping_in_flight") + "created_at", "trained_at", "ttl_from", "ping_in_flight", "parent", + "lineage_rebases", "rebase_job", "superseded_by", "overrides", "fingerprint") def cmd_list(args) -> tuple[dict, int]: @@ -2806,11 +3316,17 @@ def cmd_list(args) -> tuple[dict, int]: now = _now() cfg = reg["config"] bases = [] + bcache: dict = {} for name, b in sorted(reg["bases"].items()): row = {k: b.get(k) for k in LIST_BASE_KEYS} row["name"] = name row.update(base_warmth(b, cfg, now)) - row.update({"rebase_due": False, "retrain_due": False, "reasons": []}) + # The flags are computed for active bases only (no fork's --phase or brief here, so no + # phase_boundary and delta_tokens from the delta header alone). + if b.get("status") == "active": + row.update(due_flags(state, reg, name, bcache=bcache)) + else: + row.update({"rebase_due": False, "retrain_due": False, "reasons": []}) bases.append(row) jobs, running = [], [] for j, e in sorted(reg["jobs"].items()): @@ -2866,7 +3382,7 @@ HANDLERS = { "expire": cmd_expire, "kill": cmd_kill, "_supervise": cmd_supervise, - "rebase": _not_implemented(phase=2), + "rebase": cmd_rebase, "usage": _not_implemented(phase=2), "tick": cmd_tick, } diff --git a/plugins/spire/skills/forks/templates/rebase.md b/plugins/spire/skills/forks/templates/rebase.md new file mode 100644 index 0000000..daa2345 --- /dev/null +++ b/plugins/spire/skills/forks/templates/rebase.md @@ -0,0 +1,17 @@ +# Rebase (spire:forks) + +You are a fork of base `${base}` (run log `${run}`). When you finish, this session becomes the new base `${new_base}` (phase ${phase}): every later fork of this build starts from what you know at the end of it. Your job is to absorb what changed since `${base}` was trained or last rebased. + +Your system prompt was snapshotted when the base was trained: the git status, the CLAUDE.md files and the environment it shows are **stale**. The delta header below lists what changed in the repository since then. Trust it and the files on disk over the snapshot and over what you remember. + +- Read the handoff notes and the diffs the brief lists (`git --no-optional-locks diff`, `git --no-optional-locks log -p`), and re-read the files they touch. Handoff notes live under `${notes_dir}`. +- Do not change code and do not commit. If the brief names a trunk-notes path, update that file; it is the only file you write. +- Finish with one line `READY`, then a short summary of what changed and anything the brief asked for that you could not do. + +## Delta since the base commit + +${delta} + +## Brief + +${brief} diff --git a/plugins/spire/skills/forks/tests/test_forks.py b/plugins/spire/skills/forks/tests/test_forks.py index a3e91af..0637001 100644 --- a/plugins/spire/skills/forks/tests/test_forks.py +++ b/plugins/spire/skills/forks/tests/test_forks.py @@ -102,14 +102,79 @@ class PathTests(Base): "/c": "C:\\", "/c/": "C:\\", "../other/./z.md": "G:\\dev\\other\\z.md", - "\\rooted\\a": "G:\\rooted\\a", - "/tmp/x": "G:\\tmp\\x", # not a drive form: rooted on the cwd's drive + "\\rooted\\a": "G:\\rooted\\a", # native rooted path: the cwd's drive (Windows meaning) " G:/x/y.md ": "G:\\x\\y.md", } for src, want in cases.items(): with self.subTest(src=src): self.assertEqual(forks.norm_path(src, cwd=cwd, windows=True), want) + def fake_cygpath(self, code=0): + """A fake `cygpath` (python behind a wrapper) that maps `/a/b` to `C:\\Fake\\a\\b` and logs + its arguments.""" + d = os.path.join(self.tmp, "cyg") + os.makedirs(d, exist_ok=True) + py = os.path.join(d, "cygpath.py") + log = os.path.join(d, "calls.txt") + with open(py, "w", newline="\n") as f: + f.write("import sys\nopen(%r, 'a').write(' '.join(sys.argv[1:]) + '\\n')\n" + "print('C:\\\\Fake' + sys.argv[-1].replace('/', '\\\\'))\nsys.exit(%d)\n" % (log, code)) + if IS_WINDOWS: + exe = os.path.join(d, "cygpath.cmd") + with open(exe, "w", newline="\r\n") as f: + f.write('@echo off\n"%s" -I "%s" %%*\n' % (stub_interpreter(), py)) + else: + exe = os.path.join(d, "cygpath") + with open(exe, "w", newline="\n") as f: + f.write('#!/bin/sh\nexec "%s" -I "%s" "$@"\n' % (stub_interpreter(), py)) + os.chmod(exe, 0o755) + return exe, log + + def test_gitbash_rooted_without_cygpath(self): + """M13: a Git Bash rooted path that is not a drive form is refused (exit 2) when cygpath is + not on PATH; it is never mapped onto the cwd's drive. Drive forms need no cygpath.""" + cwd = "G:\\dev\\proj" + with mock.patch.object(forks, "find_cygpath", lambda: None): + for src in ("/tmp/x", "/usr/share/doc", "/tmp/x-run.md"): + with self.subTest(src=src): + with self.assertRaises(forks.ForksError) as cm: + forks.norm_path(src, cwd=cwd, windows=True) + self.assertEqual(cm.exception.code, 2) + self.assertIn("cygpath", cm.exception.payload["error"]) + self.assertEqual(cm.exception.payload["path"], src) + self.assertEqual(forks.norm_path("/g/x/y.md", cwd=cwd, windows=True), "G:\\x\\y.md") + self.assertEqual(forks.norm_path("/c", cwd=cwd, windows=True), "C:\\") + self.assertEqual(forks.norm_path("\\rooted\\a", cwd=cwd, windows=True), "G:\\rooted\\a") + if IS_WINDOWS: # end to end: --run /tmp/... is refused, nothing written on the cwd's drive + drive = os.path.splitdrive(os.getcwd())[0] + code, out, raw = call(["init", "--run", "/tmp/forks-m13-x-run.md", "--model", + "claude-opus-5-5", "--cwd", self.tmp]) + self.assertEqual(code, 2, out) + self.assert_one_json_line(raw) + self.assertIn("cygpath", out["error"]) + self.assertFalse(os.path.exists(drive + "\\tmp\\forks-m13-x-run-forks")) + + def test_gitbash_rooted_with_cygpath(self): + """M13: with cygpath on PATH the path is converted by `cygpath -w`; a failing cygpath + refuses.""" + exe, log = self.fake_cygpath() + with mock.patch.object(forks, "find_cygpath", lambda: exe): + self.assertEqual(forks.norm_path("/tmp/x", cwd="G:\\dev\\proj", windows=True), + "C:\\Fake\\tmp\\x") + self.assertEqual(forks.norm_path("/g/x", cwd="G:\\dev\\proj", windows=True), "G:\\x") + with open(log) as f: + self.assertEqual(f.read().splitlines(), ["-w /tmp/x"]) # drive forms skip cygpath + bad, _ = self.fake_cygpath(code=1) + with mock.patch.object(forks, "find_cygpath", lambda: bad): + with self.assertRaises(forks.ForksError) as cm: + forks.norm_path("/tmp/x", cwd="G:\\dev\\proj", windows=True) + self.assertEqual(cm.exception.payload["error"], "cygpath failed") + real = shutil.which("cygpath") + if IS_WINDOWS and real: # the real one (Git Bash): never the cwd's drive + \tmp + got = forks.norm_path("/tmp/x", cwd="G:\\dev\\proj", windows=True) + self.assertNotEqual(got, "G:\\tmp\\x") + self.assertRegex(got, r"^[A-Z]:\\.+\\x$") + def test_windows_cwd_in_gitbash_form(self): self.assertEqual(forks.norm_path("a.md", cwd="/g/dev/proj", windows=True), "G:\\dev\\proj\\a.md") @@ -632,22 +697,56 @@ class InitTests(Base): self.assertEqual(cfg["orphan_after"], 1500) self.assertNotEqual(cfg["fingerprint"], first["config"]["fingerprint"]) - def test_changed_values_after_base_refused(self): + def test_changed_values_after_base_allowed(self): + """Block 2.2 replaces Phase 1's refusal: `init` may change values after a base exists (the + input path of `retrain_due: defining:`). The base keeps its own launch record; a base from + before per-base records gets the config it was trained with as its record.""" self.assertEqual(self.init()[0], 0) self.add_fake_base() - before = self.reg_bytes() code, out, raw = self.init("--effort", "medium") - self.assertEqual(code, 2) + self.assertEqual(code, 0, out) self.assert_one_json_line(raw) - self.assertIn("base exists", out["error"]) + self.assertTrue(out["changed"]) + self.assertIn("launch.effort", out["changed_fields"]) self.assertEqual(out["bases"], ["B1"]) - self.assertIn("launch.effort", out["fields"]) - self.assertEqual(self.reg_bytes(), before) + reg = forks.read_registry(self.state) + self.assertEqual(reg["config"]["launch"]["effort"], "medium") + self.assertEqual(reg["bases"]["B1"]["launch"]["effort"], "high") + self.assertEqual(reg["bases"]["B1"]["overrides"], []) self.assertFalse(os.path.exists(forks.lock_path(self.state))) - # identical values stay fine with a base - code, out, _ = self.init() + # identical values stay unchanged with a base + before = self.reg_bytes() + code, out, _ = self.init("--effort", "medium") self.assertEqual(code, 0, out) self.assertFalse(out["changed"]) + self.assertEqual(self.reg_bytes(), before) + + def test_fingerprint_hash_stable_across_runs(self): + """M7: the hash covers the mcp config's content, not its per-run path: two runs with the + same launch fields hash the same; the path stays in the config, outside the hash.""" + code, a, _ = self.init() + self.assertEqual(code, 0, a) + other = os.path.join(self.tmp, "elsewhere", "2026-10-16-09-u-run.md") + os.makedirs(os.path.dirname(other)) + code, b, _ = call(["init", "--run", other, "--model", "claude-opus-5-5[1m]", "--cwd", self.cwd, + "--binary", self.stub]) + self.assertEqual(code, 0, b) + la, lb = a["config"]["launch"], b["config"]["launch"] + self.assertNotEqual(la["mcp"]["config"], lb["mcp"]["config"]) + self.assertEqual(la["mcp"]["config"], forks.mcp_empty_path(self.state)) + self.assertEqual(a["config"]["fingerprint"], b["config"]["fingerprint"]) + self.assertEqual(forks.fingerprint_view(la)["mcp"], {"strict": True, "servers": {}}) + self.assertNotIn("config", forks.fingerprint_view(la)["mcp"]) + # a second process computes the same hash (no per-process input) + r = subprocess.run([sys.executable, "-I", "-c", + "import importlib.util,json,sys;" + "s=importlib.util.spec_from_file_location('f',sys.argv[1]);m=importlib.util.module_from_spec(s);" + "s.loader.exec_module(m);print(m.fingerprint_hash(json.loads(sys.argv[2])))", + FORKS_PY, json.dumps(lb)], capture_output=True, timeout=60) + self.assertEqual(r.stdout.decode().strip(), a["config"]["fingerprint"], r.stderr) + # the content does count + self.assertNotEqual(forks.fingerprint_hash(dict(la, mcp=dict(la["mcp"], strict=False))), + a["config"]["fingerprint"]) # --- refusals --- def assert_refused_no_state(self, *extra, model="claude-opus-5-5[1m]"): @@ -759,7 +858,7 @@ class InitTests(Base): class CliTests(Base): def test_phase2_not_implemented(self): run_log = os.path.join(self.tmp, "r-run.md") - for argv in (("rebase",), ("usage", "--phase", "1")): # `tick` is implemented (2.1) + for argv in (("usage", "--phase", "1"),): # `tick` (2.1) and `rebase` (2.2) are implemented with self.subTest(cmd=argv[0]): code, out, raw = call([argv[0], "--run", run_log, *argv[1:]]) self.assertEqual(code, 2) @@ -777,7 +876,8 @@ class CliTests(Base): ("fork", "--base", "B1", "--block", "1.1", "--brief", brief), ("continue", "--job", "j0001", "--brief", brief), ("wait", "--all"), ("list",), ("expire", "--all"), ("kill", "--job", "j0001"), - ("_supervise", "--job", "j0001"), ("tick",)) + ("_supervise", "--job", "j0001"), ("tick",), + ("rebase", "--base", "B1", "--as", "B2", "--brief", brief)) for argv in cases: with self.subTest(cmd=argv[0]): code, out, raw = call([argv[0], "--run", run_log, *argv[1:]]) @@ -1015,7 +1115,10 @@ def main(): "permissionMode": opt(a, "--permission-mode"), "tools": tools, "mcp_servers": [], "apiKeySource": d.get("apikey", "none"), "claude_code_version": VERSION, "plugins": [{"name": "spire", "path": "x", "source": "spire@reliquary", - "version": d.get("plugin", "0.2.2")}], + "version": d.get("plugin", "0.2.2")}, + {"name": "skill-creator", "source": "skill-creator@official", + "path": "C:\\plugins\\cache\\skill-creator\\" + d.get("nover", "abc123")}], + "skills": sorted(["implementation", "workpad"] + ([d["skill"]] if "skill" in d else [])), "agents": ["claude", "general-purpose"]}) used = model if "restricted" in d: @@ -1200,6 +1303,9 @@ class LaunchBase(Base): p = mock.patch.object(forks, "USER_SETTINGS_PATH", os.path.join(self.tmp, "home", "settings.json")) p.start() self.addCleanup(p.stop) + p = mock.patch.object(forks, "CLAUDE_PROJECTS_DIR", os.path.join(self.tmp, "home", "projects")) + p.start() + self.addCleanup(p.stop) self.cwd = os.path.join(self.tmp, "proj") os.makedirs(self.cwd) self.git("init", "-q") @@ -1446,6 +1552,46 @@ class RuleTests(Base): self.assertEqual(forks.runtime_drift(other, rt, {}), ["apiKeySource", "plugins"]) self.assertEqual(forks.runtime_drift(rt, None, {"permission_mode": "default"}), ["permissionMode"]) + def test_runtime_fingerprint_skills_and_versionless_plugins(self): + """M6: sorted `skills` are compared; a plugin without a version is identified by the last + segment of its path. Both drift.""" + init = {"tools": ["Read"], "permissionMode": "dontAsk", "apiKeySource": "none", + "agents": ["claude"], "skills": ["workpad", "implementation"], + "plugins": [{"name": "spire", "version": "0.2.2", "path": "C:\\c\\spire\\0.2.2"}, + {"name": "skill-creator", + "path": "C:\\Users\\u\\.claude\\plugins\\cache\\skill-creator\\abc123\\"}, + {"name": "builtin"}]} + rt = forks.runtime_fingerprint(init) + self.assertEqual(rt["plugins"], ["builtin@", "skill-creator@path:abc123", "spire@0.2.2"]) + self.assertEqual(rt["skills"], ["implementation", "workpad"]) + self.assertEqual(sorted(rt), sorted(forks.RUNTIME_FIELDS)) + upd = json.loads(json.dumps(init)) + upd["plugins"][1]["path"] = "/home/u/.claude/plugins/cache/skill-creator/def456" + self.assertEqual(forks.runtime_drift(forks.runtime_fingerprint(upd), rt, {}), ["plugins"]) + more = dict(init, skills=["workpad", "implementation", "new-skill"]) + self.assertEqual(forks.runtime_drift(forks.runtime_fingerprint(more), rt, {}), ["skills"]) + same = dict(init, skills=["implementation", "workpad"]) # order does not count + self.assertEqual(forks.runtime_drift(forks.runtime_fingerprint(same), rt, {}), []) + old_ref = {k: v for k, v in rt.items() if k != "skills"} # a base from before M6 + self.assertEqual(forks.runtime_drift(forks.runtime_fingerprint(more), old_ref, {}), []) + + def test_note_written_by(self): + """M11: a note counts only when its mtime is at or after the job's creation (less the + file-time slack).""" + p = os.path.join(self.tmp, "n.md") + start = time.time() - 100 + entry = {"created_ts": start, "created_at": forks._iso(start)} + self.assertFalse(forks.note_written_by(p, entry)) # no file + with open(p, "w") as f: + f.write("note\n") + for delta, want in ((-forks.NOTE_MTIME_SLACK_S - 1, False), (-forks.NOTE_MTIME_SLACK_S + 0.5, True), + (0, True), (50, True)): + with self.subTest(delta=delta): + os.utime(p, (start + delta, start + delta)) + self.assertEqual(forks.note_written_by(p, entry), want) + os.utime(p, (start - 60, start - 60)) + self.assertFalse(forks.note_written_by(p, {"created_at": forks._iso(start)})) # ISO fallback + class WarmthTests(Base): """Design 5 under an injected clock (block 2.1): warmth boundaries, TTL inference, the ping rule.""" @@ -1798,9 +1944,9 @@ class LaunchTests(LaunchBase): # --- fork, after a commit on the trunk --- self.commit_file("src.py", "x = 1\n", "add src after training") os.makedirs(os.path.join(self.state, "notes"), exist_ok=True) - with open(os.path.join(self.state, "notes", "1.1.md"), "w") as f: + f_out = self.fork("turns=2", "sleep=1", extra=("--phase", "1", "--paths", "src.py", "docs/a.md")) + with open(os.path.join(self.state, "notes", "1.1.md"), "w") as f: # M11: written during the job f.write("note\n") - f_out = self.fork("turns=2", extra=("--phase", "1", "--paths", "src.py", "docs/a.md")) self.assertEqual(set(f_out), {"job", "base", "base_state", "warm_until", "rebase_due", "retrain_due", "reasons"}) self.assertEqual(f_out["base_state"], "warm") @@ -2213,7 +2359,8 @@ class LaunchTests(LaunchBase): for name in forks.LAUNCH_FIELDS: flag = "--" + name.replace("_", "-") for argv in (["fork", "--base", "B1", "--block", "1", "--brief", "b.md"], - ["continue", "--job", "j0001", "--brief", "b.md"]): + ["continue", "--job", "j0001", "--brief", "b.md"], + ["rebase", "--base", "B1", "--as", "B2", "--brief", "b.md"]): with self.subTest(field=name, cmd=argv[0]): out = self.cmd(*argv, flag, values[name], code=2) self.assertEqual(out["fields"], [name]) @@ -2499,6 +2646,358 @@ class LaunchTests(LaunchBase): self.assertIn("messages", self.runtime(job)) # M16: the job file's counter self.assertNotIn("turns", self.runtime(job)) + # --- block 2.2 ----------------------------------------------------------------------- + + REBASE_REASONS = ("phase_boundary", "commits", "lines", "delta_tokens", "binary_drift", "runtime_drift") + + def dead_spawn(self) -> dict: + p = subprocess.Popen([sys.executable, "-c", "pass"]) + dead = forks.pid_record(p.pid) + p.wait() + return {"ok": True, "method": "test", "pid": dead["pid"], "ctime": dead["ctime"]} + + def fake_fork(self, base, *extra, brief=None, code=0): + """A fork whose supervisor is never started (spawn stubbed with a dead pid), finalised at once + as killed: only its launch return is looked at.""" + fake = self.dead_spawn() + with mock.patch.object(forks, "spawn_supervisor", lambda *a: fake): + out = self.cmd("fork", "--base", base, "--block", "d", "--brief", brief or self.brief(), + *extra, code=code) + if code == 0: + forks.finalize_job(self.state, out["job"], ss=None, hint="killed", reason="kill") + return out + + def clone_base(self, name, src="B1", **changes): + """A copy of base `src` under a new name, at the current HEAD unless `base_commit` is given.""" + head = self.git("rev-parse", "HEAD") + + def f(reg): + b = json.loads(json.dumps(reg["bases"][src])) + b.update({"name": name, "forks": [], "base_commit": head}) + b.update(changes) + reg["bases"][name] = b + self.set_registry(f) + + def assert_due(self, out, *reasons): + self.assertEqual(out["reasons"], list(reasons), out) + rebase = [r for r in reasons if r.split(":")[0] in self.REBASE_REASONS] + self.assertEqual(out["rebase_due"], bool(rebase), out) + self.assertEqual(out["retrain_due"], len(rebase) < len(reasons), out) + + def test_due_reasons_each_alone(self): + """Design 5 / digest 2.2: every rebase_due and retrain_due reason triggered on its own; the + fork's launch return lists exactly it (and `list` the base-level ones).""" + self.setup_run(cap=4) + self.train("B1") + self.finish("j0001") + c0 = self.git("rev-parse", "HEAD") + self.assert_due(self.fake_fork("B1")) # nothing due + # phase_boundary: above the highest phase of the base's forks, or of the base itself + self.clone_base("Bp") + self.assert_due(self.fake_fork("Bp", "--phase", "1")) # nothing known yet + self.assert_due(self.fake_fork("Bp", "--phase", "1")) + self.assert_due(self.fake_fork("Bp", "--phase", "2"), "phase_boundary") + self.assert_due(self.fake_fork("Bp")) # no --phase + self.clone_base("Bq", phase=3) + self.assert_due(self.fake_fork("Bq", "--phase", "3")) + self.assert_due(self.fake_fork("Bq", "--phase", "4"), "phase_boundary") + self.assert_due(self.fake_fork("Bq", "--phase", "4"), "phase_boundary") # still above the base's 3 + self.assert_due(self.fake_fork("Bq", "--phase", "2")) + # commits: 7 > rebase_commits 6 (seven one-line files) + for i in range(7): + self.commit_file("c%d.txt" % i, "x\n", "commit %d" % i) + self.clone_base("Bc", base_commit=c0) + self.assert_due(self.fake_fork("Bc"), "commits:7") + # lines: 1600 > rebase_lines 1500, in one commit + c7 = self.git("rev-parse", "HEAD") + self.commit_file("big.txt", "line\n" * 1600, "big change") + self.clone_base("Bl", base_commit=c7) + self.assert_due(self.fake_fork("Bl"), "lines:1600") + # every clone from here on starts at HEAD + # delta_tokens: chars / 4 of the delta header plus the notes the brief lists > 12000 + with open(os.path.join(self.state, "notes", "big-note.md"), "w") as f: + f.write("n" * 60000) + self.clone_base("Bd") + self.assert_due(self.fake_fork("Bd")) # brief lists no note + out = self.fake_fork("Bd", brief=self.brief(text="Read notes/big-note.md first.")) + self.assertEqual(len(out["reasons"]), 1, out) + self.assertRegex(out["reasons"][0], r"^delta_tokens:\d+$") + self.assertGreater(int(out["reasons"][0].split(":")[1]), 15000) + self.assertEqual((out["rebase_due"], out["retrain_due"]), (True, False)) + # binary_drift: a clone with its own copy of the binary, changed after it was recorded + ext = os.path.splitext(self.stub)[1] + b2 = os.path.join(os.path.dirname(self.stub), "claude2" + ext) + shutil.copy2(self.stub, b2) + info = forks.binary_info(forks.norm_path(b2)) + b1_launch = self.reg()["bases"]["B1"]["launch"] + self.clone_base("Bb", launch=dict(b1_launch, binary=info), binary=info) + self.assert_due(self.fake_fork("Bb")) + with open(b2, "a", newline="\r\n" if IS_WINDOWS else "\n") as f: + f.write("rem drift\n" if IS_WINDOWS else "# drift\n") + self.assert_due(self.fake_fork("Bb"), "binary_drift") + # runtime_drift (M6): the base's last call saw a versionless plugin and the skills change + self.clone_base("Br") + r = self.finish(self.fork("turns=1", "nover=def456", "skill=extra", base="Br")["job"]) + self.assertEqual((r["status"], r["fingerprint"]), ("ok", "drift:plugins,skills"), r) + self.assert_due(self.fake_fork("Br"), "runtime_drift:plugins,skills") + # defining: the base's record differs from the config in a field it did not override + self.clone_base("Bm", launch=dict(b1_launch, model="claude-opus-4-1")) + self.assert_due(self.fake_fork("Bm"), "defining:model") + # max_base_tokens: from the stored context, and from the transcript when it exists (P0.7) + self.clone_base("Bx", context_tokens=300000) + self.assert_due(self.fake_fork("Bx"), "max_base_tokens:300000") + sid = "11111111-2222-3333-4444-555555555555" + self.clone_base("Bt", session=sid, context_tokens=1000) + tp = forks.transcript_path(b1_launch["cwd"], sid) + self.assertTrue(tp.startswith(os.path.join(self.tmp, "home", "projects")), tp) + os.makedirs(os.path.dirname(tp)) + lines = [{"type": "user", "message": {"role": "user", "content": "x" * 300}}, + {"type": "assistant", "message": {"id": "m1", "usage": { + "input_tokens": 2, "cache_creation_input_tokens": 500, "cache_read_input_tokens": 1000}}}, + {"type": "assistant", "message": {"id": "m2", "usage": { + "input_tokens": 2, "cache_creation_input_tokens": 1000, "cache_read_input_tokens": 260000}}}, + {"type": "system", "subtype": "informational", "content": "y" * 300}] + with open(tp, "w", newline="\n") as f: + f.write("".join(json.dumps(x) + "\n" for x in lines)) + with mock.patch.object(forks, "TRANSCRIPT_TAIL_BYTES", 64): # the tail grows + self.assert_due(self.fake_fork("Bt"), "max_base_tokens:261002") + # rebases: lineage_rebases >= max_rebases 4 + self.clone_base("Bn3", lineage_rebases=3) + self.assert_due(self.fake_fork("Bn3")) + self.clone_base("Bn", lineage_rebases=4) + self.assert_due(self.fake_fork("Bn"), "rebases:4") + # binary_missing: the stored path is gone; launches refuse (exit 2) + gone = dict(info, path=os.path.join(self.tmp, "gone", "claude" + ext)) + self.clone_base("Bz", launch=dict(b1_launch, binary=gone), binary=gone) + out = self.fake_fork("Bz", code=2) + self.assertEqual((out["reason"], out["retrain_due"], out["reasons"]), + ("binary_missing", True, ["binary_missing"]), out) + # list: the base-level reasons, per active base + rows = {b["name"]: b for b in self.cmd("list")["bases"]} + for name, want in (("Bl", ["lines:1600"]), ("Bd", []), ("Bb", ["binary_drift"]), + ("Br", ["runtime_drift:plugins,skills"]), ("Bm", ["defining:model"]), + ("Bx", ["max_base_tokens:300000"]), ("Bt", ["max_base_tokens:261002"]), + ("Bn", ["rebases:4"]), ("Bz", ["binary_missing"])): + with self.subTest(base=name): + self.assert_due(rows[name], *want) + + def test_train_overrides_and_init_model_change(self): + """M10 (plan 4.3): `train` takes init's launch flags as overrides; the base records its + launch fields and their hash, and every later launch on it uses that record. `init` + changing the model afterwards: a base that did not override it reports `retrain_due: + defining:model`; one that did reports nothing.""" + self.setup_run(cap=4) + t = self.cmd("train", "--base", "B1", "--brief", self.brief(), "--model", "claude-sonnet-5-5", + "--effort", "low", "--tools", "Read,Bash") + r = self.finish(t["job"]) + self.assertEqual((r["status"], r["model_requested"], r["models_used"]), + ("ok", "claude-sonnet-5-5", ["claude-sonnet-5-5"]), r) + reg = self.reg() + cfg, b1 = reg["config"], reg["bases"]["B1"] + rec = b1["launch"] + self.assertEqual((rec["model"], rec["effort"], rec["tools"]), ("claude-sonnet-5-5", "low", ["Read", "Bash"])) + self.assertEqual(rec["allowed_tools"], cfg["launch"]["allowed_tools"]) # not overridden + self.assertEqual(rec["cwd"], cfg["launch"]["cwd"]) + self.assertEqual(b1["overrides"], ["effort", "model", "tools"]) + self.assertEqual(b1["fingerprint"], forks.fingerprint_hash(rec)) + self.assertNotEqual(b1["fingerprint"], cfg["fingerprint"]) + a = stub_calls(self.stub)[0]["argv"] + self.assertEqual((a[a.index("--model") + 1], a[a.index("--effort") + 1], a[a.index("--tools") + 1]), + ("claude-sonnet-5-5", "low", "Read,Bash")) + # a base trained without overrides records the config's fields + self.train("B2") + self.assertEqual(self.finish("j0002")["status"], "ok") + b2 = self.reg()["bases"]["B2"] + self.assertEqual((b2["launch"], b2["overrides"], b2["fingerprint"]), + (cfg["launch"], [], cfg["fingerprint"])) + # fork and continue of B1 build their command line from B1's record + f = self.fork("turns=1", base="B1") + self.assert_due(f) + rf = self.finish(f["job"]) + self.assertEqual((rf["status"], rf["model_requested"], rf["models_used"]), + ("ok", "claude-sonnet-5-5", ["claude-sonnet-5-5"]), rf) + c = self.cmd("continue", "--job", f["job"], "--brief", self.brief("turns=1")) + rc = self.finish(c["job"]) + self.assertEqual((rc["status"], rc["model_requested"]), ("ok", "claude-sonnet-5-5"), rc) + for job in (f["job"], c["job"]): + with self.subTest(job=job): + argv = self.reg()["jobs"][job]["argv"] + self.assertEqual((argv[argv.index("--model") + 1], argv[argv.index("--effort") + 1], + argv[argv.index("--tools") + 1]), ("claude-sonnet-5-5", "low", "Read,Bash")) + # init changes the model after the bases exist + out = self.setup_run("--model", "claude-opus-4-1", cap=4) + self.assertTrue(out["changed"], out) + self.assertIn("launch.model", out["changed_fields"]) + rows = {b["name"]: b for b in self.cmd("list")["bases"]} + self.assert_due(rows["B2"], "defining:model") + self.assert_due(rows["B1"]) + f2 = self.fork("turns=1", base="B2") + self.assert_due(f2, "defining:model") # advisory: the fork still launches + r2 = self.finish(f2["job"]) + self.assertEqual((r2["status"], r2["model_requested"]), ("ok", TEST_MODEL), r2) # B2's record + f1 = self.fork("turns=1", base="B1") + self.assert_due(f1) + self.assertEqual(self.finish(f1["job"])["model_requested"], "claude-sonnet-5-5") + # overrides pass init's checks: aliases, plan mode, availableModels + self.cmd("train", "--base", "B3", "--brief", self.brief(), "--model", "sonnet", code=2) + self.cmd("train", "--base", "B3", "--brief", self.brief(), "--permission-mode", "plan", code=2) + settings = os.path.join(self.tmp, "home", "settings.json") + os.makedirs(os.path.dirname(settings), exist_ok=True) + with open(settings, "w") as fh: + json.dump({"availableModels": ["opus"]}, fh) + out = self.cmd("train", "--base", "B3", "--brief", self.brief(), "--model", "claude-sonnet-5-5", code=2) + self.assertEqual(out["error"], "model not admitted by availableModels") + self.assertNotIn("B3", self.reg()["bases"]) + + def test_note_only_from_this_job(self): + """M11: `note` is set only when the note file was written during this job; a note left by an + earlier fork of the block, or by the fork a continue resumes, does not count.""" + self.setup_run() + self.train("B1") + self.finish("j0001") + note = os.path.join(self.state, "notes", "1.1.md") + with open(note, "w") as f: + f.write("old note\n") + old = time.time() - 600 + os.utime(note, (old, old)) + r = self.finish(self.fork("turns=1")["job"]) + self.assertEqual((r["status"], r["note"]), ("ok", None), r) + j = self.fork("sleep=1", "turns=2")["job"] + with open(note, "w") as f: # written while the job runs + f.write("new note\n") + r = self.finish(j) + self.assertEqual((r["status"], r["note"]), ("ok", "notes/1.1.md"), r) + os.utime(note, (old, old)) + rc = self.finish(self.cmd("continue", "--job", j, "--brief", self.brief("turns=1"))["job"]) + self.assertEqual((rc["status"], rc["note"]), ("ok", None), rc) + led = {x["job"]: x for x in self.ledger()} + self.assertEqual((led[j]["note"], led[rc["job"]]["note"]), ("notes/1.1.md", None)) + + def test_rebase_end_to_end(self): + """`rebase` with the stub: the new base is active (parent, lineage, HEAD, context, launch + record), the old one superseded; a fork of the new base hits the new context; a failed + rebase leaves the base active and creates nothing; a binary drift is recorded by the new + base.""" + self.setup_run() + self.train("B1") + self.finish("j0001") + b1 = self.reg()["bases"]["B1"] + self.commit_file("src.py", "x = 1\n", "work of phase 1") + f1 = self.fork("turns=1", extra=("--phase", "1")) + self.assert_due(f1) + self.finish(f1["job"]) + f2 = self.fork("turns=1", extra=("--phase", "2")) + self.assert_due(f2, "phase_boundary") + self.finish(f2["job"]) + out = self.cmd("rebase", "--base", "B1", "--as", "B2", "--phase", "2", + "--brief", self.brief("turns=2", text="Absorb phase 1.")) + self.assertEqual(set(out), {"job", "base", "parent", "session_id"}) + self.assertEqual((out["base"], out["parent"]), ("B2", "B1")) + rr = self.finish(out["job"]) + self.assertEqual((rr["kind"], rr["status"], rr["base"], rr["cache"], rr["phase"], rr["fingerprint"]), + ("rebase", "ok", "B1", "hit", 2, "ok"), rr) + self.assertEqual((rr["session"], rr["parent_session"]), (out["session_id"], b1["session"])) + reg = self.reg() + old, new = reg["bases"]["B1"], reg["bases"]["B2"] + self.assertEqual((old["status"], old["superseded_by"]), ("superseded", "B2")) + self.assertEqual((new["status"], new["parent"], new["lineage_rebases"], new["session"], new["phase"], + new["rebase_job"]), ("active", "B1", 1, out["session_id"], 2, out["job"])) + self.assertEqual(new["base_commit"], self.git("rev-parse", "HEAD")) + self.assertEqual((new["launch"], new["fingerprint"], new["overrides"]), + (old["launch"], old["fingerprint"], [])) + self.assertGreater(new["context_tokens"], old["context_tokens"]) + self.assertIsNotNone(new["last_hit"]) + line = {x["job"]: x for x in self.ledger()}[out["job"]] + self.assertEqual((line["kind"], line["status"]), ("rebase", "ok")) + self.assertAlmostEqual(new["cost_cum_usd"], line["cost_cumulative_usd"], places=6) + a = [c for c in stub_calls(self.stub) if c["session"] == out["session_id"]][0]["argv"] + self.assertEqual(a[a.index("--resume") + 1], b1["session"]) + self.assertIn("--fork-session", a) + self.assertEqual(a[a.index("--session-id") + 1], out["session_id"]) + with open(os.path.join(self.state, "briefs", out["job"] + ".md"), encoding="utf-8") as fh: + txt = fh.read() + for s in ("READY", "Do not change code", "work of phase 1", "`B2`", "Absorb phase 1.", "stale"): + self.assertIn(s, txt) + # a fork of the new base: a hit against the new base's context + f3 = self.fork("turns=1", base="B2", extra=("--phase", "2")) + self.assert_due(f3) + r3 = self.finish(f3["job"]) + self.assertEqual((r3["status"], r3["cache"], r3["parent_session"]), ("ok", "hit", new["session"]), r3) + self.assertEqual(self.reg()["jobs"][f3["job"]]["base_context"], new["context_tokens"]) + self.assertGreaterEqual(r3["tokens"]["cache_read"], new["context_tokens"]) + a = [c for c in stub_calls(self.stub) if c["session"] == r3["session"]][0]["argv"] + self.assertEqual(a[a.index("--resume") + 1], new["session"]) + # the superseded base: no fork, no rebase + self.fork(base="B1", code=2) + self.cmd("rebase", "--base", "B1", "--as", "B9", "--brief", self.brief(), code=2) + # a failed rebase: the base stays active, the new base is never created + rf = self.finish(self.cmd("rebase", "--base", "B2", "--as", "B3", "--brief", self.brief("error"))["job"]) + self.assertEqual((rf["status"], rf["reason"]), ("failed", "is_error"), rf) + reg = self.reg() + self.assertEqual(reg["bases"]["B2"]["status"], "active") + self.assertNotIn("B3", reg["bases"]) + rows = {b["name"]: b for b in self.cmd("list")["bases"]} + self.assertEqual((rows["B1"]["status"], rows["B1"]["superseded_by"], rows["B2"]["parent"]), + ("superseded", "B2", "B1")) + # refusals + for argv, why in ((["--base", "B2", "--as", "B2"], "new base"), + (["--base", "B2", "--as", "B1"], "base exists"), + (["--base", "B9", "--as", "B4"], "unknown base"), + (["--base", "B2", "--as", "a/b"], "bad base name")): + with self.subTest(argv=argv): + self.assertIn(why, self.cmd("rebase", *argv, "--brief", self.brief(), code=2)["error"]) + os.environ[forks.CHILD_MARKER] = "1" + try: + out = self.cmd("rebase", "--base", "B2", "--as", "B4", "--brief", self.brief(), code=2) + self.assertIn("inside a fork", out["error"]) + finally: + os.environ.pop(forks.CHILD_MARKER, None) + # binary drift: the rebase runs under the current binary, and the new base records it + with open(self.stub, "a", newline="\r\n" if IS_WINDOWS else "\n") as f: + f.write("rem drift\n" if IS_WINDOWS else "# drift\n") + fd = self.fork("turns=1", base="B2") + self.assert_due(fd, "binary_drift") + self.finish(fd["job"]) + out = self.cmd("rebase", "--base", "B2", "--as", "B4", "--brief", self.brief("turns=1")) + rb = self.finish(out["job"]) + self.assertEqual((rb["status"], rb["fingerprint"]), ("ok", "drift:binary"), rb) + b2, b4 = self.reg()["bases"]["B2"], self.reg()["bases"]["B4"] + cur = forks.binary_info(forks.norm_path(self.stub)) + for k in ("version", "size", "mtime"): + self.assertEqual((b4["binary"][k], b4["launch"]["binary"][k]), (cur[k], cur[k]), k) + self.assertEqual((b4["parent"], b4["lineage_rebases"], b2["status"]), ("B2", 2, "superseded")) + self.assertEqual(b4["fingerprint"], forks.fingerprint_hash(b4["launch"])) + self.assertNotEqual(b4["fingerprint"], b2["fingerprint"]) + f4 = self.fork("turns=1", base="B4") + self.assert_due(f4) + r4 = self.finish(f4["job"]) + self.assertEqual((r4["status"], r4["fingerprint"], r4["cache"]), ("ok", "ok", "hit"), r4) + + def test_rebase_counts_toward_cap(self): + """A running rebase counts toward parallel_cap, blocks a second rebase of its base and a + train of its new name; killed, it leaves the base active and creates nothing.""" + self.setup_run(cap=1) + self.train("B1") + self.finish("j0001") + nxt = forks._next_job_id(self.reg()) + out = self.cmd("rebase", "--base", "B1", "--as", "B2", + "--brief", self.brief("hang=1", text="RUN-%s" % nxt)) + job = out["job"] + self.assertEqual(job, nxt) + self.until_running(job) + refused = self.fork(code=2) + self.assertEqual((refused["error"], refused["running"]), ("parallel cap reached", [job])) + self.assertIn("rebased", self.cmd("rebase", "--base", "B1", "--as", "B3", "--brief", self.brief(), + code=2)["error"]) + self.assertIn("rebase creates", self.cmd("train", "--base", "B2", "--brief", self.brief(), + code=2)["error"]) + self.assertTrue(self.cmd("kill", "--job", job)["ok"]) + r = self.finish(job) + self.assertEqual((r["status"], r["kind"]), ("killed", "rebase")) + reg = self.reg() + self.assertEqual(reg["bases"]["B1"]["status"], "active") + self.assertNotIn("B2", reg["bases"]) + def tearDownModule(): """No supervisor or stub of this suite may survive it (checked by command line, never by name)."""