discovery.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/discovery.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_storage_foundation.py"
"""
Registry location discovery.
 
Purpose
-------
Resolve the SQLite registry location in the order specified by the
orchestration registry design-lock document.
 
Responsibilities
----------------
- Honor ``OC_AGENT_REGISTRY`` when it points at an absolute path.
- Detect ``.option_c_workspace.json`` and derive the shared registry path.
- Fall back through git common-dir and project-root locations.
- Return workspace metadata needed by the init seed path.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events
    L3 - semantic details
    L4 - deep tracing
 
Contracts
---------
- Discovery is read-only and does not create directories or files.
- Environment override paths must be absolute.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core._paths (workspace root fallback)
"""
from __future__ import annotations
 
from dataclasses import dataclass
import json
import os
from pathlib import Path
import subprocess
from typing import Mapping
 
from oc_diagnostics import _dbg
 
from oc_chat.core._paths import find_workspace_root
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
REGISTRY_ENV_VAR = "OC_AGENT_REGISTRY"
WORKER_ENV_VAR = "OC_AGENT_WORKER"
WORKSPACE_MANIFEST = ".option_c_workspace.json"
DEFAULT_REGISTRY_RELATIVE = Path(".option_c") / "orchestration" / "registry.sqlite"
DEFAULT_ORCHESTRATION_RELATIVE = Path(".option_c") / "orchestration"
DEFAULT_CHECKPOINT_NAME = "state.json"
 
 
class WorkerRegistryError(RuntimeError):
    """Fail-closed worker registry-discovery boundary error (Plan 142j R7).
 
    Raised under ``OC_AGENT_WORKER=1`` when ``OC_AGENT_REGISTRY`` is unset/empty,
    not absolute, or resolves OUTSIDE the current worktree's
    ``.option_c/orchestration/`` boundary — the cross-worktree contamination the
    per-worktree resolved-path model exists to prevent. NOT a §8 source/regenerate
    envelope condition (discovery runs before any source operation); the CLI
    surfaces it as exit 2 + a stderr message naming both ``OC_AGENT_REGISTRY`` and
    the worker context.
    """
 
 
@dataclass(frozen=True)
class RegistryLocation:
    """Resolved registry path plus workspace metadata for init seeding."""
 
    registry_path: Path
    workspace_root: Path
    git_common_dir: Path
    checkpoint_path: Path
 
 
def discover_registry_path(
    start: str | os.PathLike[str] | None = None,
    env: Mapping[str, str] | None = None,
) -> Path:
    """Resolve the registry SQLite path using spec section 6.1 order."""
 
    return discover_registry_location(start=start, env=env).registry_path
 
 
def discover_registry_location(
    start: str | os.PathLike[str] | None = None,
    env: Mapping[str, str] | None = None,
    registry_path: str | os.PathLike[str] | None = None,
    workspace_root: str | os.PathLike[str] | None = None,
) -> RegistryLocation:
    """Resolve registry path and metadata needed by ``init``."""
 
    mapping = os.environ if env is None else env
    anchor = _anchor_path(start)
    resolved_workspace = _resolve_workspace_root(anchor, workspace_root)
    _worker_registry_guard(mapping, resolved_workspace)
 
    if registry_path is not None:
        resolved_registry = _absolute_path(Path(registry_path), anchor)
    else:
        resolved_registry = _discover_path(anchor, mapping, resolved_workspace)
 
    git_common_dir = _git_common_dir(anchor) or (resolved_workspace / ".git").resolve()
    checkpoint_path = resolved_registry.parent / DEFAULT_CHECKPOINT_NAME
    return RegistryLocation(
        registry_path=resolved_registry,
        workspace_root=resolved_workspace,
        git_common_dir=git_common_dir,
        checkpoint_path=checkpoint_path,
    )
 
 
def _worker_registry_guard(env: Mapping[str, str], workspace_root: Path) -> None:
    """Plan 142j R7: in worker context (``OC_AGENT_WORKER=1``), ``OC_AGENT_REGISTRY``
    MUST be set, absolute, and resolve INSIDE this worktree's
    ``.option_c/orchestration/`` boundary; else fail closed (``WorkerRegistryError``).
 
    A not-yet-created ``registry.sqlite`` **inside** the boundary is VALID (the
    per-worktree store is created on first use — location, not existence, is
    checked). Unset/empty OR resolving OUTSIDE the boundary (main repo / sibling
    worktree / arbitrary temp) is fail-closed. Non-worker local CLI is unaffected
    (the existing workspace/git fallback still applies).
    """
    if env.get(WORKER_ENV_VAR) != "1":
        return
    boundary = (Path(workspace_root) / DEFAULT_ORCHESTRATION_RELATIVE).resolve()
    raw = env.get(REGISTRY_ENV_VAR)
    if not raw:
        raise WorkerRegistryError(
            f"worker context ({WORKER_ENV_VAR}=1) requires {REGISTRY_ENV_VAR} set to an "
            f"absolute path inside this worktree's boundary {boundary} (got unset/empty)"
        )
    path = Path(raw).expanduser()
    if not path.is_absolute():
        raise WorkerRegistryError(
            f"worker {REGISTRY_ENV_VAR} must be an absolute path (got {raw!r})"
        )
    resolved = path.resolve()
    try:
        resolved.relative_to(boundary)
    except ValueError as error:
        raise WorkerRegistryError(
            f"worker {REGISTRY_ENV_VAR} {resolved} resolves OUTSIDE this worktree's "
            f"boundary {boundary} — refusing to read/write another worktree's store"
        ) from error
 
 
def _discover_path(
    anchor: Path,
    env: Mapping[str, str],
    workspace_root: Path,
) -> Path:
    env_value = env.get(REGISTRY_ENV_VAR)
    if env_value:
        path = Path(env_value).expanduser()
        if not path.is_absolute():
            raise ValueError(f"{REGISTRY_ENV_VAR} must be an absolute path")
        return path.resolve()
 
    manifest = _find_workspace_manifest(anchor)
    if manifest is not None:
        root = _workspace_root_from_manifest(manifest)
        return (root / DEFAULT_REGISTRY_RELATIVE).resolve()
 
    common_dir = _git_common_dir(anchor)
    if common_dir is not None:
        return (common_dir.parent / DEFAULT_REGISTRY_RELATIVE).resolve()
 
    return (workspace_root / DEFAULT_REGISTRY_RELATIVE).resolve()
 
 
def _resolve_workspace_root(
    anchor: Path,
    workspace_root: str | os.PathLike[str] | None,
) -> Path:
    if workspace_root is not None:
        return _absolute_path(Path(workspace_root), anchor)
    manifest = _find_workspace_manifest(anchor)
    if manifest is not None:
        return _workspace_root_from_manifest(manifest)
    try:
        return find_workspace_root(anchor).resolve()
    except FileNotFoundError:
        return anchor.resolve()
 
 
def _workspace_root_from_manifest(manifest: Path) -> Path:
    try:
        data = json.loads(manifest.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return manifest.parent.resolve()
    for key in ("workspace_root", "root", "workspaceRoot"):
        raw_value = data.get(key) if isinstance(data, dict) else None
        if isinstance(raw_value, str) and raw_value.strip():
            return _absolute_path(Path(raw_value).expanduser(), manifest.parent)
    return manifest.parent.resolve()
 
 
def _find_workspace_manifest(anchor: Path) -> Path | None:
    current = anchor if anchor.is_dir() else anchor.parent
    for candidate in (current, *current.parents):
        manifest = candidate / WORKSPACE_MANIFEST
        if manifest.is_file():
            return manifest.resolve()
    return None
 
 
def _git_common_dir(anchor: Path) -> Path | None:
    try:
        result = subprocess.run(
            ["git", "rev-parse", "--git-common-dir"],
            cwd=str(anchor),
            check=False,
            capture_output=True,
            text=True,
        )
    except (OSError, ValueError):
        return None
    if result.returncode != 0:
        return None
    raw_value = result.stdout.strip()
    if not raw_value:
        return None
    return _absolute_path(Path(raw_value), anchor)
 
 
def _anchor_path(start: str | os.PathLike[str] | None) -> Path:
    return Path.cwd().resolve() if start is None else Path(start).expanduser().resolve()
 
 
def _absolute_path(path: Path, base: Path) -> Path:
    if path.is_absolute():
        return path.resolve()
    return (base / path).resolve()