activity-core/src/activity_core/runtime_paths.py

"""Resolve repo-relative runtime paths for Custodian-owned assets."""

from __future__ import annotations

import os
from pathlib import Path

_DEFAULT_CUSTODIAN_ROOT = Path("/home/worsch/the-custodian")
_DEFAULT_ACTIVITY_CORE_ROOT = Path("/etc/activity-core")

_CUSTODIAN_SCHEME = "custodian://"
_ACTIVITY_CORE_SCHEME = "activity-core://"


def custodian_repo_root() -> Path:
    raw = os.environ.get("CUSTODIAN_REPO_ROOT", "").strip()
    return Path(raw).expanduser() if raw else _DEFAULT_CUSTODIAN_ROOT


def activity_core_root() -> Path:
    raw = os.environ.get("ACTIVITY_CORE_ROOT", "").strip()
    return Path(raw).expanduser() if raw else _DEFAULT_ACTIVITY_CORE_ROOT


def resolve_runtime_path(raw_path: str) -> Path:
    """Map custodian:// and activity-core:// URIs to mounted runtime paths."""
    value = str(raw_path or "").strip()
    if not value:
        return Path(value)

    if value.startswith(_CUSTODIAN_SCHEME):
        return (custodian_repo_root() / value.removeprefix(_CUSTODIAN_SCHEME)).resolve()

    if value.startswith(_ACTIVITY_CORE_SCHEME):
        return (activity_core_root() / value.removeprefix(_ACTIVITY_CORE_SCHEME)).resolve()

    return Path(value).expanduser()


def custodian_working_memory_dir() -> Path:
    return resolve_runtime_path("custodian://memory/working")


def custodian_repo_relative(path: Path) -> str:
    """Return a repo-relative POSIX path when under CUSTODIAN_REPO_ROOT."""
    root = custodian_repo_root().resolve()
    try:
        return path.resolve().relative_to(root).as_posix()
    except ValueError:
        return path.as_posix()


def default_ops_inventory_path() -> Path:
    raw = os.environ.get("OPS_INVENTORY_PATH", "").strip()
    if raw:
        return resolve_runtime_path(raw)
    return resolve_runtime_path("custodian://ops/service-inventory.yml")