oct_install.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/install/oct_install.py
 
"""
Purpose
-------
Wrap ``pip install -e <target>`` with safety preflight checks that
catch the most common operator-error patterns: running the install
from the wrong terminal window with the wrong venv active, in the
wrong CWD, or against a worktree that would shadow a sibling worktree.
 
Responsibilities
----------------
- Define the :func:`oct_install_cmd` Click command, mounted on the
  top-level ``cli`` group.
- Resolve and validate the install target (an Option C project root).
- Verify the active venv matches the target's expected venv.
- Detect existing-shadow installs from another worktree.
- Refuse to install from inside a Claude Code worktree without an
  explicit override.
- Delegate to ``pip install -e <target>`` via :mod:`subprocess` after
  preflights succeed; print post-install diagnostics so the user
  sees exactly which ``oct`` binary will be invoked next.
 
Diagnostics
-----------
Domain: INSTALL
Levels:
    L1 - errors (pip failure, preflight refusal)
    L2 - lifecycle (preflight start/end, install start/end)
    L3 - details (venv path, target path, shadow location)
    L4 - deep trace (each preflight check)
 
Contracts
---------
- All subprocess calls use ``shell=False`` and explicit argv lists.
- No filesystem writes are performed by this command itself; only
  ``pip`` (the delegated subprocess) writes to the venv's
  ``site-packages``.
- Audit trail via the existing :func:`oct.git.audit.audited` decorator
  is intentionally NOT applied here, because the audit module emits
  to ``logs/git-audit-*.jsonl`` which lives under the project root -
  but ``oct install`` may run before any project root exists. A
  lightweight ``_dbg`` trace is kept instead.
 
Dependencies
------------
- oct.core.diagnostics (_dbg structured logger)
- click (CLI framework)
"""
 
from __future__ import annotations
 
import json
import subprocess
import sys
import tomllib
from pathlib import Path
 
import click
 
from oct.core.diagnostics import _dbg
 
 
# ---------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------
 
#: Subdirectories under a project root that are accepted as the local
#: venv. Both names are common; the user's `oct health` Venv-First Rule
#: documents `.venv`, but pre-existing setups may use `venv`.
_LOCAL_VENV_NAMES: tuple[str, ...] = (".venv", "venv")
 
#: Default timeout for the pip subprocess, in seconds. Generous so a
#: cold-cache resolution doesn't trip.
_PIP_TIMEOUT: int = 600
 
 
# ---------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------
 
 
def _is_option_c_project(target: Path) -> bool:
    """Return True if *target* looks like an Option C project root.
 
    Accept both legacy (``.octrc.json`` at root) and migrated
    (``.option_c/`` directory) layouts. ``pyproject.toml`` must also
    be present so ``pip install -e`` has something to install.
    """
    if not (target / "pyproject.toml").is_file():
        return False
    return (target / ".octrc.json").is_file() \
        or (target / ".option_c").is_dir()
 
 
def _read_project_name(target: Path) -> str | None:
    """Return the ``[project] name`` from *target*'s pyproject.toml."""
    func = "_read_project_name"
    pyproject = target / "pyproject.toml"
    if not pyproject.is_file():
        return None
    try:
        with pyproject.open("rb") as fh:
            data = tomllib.load(fh)
    except (OSError, tomllib.TOMLDecodeError) as exc:
        _dbg("INSTALL", func, f"pyproject parse failed: {exc}", 1, override=True)
        return None
    project = data.get("project")
    if not isinstance(project, dict):
        return None
    name = project.get("name")
    return name if isinstance(name, str) else None
 
 
def _active_venv_root() -> Path | None:
    """Return the root of the currently active venv, or ``None``.
 
    ``sys.prefix`` is the venv root for an active virtual environment,
    ``sys.base_prefix`` is the system Python. When they differ, a venv
    is active. If they match, no venv is active.
    """
    if sys.prefix != sys.base_prefix:
        return Path(sys.prefix).resolve()
    return None
 
 
def _expected_venv_under(target: Path) -> Path | None:
    """Return the first existing local venv directory under *target*."""
    target_resolved = target.resolve()
    for name in _LOCAL_VENV_NAMES:
        candidate = target_resolved / name
        if candidate.is_dir():
            return candidate
    return None
 
 
def _is_under(child: Path, parent: Path) -> bool:
    """Return True if *child* is at or below *parent*."""
    try:
        child.resolve().relative_to(parent.resolve())
        return True
    except ValueError:
        return False
 
 
def _pip_show_location(pkg: str, venv: Path | None) -> Path | None:
    """Return the ``Location:`` from ``pip show <pkg>`` or ``None``.
 
    Runs ``<venv>/Scripts/python -m pip show <pkg>`` (Windows) or
    ``<venv>/bin/python -m pip show <pkg>`` (Unix) when *venv* is
    given; falls back to ``sys.executable`` otherwise.
    """
    func = "_pip_show_location"
    python = _venv_python(venv) if venv is not None else Path(sys.executable)
    try:
        proc = subprocess.run(
            [str(python), "-m", "pip", "show", pkg],
            shell=False, capture_output=True, text=True, timeout=30,
        )
    except (subprocess.TimeoutExpired, FileNotFoundError) as exc:
        _dbg("INSTALL", func, f"pip show failed: {exc}", 1, override=True)
        return None
    if proc.returncode != 0:
        return None
    for line in proc.stdout.splitlines():
        if line.startswith("Location:"):
            loc = line.split(":", 1)[1].strip()
            if loc:
                return Path(loc).resolve()
    return None
 
 
def _venv_python(venv: Path) -> Path:
    """Return the python executable path inside *venv*."""
    if (venv / "Scripts" / "python.exe").is_file():
        return (venv / "Scripts" / "python.exe").resolve()
    if (venv / "bin" / "python").is_file():
        return (venv / "bin" / "python").resolve()
    return Path(sys.executable).resolve()
 
 
def _which_oct(venv: Path | None) -> Path | None:
    """Return the path to the ``oct`` script, preferring *venv*'s bin."""
    if venv is not None:
        for sub in ("Scripts/oct.exe", "Scripts/oct", "bin/oct"):
            candidate = venv / sub
            if candidate.is_file():
                return candidate.resolve()
    # Fallback: scan PATH.
    import shutil as _sh
    found = _sh.which("oct")
    return Path(found).resolve() if found else None
 
 
def _is_inside_worktree(target: Path) -> bool:
    """Return True if *target* lives under any ``.claude/worktrees/`` ancestor."""
    target_resolved = target.resolve()
    for parent in [target_resolved, *target_resolved.parents]:
        if parent.name == "worktrees" and parent.parent.name == ".claude":
            return True
    return False
 
 
# ---------------------------------------------------------------------
# Click command
# ---------------------------------------------------------------------
 
 
@click.command(name="install")
@click.argument(
    "target", type=click.Path(file_okay=False, dir_okay=True),
    default=".",
)
@click.option(
    "--editable", "-e", "editable", is_flag=True,
    help="Install in editable mode (pip install -e). Required for now.",
)
@click.option(
    "--reinstall", is_flag=True,
    help="Allow re-install over an existing shadow install.",
)
@click.option(
    "--use-active-venv", is_flag=True,
    help="Bypass the venv-match check and use whichever venv is active.",
)
@click.option(
    "--allow-worktree", is_flag=True,
    help="Allow installing from inside a Claude Code worktree.",
)
@click.option(
    "--allow-outside-project", is_flag=True,
    help="Allow running outside a recognised Option C project.",
)
@click.option(
    "--json", "json_mode", is_flag=True, help="Emit JSON output.",
)
@click.pass_context
def oct_install_cmd(
    ctx: click.Context,
    target: str,
    editable: bool,
    reinstall: bool,
    use_active_venv: bool,
    allow_worktree: bool,
    allow_outside_project: bool,
    json_mode: bool,
) -> None:
    """Install an Option C project as an editable package, with safety checks.
 
    Wraps ``pip install -e <target>`` after verifying:
 
    1. The target is an Option C project (pyproject.toml + .octrc.json
       or .option_c/).
    2. The active venv matches the target's expected venv (refuse with
       a precise diagnostic otherwise).
    3. No existing shadow install from a different worktree is about
       to be silently overwritten.
    4. The target is not inside a Claude Code worktree (refusable
       with --allow-worktree).
    """
    func = "oct_install_cmd"
    _dbg(
        "INSTALL", func,
        f"target={target} editable={editable} reinstall={reinstall} "
        f"use_active_venv={use_active_venv} allow_worktree={allow_worktree}",
        2,
    )
 
    if not editable:
        click.echo(
            "Error: --editable / -e is required. Non-editable installs "
            "are out of scope for `oct install`.",
            err=True,
        )
        ctx.exit(1)
        return
 
    target_path = Path(target).resolve()
    refusals: list[str] = []
    warnings: list[str] = []
 
    # -- Preflight 1: target is an Option C project --------------------
    if not _is_option_c_project(target_path):
        refusals.append(
            f"target {target_path} is not an Option C project "
            f"(missing pyproject.toml + (.octrc.json | .option_c/))"
        )
 
    # -- Preflight 2: not inside a Claude Code worktree (unless override) --
    if _is_inside_worktree(target_path) and not allow_worktree:
        refusals.append(
            f"target {target_path} is inside a Claude Code worktree "
            f"(.claude/worktrees/...). Pass --allow-worktree to override."
        )
 
    # -- Preflight 3: CWD is also an Option C project (escape hatch) ---
    cwd = Path.cwd().resolve()
    if not allow_outside_project and not _is_option_c_project(cwd):
        # Allow the target to be the CWD even if cwd lacks the markers
        # outright - we already validate the target above.
        if cwd != target_path:
            warnings.append(
                f"CWD {cwd} is not an Option C project root. "
                f"Continuing because target {target_path} is."
            )
 
    # -- Preflight 4: active venv matches expected venv ----------------
    expected_venv = _expected_venv_under(target_path)
    active_venv = _active_venv_root()
 
    if not use_active_venv:
        if expected_venv is None:
            warnings.append(
                f"target has no local venv at {target_path}/.venv "
                f"or {target_path}/venv; skipping venv-match check"
            )
        elif active_venv is None:
            refusals.append(
                f"no active venv detected; project expects {expected_venv}. "
                f"Activate the venv (.venv\\Scripts\\activate or "
                f"source .venv/bin/activate), or pass --use-active-venv."
            )
        else:
            try:
                same = active_venv.samefile(expected_venv)
            except OSError:
                same = active_venv == expected_venv
            if not same:
                refusals.append(
                    f"active venv is at {active_venv}; project expects "
                    f"{expected_venv}. Activate the right venv, or pass "
                    f"--use-active-venv to override."
                )
 
    # -- Preflight 5: existing-shadow detection ------------------------
    pkg_name = _read_project_name(target_path)
    shadow_loc: Path | None = None
    if not refusals and pkg_name:
        shadow_loc = _pip_show_location(
            pkg_name,
            active_venv if use_active_venv or expected_venv is None
            else expected_venv,
        )
        if shadow_loc is not None:
            # Compare against the target's package directory. For a
            # standard layout that's <target>/<pkg_name>/.
            expected_loc = target_path
            if not _is_under(shadow_loc, expected_loc):
                if not reinstall:
                    refusals.append(
                        f"package '{pkg_name}' is already installed "
                        f"with Location={shadow_loc}, which is NOT under "
                        f"the install target {target_path}. This would "
                        f"silently shadow another worktree's install. "
                        f"Pass --reinstall to overwrite, or activate the "
                        f"right venv."
                    )
                else:
                    warnings.append(
                        f"overwriting existing install at {shadow_loc} "
                        f"(--reinstall acknowledged)"
                    )
 
    # -- Render preflight result ---------------------------------------
    if refusals:
        if json_mode:
            click.echo(json.dumps({
                "operation": "install",
                "target": str(target_path),
                "exit_code": 1,
                "errors": refusals,
                "warnings": warnings,
            }, indent=2))
        else:
            click.echo("oct install preflight FAILED:", err=True)
            for r in refusals:
                click.echo(f"  - {r}", err=True)
            for w in warnings:
                click.echo(f"  ! {w}", err=True)
        ctx.exit(1)
        return
 
    if warnings and not json_mode:
        for w in warnings:
            click.echo(f"Warning: {w}", err=True)
 
    # -- Execute pip install -e ----------------------------------------
    pip_python = _venv_python(active_venv) if active_venv else Path(sys.executable)
    pip_argv = [str(pip_python), "-m", "pip", "install", "-e", str(target_path)]
    _dbg("INSTALL", func, f"running: {pip_argv}", 2)
 
    try:
        proc = subprocess.run(
            pip_argv, shell=False, timeout=_PIP_TIMEOUT,
            text=True, capture_output=json_mode,
        )
    except subprocess.TimeoutExpired:
        click.echo(
            f"pip install timed out after {_PIP_TIMEOUT}s.",
            err=True,
        )
        ctx.exit(2)
        return
    except FileNotFoundError as exc:
        click.echo(f"pip executable not found: {exc}", err=True)
        ctx.exit(2)
        return
 
    if proc.returncode != 0:
        if json_mode:
            click.echo(json.dumps({
                "operation": "install",
                "target": str(target_path),
                "exit_code": 2,
                "errors": [f"pip install failed (exit {proc.returncode})"],
                "stdout": proc.stdout, "stderr": proc.stderr,
            }, indent=2))
        else:
            click.echo(
                f"pip install failed (exit {proc.returncode}).",
                err=True,
            )
        ctx.exit(2)
        return
 
    # -- Post-install verification -------------------------------------
    new_loc = _pip_show_location(pkg_name, active_venv) if pkg_name else None
    oct_path = _which_oct(active_venv)
 
    if json_mode:
        click.echo(json.dumps({
            "operation": "install",
            "target": str(target_path),
            "package": pkg_name,
            "venv": str(active_venv) if active_venv else None,
            "installed_location": str(new_loc) if new_loc else None,
            "oct_binary": str(oct_path) if oct_path else None,
            "exit_code": 0,
            "warnings": warnings,
        }, indent=2))
    else:
        click.echo(f"\noct install: {pkg_name or '<pkg>'} installed.")
        if new_loc:
            click.echo(f"  Location: {new_loc}")
        if oct_path:
            click.echo(f"  oct binary: {oct_path}")
        if active_venv:
            click.echo(f"  venv: {active_venv}")