istosandClaude Fable 5
29cbb11eb5
Everything core from cicero-pas's .task-manager, with instance data reduced to skeleton: empty stage directories, the task template, empty local/ scaffolding, and a README covering install (clone into .task-manager/, vendored on purpose) and update (update.sh replaces core wholesale; local/ and tasks/ survive). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
146 lines
5.2 KiB
Python
146 lines
5.2 KiB
Python
"""Paths, stages and launch configuration for the task manager.
|
||
|
||
Core knows about tasks, worktrees, PRs and events. It knows nothing about
|
||
any particular app (drivers do), agent vendor (adapters do), or project
|
||
(local/ does). Everything the other modules need to know about *where
|
||
things are* lives here. No state, no behaviour.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import subprocess
|
||
from pathlib import Path
|
||
|
||
CORE = Path(__file__).resolve().parent # manager/core — replaceable
|
||
MANAGER = CORE.parent # manager
|
||
LOCAL = MANAGER / "local" # project-owned, never replaced
|
||
TM_ROOT = MANAGER.parent # .task-manager
|
||
TASKS = TM_ROOT / "tasks" # stage directories only
|
||
|
||
STATE = LOCAL / "state" # runtime data (gitignored)
|
||
SESSIONS_DIR = STATE / "sessions" # per-session event logs, JSONL
|
||
AGENT_DIR = STATE / "agent" # headless-agent stdout logs
|
||
DRIVES_DIR = STATE / "drives" # driver stdout logs
|
||
|
||
# Ordered — this is the column order on the board.
|
||
STAGES = [
|
||
("backlog", "Backlog"),
|
||
("to-do", "To Do"),
|
||
("in-progress", "In Progress"),
|
||
("review", "Review"),
|
||
("done", "Done"),
|
||
]
|
||
STAGE_DIRS = {slug for slug, _ in STAGES}
|
||
STAGE_LABELS = dict(STAGES)
|
||
|
||
|
||
def _repo_root() -> Path:
|
||
try:
|
||
out = subprocess.check_output(
|
||
["git", "-C", str(MANAGER), "rev-parse", "--show-toplevel"],
|
||
text=True, stderr=subprocess.DEVNULL,
|
||
).strip()
|
||
return Path(out)
|
||
except (subprocess.CalledProcessError, OSError):
|
||
return TM_ROOT.parent
|
||
|
||
|
||
REPO = _repo_root()
|
||
|
||
|
||
def _load_env() -> dict[str, str]:
|
||
"""local/.env, overridden by the process environment. Stdlib-only
|
||
parser: KEY=VALUE lines, # comments, optional quotes around the value."""
|
||
values: dict[str, str] = {}
|
||
path = LOCAL / ".env"
|
||
if path.is_file():
|
||
for line in path.read_text(encoding="utf-8").splitlines():
|
||
line = line.strip()
|
||
if not line or line.startswith("#") or "=" not in line:
|
||
continue
|
||
key, _, value = line.partition("=")
|
||
values[key.strip()] = value.strip().strip("'\"")
|
||
values.update(os.environ)
|
||
return values
|
||
|
||
|
||
_ENV = _load_env()
|
||
|
||
|
||
def setting(key: str, default: str) -> str:
|
||
return _ENV.get(key, default)
|
||
|
||
|
||
def child_env() -> dict[str, str]:
|
||
"""Environment for adapter/driver child processes: the real environment
|
||
with local/.env settings folded in (process env still wins), so
|
||
BOARD_* settings reach the scripts that read them directly."""
|
||
return dict(_ENV)
|
||
|
||
|
||
# Pinned so the board is always at the same bookmarkable URL. Sits in the
|
||
# ephemeral-safe 10000–30000 range, clear of the other local dev servers.
|
||
PORT = int(setting("BOARD_PORT", "26071"))
|
||
# One isolated checkout per running work agent, relative to the repo root.
|
||
WORKTREES = REPO / setting("BOARD_WORKTREES", ".worktrees")
|
||
|
||
# Which agent adapter runs headless jobs. Resolution ladder: local wins.
|
||
ADAPTER = setting("BOARD_AGENT_ADAPTER", "claude")
|
||
|
||
# GitHub plumbing: the gh CLI (stub-able for tests) and the git remote PRs
|
||
# go to. Empty remote = auto-detect the first remote; no remote = no PRs.
|
||
GH_BIN = setting("BOARD_GH_BIN", "gh")
|
||
GIT_REMOTE = setting("BOARD_GIT_REMOTE", "")
|
||
PR_POLL_INTERVAL = float(setting("BOARD_PR_POLL_INTERVAL", "60"))
|
||
|
||
WATCH_INTERVAL = float(setting("BOARD_WATCH_INTERVAL", "2"))
|
||
EVENTS_CAP = int(setting("BOARD_EVENTS_CAP", "800"))
|
||
BOARD_EVENTS_CAP = int(setting("BOARD_HISTORY_CAP", "300"))
|
||
|
||
|
||
def prompt(name: str) -> str:
|
||
"""Prompt templates: core ships defaults, local/prompts/ overrides win.
|
||
Read fresh on every launch so edits apply without a restart."""
|
||
override = LOCAL / "prompts" / name
|
||
if override.is_file():
|
||
return override.read_text(encoding="utf-8")
|
||
return (CORE / "prompts" / name).read_text(encoding="utf-8")
|
||
|
||
|
||
def adapter_dir() -> Path | None:
|
||
"""The configured agent adapter's directory — local overrides core."""
|
||
for base in (LOCAL / "adapters", CORE / "adapters"):
|
||
candidate = base / ADAPTER
|
||
if (candidate / "run").is_file():
|
||
return candidate
|
||
return None
|
||
|
||
|
||
def driver_path() -> Path | None:
|
||
"""The project's app driver, if it has one."""
|
||
candidate = LOCAL / "driver" / "start"
|
||
return candidate if candidate.is_file() else None
|
||
|
||
|
||
def commands() -> list[dict]:
|
||
"""Project-owned commands: executables in local/commands/, surfaced as
|
||
chips on cards and run against the task's worktree. A `# help:` line
|
||
near the top becomes the tooltip."""
|
||
directory = LOCAL / "commands"
|
||
found = []
|
||
if directory.is_dir():
|
||
for path in sorted(directory.iterdir()):
|
||
if not path.is_file() or path.name.startswith(".") or not os.access(path, os.X_OK):
|
||
continue
|
||
help_text = ""
|
||
try:
|
||
for line in path.read_text(encoding="utf-8").splitlines()[:8]:
|
||
if line.startswith("# help:"):
|
||
help_text = line[len("# help:"):].strip()
|
||
break
|
||
except OSError:
|
||
pass
|
||
found.append({"name": path.name, "help": help_text})
|
||
return found
|