path: root/wflib/config.py

"""Find the project (nearest ancestor with workflow.toml) and load its config."""
from __future__ import annotations

import tomllib
from dataclasses import dataclass, field
from pathlib import Path

from . import lanes as lanes_mod

NAME = "workflow.toml"
WF_HOME = ".wf-home"
FORMAT = 1
KEYS = {"format", "tasks", "archive", "docs", "verify", "done", "ledgers", "anchors", "ctx_hint", "worktree_setup", "quick_gate",
        "slice_above", "lanes", "areas", "code_root", "area_stale_commits", "area_ignore",
        "cloud", "cloud_include", "cloud_note", "push"}
AREA_IGNORE = ("tests", "test", "docs", "doc")
ANCHOR_KEYS = {"index", "index_section", "specs"}


class ConfigError(Exception):
    pass


@dataclass
class Config:
    root: Path
    format: int
    tasks: Path
    archive: Path
    docs: list[Path] = field(default_factory=list)
    verify: list[str] = field(default_factory=list)
    done: list[str] = field(default_factory=list)
    ledgers: Path | None = None
    anchors_index: Path | None = None
    anchors_section: str | None = None
    anchors_specs: Path | None = None
    ctx_hint: int = 100_000             # tokens; wf done/next suggest /clear above it (0 = off)
    worktree_setup: list[str] = field(default_factory=list)  # `wf setup`: cwd = worktree, env WF_MAIN
    quick_gate: list[str] = field(default_factory=list)      # `wf gate`: worker runs it before `wf done`
    slice_above: str = "1h"
    lanes: tuple = lanes_mod.DEFAULT_LANES
    areas: Path | None = None               # area notes file; None = CLAUDE.md
    code_root: Path | None = None           # repo area anchors / Checked / staleness refer to; None = root
    area_stale_commits: int = 20
    area_ignore: list[str] = field(default_factory=lambda: list(AREA_IGNORE))  # never an uncovered area
    local: Path | None = None               # lane worktree's copy of root whose LOCAL_KEYS were used
    cloud: bool = False                     # cloud lane opt-in (wf cloud send)
    cloud_include: list[str] = field(default_factory=list)   # git-ignored paths force-added to the snapshot
    cloud_note: str | None = None           # appended to the cloud prompt
    push: list[str] = field(default_factory=list)   # wf finish/merge/push: run in root after a merge (else push home)

    @property
    def areas_file(self) -> Path:
        return self.areas or (self.local or self.root) / "CLAUDE.md"

    @property
    def split(self) -> bool:
        """code_root is another git repo than root: private home repo + code repo (split project)."""
        return bool(self.code_root) and git_top(self.code_root) not in (None, git_top(self.root))

    @property
    def code(self) -> Path:
        return self.code_root or self.root

    def rel(self, path: Path) -> str:
        try:
            return str(path.relative_to(self.root))
        except ValueError:
            return str(path)

    @property
    def areas_rel(self) -> str:
        """areas_file relative to its project folder (the lane worktree's copy or root)."""
        try:
            return str(self.areas_file.relative_to(self.local or self.root))
        except ValueError:
            return str(self.areas_file)


def git_top(folder: Path) -> Path | None:
    """The work tree top holding folder (.git dir or file), or None."""
    for f in (folder, *folder.parents):
        if (f / ".git").exists():
            return f
    return None


def git_branch(gitdir: Path) -> str | None:
    try:
        head = (gitdir / "HEAD").read_text().strip()
    except OSError:
        return None
    return head.removeprefix("ref: refs/heads/") if head.startswith("ref: refs/heads/") else None


def linked_worktree(folder: Path) -> tuple[Path, Path, Path] | None:
    """(worktree top, its gitdir, main tree top) when folder is in a linked git worktree."""
    top = git_top(folder)
    if not top or not (top / ".git").is_file():
        return None
    try:
        line = (top / ".git").read_text().strip()
        gitdir = (top / line.removeprefix("gitdir:").strip()).resolve()
        common = (gitdir / (gitdir / "commondir").read_text().strip()).resolve()
    except OSError:
        return None
    if common.name != ".git" or not line.startswith("gitdir:"):
        return None
    return top, gitdir, common.parent


def find_root(start: Path) -> Path:
    """Folder with workflow.toml at or above start; in a linked git worktree the main tree's
    same folder, when it is a project too (one TASKS.md for all worktrees). A code worktree of a
    split project: its .wf-home."""
    start = start.resolve()
    top = git_top(start)
    if top and (top / WF_HOME).is_file():
        home = Path((top / WF_HOME).read_text().strip())
        if not (home / NAME).is_file():
            raise ConfigError(f"{WF_HOME} points to {home}: no {NAME} there (fix or delete {top / WF_HOME})")
        return home
    for folder in (start, *start.parents):
        if (folder / NAME).is_file():
            wt = linked_worktree(folder)
            if wt:
                main = wt[2] / folder.relative_to(wt[0])
                if (main / NAME).is_file():
                    return main
            return folder
    raise ConfigError(f"no {NAME} in {start} or above: not a wf project (wf init makes one)")


def code_main(cfg: "Config") -> Path | None:
    """Main tree top of the code repo of a split project, else None."""
    return git_top(cfg.code_root) if cfg.split else None


def find_local(start: Path) -> Path | None:
    """In a linked worktree whose project root is the main tree's (find_root): the worktree's own copy
    of that folder when it has a workflow.toml, else None."""
    start = start.resolve()
    wt = linked_worktree(start)
    if not wt:
        return None
    root = find_root(start)
    try:
        local = wt[0] / root.relative_to(wt[2])
    except ValueError:
        return None
    return local if local != root and (local / NAME).is_file() else None


def load_at(start: Path) -> "Config":
    """Config for a command run in start: the main tree's books, branch-testable keys (LOCAL_KEYS:
    worktree_setup, quick_gate, areas file, area settings) from a lane worktree's own workflow.toml."""
    return load(find_root(start), find_local(start))


def _strings(data: dict, key: str) -> list[str]:
    value = data.get(key, [])
    if not isinstance(value, list) or not all(isinstance(v, str) for v in value):
        raise ConfigError(f"{NAME}: '{key}' must be a list of strings")
    return value


def _string(data: dict, key: str, shown: str | None = None) -> str | None:
    value = data.get(key)
    if value is not None and not isinstance(value, str):
        raise ConfigError(f"{NAME}: '{shown or key}' must be a string")
    return value


LOCAL_KEYS = ("worktree_setup", "quick_gate", "areas", "area_stale_commits", "area_ignore")


def _read(path: Path) -> dict:
    try:
        return tomllib.loads(path.read_text(encoding="utf-8"))
    except (tomllib.TOMLDecodeError, OSError) as e:
        raise ConfigError(f"{path.name}: {e}") from None


def load(root: Path, local: Path | None = None) -> Config:
    """root's workflow.toml; with local (a lane worktree's copy of root), LOCAL_KEYS come from local's
    workflow.toml and the areas file resolves there (a branch can test and --mark them before merge)."""
    data = _read(root / NAME)
    if local:
        mine = _read(local / NAME)
        for key in LOCAL_KEYS:
            data.pop(key, None)
            if key in mine:
                data[key] = mine[key]
    for key in data:
        if key not in KEYS:
            raise ConfigError(f"{NAME}: unknown key '{key}'")
    anchors = data.get("anchors", {})
    if not isinstance(anchors, dict):
        raise ConfigError(f"{NAME}: 'anchors' must be a table")
    for key in anchors:
        if key not in ANCHOR_KEYS:
            raise ConfigError(f"{NAME}: unknown key 'anchors.{key}'")
    for key in ("format", "tasks", "archive"):
        if key not in data:
            raise ConfigError(f"{NAME}: missing '{key}'")
    if not isinstance(data["format"], int) or isinstance(data["format"], bool):
        raise ConfigError(f"{NAME}: 'format' must be a number")
    ctx_hint = data.get("ctx_hint", 100_000)
    if not isinstance(ctx_hint, int) or isinstance(ctx_hint, bool) or ctx_hint < 0:
        raise ConfigError(f"{NAME}: 'ctx_hint' must be a number of tokens (0 = off)")
    index = _string(anchors, "index", "anchors.index")
    specs = _string(anchors, "specs", "anchors.specs")
    ledgers = _string(data, "ledgers")
    stale = data.get("area_stale_commits", 20)
    if not isinstance(stale, int) or isinstance(stale, bool) or stale < 1:
        raise ConfigError(f"{NAME}: 'area_stale_commits' must be a number ≥ 1")
    slice_above = _string(data, "slice_above") or "1h"
    try:
        lane_list = lanes_mod.parse_lanes(data.get("lanes"), slice_above)
    except ValueError as e:
        raise ConfigError(f"{NAME}: {e}") from None
    areas = _string(data, "areas")
    code_root = _string(data, "code_root")
    if not isinstance(data.get("cloud", False), bool):
        raise ConfigError(f"{NAME}: 'cloud' must be true or false")
    return Config(
        root=root, format=data["format"],
        tasks=root / _string(data, "tasks"), archive=root / _string(data, "archive"),
        docs=[root / d for d in _strings(data, "docs")],
        verify=_strings(data, "verify"), done=_strings(data, "done"),
        ledgers=root / ledgers if ledgers else None,
        anchors_index=root / index if index else None,
        anchors_section=_string(anchors, "index_section", "anchors.index_section"),
        anchors_specs=root / specs if specs else None,
        ctx_hint=ctx_hint,
        worktree_setup=_strings(data, "worktree_setup"), quick_gate=_strings(data, "quick_gate"),
        slice_above=slice_above, lanes=lane_list, areas=(local or root) / areas if areas else None,
        code_root=(root / code_root).resolve() if code_root else None,
        area_stale_commits=stale,
        area_ignore=_strings(data, "area_ignore") if "area_ignore" in data else list(AREA_IGNORE),
        local=local,
        cloud=data.get("cloud", False), cloud_include=_strings(data, "cloud_include"),
        cloud_note=_string(data, "cloud_note"),
        push=_strings(data, "push"),
    )