workspace.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/core/workspace.py
 
"""
Purpose
-------
Workspace (monorepo) support for OCT (F1, leanest first cut). A
workspace is a directory containing a ``.option_c_workspace.json``
manifest and one or more Option C subprojects. ``oct git`` commands
invoked from a workspace root may operate on the entire repo without
the ``--root-dir <subproj> --allow-outside-project`` escape hatch
required pre-F1.
 
Responsibilities
----------------
- Define :class:`Workspace` — the loaded representation of the
  manifest.
- Provide :func:`load_workspace` — read and validate the manifest at
  ``<root>/.option_c_workspace.json``.
- Provide :func:`find_workspace_root` — walk upward from a starting
  directory to locate the nearest workspace manifest.
- Be intentionally conservative: invalid manifests raise
  :class:`WorkspaceError` rather than degrading silently.
 
Diagnostics
-----------
Domain: WORKSPACE
Levels:
    L1 — errors (invalid manifest, IO failure)
    L2 — lifecycle (load start / end)
    L3 — details (resolved subproject paths)
 
Contracts
---------
- This module imports nothing from ``oct.git`` or ``oct.cli``; it sits
  beneath them in the dependency graph.
- Manifest schema is JSON-only (no TOML); the file is small enough
  that adding a TOML parser is unwarranted.
- Subproject paths in the manifest are interpreted relative to the
  workspace root. Absolute paths in the manifest are rejected.
- F1 fan-out (v0.25.0): the manifest now also carries
  ``workspace_profile`` (default ``"compact"``), ``workspace_files``
  (gitignore-style globs of files owned by the workspace itself
  rather than any subproject), and ``ignore_at_workspace_scope``
  (extra ignore patterns applied at the workspace ``.gitignore``).
  All three are optional with documented defaults.
"""
 
from __future__ import annotations
 
import functools
import json
import re
from dataclasses import dataclass, field
from pathlib import Path
 
from oct.core.diagnostics import _dbg
 
 
#: Filename of the workspace manifest at the workspace root.
WORKSPACE_MANIFEST: str = ".option_c_workspace.json"
 
#: Manifest schema version this loader understands.
_SCHEMA_VERSION: str = "1.0"
 
#: Valid lint profile names.
_VALID_PROFILES: tuple[str, ...] = ("proto", "compact", "strict")
 
#: Default profile for files matched by ``workspace_files`` when the
#: manifest does not set ``workspace_profile`` explicitly.
_DEFAULT_WORKSPACE_PROFILE: str = "compact"
 
 
class WorkspaceError(Exception):
    """Raised when a workspace manifest is invalid or unreadable."""
 
 
@dataclass(frozen=True)
class Subproject:
    """One Option C subproject under a workspace.
 
    Attributes
    ----------
    path
        Absolute path to the subproject root (resolved relative to the
        workspace root at load time).
    name
        Display name for the subproject.
    profile
        Optional override for the lint profile applied to this
        subproject's files when scoped via the workspace. ``None``
        means "use the subproject's own ``.octrc.json``".
    """
 
    path: Path
    name: str
    profile: str | None = None
 
 
@dataclass(frozen=True)
class Workspace:
    """Loaded representation of an Option C workspace manifest.
 
    Attributes
    ----------
    root
        Absolute resolved path to the workspace root.
    name
        Display name from the manifest.
    subprojects
        Tuple of :class:`Subproject` entries, in manifest order.
    workspace_profile
        Lint profile applied to files matched by *workspace_files*.
        Defaults to ``"compact"``.
    workspace_files
        Tuple of gitignore-style globs identifying files at the
        workspace level (i.e. owned by no subproject). Examples:
        ``("CLAUDE.md", ".github/**", "docs/**")``.
    ignore_at_workspace_scope
        Tuple of additional ignore patterns to be inserted into the
        OCT-managed block of the workspace-root ``.gitignore`` when
        ``oct git init`` runs at workspace scope.
    """
 
    root: Path
    name: str
    subprojects: tuple[Subproject, ...] = field(default_factory=tuple)
    workspace_profile: str = _DEFAULT_WORKSPACE_PROFILE
    workspace_files: tuple[str, ...] = field(default_factory=tuple)
    ignore_at_workspace_scope: tuple[str, ...] = field(default_factory=tuple)
 
    @property
    def subprojects_by_name(self) -> dict[str, "Subproject"]:
        """Lookup index from subproject name to :class:`Subproject`.
 
        Computed eagerly on first access; cheap because subprojects
        are bounded to a small N. The dict is regenerated per call —
        the dataclass is frozen so we can't cache it.
        """
        return {sp.name: sp for sp in self.subprojects}
 
 
def load_workspace(root: Path) -> Workspace:
    """Load and validate the workspace manifest at ``<root>``.
 
    Parameters
    ----------
    root
        Path to the workspace root (the directory containing
        ``.option_c_workspace.json``).
 
    Returns
    -------
    :class:`Workspace`
 
    Raises
    ------
    WorkspaceError
        If the manifest is missing, malformed, or contains invalid
        subproject paths.
    """
    func = "load_workspace"
    _dbg("WORKSPACE", func, f"root={root}", 2)
 
    manifest_path = Path(root) / WORKSPACE_MANIFEST
    if not manifest_path.is_file():
        raise WorkspaceError(
            f"workspace manifest not found at {manifest_path}",
        )
 
    try:
        raw = manifest_path.read_text(encoding="utf-8")
        data = json.loads(raw)
    except (OSError, json.JSONDecodeError) as exc:
        raise WorkspaceError(
            f"failed to read workspace manifest {manifest_path}: {exc}",
        ) from exc
 
    if not isinstance(data, dict):
        raise WorkspaceError(
            f"workspace manifest must be a JSON object, "
            f"got {type(data).__name__}",
        )
 
    version = data.get("version")
    if version != _SCHEMA_VERSION:
        raise WorkspaceError(
            f"unsupported workspace manifest version "
            f"{version!r} (expected {_SCHEMA_VERSION!r})",
        )
 
    name_raw = data.get("name")
    if not isinstance(name_raw, str) or not name_raw.strip():
        raise WorkspaceError(
            "workspace manifest 'name' must be a non-empty string",
        )
 
    subprojects_raw = data.get("subprojects", [])
    if not isinstance(subprojects_raw, list):
        raise WorkspaceError(
            f"workspace 'subprojects' must be a list, "
            f"got {type(subprojects_raw).__name__}",
        )
 
    root_resolved = Path(root).resolve()
    subprojects: list[Subproject] = []
    seen_paths: set[Path] = set()
 
    for idx, entry in enumerate(subprojects_raw):
        if not isinstance(entry, dict):
            raise WorkspaceError(
                f"subprojects[{idx}] must be an object, "
                f"got {type(entry).__name__}",
            )
        sp_path_raw = entry.get("path")
        sp_name = entry.get("name")
        sp_profile = entry.get("profile")
 
        if not isinstance(sp_path_raw, str) or not sp_path_raw.strip():
            raise WorkspaceError(
                f"subprojects[{idx}].path must be a non-empty string",
            )
        sp_path_obj = Path(sp_path_raw)
        if sp_path_obj.is_absolute():
            raise WorkspaceError(
                f"subprojects[{idx}].path must be relative to the "
                f"workspace root, got absolute path {sp_path_raw!r}",
            )
        sp_resolved = (root_resolved / sp_path_raw).resolve()
 
        # Path-escape: subproject must live inside the workspace root.
        try:
            sp_resolved.relative_to(root_resolved)
        except ValueError:
            raise WorkspaceError(
                f"subprojects[{idx}].path {sp_path_raw!r} escapes the "
                f"workspace root {root_resolved}",
            )
 
        if sp_resolved in seen_paths:
            raise WorkspaceError(
                f"duplicate subproject path {sp_path_raw!r}",
            )
        seen_paths.add(sp_resolved)
 
        if not isinstance(sp_name, str) or not sp_name.strip():
            raise WorkspaceError(
                f"subprojects[{idx}].name must be a non-empty string",
            )
 
        if sp_profile is not None and (
            not isinstance(sp_profile, str)
            or sp_profile not in ("proto", "compact", "strict")
        ):
            raise WorkspaceError(
                f"subprojects[{idx}].profile must be one of "
                f"'proto'/'compact'/'strict' or omitted, got "
                f"{sp_profile!r}",
            )
 
        subprojects.append(
            Subproject(
                path=sp_resolved, name=sp_name, profile=sp_profile,
            ),
        )
 
    # -- workspace_profile ----------------------------------------------
    wp_raw = data.get("workspace_profile")
    if wp_raw is None:
        workspace_profile = _DEFAULT_WORKSPACE_PROFILE
    elif isinstance(wp_raw, str) and wp_raw in _VALID_PROFILES:
        workspace_profile = wp_raw
    else:
        raise WorkspaceError(
            f"'workspace_profile' must be one of "
            f"{', '.join(repr(p) for p in _VALID_PROFILES)} or omitted, "
            f"got {wp_raw!r}",
        )
 
    # -- workspace_files -------------------------------------------------
    wf_raw = data.get("workspace_files", [])
    if not isinstance(wf_raw, list):
        raise WorkspaceError(
            f"'workspace_files' must be a list of strings, "
            f"got {type(wf_raw).__name__}",
        )
    workspace_files: list[str] = []
    for idx, entry in enumerate(wf_raw):
        valid, msg = _validate_workspace_pattern(entry)
        if not valid:
            raise WorkspaceError(
                f"workspace_files[{idx}]: {msg}",
            )
        workspace_files.append(entry)
 
    # -- ignore_at_workspace_scope --------------------------------------
    iaw_raw = data.get("ignore_at_workspace_scope", [])
    if not isinstance(iaw_raw, list):
        raise WorkspaceError(
            f"'ignore_at_workspace_scope' must be a list of strings, "
            f"got {type(iaw_raw).__name__}",
        )
    ignore_at_workspace_scope: list[str] = []
    for idx, entry in enumerate(iaw_raw):
        valid, msg = _validate_workspace_pattern(entry)
        if not valid:
            raise WorkspaceError(
                f"ignore_at_workspace_scope[{idx}]: {msg}",
            )
        ignore_at_workspace_scope.append(entry)
 
    workspace = Workspace(
        root=root_resolved,
        name=name_raw,
        subprojects=tuple(subprojects),
        workspace_profile=workspace_profile,
        workspace_files=tuple(workspace_files),
        ignore_at_workspace_scope=tuple(ignore_at_workspace_scope),
    )
    _dbg(
        "WORKSPACE", func,
        f"loaded: {len(subprojects)} subproject(s), "
        f"{len(workspace_files)} workspace-files glob(s), "
        f"{len(ignore_at_workspace_scope)} workspace-ignore pattern(s)",
        3,
    )
    return workspace
 
 
def find_workspace_root(start: Path) -> Path | None:
    """Walk upward from *start* to locate a workspace manifest.
 
    Returns the workspace root, or ``None`` when no manifest is found
    by the time the filesystem root is reached.
    """
    func = "find_workspace_root"
    current = Path(start).resolve()
    for parent in [current] + list(current.parents):
        if (parent / WORKSPACE_MANIFEST).is_file():
            _dbg("WORKSPACE", func, f"found at {parent}", 3)
            return parent
    _dbg("WORKSPACE", func, f"no manifest above {start}", 4)
    return None
 
 
# ---------------------------------------------------------------------
# F1 fan-out helpers (v0.25.0)
# ---------------------------------------------------------------------
 
#: Regex of characters allowed inside a workspace pattern. Matches the
#: contract enforced by ``oct git ignore`` validation
#: (``oct/git/policy.py:_VALID_PATTERN_RE``) so the two surfaces stay
#: consistent. ``**`` glob is allowed (it's two ``*``s in the char set).
_VALID_WORKSPACE_PATTERN_RE: re.Pattern[str] = re.compile(
    r"^[A-Za-z0-9_./*?\‍[\‍]\-]+$",
)
 
 
def _validate_workspace_pattern(pattern: object) -> tuple[bool, str]:
    """Return ``(valid, message)`` for a workspace_files / ignore pattern.
 
    Mirrors ``oct.git.policy.validate_ignore_pattern`` so the manifest
    surface is held to the same safety bar as ``oct git ignore``: no
    negation, no absolute paths, no shell metacharacters, no newlines.
    """
    if not isinstance(pattern, str):
        return False, (
            f"pattern must be a string, got {type(pattern).__name__}"
        )
    if not pattern.strip():
        return False, "pattern is empty"
    if "\n" in pattern or "\r" in pattern:
        return False, "pattern contains a newline"
    if pattern.startswith("!"):
        return False, "negation patterns ('!...') are not allowed"
    if pattern.startswith("/"):
        return False, "absolute patterns (leading '/') are not allowed"
    if not _VALID_WORKSPACE_PATTERN_RE.match(pattern):
        return False, (
            "pattern contains characters outside the safe set "
            "[A-Za-z0-9_./*?[]-]"
        )
    return True, ""
 
 
@functools.lru_cache(maxsize=None)
def _glob_to_regex(pattern: str) -> re.Pattern[str]:
    """Compile a gitignore-style glob to a regex.
 
    Supports::
 
        *   matches any sequence of chars within one path segment (no /)
        **  matches across path segments, including /
        ?   matches a single non-/ char
        []  literal char class (treated character-by-character)
 
    Patterns are anchored: the resulting regex must match the entire
    relative path. Caching is global and unbounded — the working set
    of patterns per session is small and stable.
    """
    parts: list[str] = []
    i = 0
    while i < len(pattern):
        c = pattern[i]
        if c == "*":
            if i + 1 < len(pattern) and pattern[i + 1] == "*":
                # ``**`` matches zero or more path segments.
                parts.append(".*")
                i += 2
                # Eat a trailing ``/`` so ``foo/**/bar`` works.
                if i < len(pattern) and pattern[i] == "/":
                    i += 1
            else:
                parts.append("[^/]*")
                i += 1
        elif c == "?":
            parts.append("[^/]")
            i += 1
        elif c in r".+()|^$\\":
            parts.append(re.escape(c))
            i += 1
        else:
            parts.append(c)
            i += 1
    return re.compile("^" + "".join(parts) + "$")
 
 
def matches_workspace_files(
    rel_path: str,
    patterns: tuple[str, ...],
) -> bool:
    """Return True if *rel_path* matches any *patterns* glob.
 
    *rel_path* must be a forward-slash POSIX-style relative path
    (e.g. ``"docs/README.md"``). Backslashes — common on Windows —
    must be normalised by the caller.
    """
    if not patterns:
        return False
    for pat in patterns:
        if _glob_to_regex(pat).match(rel_path):
            return True
    return False
 
 
def assign_to_subproject(
    file_path: Path,
    workspace: Workspace,
) -> Subproject | None:
    """Return the subproject containing *file_path*, or ``None``.
 
    Uses longest-path-prefix match so nested subprojects (rare)
    resolve to the deepest match. The returned :class:`Subproject` is
    one of ``workspace.subprojects`` (identity-comparable). Returns
    ``None`` when *file_path* falls outside every subproject — the
    caller then decides whether the file matches *workspace_files*
    or should be dropped.
 
    Performance: O(N subprojects) per call; N is bounded by manifest
    size and is small in practice.
    """
    try:
        file_resolved = Path(file_path).resolve()
    except OSError:
        return None
    matches: list[Subproject] = []
    for sp in workspace.subprojects:
        try:
            file_resolved.relative_to(sp.path)
            matches.append(sp)
        except ValueError:
            continue
    if not matches:
        return None
    return max(matches, key=lambda s: len(str(s.path)))