Enforce literal substrate convergence

apps/vision-board-studio/photos-restore/2_build_gallery.py

@@ -13,7 +13,7 @@
         if u not in uset: continue
         p=os.path.join(root,f); sz=os.path.getsize(p)
         if u not in best or sz>best[u][1]: best[u]=(p,sz)
-board=json.load(open('/Users/4jp/Workspace/limen/.claude/worktrees/feat-vision-board-studio/apps/vision-board-studio/boards/tony-2017.json'))
+board=json.load(open('/Users/4jp/Workspace/library/engine/organvm/limen/.claude/worktrees/feat-vision-board-studio/apps/vision-board-studio/boards/tony-2017.json'))
 tiles=board['tiles']
 # image embeddings of salvage crops
 def embed_imgs(paths):

cli/src/limen/capacity.py

@@ -1,7 +1,8 @@
 from __future__ import annotations
 
-import os
 import json
+import math
+import os
 import re
 import shlex
 import shutil
@@ -392,8 +393,11 @@ def _usage_int(value: object) -> int | None:
     if isinstance(value, bool):
         return None
     try:
-        return int(float(str(value)))
-    except (TypeError, ValueError):
+        parsed = float(str(value))
+        if not math.isfinite(parsed) or parsed < 0:
+            return None
+        return int(parsed)
+    except (OverflowError, TypeError, ValueError):
         return None
 
 
@@ -420,11 +424,16 @@ def _live_usage_capacity(
         return None
     cap = _usage_int(info.get("possible"))
     remaining = _usage_int(info.get("remaining"))
-    if cap is None or cap <= 0 or remaining is None:
+    if cap is None or cap <= 0 or remaining is None or remaining < 0 or remaining > cap:
+        return None
+    raw_consumed = info.get("consumed")
+    consumed = _usage_int(raw_consumed)
+    if raw_consumed not in (None, "") and consumed is None:
+        return None
+    if consumed is not None and (consumed > cap or consumed + remaining > cap):
         return None
-    consumed = _usage_int(info.get("consumed"))
     spent = consumed if consumed is not None else max(0, cap - remaining)
-    return cap, max(0, spent), max(0, remaining)
+    return cap, spent, remaining
 
 
 def capacity_census(board: object = None, budget_limit: int | None = None) -> list[CapacityRow]:
@@ -522,7 +531,11 @@ def format_capacity_census(rows: list[CapacityRow]) -> str:
 
 
 def _root() -> Path:
-    return Path(os.environ.get("LIMEN_ROOT", str(Path.home() / "Workspace" / "limen")))
+    configured = os.environ.get("LIMEN_ROOT")
+    if configured:
+        return Path(configured).expanduser()
+    workspace = Path(os.environ.get("WORKSPACE_ROOT", str(Path.home() / "Workspace"))).expanduser()
+    return workspace / "library" / "engine" / "organvm" / "limen"
 
 
 def _load_usage(root: Path | None = None) -> dict[str, object]:

cli/src/limen/cli.py

@@ -67,6 +67,45 @@ def main():
 main.add_command(fanout_group)
 
 
+@main.group("worktree")
+def worktree_group() -> None:
+    """Inspect and reclaim disposable worktree lifecycle units."""
+
+
+@worktree_group.command(
+    "reap",
+    context_settings={
+        "ignore_unknown_options": True,
+        "allow_extra_args": True,
+        "help_option_names": [],
+    },
+)
+@click.argument("reaper_args", nargs=-1, type=click.UNPROCESSED)
+def worktree_reap(reaper_args: tuple[str, ...]) -> None:
+    """Delegate unchanged arguments to the bounded worktree reaper."""
+
+    root = resolve_limen_repo_root()
+    script = root / "scripts" / "reclaim-worktrees.py"
+    if not script.is_file():
+        raise click.ClickException(f"worktree reaper is missing: {script}")
+    try:
+        result = subprocess.run(
+            [sys.executable, str(script), *reaper_args],
+            cwd=root,
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+    except OSError as exc:
+        raise click.ClickException(f"could not launch worktree reaper: {exc}") from exc
+    if result.stdout:
+        click.echo(result.stdout, nl=False)
+    if result.stderr:
+        click.echo(result.stderr, nl=False, err=True)
+    if result.returncode:
+        raise click.exceptions.Exit(result.returncode)
+
+
 def _host_owner() -> tuple[str, int]:
     pid = os.getppid()
     label = os.environ.get("LIMEN_HOST_ADMISSION_OWNER") or os.environ.get("LIMEN_SESSION_ID")

cli/src/limen/conduct/campaign_relay.py

@@ -24,6 +24,7 @@
     ContractError,
     validate_contract,
 )
+from limen.worktree_layout import runtime_worktree_path
 
 _RECEIPT_CEILING = 65_536
 _PREDECESSOR_RECEIPT_CEILING = 262_144
@@ -379,12 +380,28 @@ def _relay_worktree(
 ) -> Path:
     """Resolve the generated successor worktree and bind it to the same repository."""
 
-    primary = _primary_checkout(
-        root,
+    primary = _primary_checkout(root, deadline_monotonic=deadline_monotonic)
+    listing = _git(
+        primary,
+        "worktree",
+        "list",
+        "--porcelain",
         deadline_monotonic=deadline_monotonic,
     )
-    candidate = primary / ".worktrees" / successor_slug
+    expected_branch = f"refs/heads/work/{successor_slug}"
+    candidates: list[Path] = []
+    for record in listing.split("\n\n"):
+        fields = dict(line.split(" ", 1) for line in record.splitlines() if " " in line)
+        if fields.get("branch") == expected_branch and fields.get("worktree"):
+            candidates.append(Path(fields["worktree"]))
+    if len(candidates) != 1:
+        raise CampaignRelayError(
+            "relay_worktree_invalid",
+            "campaign relay successor worktree is unavailable",
+        )
+    candidate = candidates[0]
     try:
+        expected_candidate = runtime_worktree_path(primary, successor_slug)
         resolved = candidate.resolve(strict=True)
         top_level = Path(
             _git(
@@ -403,12 +420,18 @@ def _relay_worktree(
             primary,
             deadline_monotonic=deadline_monotonic,
         )
-    except OSError as exc:
+    except (OSError, ValueError) as exc:
         raise CampaignRelayError(
             "relay_worktree_invalid",
             "campaign relay successor worktree is unavailable",
         ) from exc
-    if candidate.is_symlink() or not resolved.is_dir() or top_level != resolved or worktree_common != primary_common:
+    if (
+        candidate.is_symlink()
+        or not resolved.is_dir()
+        or candidate != expected_candidate
+        or top_level != resolved
+        or worktree_common != primary_common
+    ):
         raise CampaignRelayError(
             "relay_worktree_invalid",
             "campaign relay successor worktree does not match the primary checkout",

cli/src/limen/dispatch.py

@@ -130,6 +130,14 @@
 )
 
 
+def _limen_root() -> Path:
+    configured = os.environ.get("LIMEN_ROOT")
+    if configured:
+        return Path(configured).expanduser()
+    workspace = Path(os.environ.get("WORKSPACE_ROOT", str(Path.home() / "Workspace"))).expanduser()
+    return workspace / "library" / "engine" / "organvm" / "limen"
+
+
 def _int_or_default(raw: object, default: int) -> int:
     if isinstance(raw, bool):
         return default
@@ -213,7 +221,7 @@ def _usage_dead_lanes() -> set[str]:
     signal, never pinned: a lane auto-rejoins the instant its rolling window refills (no manual edit).
     This is what makes dispatch HONEST — we never assign a task to a lane that physically cannot
     produce, and we never burn one to 0."""
-    f = Path(os.environ.get("LIMEN_ROOT", str(Path.home() / "Workspace" / "limen"))) / "logs" / "usage.json"
+    f = _limen_root() / "logs" / "usage.json"
     try:
         vendors = (json.loads(f.read_text()) or {}).get("vendors", {})
     except (OSError, ValueError):
@@ -334,7 +342,7 @@ def _down_lanes() -> set[str]:
          when every observed provider is in auth/rate cooldown.
     Rebalance + dispatch + route skip these so tasks aren't wasted on a lane that can't produce.
     Sources 2–4 self-heal; remove a line from source 1 when that lane is healthy again."""
-    f = Path(os.environ.get("LIMEN_ROOT", str(Path.home() / "Workspace" / "limen"))) / "logs" / "lanes-down.txt"
+    f = _limen_root() / "logs" / "lanes-down.txt"
     manual: set[str] = set()
     try:
         manual = {ln.split("#")[0].strip() for ln in f.read_text().splitlines() if ln.split("#")[0].strip()}
@@ -393,7 +401,7 @@ def run_always_working_before_dispatch(tasks_path: Path, *, dry_run: bool = Fals
     """
     if dry_run or os.environ.get("LIMEN_ALWAYS_WORKING_BEFORE_DISPATCH", "1") != "1":
         return True
-    root = Path(os.environ.get("LIMEN_ROOT", str(Path.home() / "Workspace" / "limen"))).resolve()
+    root = _limen_root().resolve()
     try:
         if tasks_path.resolve() != (root / "tasks.yaml").resolve():
             return True
@@ -1159,7 +1167,7 @@ def _value_tier_repos() -> set[str]:
     repos: set[str] = {r.strip() for r in os.environ.get("LIMEN_VALUE_REPOS", "").split(",") if r.strip()}
     fpath = os.environ.get(
         "LIMEN_VALUE_REPOS_FILE",
-        str(Path(os.environ.get("LIMEN_ROOT", str(Path.home() / "Workspace" / "limen"))) / "value-repos.json"),
+        str(_limen_root() / "value-repos.json"),
     )
     try:
         data = json.loads(Path(fpath).read_text())
@@ -1718,21 +1726,21 @@ def _journaled_agent_dispatch(
 
 
 def _remote_receipt_store() -> ReceiptStore:
-    root = Path(os.environ.get("LIMEN_ROOT", str(Path.home() / "Workspace" / "limen")))
+    root = _limen_root()
     configured = os.environ.get("LIMEN_REMOTE_RECEIPT_ROOT")
     return ReceiptStore(Path(configured).expanduser() if configured else root / "logs" / "remote-execution")
 
 
 def _receipt_path_for_board(path: Path) -> str:
-    root = Path(os.environ.get("LIMEN_ROOT", str(Path.home() / "Workspace" / "limen")))
+    root = _limen_root()
     try:
         return str(path.resolve().relative_to(root.resolve()))
     except ValueError:
         return str(path)
 
 
 def _authoritative_remote_verification(task: Task) -> tuple[Task, dict[str, object]]:
-    root = Path(os.environ.get("LIMEN_ROOT", str(Path.home() / "Workspace" / "limen")))
+    root = _limen_root()
     tasks_path = Path(os.environ.get("LIMEN_TASKS", str(root / "tasks.yaml"))).expanduser()
     if not tasks_path.is_file():
         raise RemoteExecutionError("authoritative task board is unavailable for verification-only dispatch")
@@ -1903,7 +1911,7 @@ def _flame_preamble() -> str:
     a missing/unreadable kernel must NEVER block a dispatch (derive-never-pin, fail-open)."""
     if os.environ.get("LIMEN_FLAME_KERNEL", "1") != "1":
         return ""
-    root = Path(os.environ.get("LIMEN_ROOT", str(Path.home() / "Workspace" / "limen")))
+    root = _limen_root()
     f = root / os.environ.get("LIMEN_FLAME_FILE", "FLAME.md")
     try:
         key = f"{f}:{f.stat().st_mtime_ns}"
@@ -3049,7 +3057,7 @@ def _local_repo_matches(repo: str | None, identity: str) -> bool:
         for raw in (
             os.environ.get("LIMEN_ROOT"),
             os.environ.get("LIMEN_LIVE_ROOT"),
-            str(Path.home() / "Workspace" / "limen"),
+            str(Path.home() / "Workspace" / "library" / "engine" / "organvm" / "limen"),
         )
         if raw
     }
@@ -4286,7 +4294,7 @@ def _bridge_agy_scratch(task: Task, wt: Path) -> None:
 def _lane_run_env(agent: str, wt: Path | None = None, task: Task | None = None) -> dict[str, str]:
     run_env = os.environ.copy()
     if wt is not None:
-        live_root = os.environ.get("LIMEN_ROOT", str(Path.home() / "Workspace" / "limen"))
+        live_root = str(_limen_root())
         run_env["LIMEN_LIVE_ROOT"] = live_root
         run_env["LIMEN_ROOT"] = str(wt)
         run_env["LIMEN_TASKS"] = str(wt / "tasks.yaml")
@@ -4300,9 +4308,7 @@ def _lane_run_env(agent: str, wt: Path | None = None, task: Task | None = None)
     # agy/antigravity defense-in-depth: if auth falls through to browser opening mid-run,
     # make the opener a no-op inside the lane subprocess only.
     if agent in ("agy", "antigravity"):
-        shim = str(
-            Path(os.environ.get("LIMEN_ROOT", str(Path.home() / "Workspace" / "limen"))) / "scripts" / "agy-noop-shim"
-        )
+        shim = str(_limen_root() / "scripts" / "agy-noop-shim")
         run_env["PATH"] = shim + os.pathsep + run_env.get("PATH", "")
         run_env["BROWSER"] = "true"
     # claude fleet auth must not share or mutate the interactive session's macOS Keychain token.
@@ -4585,7 +4591,7 @@ def _record_worktree_lifecycle(
     generated_cleanup: str,
     pushed: bool,
 ) -> None:
-    root = Path(os.environ.get("LIMEN_ROOT", str(Path.home() / "Workspace" / "limen")))
+    root = _limen_root()
     try:
         log = root / "logs" / "worktree-lifecycle.jsonl"
         log.parent.mkdir(parents=True, exist_ok=True)
@@ -4966,7 +4972,7 @@ def _window_hours(agent: str) -> float:
     (5h rolling windows) refill ~5x/day instead of being throttled by a once-a-day cap,
     while jules/gemini/opencode/agy refill daily."""
     try:
-        root = Path(os.environ.get("LIMEN_ROOT", str(Path.home() / "Workspace" / "limen")))
+        root = _limen_root()
         with open(root / "logs" / "usage-limits.json") as fh:
             limits = json.load(fh)
         window = str((limits.get(agent) or {}).get("window", ""))
@@ -5505,7 +5511,7 @@ def _apply_result(
 def _ledger_lanes() -> dict[str, dict[str, list[str]]]:
     """logs/ledger.json lanes map (waste_classes/win_classes per lane) — fail-open to {}."""
     try:
-        root = Path(os.environ.get("LIMEN_ROOT", str(Path.home() / "Workspace" / "limen")))
+        root = _limen_root()
         return json.loads((root / "logs" / "ledger.json").read_text()).get("lanes", {}) or {}
     except Exception:
         return {}
@@ -5514,7 +5520,7 @@ def _ledger_lanes() -> dict[str, dict[str, list[str]]]:
 def _lane_fitness() -> dict[str, dict[str, dict]]:
     """logs/lane-fitness.json per-(agent, task_class) fitness map — fail-open to {}."""
     try:
-        root = Path(os.environ.get("LIMEN_ROOT", str(Path.home() / "Workspace" / "limen")))
+        root = _limen_root()
         return json.loads((root / "logs" / "lane-fitness.json").read_text()).get("pairs", {}) or {}
     except Exception:
         return {}
@@ -5673,7 +5679,7 @@ def _claude_tier_overrides() -> dict[str, list[str]]:
     Fail-open to {} (→ the ledger-DISCOVERED default). Demoted to an override: the default
     pre-assign set is discovered from the ledger, not pinned. Same read pattern as _ledger_lanes()."""
     try:
-        root = Path(os.environ.get("LIMEN_ROOT", str(Path.home() / "Workspace" / "limen")))
+        root = _limen_root()
         return json.loads((root / "logs" / "model-tiers.json").read_text()).get("claude") or {}
     except Exception:
         return {}

cli/src/limen/provider_health.py

@@ -91,7 +91,9 @@ def provider_health_policy() -> ProviderHealthPolicy:
 
 
 def provider_outcome_ledger_path() -> Path:
-    root = Path(os.environ.get("LIMEN_ROOT", Path.home() / "Workspace" / "limen")).expanduser()
+    root = Path(
+        os.environ.get("LIMEN_ROOT", Path.home() / "Workspace" / "library" / "engine" / "organvm" / "limen")
+    ).expanduser()
     raw = str(parameter("LIMEN_PROVIDER_OUTCOME_LEDGER", str(root / "logs" / "provider-outcomes.jsonl")))
     return Path(raw.replace("$LIMEN_ROOT", str(root))).expanduser()
 

cli/src/limen/repository_ignored.py

@@ -0,0 +1,41 @@
+"""Shared loss-free classification for ignored repository entries.
+
+The literal-substrate court and the clone reaper must agree about which ignored
+working-tree payloads are reproducible. Anything outside this positive
+allowlist requires custody and therefore prevents checkout removal.
+"""
+
+from __future__ import annotations
+
+
+REGENERABLE_DIRS = frozenset(
+    "node_modules .venv venv .venv-demucs __pycache__ .pytest_cache .mypy_cache .ruff_cache "
+    ".tox dist build .next .nuxt .svelte-kit .astro .turbo .parcel-cache .vercel .wrangler "
+    ".gradle coverage .nyc_output .eggs .ipynb_checkpoints".split()
+)
+REGENERABLE_SUFFIXES = (".pyc", ".pyo")
+REGENERABLE_FILES = frozenset({".DS_Store"})
+
+
+def ignored_entries_from_porcelain(output: str) -> tuple[str, ...]:
+    """Return raw relative paths from ``git status --porcelain=v1 -z --ignored``."""
+
+    return tuple(
+        record[3:].rstrip("/") for record in output.split("\0") if record.startswith("!! ") and record[3:].rstrip("/")
+    )
+
+
+def ignored_entry_is_regenerable(path: str) -> bool:
+    """Whether one enumerated ignored entry is safe to recreate after cloning."""
+
+    normalized = path.rstrip("/")
+    if not normalized or normalized.startswith("/") or "\0" in normalized:
+        return False
+    parts = normalized.split("/")
+    if any(part in {"", ".", ".."} for part in parts):
+        return False
+    top = parts[0]
+    base = parts[-1]
+    if top in REGENERABLE_DIRS:
+        return True
+    return len(parts) == 1 and (base in REGENERABLE_FILES or base.endswith(REGENERABLE_SUFFIXES))