workspace.py

Go to the documentation of this file.

"""Workspace-anchored path resolution for every piece of connector state.
 
CONNECT-02 defect 1 (proposal V1 §7). Before this module the state directory
was a module-level ``Path(os.environ.get(..., ".option_c/orchestration/oc_connect"))``
evaluated once at import time and interpreted against the *current working
directory*. Two failures followed from the one expression: a send issued from
any subdirectory looked for connector state in a directory that has none and
died with ``Telegram destination 'operator' is not configured``, and the value
could not be re-resolved after import even when the process changed directory.
 
Resolution here is per call and anchored by marker walk, the way OCT's own
project-root discovery works:
 
1. an absolute ``OC_CONNECT_STATE_DIR`` wins outright (an explicit operator
   override needs no discovery);
2. otherwise walk upward from *start* for ``.option_c_workspace.json`` — the
   workspace marker wins even when a nearer subproject carries its own
   ``.option_c/``, because connector authority is workspace-wide, not
   per-subproject;
3. failing that, the nearest ancestor holding an ``.option_c/`` directory;
4. failing that, raise. Guessing a root would put operator grants somewhere the
   operator cannot find them.
 
A *relative* ``OC_CONNECT_STATE_DIR`` resolves against the workspace root, not
the cwd — reinstating cwd-relative behaviour through the override would
reintroduce exactly the defect this module removes.
 
Dependencies
------------
- oc_diagnostics (_dbg structured logger)
"""
from __future__ import annotations
 
import os
from pathlib import Path
 
from oc_diagnostics import _dbg
 
WORKSPACE_MARKER = ".option_c_workspace.json"
PROJECT_MARKER = ".option_c"
STATE_SUBPATH = (".option_c", "orchestration", "oc_connect")
CREDENTIAL_SUBPATH = (".option_c", "connector_secrets.json")
 
#: The comms bus lives beside the dev engine's orchestration state
#: (proposal V1 §3, verbatim layout). Workspaces without an ``oc_chat``
#: subproject fall back to the workspace-level ``.option_c/`` so the bus is
#: still anchored rather than cwd-relative.
COMMS_ANCHOR = "oc_chat"
COMMS_SUBPATH = (".option_c", "comms")
 
STATE_DIR_ENV = "OC_CONNECT_STATE_DIR"
CREDENTIAL_ENV = "OC_CONNECT_SECRETS"
 
 
class WorkspaceNotFoundError(RuntimeError):
    """No workspace or Option C project marker was found above the start path."""
 
 
def _resolved(path: str | Path) -> Path:
    candidate = Path(path).expanduser()
    try:
        return candidate.resolve()
    except OSError:
        return candidate.absolute()
 
 
def find_workspace_root(start: str | Path | None = None) -> Path:
    """Return the workspace root above *start*, failing loudly when absent."""
    func = "find_workspace_root"
    origin = _resolved(start if start is not None else Path.cwd())
    project_fallback: Path | None = None
    for parent in (origin, *origin.parents):
        if (parent / WORKSPACE_MARKER).is_file():
            _dbg(
                "OC_CONNECT.WORKSPACE",
                func,
                f"anchored on workspace marker root={parent}",
                3,
            )
            return parent
        if project_fallback is None and (parent / PROJECT_MARKER).is_dir():
            project_fallback = parent
    if project_fallback is not None:
        _dbg(
            "OC_CONNECT.WORKSPACE",
            func,
            f"anchored on project marker root={project_fallback}",
            3,
        )
        return project_fallback
    raise WorkspaceNotFoundError(
        f"no Option C workspace marker ({WORKSPACE_MARKER}) or project marker "
        f"({PROJECT_MARKER}/) was found at or above {origin}"
    )
 
 
def resolve_root(root: str | Path | None = None) -> Path:
    """Return an explicit root when given, otherwise discover one."""
    if root is not None:
        return _resolved(root)
    return find_workspace_root()
 
 
def _from_env(name: str, root: str | Path | None) -> Path | None:
    raw = os.environ.get(name)
    if raw is None or not raw.strip():
        return None
    override = Path(raw.strip()).expanduser()
    if override.is_absolute():
        return _resolved(override)
    return resolve_root(root) / override
 
 
def state_dir(root: str | Path | None = None) -> Path:
    """Return the connector state directory for grants and destinations."""
    override = _from_env(STATE_DIR_ENV, root)
    if override is not None:
        return override
    return resolve_root(root).joinpath(*STATE_SUBPATH)
 
 
def credential_path(root: str | Path | None = None) -> Path:
    """Return the credential sidecar path, anchored the same way as state."""
    override = _from_env(CREDENTIAL_ENV, root)
    if override is not None:
        return override
    return resolve_root(root).joinpath(*CREDENTIAL_SUBPATH)
 
 
def comms_root(root: str | Path | None = None) -> Path:
    """Return the comms bus root for this workspace."""
    anchor = resolve_root(root)
    if (anchor / COMMS_ANCHOR).is_dir():
        return anchor.joinpath(COMMS_ANCHOR, *COMMS_SUBPATH)
    return anchor.joinpath(*COMMS_SUBPATH)
 
 
__all__ = [
    "COMMS_ANCHOR",
    "COMMS_SUBPATH",
    "CREDENTIAL_ENV",
    "CREDENTIAL_SUBPATH",
    "PROJECT_MARKER",
    "STATE_DIR_ENV",
    "STATE_SUBPATH",
    "WORKSPACE_MARKER",
    "WorkspaceNotFoundError",
    "comms_root",
    "credential_path",
    "find_workspace_root",
    "resolve_root",
    "state_dir",
]