diff options
Diffstat (limited to 'wflib/config.py')
| -rw-r--r-- | wflib/config.py | 225 |
1 files changed, 225 insertions, 0 deletions
diff --git a/wflib/config.py b/wflib/config.py new file mode 100644 index 0000000..8f75c89 --- /dev/null +++ b/wflib/config.py @@ -0,0 +1,225 @@ +"""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" +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"} +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 + + @property + def areas_file(self) -> Path: + return self.areas or (self.local or self.root) / "CLAUDE.md" + + @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).""" + start = start.resolve() + 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 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"), + ) |