dependency_healer.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/core/dependency_healer.py
 
"""
Self-healing dependency handler.
 
Purpose
-------
Intercept ``missing_dependency`` errors from tool calls and present an
interactive menu so the user can install the missing package or skip.
Lets the agent recover from optional-dep failures (e.g. a tool needs
``httpx`` but it isn't installed) without restarting the REPL. (The
legacy "custom command" ``shell=True`` option was removed in M8.0 WU-3d.)
 
Responsibilities
----------------
- :func:`detect_missing_dependency` — recognise the structured
  ``{"missing_dependency": ...}`` shape returned by tools.
- :func:`show_dependency_menu` — render the install / skip / custom
  prompt and read the user's choice.
- :func:`handle_dependency_issue` — orchestrate detection + menu +
  optional ``pip install`` retry of the failing tool.
 
Usage from the ReAct loop (``main.py``)::
 
    from oc_chat.core.dependency_healer import handle_dependency_issue
    result = handle_dependency_issue(result, presenter, tool_registry, tc)
 
Diagnostics
-----------
Domain: OC_CHAT.TOOLS
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- ``detect_missing_dependency`` returns ``None`` when the result does
  not match the structured shape.
- The handler never raises on a failed install — it surfaces the
  error to the user and falls back to ``skip``.
- ``pip install`` runs in a child process via :mod:`subprocess`; the
  REPL's environment is not modified outside that call.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core._paths (lazy; workspace-root fallback when no manifest sits beside the package)
- oc_ui.presenter_base (lazy; Presenter type for the install menu and status messages)
- pip (not imported; python -m pip install of the missing package in a child process)
"""
from __future__ import annotations
 
import json
import subprocess
import sys
from pathlib import Path
from typing import TYPE_CHECKING
 
from oc_diagnostics import _dbg
 
if TYPE_CHECKING:
    from oc_ui.presenter_base import Presenter
 
# ── Detection ───────────────────────────────────────────────────────────────
 
_MISSING_DEP_KEY = "missing_dependency"
 
 
def detect_missing_dependency(result: dict) -> dict | None:
    """If *result* contains a structured missing-dependency error, return the
    error sub-dict, otherwise return ``None``."""
    if isinstance(result, dict) and _MISSING_DEP_KEY in result:
        return {
            "package": result[_MISSING_DEP_KEY],
            "pip_install": result.get("pip_install", result[_MISSING_DEP_KEY]),
            "optional_group": result.get("optional_group"),
            "error": result.get("error", "Missing dependency."),
        }
    return None
 
 
# ── Interactive Menu ────────────────────────────────────────────────────────
 
def show_dependency_menu(presenter: Presenter, dep_info: dict) -> str:
    """Show an interactive menu and return the user's choice.
 
    Returns ``"install"`` or ``"continue"`` (M8.0 WU-3d removed the custom-command
    ``shell=True`` option per LD-4).
    """
    pkg = dep_info["package"]
    group = dep_info["optional_group"]
 
    print(f"\n{'─' * 60}")
    print(f"  ⚠  Missing Dependency: {pkg}")
    print(f"{'─' * 60}")
    if group:
        print(f"  The {group} feature requires '{pkg}'.")
    else:
        print(f"  The current tool requires '{pkg}'.")
    print(f"\n  Options:")
    print(f"    1) Install '{pkg}' via pip + update project files")
    print(f"    2) Continue without (report error to model)")
    print(f"{'─' * 60}")
 
    while True:
        try:
            choice = input("  Your choice (1-2) [1]: ").strip()
        except (EOFError, KeyboardInterrupt):
            return "continue"
 
        if not choice:
            return "install"
        if choice == "1":
            return "install"
        if choice == "2":
            return "continue"
        # The legacy "3) custom command" option (subprocess shell=True) was
        # removed in M8.0 WU-3d (LD-4: no shell=True path).
        print(f"  Invalid choice '{choice}'. Enter 1 or 2.")
 
 
# ── Installation ────────────────────────────────────────────────────────────
 
def _project_root() -> Path:
    """Locate the OC Chat workspace root that holds ``requirements*.txt``.
 
    Defers to :func:`oc_chat.core._paths.find_workspace_root` so the
    lookup is install-survivable: it walks up from this module's path
    rather than depending on the caller's CWD, and skips past the
    per-package ``oc_chat/pyproject.toml`` (D.1c) to the repo root that
    actually holds the manifest files.
 
    Falls back to ``Path.cwd()`` only if the walk fails (e.g. running
    from a wheel install with no source tree present), in which case
    manifest updates simply skip files that don't exist.
 
    Plan 151 T2 step 11(c) — BOUNDED subproject-root resolution first (the
    session_log V23 pattern): post-monorepo-migration ``find_workspace_root()``
    returns the OUTERMOST ``.git`` (the option_c monorepo root), which holds
    NO ``requirements*.txt``/``pyproject.toml`` — the healer would silently
    degrade. The manifests live at the oc_chat SUBPROJECT root (the parent of
    the ``oc_chat`` package dir); one adjacency check, no walk, then the old
    resolver as fallback.
    """
    pkg_parent = Path(__file__).resolve().parent.parent.parent
    if (pkg_parent / "requirements.txt").is_file() or (
        pkg_parent / "pyproject.toml"
    ).is_file():
        return pkg_parent
    from oc_chat.core._paths import find_workspace_root
    try:
        return find_workspace_root()
    except FileNotFoundError:
        return Path.cwd().resolve()
 
 
_PROJECT_ROOT = _project_root()
 
 
def install_dependency(dep_info: dict) -> bool:
    """Install the missing package via pip. Returns ``True`` on success."""
    package = dep_info["pip_install"]
    print(f"  Installing '{package}'...")
    try:
        subprocess.run(
            [sys.executable, "-m", "pip", "install", package],
            check=True,
            capture_output=False,
        )
        print(f"  ✓ Successfully installed '{package}'.")
        return True
    except subprocess.CalledProcessError as exc:
        print(f"  ✗ pip install failed (return code {exc.returncode}).")
        return False
 
 
# ── Manifest Updates ────────────────────────────────────────────────────────
 
def update_manifest_files(dep_info: dict) -> dict:
    """Add the missing package to project dependency manifests where missing.
 
    Returns ``{"updated": [filenames...], "errors": [...]}``.
    """
    pkg = dep_info["package"]
    group = dep_info.get("optional_group")
    updated = []
    errors = []
 
    # 1. pyproject.toml — ensure package is in optional-dependencies
    pyproject = _PROJECT_ROOT / "pyproject.toml"
    if pyproject.exists():
        try:
            content = pyproject.read_text(encoding="utf-8")
            if pkg not in content:
                if group:
                    marker = f"{group} = ["
                    if marker in content:
                        lines = content.split("\n")
                        new_lines = []
                        added = False
                        for i, line in enumerate(lines):
                            new_lines.append(line)
                            if not added and marker in line:
                                next_line = lines[i + 1] if i + 1 < len(lines) else ""
                                if pkg not in "".join(lines[i:]):
                                    new_lines.append(f'    "{pkg}",')
                                added = True
                        if not added:
                            # Group not found; append to optional-dependencies
                            new_lines.append(f'{group} = [')
                            new_lines.append(f'    "{pkg}",')
                            new_lines.append(']')
                        content = "\n".join(new_lines)
                    else:
                        content += f'\n{group} = [\n    "{pkg}",\n]\n'
                pyproject.write_text(content, encoding="utf-8")
                updated.append("pyproject.toml")
        except Exception as e:
            errors.append(f"pyproject.toml: {e}")
 
    # 2. requirements-all.txt — ensure package is present (uncommented)
    req_all = _PROJECT_ROOT / "requirements-all.txt"
    if req_all.exists():
        try:
            lines = req_all.read_text(encoding="utf-8").split("\n")
            changed = False
            found = False
            new_lines = []
            for line in lines:
                stripped = line.strip()
                if stripped.lstrip("#").strip() == pkg and stripped.startswith("#"):
                    new_lines.append(pkg)
                    changed = True
                    found = True
                else:
                    new_lines.append(line)
                if stripped == pkg:
                    found = True
            if not found:
                new_lines.append(pkg)
                changed = True
            if changed:
                req_all.write_text("\n".join(new_lines) + "\n", encoding="utf-8")
                updated.append("requirements-all.txt")
        except Exception as e:
            errors.append(f"requirements-all.txt: {e}")
 
    # 3. requirements.txt — ensure package is present (uncommented)
    req_txt = _PROJECT_ROOT / "requirements.txt"
    if req_txt.exists():
        try:
            lines = req_txt.read_text(encoding="utf-8").split("\n")
            changed = False
            found = False
            new_lines = []
            for line in lines:
                stripped = line.strip()
                if stripped.lstrip("#").strip() == pkg and stripped.startswith("#"):
                    new_lines.append(f"{pkg}  # {group or 'optional'}")
                    changed = True
                    found = True
                else:
                    new_lines.append(line)
                if stripped == pkg or stripped.startswith(f"{pkg} #") or stripped.startswith(f"{pkg}  #"):
                    found = True
            if not found:
                new_lines.append(f"{pkg}  # {group or 'optional'}")
                changed = True
            if changed:
                req_txt.write_text("\n".join(new_lines) + "\n", encoding="utf-8")
                updated.append("requirements.txt")
        except Exception as e:
            errors.append(f"requirements.txt: {e}")
 
    return {"updated": updated, "errors": errors}
 
 
# ── Orchestrator ────────────────────────────────────────────────────────────
 
def handle_dependency_issue(result: dict, presenter: Presenter, tool_registry: dict | None = None, tc: dict | None = None) -> dict:
    """Check if *result* reports a missing dependency and, if so, offer an
    interactive repair menu.
 
    Args:
        result: The raw tool result dict.
        presenter: A :class:`~oc_ui.presenter_base.Presenter` instance.
        tool_registry: The ``{name: callable}`` registry (needed for re-execution).
        tc: The original tool-call dict (needed for re-execution).
 
    Returns:
        The (possibly re-executed) tool result.
    """
    func = "handle_dependency_issue"
    dep_info = detect_missing_dependency(result)
    if dep_info is None:
        return result  # No dependency issue
    _dbg("OC_CHAT.TOOLS", func,
         f"missing={dep_info.get('package')} tc={tc.get('name') if tc else '-'}", 2)
 
    # Show the dep info via the presenter
    presenter.show_info(f"⚠ Missing dependency: {dep_info['error']}")
 
    choice = show_dependency_menu(presenter, dep_info)
 
    if choice == "continue":
        return result  # Return original error to the model
 
    if choice == "install":
        ok = install_dependency(dep_info)
        if not ok:
            presenter.show_error("Installation failed. Continuing with original error.")
            return result
        update_result = update_manifest_files(dep_info)
        if update_result.get("updated"):
            presenter.show_info(f"Updated: {', '.join(update_result['updated'])}")
        if update_result.get("errors"):
            presenter.show_error(f"Manifest update warnings: {', '.join(update_result['errors'])}")
 
        # Re-execute the tool if we have the registry and tool-call info
        if tool_registry is not None and tc is not None:
            tool_name = tc["name"]
            tool_fn = tool_registry.get(tool_name)
            if tool_fn:
                presenter.show_info(f"Re-executing '{tool_name}'...")
                try:
                    args = json.loads(tc["arguments"] or "{}")
                    if isinstance(args, dict):
                        return tool_fn(**args)
                    else:
                        return tool_fn(args)
                except Exception as e:
                    return {"error": f"Tool re-execution failed after dependency install: {e}"}
        return {"error": "Dependency installed. Please retry the operation."}
 
    # M8.0 WU-3d (LD-4): the legacy "custom command" option ran
    # subprocess.run(choice, shell=True) -- an arbitrary-shell ACE path. There is
    # no shell=True lane anywhere in M8; the menu now offers only install/continue,
    # so any other choice is treated as continue (never executed).
    return result