forked from Keelrleek/Harbor
KeelrLeek
dbfd1720c5
Messages, threads, acknowledgements, claims, bands, tasks, review items and stop flags for a team of AI agents; the chat page and board; host hooks and MCP server; autopilot; tools/watch.py. Put under version control after the weekend's changes so they have history and can be rolled back. Local only.
300 lines
13 KiB
Python
300 lines
13 KiB
Python
"""Projects: which team, roster and database a Harbor command talks to.
|
|
|
|
A project is a TOML file. Registered projects live in <home>/projects/<name>.toml, where <home> is $HARBOR_HOME, or
|
|
the `home` set in local.toml next to this package, or ~/.harbor. A directory tree joins a project with a
|
|
`.harbor.toml` file at its root (found by walking up from the working directory, the way git finds a repository);
|
|
that file either names a registered project (`project = "open-seas"`) or holds a full definition. A marker that says
|
|
`boundary = true` ends the walk: nothing below it belongs to a project marked further up.
|
|
|
|
name = "open-seas"
|
|
title = "Open Sea's decomp"
|
|
database = 'R:\\Data\\open-seas\\harbor.db' # default: <home>/projects/<name>/harbor.db
|
|
mirror = 'R:\\OpenSeas\\agents-chat.md' # optional plain-text copy of every message
|
|
human = "User" # the person; only the web page speaks as them
|
|
coordinators = ["Claude"] # may set bands, stop others and land review items
|
|
reviewers = ["Claude"] # default reviewers of submitted items
|
|
|
|
[identities] # optional: working-directory name -> agent
|
|
"open-seas-decomp-deepseek" = "DeepSeek"
|
|
|
|
[git] # optional: the branch work lands on
|
|
main = "main"
|
|
|
|
[policy] # optional rules Harbor enforces
|
|
land_requires = ["Kimi"] # approvals needed (since the last submit) before landing
|
|
all_per_hour = 12 # cap on messages to everyone per agent per hour
|
|
|
|
[templates.decomp] # optional task pipelines: harbor task template decomp fn:0x...
|
|
steps = [
|
|
{name = "draft", title = "Make {scope} readable", assignee = "{owner}"},
|
|
{name = "names", title = "Name the helpers {scope} calls", assignee = "{owner}", after = ["draft"]},
|
|
]
|
|
|
|
[attention] # optional thresholds in minutes
|
|
ack_overdue = 30
|
|
item_stale = 120
|
|
agent_quiet = 60
|
|
|
|
[[agents]]
|
|
name = "Claude"
|
|
role = "Integration and review"
|
|
initials = "CL"
|
|
color = "#cc8a65"
|
|
host = "claude" # optional: codex, opencode, deepseek, kimi, claude
|
|
workdir = 'R:/code/project' # optional: shown in restart prompts
|
|
instructions = "Owns main; reviews everything." # optional: standing instructions for restart prompts
|
|
[[agents]]
|
|
name = "User"
|
|
kind = "human"
|
|
role = "Project owner"
|
|
initials = "You"
|
|
"""
|
|
import os
|
|
import re
|
|
import tomllib
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
|
|
PACKAGE_ROOT = Path(__file__).resolve().parents[1]
|
|
MARKER = ".harbor.toml"
|
|
NAME = re.compile(r"[a-z0-9][a-z0-9._-]{0,63}")
|
|
PALETTE = ("#cc8a65", "#4688e6", "#8071ce", "#39a48d", "#b45b9b", "#d0913c", "#5f8f3a", "#c0564f", "#3d8fa8")
|
|
|
|
|
|
class ProjectError(Exception):
|
|
pass
|
|
|
|
|
|
def harbor_home():
|
|
if os.environ.get("HARBOR_HOME"):
|
|
return Path(os.environ["HARBOR_HOME"])
|
|
local = PACKAGE_ROOT / "local.toml"
|
|
if local.exists():
|
|
home = tomllib.loads(local.read_text(encoding="utf-8")).get("home")
|
|
if home:
|
|
return Path(home)
|
|
return Path.home() / ".harbor"
|
|
|
|
|
|
@dataclass
|
|
class Project:
|
|
name: str
|
|
title: str
|
|
db: Path
|
|
mirror: Path | None
|
|
agents: list = field(default_factory=list)
|
|
human: str = "User"
|
|
coordinators: tuple = ()
|
|
reviewers: tuple = ()
|
|
identities: dict = field(default_factory=dict)
|
|
attention: dict = field(default_factory=dict)
|
|
source: Path | None = None
|
|
git_main: str = "main"
|
|
templates: dict = field(default_factory=dict)
|
|
policy: dict = field(default_factory=dict)
|
|
autopilot: bool = False # running agents keep working instead of ending their turns (hooks)
|
|
shared_files: tuple = () # glob patterns (repo-relative) every agent edits and the integrator merges: not collisions
|
|
|
|
@property
|
|
def names(self):
|
|
return [a["name"] for a in self.agents]
|
|
|
|
@property
|
|
def ai_names(self):
|
|
return [a["name"] for a in self.agents if a["kind"] != "human"]
|
|
|
|
@property
|
|
def attachments(self):
|
|
return self.db.parent / "attachments"
|
|
|
|
@property
|
|
def backups(self):
|
|
return self.db.parent / "backups"
|
|
|
|
def identity_for(self, cwd):
|
|
"""The agent whose working directory this is (from [identities]), or None."""
|
|
parts = [p.lower() for p in Path(cwd).resolve().parts]
|
|
for key, agent in self.identities.items():
|
|
if key.lower() in parts:
|
|
return agent
|
|
return None
|
|
|
|
def threshold(self, key, default):
|
|
return int(self.attention.get(key, default))
|
|
|
|
def public(self):
|
|
return {"name": self.name, "title": self.title, "human": self.human, "agents": self.agents,
|
|
"coordinators": list(self.coordinators)}
|
|
|
|
|
|
def _normalise_agents(raw, human):
|
|
agents, seen = [], set()
|
|
for i, a in enumerate(raw):
|
|
name = str(a.get("name", "")).strip()
|
|
if not re.fullmatch(r"[A-Za-z][A-Za-z0-9_-]{0,31}", name):
|
|
raise ProjectError(f"agent name {name!r}: letters, digits, _ or -, starting with a letter")
|
|
if name.lower() == "all" or name in seen:
|
|
raise ProjectError(f"agent name {name!r} is reserved or repeated")
|
|
seen.add(name)
|
|
kind = "human" if (a.get("kind") == "human" or name == human) else "ai"
|
|
agents.append({"name": name, "role": str(a.get("role", "")), "kind": kind,
|
|
"initials": str(a.get("initials") or ("You" if kind == "human" else name[:2].upper())),
|
|
"color": str(a.get("color") or ("#536070" if kind == "human" else PALETTE[i % len(PALETTE)])),
|
|
"workdir": str(a.get("workdir", "")), "instructions": str(a.get("instructions", "")),
|
|
"host": str(a.get("host", "")), "git_submits": a.get("git_submits", True)})
|
|
if human not in seen:
|
|
agents.append({"name": human, "role": "Project owner", "kind": "human", "initials": "You", "color": "#536070"})
|
|
return agents
|
|
|
|
|
|
def _templates(raw, names):
|
|
out = {}
|
|
for name, spec in (raw or {}).items():
|
|
steps = spec.get("steps") if isinstance(spec, dict) else None
|
|
if not isinstance(steps, list) or not steps:
|
|
raise ProjectError(f"template {name!r} needs a list of steps")
|
|
seen = set()
|
|
for step in steps:
|
|
if not isinstance(step, dict) or not step.get("title"):
|
|
raise ProjectError(f"template {name!r}: every step needs a title")
|
|
for dep in step.get("after", []):
|
|
if dep not in seen:
|
|
raise ProjectError(f"template {name!r}: step {step.get('name')!r} comes after unknown step {dep!r}")
|
|
assignee = step.get("assignee", "")
|
|
if assignee and "{" not in assignee and assignee not in names:
|
|
raise ProjectError(f"template {name!r}: {assignee!r} is not on the roster")
|
|
seen.add(step.get("name", ""))
|
|
out[name] = {"steps": steps}
|
|
return out
|
|
|
|
|
|
def _policy(raw, names):
|
|
policy = dict(raw or {})
|
|
if "land_requires" in policy:
|
|
value = policy["land_requires"]
|
|
value = [value] if isinstance(value, str) else value
|
|
if not isinstance(value, list) or not all(v in names for v in value):
|
|
raise ProjectError("policy land_requires must list agents on the roster")
|
|
policy["land_requires"] = value
|
|
if "all_per_hour" in policy:
|
|
try:
|
|
policy["all_per_hour"] = int(policy["all_per_hour"])
|
|
except (TypeError, ValueError):
|
|
raise ProjectError("policy all_per_hour must be a whole number")
|
|
return policy
|
|
|
|
|
|
def project_from_dict(data, source=None, home=None):
|
|
home = Path(home) if home else harbor_home()
|
|
name = str(data.get("name", "")).strip()
|
|
if not NAME.fullmatch(name):
|
|
raise ProjectError(f"project name {name!r}: lower-case letters, digits, '.', '_' or '-'")
|
|
human = str(data.get("human", "User"))
|
|
agents = _normalise_agents(data.get("agents", []), human)
|
|
names = {a["name"] for a in agents}
|
|
coordinators = tuple(data.get("coordinators", ()))
|
|
reviewers = tuple(data.get("reviewers", coordinators[:1]))
|
|
for who in (*coordinators, *reviewers, *data.get("identities", {}).values()):
|
|
if who not in names:
|
|
raise ProjectError(f"{who!r} is named in the project but is not in its agents")
|
|
base = source.parent if source else home
|
|
db = Path(data["database"]) if data.get("database") else home / "projects" / name / "harbor.db"
|
|
mirror = data.get("mirror")
|
|
return Project(name=name, title=str(data.get("title", name)), db=db if db.is_absolute() else base / db,
|
|
mirror=(Path(mirror) if Path(mirror).is_absolute() else base / mirror) if mirror else None,
|
|
agents=agents, human=human, coordinators=coordinators, reviewers=reviewers,
|
|
identities=dict(data.get("identities", {})), attention=dict(data.get("attention", {})),
|
|
source=source, git_main=str(data.get("git", {}).get("main", "main")),
|
|
templates=_templates(data.get("templates", {}), names), policy=_policy(data.get("policy"), names),
|
|
autopilot=bool(data.get("autopilot", False)),
|
|
shared_files=tuple(str(p).replace("\\", "/") for p in data.get("shared_files", [])))
|
|
|
|
|
|
def load_file(path, home=None):
|
|
path = Path(path)
|
|
try:
|
|
data = tomllib.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, tomllib.TOMLDecodeError) as error:
|
|
raise ProjectError(f"{path}: {error}")
|
|
if "project" in data and "name" not in data: # a marker that points at a registered project
|
|
return load_named(data["project"], home)
|
|
return project_from_dict(data, source=path, home=home)
|
|
|
|
|
|
def load_named(name, home=None):
|
|
home = Path(home) if home else harbor_home()
|
|
path = home / "projects" / f"{name}.toml"
|
|
if not path.exists():
|
|
known = ", ".join(list_names(home)) or "none"
|
|
raise ProjectError(f"no project {name!r} in {home / 'projects'} (registered: {known})")
|
|
return load_file(path, home)
|
|
|
|
|
|
def list_names(home=None):
|
|
folder = (Path(home) if home else harbor_home()) / "projects"
|
|
return sorted(p.stem for p in folder.glob("*.toml")) if folder.exists() else []
|
|
|
|
|
|
def find_marker(cwd=None):
|
|
"""The nearest .harbor.toml at or above cwd, or None (also when the nearest one is a boundary)."""
|
|
here = Path(cwd or os.getcwd()).resolve()
|
|
for folder in (here, *here.parents):
|
|
candidate = folder / MARKER
|
|
if candidate.is_file():
|
|
try:
|
|
boundary = tomllib.loads(candidate.read_text(encoding="utf-8")).get("boundary") is True
|
|
except (OSError, tomllib.TOMLDecodeError):
|
|
boundary = False # load_file reports the broken marker
|
|
return None if boundary else candidate
|
|
return None
|
|
|
|
|
|
def default_name(home=None):
|
|
config = (Path(home) if home else harbor_home()) / "config.toml"
|
|
if config.exists():
|
|
return tomllib.loads(config.read_text(encoding="utf-8")).get("default_project")
|
|
return None
|
|
|
|
|
|
def resolve(spec=None, cwd=None, home=None, use_default=True):
|
|
"""--project value, else $HARBOR_PROJECT, else the nearest .harbor.toml, else (when use_default) the home's
|
|
default project. Host hooks pass use_default=False so they stay silent outside Harbor projects."""
|
|
spec = spec or os.environ.get("HARBOR_PROJECT")
|
|
if spec:
|
|
return load_file(spec, home) if spec.endswith(".toml") else load_named(spec, home)
|
|
marker = find_marker(cwd)
|
|
if marker:
|
|
return load_file(marker, home)
|
|
name = default_name(home) if use_default else None
|
|
if name:
|
|
return load_named(name, home)
|
|
raise ProjectError("no Harbor project here: pass --project, set HARBOR_PROJECT, add a .harbor.toml, "
|
|
"or create one with `harbor init`")
|
|
|
|
|
|
def write_project(path, name, title, agents, human="User", coordinators=(), database=None, mirror=None,
|
|
identities=None):
|
|
"""Write a new project file (refuses to overwrite)."""
|
|
path = Path(path)
|
|
if path.exists():
|
|
raise ProjectError(f"{path} already exists")
|
|
q = lambda s: "'" + str(s).replace("'", "") + "'"
|
|
lines = [f'name = "{name}"', f"title = {q(title)}"]
|
|
if database:
|
|
lines.append(f"database = {q(database)}")
|
|
if mirror:
|
|
lines.append(f"mirror = {q(mirror)}")
|
|
lines.append(f'human = "{human}"')
|
|
lines.append("coordinators = [" + ", ".join(f'"{c}"' for c in coordinators) + "]")
|
|
if identities:
|
|
lines.append("\n[identities]")
|
|
lines += [f"{q(k)} = \"{v}\"" for k, v in identities.items()]
|
|
for a in agents:
|
|
lines.append("\n[[agents]]")
|
|
lines += [f"{k} = {q(v)}" for k, v in a.items() if v]
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
text = "\n".join(lines) + "\n"
|
|
project_from_dict(tomllib.loads(text), source=path) # validate before writing
|
|
path.write_text(text, encoding="utf-8")
|
|
return path
|