worktree_env.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_orchestrator/worktree_env.py
 
# OCT-LINT: disable=tests_exist reason="Covered by test_241_worktree_env.py"
"""
Per-worktree uv environments: prepare and teardown.
 
Purpose
-------
Plan 241 (ENV-01). The shared ``G:\\PythonApps\\option_c\\.venv`` carries
thirteen editable installs that all point at the MAIN tree, so a worker's
probe or suite silently imports the main tree unless ``PYTHONPATH`` lists
the worktree first — the F17 shadow. This module ends that class: it builds
a venv INSIDE the worktree whose editable installs point at that worktree,
and it VERIFIES the result by importing every package in a clean
interpreter with no ``PYTHONPATH`` before it reports success.
 
Responsibilities
----------------
- :data:`EDITABLE_PACKAGES` — the one declared table of the monorepo's
  editable packages (path, distribution name, import name, extras, and
  the modules those extras must make importable).
- :data:`SMOKE_CHECKS` — plan 243 P4: the two commands every worker runs
  first, executed inside the prepared worktree on ``--smoke``.
- :func:`ensure_branch_and_worktree` — the idempotent git anchors; the ONE
  implementation, shared with :mod:`oc_orchestrator.dispatch`.
- :func:`prepare_worktree` — branch, worktree, ``uv venv``, the editable
  installs FROM THE WORKTREE, the ``sitecustomize.py`` copy, and the
  verification. Idempotent: a second run reports "already prepared" and
  still repairs a missing ``sitecustomize.py``.
- :func:`teardown_worktree` — the symmetric removal, which names the known
  Windows ``.git/worktrees/<name>`` permission failure rather than leaving
  the orchestrator to recognise it.
- :class:`WorktreeEnvError` — every refusal, named, raised BEFORE the step
  it refuses so a rerun can always repair what a failure left behind.
- :data:`OctHook` / :func:`use_oct_hook` — plan 270's seam: the optional
  ``oct`` timing ledger, absent by default, registered by a consumer
  that has it.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — lifecycle events (prepare/teardown attempted, refused, completed).
    L3 — semantic details (discovery, install, verification).
    L4 — deep tracing.
 
Contracts
---------
- Verification IS the contract (plan 243 P3): a prepare that prints
  "verified" means STEP-0 and both suites can run in that worktree, and a
  prepare that cannot say so refuses naming EVERY module that failed —
  never the first one only. No worker repairs a prepared venv by hand.
- ``uv`` is an external tool the operator installs; this module CHECKS for
  it and refuses by name when it is absent. It never installs it, and it
  never silently falls back to the shared venv, which would re-open the
  shadow this module exists to close.
- The uv cache must live on the same drive as the venv it fills: across
  drives uv cannot hardlink and silently copies. A cross-drive cache is
  refused with both paths named.
- Nothing outside ``worktree`` (and the git metadata of ``repo_root``) is
  written. The shared main venv is never touched.
- ``oct`` is optional and absent by default (plan 270 P1/P4). Every edge
  to it goes through :data:`OctHook`, whose default refuses, so the
  timing ledger degrades to a null context and nothing in this package's
  import graph names the toolchain.
- :data:`EDITABLE_PACKAGES` and :data:`SMOKE_CHECKS` are DATA describing
  the Option C monorepo, not a contract of this package: a different
  consumer supplies a different tree and would not use them. They are
  the honest residue of the split, named here rather than hidden.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg; also the home of the venv repair file this module copies into a fresh venv)
- uv (not imported; external tool run as a subprocess for the venv and its editable installs, checked for, never installed)
"""
 
from __future__ import annotations
 
from dataclasses import dataclass, field
import json
import os
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
import time
from typing import Any, Callable
 
try:  # pragma: no cover - 3.11+ everywhere this module runs
    import tomllib
except ModuleNotFoundError:  # pragma: no cover - 3.10 has no tomllib
    tomllib = None  # type: ignore[assignment]
 
import oc_diagnostics
from oc_diagnostics import _dbg
 
_dbg("OC_CHAT.MAIN", "module", "worktree_env loaded", 4)
 
 
class WorktreeEnvError(RuntimeError):
    """A named refusal at the worktree-environment boundary."""
 
 
#: Resolves one member of the optional ``oct`` sibling by name.
OctHook = Callable[[str], Any]
 
#: The one target the hook still resolves. Plan 270 P1 retired the second
#: (``"package"``): the venv repair file lives in ``oc_diagnostics`` now,
#: so :func:`sitecustomize_source` reaches ``oct`` not at all.
OCT_HOOK_TARGETS = ("timed_stage",)
 
 
def _default_oct_hook(target: str) -> Any:
    """Refuse every target: this package does not know ``oct`` exists.
 
    Plan 257 P5 made ``oct`` reachable through exactly one seam so that a
    packaging plan could make it a genuine optional extra by replacing
    this default and changing nothing else. Plan 270 is that plan, and
    this is that replacement: the default now raises ``ImportError`` for
    a known target, which is precisely what :func:`_engine_timing` reads
    as "oct is absent" and degrades to a null context for. A standalone
    dispatch therefore records no timings and reaches no toolchain, and
    ``oct`` appears nowhere in this package's import graph.
 
    A consumer that HAS ``oct`` installs the live hook at its boundary
    with :func:`use_oct_hook` — ``oc_chat`` does, in its
    ``core/worktree_env.py`` shim — and the timing ledger fills exactly
    as it did before the split.
 
    An UNKNOWN target is still a :class:`WorktreeEnvError`, deliberately
    not an ``ImportError`` subclass, so the timing edge's ``except
    ImportError`` cannot swallow a programming mistake as "oct is absent".
    """
    func = "_default_oct_hook"
    if target in OCT_HOOK_TARGETS:
        _dbg("OC_CHAT.MAIN", func, f"no oct hook installed for {target!r}", 4)
        raise ImportError(
            f"no oct hook is installed, so {target!r} cannot be resolved; "
            f"a consumer that has oct registers one with use_oct_hook()"
        )
    _dbg("OC_CHAT.MAIN", func, f"VALIDATION_ERROR: unknown target {target!r}", 3)
    raise WorktreeEnvError(f"unknown oct hook target {target!r}")
 
 
#: The module default; a caller passes ``oct_hook=`` to substitute one for
#: a single call, or :func:`use_oct_hook` to substitute the default.
_oct_hook: OctHook = _default_oct_hook
 
 
def use_oct_hook(hook: OctHook | None) -> OctHook:
    """Install *hook* as this module's default, returning the previous one.
 
    The registration point for a consumer that ships ``oct`` (plan 270
    P4). Passing ``None`` restores :func:`_default_oct_hook`, which is
    what a test that borrowed the seam uses to give it back.
    """
    func = "use_oct_hook"
    global _oct_hook
    previous = _oct_hook
    _oct_hook = hook if hook is not None else _default_oct_hook
    _dbg("OC_CHAT.MAIN", func, f"oct hook is now {getattr(_oct_hook, '__name__', _oct_hook)!r}", 3)
    return previous
 
 
@dataclass(frozen=True)
class EditablePackage:
    """One editable install: where it lives, what it is called, what it needs."""
 
    relative_path: str
    dist_name: str
    import_name: str
    extras: tuple[str, ...] = ()
    #: Plan 243 P1. Modules the declared ``extras`` must make importable,
    #: beyond ``import_name`` itself: the imports the dev-engine tools and
    #: the two suites actually make. These live in the venv's
    #: ``site-packages``, not in the worktree's source tree, so they are
    #: verified for importability and venv residence rather than for
    #: resolving inside the checkout.
    verify_imports: tuple[str, ...] = ()
 
    def source_dir(self, root: Path) -> Path:
        """This package's project directory beneath *root*."""
        return root / Path(self.relative_path)
 
    def install_spec(self, root: Path) -> str:
        """The ``uv pip install -e`` argument for this package under *root*."""
        target = str(self.source_dir(root))
        if self.extras:
            return f"{target}[{','.join(self.extras)}]"
        return target
 
    def declares_its_extras(self, root: Path) -> bool:
        """Whether this package's own pyproject declares every named extra.
 
        A stand-in package built by a fixture (and any checkout that has
        renamed an extra) declares none of them: ``uv`` only warns for an
        extra that does not exist, so nothing is installed and there is
        nothing to verify. Saying so is honest; demanding the modules
        anyway would refuse every fixture tree. The live checkout is
        fenced separately by the plan-243 W5 witness, which asserts each
        declared extra really exists.
        """
        if not self.extras:
            return True
        if tomllib is None:  # pragma: no cover - Python 3.10 has no tomllib
            return False
        path = self.source_dir(root) / "pyproject.toml"
        try:
            with path.open("rb") as handle:
                loaded = tomllib.load(handle)
        except (OSError, ValueError):
            return False
        project = loaded.get("project")
        declared = project.get("optional-dependencies") if isinstance(project, dict) else None
        if not isinstance(declared, dict):
            return False
        return all(extra in declared for extra in self.extras)
 
 
#: The monorepo's editable packages, in install order. Verified against the
#: checkout by ``test_w1_declared_package_roots_all_exist_in_this_checkout``.
#: Plan 243 P1 (ENV-02): the table declares what a DEVELOPMENT checkout
#: needs, so ``oct`` installs its aggregate ``dev`` extra (test, docs, mcp,
#: metrics, claude) and ``oc_chat`` its ``terminal`` and ``test`` extras.
#: The ``verify_imports`` columns are the imports the dev-engine tools and
#: the two suites actually make: ``yaml`` for the STEP-0 ``plan_status``
#: tool, ``mcp.server.mcpserver`` for ``oct/tests/tests_mcp`` collection,
#: ``jinja2`` and ``sphinx`` for ``tests_docs``, and
#: ``setuptools``/``setuptools_scm`` for the two named oc_chat tests. Both
#: worktrees prepared before this landed were short of exactly those, and
#: both workers repaired them by hand.
EDITABLE_PACKAGES: tuple[EditablePackage, ...] = (
    EditablePackage("oc_diagnostics", "oc_diagnostics", "oc_diagnostics"),
    EditablePackage(
        "oct",
        "oct",
        "oct",
        ("dev",),
        (
            "pytest",
            "xdist",
            "pytest_timeout",
            # NOT bare ``mcp``: what oct/tests/tests_mcp needs is the module
            # oct's own server imports, and mcp 2.x renamed it. A resolved
            # mcp that cannot serve oct is a refusal here rather than a
            # green prepare followed by an aborted oct collection.
            "mcp.server.mcpserver",  # plan 244: the v2 SDK module oct-mcp builds on
            "jinja2",
            "sphinx",
        ),
    ),
    EditablePackage("oc_pricing", "oc-pricing", "oc_pricing"),
    EditablePackage("oc_connect", "oc-connect", "oc_connect"),
    # Plan 270 P2: the dispatch layer's own package. It MUST be installed
    # before oc_chat can be, because oc_chat now declares it — uv resolves
    # the whole `-e` set in one command, so the row is what makes the
    # requirement resolvable at all inside a worktree.
    EditablePackage("oc_orchestrator", "oc-orchestrator", "oc_orchestrator"),
    EditablePackage(
        "contracts/option_c_agent_contracts",
        "option-c-agent-contracts",
        "option_c_agent_contracts",
    ),
    EditablePackage(
        "contracts/option_c_connector_contracts",
        "option-c-connector-contracts",
        "option_c_connector_contracts",
    ),
    EditablePackage(
        "contracts/option_c_registry_contracts",
        "option-c-registry-contracts",
        "option_c_registry_contracts",
    ),
    EditablePackage(
        "oc_chat/oc_ui", "oc_ui", "oc_ui", ("terminal",), ("rich", "prompt_toolkit")
    ),
    EditablePackage(
        "oc_chat/oc_chat",
        "oc_chat",
        "oc_chat",
        ("terminal", "test"),
        ("yaml", "setuptools_scm", "setuptools", "rich", "prompt_toolkit"),
    ),
    EditablePackage(
        "oc_chat/providers/deepseek", "providers-deepseek", "providers.deepseek"
    ),
    EditablePackage(
        "oc_chat/providers/example", "providers-example", "providers.example"
    ),
    EditablePackage(
        "oc_chat/providers/ollama", "providers-ollama", "providers.ollama"
    ),
    EditablePackage(
        "oc_chat/providers/openai_compat",
        "providers-openai-compat",
        "providers.openai_compat",
    ),
)
 
#: Written into the prepared venv so a second run can tell "already
#: prepared" from "needs installing" without re-resolving thirteen builds.
STAMP_FILENAME = ".oc_prepare.json"
STAMP_VERSION = 1
 
#: The workspace-root cache directory plan 241 P2 pins on the venvs' drive.
UV_CACHE_DIRNAME = ".uv-cache"
 
#: Plan 243 P4: the two things every worker actually does first, run inside
#: the prepared worktree on ``--smoke``. Verification proves the modules
#: import; these prove the tool and the suite prerequisites really run.
#: Plan 270: this table, like :data:`EDITABLE_PACKAGES`, states what the
#: Option C MONOREPO's workers do. The first entry names the module path
#: the rendered brief tells the worker to run — ``oc_chat``'s shim, not
#: this package's own module. Naming the package here instead would
#: quietly stop proving that the brief's OWN command works, which is the
#: whole point of the check: a brief handed out before its command is
#: known to work is the hand-repair loop ENV-02 exists to end.
SMOKE_CHECKS: tuple[tuple[str, tuple[str, ...]], ...] = (
    (
        "STEP-0 tool (plan_status --help)",
        ("-m", "oc_chat.scripts.plan_status", "--help"),
    ),
    (
        "dev-engine imports",
        ("-c", "import yaml, mcp, jinja2, setuptools_scm, setuptools"),
    ),
)
 
_UV_INSTALL_HINT = (
    "install it (https://docs.astral.sh/uv/) and re-run, or dispatch with "
    "--shared-venv to render the legacy shared-venv brief explicitly"
)
_HARDLINK_FALLBACK_MARKER = "failed to hardlink"
 
 
# ── git plumbing (shared with oc_orchestrator.dispatch) ───────────────────────
 
 
def _run_git(
    repo_root: Path, *args: str, check: bool = True
) -> subprocess.CompletedProcess[str]:
    """Run one git command in *repo_root*, refusing by name when it fails."""
    result = subprocess.run(
        ["git", *args],
        cwd=repo_root,
        text=True,
        capture_output=True,
        check=False,
    )
    if check and result.returncode != 0:
        detail = (result.stderr or result.stdout).strip()
        raise WorktreeEnvError(f"git {' '.join(args)} failed: {detail}")
    return result
 
 
def _branch_exists(repo_root: Path, branch: str) -> bool:
    probe = _run_git(
        repo_root, "show-ref", "--verify", f"refs/heads/{branch}", check=False
    )
    return probe.returncode == 0
 
 
def _same_path(left: Path, right: Path) -> bool:
    return str(left.resolve()).casefold() == str(right.resolve()).casefold()
 
 
def worktree_entries(repo_root: Path) -> list[tuple[Path, str | None]]:
    """Return ``(path, branch)`` for every worktree registered to *repo_root*."""
    result = _run_git(repo_root, "worktree", "list", "--porcelain")
    entries: list[tuple[Path, str | None]] = []
    path: Path | None = None
    branch: str | None = None
    for line in [*result.stdout.splitlines(), ""]:
        if line.startswith("worktree "):
            path = Path(line.removeprefix("worktree ")).resolve()
        elif line.startswith("branch refs/heads/"):
            branch = line.removeprefix("branch refs/heads/")
        elif not line and path is not None:
            entries.append((path, branch))
            path = None
            branch = None
    return entries
 
 
def ensure_branch_and_worktree(
    repo_root: Path,
    branch: str,
    worktree: Path,
    *,
    base_sha: str | None = None,
) -> None:
    """Create the branch and worktree when absent; reuse them when present.
 
    *base_sha* is the anchor the branch is created at (``master`` when it is
    ``None``). When it is supplied AND the branch already exists at a
    different commit the call refuses: an operator who names an anchor is
    asserting where the branch starts, and silently accepting a branch that
    starts somewhere else is the lie plan 223 P4 removed from BASE_SHA.
    Dispatch passes ``None`` — it has already recorded the merge-base.
    """
    func = "ensure_branch_and_worktree"
    entries = worktree_entries(repo_root)
    for existing_path, existing_branch in entries:
        if existing_branch == branch:
            if _same_path(existing_path, worktree):
                _dbg("OC_CHAT.MAIN", func, f"{branch} already at {worktree}", 3)
                return
            raise WorktreeEnvError(
                f"branch {branch!r} is already checked out at {existing_path}"
            )
        if _same_path(existing_path, worktree):
            raise WorktreeEnvError(
                f"worktree path {worktree} is registered to branch {existing_branch!r}"
            )
    if worktree.exists():
        raise WorktreeEnvError(
            f"worktree path {worktree} exists but is not registered with git"
        )
    created_branch = not _branch_exists(repo_root, branch)
    if created_branch:
        _run_git(repo_root, "branch", branch, base_sha or "master")
    elif base_sha is not None:
        tip = _run_git(repo_root, "rev-parse", branch).stdout.strip()
        anchor = _run_git(repo_root, "rev-parse", base_sha).stdout.strip()
        if tip != anchor:
            raise WorktreeEnvError(
                f"branch {branch!r} exists at {tip} but the requested anchor is "
                f"{anchor}; refusing to move or reuse a branch that starts elsewhere"
            )
    try:
        worktree.parent.mkdir(parents=True, exist_ok=True)
        _run_git(repo_root, "worktree", "add", str(worktree), branch)
    except Exception:
        if created_branch:
            _run_git(repo_root, "branch", "-d", branch, check=False)
        raise
    _dbg("OC_CHAT.MAIN", func, f"{branch} checked out at {worktree}", 2)
 
 
# ── venv layout helpers ────────────────────────────────────────────────────
 
 
def venv_scripts_dir(venv: Path) -> Path:
    """The venv's executables directory (``Scripts`` on Windows, ``bin`` else)."""
    posix = venv / "bin"
    if posix.is_dir() and not (venv / "Scripts").is_dir():
        return posix
    return venv / "Scripts"
 
 
def venv_python(venv: Path) -> Path:
    """The venv's interpreter path."""
    scripts = venv_scripts_dir(venv)
    return scripts / ("python.exe" if scripts.name == "Scripts" else "python")
 
 
def venv_site_packages(venv: Path) -> Path:
    """The venv's ``site-packages`` directory, refusing when it is absent."""
    windows = venv / "Lib" / "site-packages"
    if windows.is_dir():
        return windows
    for candidate in sorted((venv / "lib").glob("python*/site-packages")):
        return candidate
    raise WorktreeEnvError(f"no site-packages directory under {venv}")
 
 
def _drive_of(path: Path) -> str:
    """The drive (or UNC share) *path* lives on; patched by the W2 witness."""
    return Path(path).drive.casefold()
 
 
def discover_packages(root: Path) -> tuple[EditablePackage, ...]:
    """Return the declared packages that actually exist beneath *root*."""
    func = "discover_packages"
    found = tuple(
        package
        for package in EDITABLE_PACKAGES
        if (package.source_dir(root) / "pyproject.toml").is_file()
    )
    _dbg("OC_CHAT.MAIN", func, f"{len(found)} package(s) under {root}", 3)
    return found
 
 
def sitecustomize_source(worktree: Path) -> Path:
    """Locate ``venv_sitecustomize.py``, preferring the worktree's own copy.
 
    A fresh venv without this file resolves ``oc_diagnostics`` as a bare
    namespace package (``__file__`` is ``None``) — the OCON incident.
 
    Plan 270 P1 moved the file into ``oc_diagnostics``, the package it
    repairs, so the order is: the worktree's own copy first (a checkout
    ships its own), then the copy inside the ``oc_diagnostics`` importable
    in THIS process, then — for one arc — the pre-270 location under the
    worktree's ``oct/oct/tools/``, which is what a checkout cut before the
    move and a fixture tree with no toolchain of its own still carry. With
    the file re-homed, ``oct`` is no longer reached from this edge at all,
    which is what takes ``oct`` out of the dispatcher package's graph.
 
    Unlike the timing edge, this one is a HARD refusal, and it stays that
    way — an unusable venv is worse than a loud failure. The refusal names
    every location that was tried, never just the first.
    """
    func = "sitecustomize_source"
    candidates = [
        worktree / "oc_diagnostics" / "oc_diagnostics" / "venv_sitecustomize.py",
        Path(oc_diagnostics.__file__).resolve().parent / "venv_sitecustomize.py"
        if oc_diagnostics.__file__
        else None,
        worktree / "oct" / "oct" / "tools" / "venv_sitecustomize.py",
    ]
    for candidate in candidates:
        if candidate is not None and candidate.is_file():
            _dbg("OC_CHAT.MAIN", func, f"venv_sitecustomize.py at {candidate}", 3)
            return candidate
    tried = ", ".join(str(candidate) for candidate in candidates if candidate is not None)
    raise WorktreeEnvError(f"cannot locate venv_sitecustomize.py: tried {tried}")
 
 
# ── prepare ────────────────────────────────────────────────────────────────
 
 
@dataclass(frozen=True)
class PrepareRequest:
    """All caller-controlled inputs to one prepare attempt."""
 
    repo_root: Path
    branch: str
    worktree: Path
    base_sha: str | None = None
    uv_executable: str = "uv"
    uv_cache_dir: Path | None = None
    interpreter: Path = field(default_factory=lambda: Path(sys.executable))
    #: Plan 243 P4. Off for the bare CLI command so the 241 witnesses'
    #: expectations of it stay byte-identical; ``dispatch`` passes it.
    smoke: bool = False
 
 
@dataclass(frozen=True)
class PrepareResult:
    """What one prepare attempt actually did."""
 
    worktree: Path
    branch: str
    venv: Path | None
    python: Path | None
    installed: tuple[str, ...]
    verified: tuple[tuple[str, str], ...]
    uv_cache_dir: Path | None
    link_mode: str
    already_prepared: bool
    repaired: tuple[str, ...]
    skipped_reason: str | None
    elapsed_seconds: float
    #: Plan 243 P1: ``(module, __file__)`` for every ``verify_imports``
    #: module the declared extras had to supply.
    verified_extras: tuple[tuple[str, str], ...] = ()
    #: Packages whose own pyproject declares none of the extras the table
    #: names for them, so nothing was installed and nothing was verified.
    extras_not_declared: tuple[str, ...] = ()
    #: Plan 243 P4: the ``--smoke`` checks that ran and passed.
    smoke_checks: tuple[str, ...] = ()
    #: Plan 243 P3: what the stamp of an older prepare disagreed about.
    stale_reason: str | None = None
 
    def report_lines(self) -> list[str]:
        """Operator-facing lines; surfaced verbatim by dispatch and the CLI."""
        if self.skipped_reason is not None:
            return [f"prepare skipped: {self.skipped_reason}"]
        state = "already prepared" if self.already_prepared else "prepared"
        lines = [
            f"Worktree env {state}: {self.worktree}",
            f"  venv:       {self.venv}",
            f"  uv cache:   {self.uv_cache_dir} (link mode: {self.link_mode})",
            f"  installed:  {len(self.installed)} editable package(s)",
            f"  verified:   {len(self.verified)} import(s) inside the worktree",
            f"  extras:     {self._extras_summary()}",
            f"  wall clock: {self.elapsed_seconds:.1f}s",
        ]
        if self.stale_reason is not None:
            lines.append(f"  stale:      {self.stale_reason}")
        if self.smoke_checks:
            lines.append(f"  smoke:      {', '.join(self.smoke_checks)}")
        for repair in self.repaired:
            lines.append(f"  repaired:   {repair}")
        return lines
 
    def _extras_summary(self) -> str:
        """One line naming what the declared extras were asked to supply."""
        summary = f"{len(self.verified_extras)} declared import(s) verified"
        if self.extras_not_declared:
            summary += (
                f" ({len(self.extras_not_declared)} package(s) declare no such "
                f"extra: {', '.join(self.extras_not_declared)})"
            )
        return summary
 
 
def _resolve_uv(executable: str) -> str:
    """Return the resolved ``uv`` path, or refuse by name with the hint."""
    resolved = shutil.which(executable)
    if resolved is None:
        raise WorktreeEnvError(
            f"{executable!r} is not on PATH; per-worktree environments need it — "
            f"{_UV_INSTALL_HINT}"
        )
    return resolved
 
 
def _resolve_cache_dir(request: PrepareRequest, worktree: Path) -> Path:
    """Pin the uv cache to the worktree's drive, or refuse naming both paths."""
    func = "_resolve_cache_dir"
    cache = (
        Path(request.uv_cache_dir).resolve()
        if request.uv_cache_dir is not None
        else (request.repo_root.resolve() / UV_CACHE_DIRNAME)
    )
    if _drive_of(cache) != _drive_of(worktree):
        raise WorktreeEnvError(
            f"uv cache {cache} is on a different drive from the venv it fills "
            f"({worktree}); across drives uv cannot hardlink and silently copies. "
            f"Point --uv-cache-dir at a directory on the venv's drive."
        )
    cache.mkdir(parents=True, exist_ok=True)
    _dbg("OC_CHAT.MAIN", func, f"uv cache at {cache}", 3)
    return cache
 
 
def _uv_env(cache: Path) -> dict[str, str]:
    env = dict(os.environ)
    env["UV_CACHE_DIR"] = str(cache)
    return env
 
 
def _run_uv(
    uv: str, args: list[str], cache: Path, *, context: str
) -> subprocess.CompletedProcess[str]:
    """Run one uv command, refusing with uv's own diagnostic when it fails."""
    func = "_run_uv"
    result = subprocess.run(
        [uv, *args],
        text=True,
        capture_output=True,
        check=False,
        env=_uv_env(cache),
    )
    if result.returncode != 0:
        detail = (result.stderr or result.stdout).strip()
        raise WorktreeEnvError(f"{context} failed: {detail}")
    _dbg("OC_CHAT.MAIN", func, f"{context} ok", 3)
    return result
 
 
def _stamp_payload(
    packages: tuple[EditablePackage, ...], worktree: Path, interpreter: Path
) -> dict[str, object]:
    return {
        "version": STAMP_VERSION,
        "worktree": str(worktree),
        "interpreter": str(interpreter),
        "packages": [package.install_spec(worktree) for package in packages],
    }
 
 
def _read_stamp(venv: Path) -> dict[str, object] | None:
    path = venv / STAMP_FILENAME
    if not path.is_file():
        return None
    try:
        loaded = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return None
    return loaded if isinstance(loaded, dict) else None
 
 
def _link_mode(install_output: str, *, installed: bool) -> str:
    """Report the link strategy uv actually used, never a guess."""
    if not installed:
        return "n/a (nothing installed this run)"
    if _HARDLINK_FALLBACK_MARKER in install_output.casefold():
        return "copy (uv fell back from hardlink)"
    configured = os.environ.get("UV_LINK_MODE")
    return configured if configured else "uv default (hardlink on Windows NTFS)"
 
 
@dataclass(frozen=True)
class _Verification:
    """What one verification run proved, split by what it proves about."""
 
    packages: tuple[tuple[str, str], ...]
    extras: tuple[tuple[str, str], ...]
    not_declared: tuple[str, ...]
 
 
def _clean_env() -> dict[str, str]:
    """The probe environment: no ``PYTHONPATH``, no user site-packages."""
    env = {key: value for key, value in os.environ.items() if key != "PYTHONPATH"}
    env["PYTHONNOUSERSITE"] = "1"
    return env
 
 
def _verify_imports(
    python: Path,
    packages: tuple[EditablePackage, ...],
    worktree: Path,
    venv: Path,
) -> _Verification:
    """Import every package and every declared extra, refusing by name.
 
    Runs with ``PYTHONPATH`` REMOVED and from a neutral temporary directory,
    so a pass means the editable installs alone resolve inside the worktree —
    which is the whole point of the exercise.
 
    Plan 243 P3 adds the second class: the modules the declared ``extras``
    exist to supply. Those live in the venv's ``site-packages`` rather than
    in the checkout, so they are held to importability and venv residence.
    EVERY failure across both classes is named in one refusal — a worker
    who has to re-run prepare once per missing package learns nothing the
    first refusal could not have told them.
    """
    func = "_verify_imports"
    script = (
        # OCT-LINT: disable=no_hardcoded_secrets reason="inline probe SOURCE passed to python -c; mixed case plus \\t/\\n escapes read as high entropy, no credential material"
        "import importlib, sys\n"
        "for name in sys.argv[1:]:\n"
        "    try:\n"
        "        module = importlib.import_module(name)\n"
        "    except Exception as exc:\n"
        "        print(name + '\\tERROR\\t' + type(exc).__name__ + ': ' + str(exc))\n"
        "        continue\n"
        "    print(name + '\\tOK\\t' + str(getattr(module, '__file__', None)))\n"
    )
    names = [package.import_name for package in packages]
    owners = {package.import_name: package.dist_name for package in packages}
    extra_names: list[str] = []
    not_declared: list[str] = []
    for package in packages:
        if not package.verify_imports:
            continue
        if not package.declares_its_extras(worktree):
            not_declared.append(package.install_spec(worktree))
            continue
        for module in package.verify_imports:
            if module not in extra_names and module not in names:
                extra_names.append(module)
                owners[module] = f"{package.dist_name}[{','.join(package.extras)}]"
    with tempfile.TemporaryDirectory(prefix="oc_verify_") as neutral:
        result = subprocess.run(
            [str(python), "-c", script, *names, *extra_names],
            text=True,
            capture_output=True,
            check=False,
            cwd=neutral,
            env=_clean_env(),
        )
    if result.returncode != 0:
        raise WorktreeEnvError(
            f"verification interpreter {python} failed: "
            f"{(result.stderr or result.stdout).strip()}"
        )
 
    resolved = worktree.resolve()
    resolved_venv = venv.resolve()
    verified: list[tuple[str, str]] = []
    verified_extras: list[tuple[str, str]] = []
    problems: list[str] = []
    seen: set[str] = set()
    for line in result.stdout.splitlines():
        if line.count("\t") < 2:
            continue
        name, status, value = line.split("\t", 2)
        seen.add(name)
        owner = owners.get(name, "?")
        is_extra = name in extra_names
        if status != "OK":
            problems.append(f"{name}: import failed ({value}) — declared by {owner}")
            continue
        if value in ("None", ""):
            problems.append(
                f"{name}: resolved as a namespace package (__file__ is None) — "
                f"the sitecustomize.py copy is missing or ineffective"
            )
            continue
        if is_extra:
            if not Path(value).resolve().is_relative_to(resolved_venv):
                problems.append(
                    f"{name}: resolves OUTSIDE the prepared venv at {value} — "
                    f"{owner} did not supply it, something else on the path did"
                )
                continue
            verified_extras.append((name, value))
            continue
        if not Path(value).resolve().is_relative_to(resolved):
            problems.append(f"{name}: resolves OUTSIDE the worktree at {value}")
            continue
        verified.append((name, value))
    problems.extend(
        f"{name}: not reported by the probe"
        for name in [*names, *extra_names]
        if name not in seen
    )
    if problems:
        raise WorktreeEnvError(
            f"prepared venv {python.parent.parent} failed verification:\n  "
            + "\n  ".join(problems)
        )
    _dbg(
        "OC_CHAT.MAIN",
        func,
        f"verified {len(verified)} package + {len(verified_extras)} extra import(s)",
        2,
    )
    return _Verification(
        packages=tuple(verified),
        extras=tuple(verified_extras),
        not_declared=tuple(not_declared),
    )
 
 
def _run_smoke(python: Path, worktree: Path) -> tuple[str, ...]:
    """Run the worker's own first two commands, refusing by name.
 
    Plan 243 P4. Verification proves the modules import in a neutral
    directory; this proves the STEP-0 tool and the suite prerequisites run
    from the worktree the worker will actually sit in. Every check runs so
    one refusal names everything that is wrong.
    """
    func = "_run_smoke"
    problems: list[str] = []
    passed: list[str] = []
    for label, argv in SMOKE_CHECKS:
        result = subprocess.run(
            [str(python), *argv],
            text=True,
            capture_output=True,
            check=False,
            cwd=str(worktree),
            env=_clean_env(),
        )
        rendered = " ".join(argv)
        if result.returncode == 0:
            passed.append(label)
            continue
        detail = (result.stderr or result.stdout).strip().splitlines()
        tail = detail[-1] if detail else "(no output)"
        problems.append(
            f"{label}: `python {rendered}` exited {result.returncode} — {tail}"
        )
    if problems:
        raise WorktreeEnvError(
            f"prepared venv {python.parent.parent} failed the --smoke checks "
            f"in {worktree}:\n  " + "\n  ".join(problems)
        )
    _dbg("OC_CHAT.MAIN", func, f"{len(passed)} smoke check(s) passed", 2)
    return tuple(passed)
 
 
def _describe_stale(previous: dict[str, object], current: dict[str, object]) -> str:
    """Say what an older prepare's stamp disagrees with the table about.
 
    Plan 243 P3. A table change (an added extra, a new package) must make
    an already-prepared worktree report itself stale rather than silently
    reinstall, so an operator reading the summary learns WHY the run was
    not the no-op they expected.
    """
    old = [str(spec) for spec in _stamp_specs(previous)]
    new = [str(spec) for spec in _stamp_specs(current)]
    parts: list[str] = []
    added = [spec for spec in new if spec not in old]
    removed = [spec for spec in old if spec not in new]
    if added:
        parts.append("now installs " + ", ".join(added))
    if removed:
        parts.append("no longer installs " + ", ".join(removed))
    if previous.get("interpreter") != current.get("interpreter"):
        parts.append(
            f"interpreter changed from {previous.get('interpreter')} to "
            f"{current.get('interpreter')}"
        )
    if not parts:
        parts.append("the recorded stamp differs from the declared table")
    return "declared spec list changed since this venv was prepared: " + "; ".join(parts)
 
 
def _stamp_specs(stamp: dict[str, object]) -> list[object]:
    packages = stamp.get("packages")
    return list(packages) if isinstance(packages, list) else []
 
 
def _engine_timing(
    repo_root: Path, stage: str, plan: str, *, oct_hook: OctHook | None = None
):
    """Load optional timing support without making oct an engine dependency."""
    from contextlib import nullcontext
    import re
    func = "_engine_timing"
    try:
        timed_stage = (oct_hook or _oct_hook)("timed_stage")
    except ImportError:
        _dbg("OC_CHAT.MAIN", func, "timing unavailable without oct", 3)
        return nullcontext(None)
    match = re.search(r"(?:^|[/_-])(\d+)(?:$|[/_-])", plan)
    return timed_stage(repo_root, stage, scope=match.group(1) if match else None,
                       files=0 if stage == "engine.prepare" else None)
 
 
def prepare_worktree(request: PrepareRequest) -> PrepareResult:
    """Create the branch, worktree and its own uv venv, then verify it."""
    func = "prepare_worktree"
    with _engine_timing(request.repo_root, "engine.prepare", request.branch) as timing:
        started = time.monotonic()
        repo_root = Path(request.repo_root).resolve()
        worktree = Path(request.worktree).resolve()
        _dbg("OC_CHAT.MAIN", func, f"prepare {request.branch} at {worktree}", 2)
 
        ensure_branch_and_worktree(
            repo_root, request.branch, worktree, base_sha=request.base_sha
        )
 
        packages = discover_packages(worktree)
        if not packages:
            reason = (
                f"no declared Option C package roots exist under {worktree}; "
                f"there is nothing to install and no editable shadow to close"
            )
            _dbg("OC_CHAT.MAIN", func, reason, 2)
            return PrepareResult(
                worktree=worktree,
                branch=request.branch,
                venv=None,
                python=None,
                installed=(),
                verified=(),
                uv_cache_dir=None,
                link_mode="n/a",
                already_prepared=False,
                repaired=(),
                skipped_reason=reason,
                elapsed_seconds=time.monotonic() - started,
            )
 
        uv = _resolve_uv(request.uv_executable)
        cache = _resolve_cache_dir(request, worktree)
        venv = worktree / ".venv"
        python = venv_python(venv)
 
        if not python.is_file():
            _run_uv(
                uv,
                ["venv", "--python", str(request.interpreter), str(venv)],
                cache,
                context=f"uv venv at {venv}",
            )
            python = venv_python(venv)
            if not python.is_file():
                raise WorktreeEnvError(f"uv venv produced no interpreter at {python}")
 
        stamp = _stamp_payload(packages, worktree, request.interpreter)
        previous = _read_stamp(venv)
        already = previous == stamp
        stale_reason = None if already or previous is None else _describe_stale(previous, stamp)
        if stale_reason is not None:
            _dbg("OC_CHAT.MAIN", func, f"stale stamp: {stale_reason}", 2)
        installed: tuple[str, ...] = ()
        install_output = ""
        if not already:
            specs = [package.install_spec(worktree) for package in packages]
            result = _run_uv(
                uv,
                ["pip", "install", "--python", str(python), *_editable(specs)],
                cache,
                context=f"uv pip install -e into {venv}",
            )
            install_output = f"{result.stdout}\n{result.stderr}"
            installed = tuple(package.dist_name for package in packages)
            if timing is not None:
                timing.files = len(installed)
            (venv / STAMP_FILENAME).write_text(
                json.dumps(stamp, indent=2, sort_keys=True), encoding="utf-8"
            )
 
        repaired = _ensure_sitecustomize(venv, worktree)
        try:
            verification = _verify_imports(python, packages, worktree, venv)
            smoke_checks = _run_smoke(python, worktree) if request.smoke else ()
        except WorktreeEnvError:
            # The stamp claims "prepared" and verification says otherwise — the
            # state a changed DECLARATION leaves behind, since the stamp records
            # the resolved spec list and an edited pyproject does not change it.
            # Leaving it would make every re-run skip the install and fail
            # identically, so the only way out would be deleting it by hand: the
            # hand repair this module exists to end. Clearing it makes the next
            # prepare reinstall and try again.
            (venv / STAMP_FILENAME).unlink(missing_ok=True)
            raise
        elapsed = time.monotonic() - started
        _dbg(
            "OC_CHAT.MAIN",
            func,
            f"prepared {worktree} in {elapsed:.1f}s "
            f"({len(verification.packages)} verified)",
            2,
        )
        return PrepareResult(
            worktree=worktree,
            branch=request.branch,
            venv=venv,
            python=python,
            installed=installed,
            verified=verification.packages,
            uv_cache_dir=cache,
            link_mode=_link_mode(install_output, installed=bool(installed)),
            already_prepared=already,
            repaired=repaired,
            skipped_reason=None,
            elapsed_seconds=elapsed,
            verified_extras=verification.extras,
            extras_not_declared=verification.not_declared,
            smoke_checks=smoke_checks,
            stale_reason=stale_reason,
        )
 
 
def _editable(specs: list[str]) -> list[str]:
    """Interleave ``-e`` before every install spec."""
    argv: list[str] = []
    for spec in specs:
        argv.extend(["-e", spec])
    return argv
 
 
def _ensure_sitecustomize(venv: Path, worktree: Path) -> tuple[str, ...]:
    """Copy ``venv_sitecustomize.py`` in, reporting a repair when it was gone."""
    func = "_ensure_sitecustomize"
    source = sitecustomize_source(worktree)
    target = venv_site_packages(venv) / "sitecustomize.py"
    existed = target.is_file()
    if existed and target.read_bytes() == source.read_bytes():
        return ()
    shutil.copyfile(source, target)
    verb = "refreshed" if existed else "installed"
    note = f"sitecustomize.py {verb} at {target}"
    _dbg("OC_CHAT.MAIN", func, note, 3)
    return (note,)
 
 
# ── teardown ───────────────────────────────────────────────────────────────
 
 
@dataclass(frozen=True)
class TeardownResult:
    """What one teardown attempt actually did."""
 
    worktree: Path
    removed: bool
    messages: tuple[str, ...]
 
    def report_lines(self) -> list[str]:
        """Operator-facing lines; surfaced verbatim by the CLI."""
        head = (
            f"Removed worktree {self.worktree}"
            if self.removed
            else f"Worktree {self.worktree} was NOT removed"
        )
        return [head, *(f"  {message}" for message in self.messages)]
 
 
def teardown_worktree(
    repo_root: Path, worktree: Path, *, branch: str | None = None
) -> TeardownResult:
    """Remove a prepared worktree and its venv, naming any residue by path."""
    func = "teardown_worktree"
    with _engine_timing(repo_root, "engine.teardown", branch or worktree.name) as timing:
        repo_root = Path(repo_root).resolve()
        worktree = Path(worktree).resolve()
        metadata = repo_root / ".git" / "worktrees" / worktree.name
        registered = any(
            _same_path(path, worktree) for path, _branch in worktree_entries(repo_root)
        )
        if not registered:
            if worktree.exists():
                raise WorktreeEnvError(
                    f"{worktree} exists but is not a worktree of {repo_root}; "
                    f"refusing to delete a directory git does not own"
                )
            _dbg("OC_CHAT.MAIN", func, f"{worktree} already gone", 2)
            return TeardownResult(
                worktree=worktree,
                removed=True,
                messages=("already removed (not registered, nothing on disk)",),
            )
 
        removal = _run_git(
            repo_root, "worktree", "remove", "--force", str(worktree), check=False
        )
        if removal.returncode != 0:
            detail = (removal.stderr or removal.stdout).strip()
            _dbg("OC_CHAT.MAIN", func, f"remove refused: {detail}", 2)
            if timing is not None:
                timing.outcome = "error"
            return TeardownResult(
                worktree=worktree,
                removed=False,
                messages=(
                    f"git worktree remove failed: {detail}",
                    f"the known Windows failure leaves {metadata} behind; nothing "
                    f"else was deleted. Remove that directory by hand, then run "
                    f"`git worktree prune` in {repo_root}.",
                ),
            )
 
        _run_git(repo_root, "worktree", "prune", check=False)
        if branch is not None and _branch_exists(repo_root, branch):
            _dbg("OC_CHAT.MAIN", func, f"branch {branch} left in place", 3)
        if worktree.exists():
            if timing is not None:
                timing.outcome = "error"
            return TeardownResult(
                worktree=worktree,
                removed=False,
                messages=(
                    f"git reported success but {worktree} is still on disk; "
                    f"a handle is open on it (an editor, a shell, or a running "
                    f"interpreter from {worktree / '.venv'}).",
                ),
            )
        _dbg("OC_CHAT.MAIN", func, f"removed {worktree}", 2)
        return TeardownResult(
            worktree=worktree,
            removed=True,
            messages=(f"venv and checkout deleted; {metadata} pruned",),
        )
 
 
__all__ = [
    "EDITABLE_PACKAGES",
    "EditablePackage",
    "PrepareRequest",
    "PrepareResult",
    "SMOKE_CHECKS",
    "STAMP_FILENAME",
    "TeardownResult",
    "OCT_HOOK_TARGETS",
    "OctHook",
    "UV_CACHE_DIRNAME",
    "WorktreeEnvError",
    "discover_packages",
    "ensure_branch_and_worktree",
    "prepare_worktree",
    "sitecustomize_source",
    "teardown_worktree",
    "use_oct_hook",
    "venv_python",
    "venv_scripts_dir",
    "venv_site_packages",
    "worktree_entries",
]