From 9b6ea3bc576b4aaca29243911594f3bc70739095 Mon Sep 17 00:00:00 2001 From: KeelrLeek <156356974+KeelrLeek@users.noreply.github.com> Date: Wed, 7 Oct 2026 08:43:49 -0500 Subject: [PATCH] hooks: any agent's hook restarts the project's web server when it is down ([server] port and start; at most once a minute, detached), so the board and chat no longer depend on the coordinator's session --- harbor/hooks.py | 37 +++ harbor/projects.py | 632 +++++++++++++++++++++++---------------------- 2 files changed, 354 insertions(+), 315 deletions(-) diff --git a/harbor/hooks.py b/harbor/hooks.py index ef6ad2f..e2d06e7 100644 --- a/harbor/hooks.py +++ b/harbor/hooks.py @@ -200,6 +200,42 @@ def _context(host, event, text): print(text) +def ensure_server(project, every=60): + """Restart the project's web server when it is down: [server] port and start (a command). Any agent's hook + does it, at most once a minute, so the board and chat come back without the coordinator's session. Detached, + so the server outlives the hook (and the app) that started it.""" + port, start = project.server.get("port"), project.server.get("start") + if not port or not start: + return + stamp = project.db.parent / "server-check.stamp" + try: + import time + if stamp.exists() and time.time() - stamp.stat().st_mtime < every: + return + stamp.touch() + import socket + with socket.socket() as connection: + connection.settimeout(0.3) + if connection.connect_ex(("127.0.0.1", int(port))) == 0: + return + import subprocess + flags = 0 + if os.name == "nt": + flags = (subprocess.DETACHED_PROCESS | subprocess.CREATE_NEW_PROCESS_GROUP + | subprocess.CREATE_BREAKAWAY_FROM_JOB | subprocess.CREATE_NO_WINDOW) + command = start if isinstance(start, list) else [start] + try: + subprocess.Popen(command, creationflags=flags, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, close_fds=True) + except OSError: # the job may refuse breakaway: start it without + subprocess.Popen(command, creationflags=flags & ~getattr(subprocess, "CREATE_BREAKAWAY_FROM_JOB", 0), + stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + close_fds=True) + print(f"harbor: the web server on port {port} was down; restarting it", file=sys.stderr) + except Exception as error: # never break the host because of this + print(f"harbor: server check failed: {error}", file=sys.stderr) + + def run(event, project_spec, agent, host="generic"): payload = _payload() try: @@ -216,6 +252,7 @@ def run(event, project_spec, agent, host="generic"): agent = project.session_identity(agent or os.environ.get("HARBOR_AGENT"), cwd) # the session's own folder if not agent or agent not in project.ai_names: return 0 + ensure_server(project) harbor = Harbor(project, busy_ms=1500, write_attempts=2) try: return _run(event, harbor, agent, host, payload) diff --git a/harbor/projects.py b/harbor/projects.py index cfe9bbc..30d8db3 100644 --- a/harbor/projects.py +++ b/harbor/projects.py @@ -1,315 +1,317 @@ -"""Projects: which team, roster and database a Harbor command talks to. - -A project is a TOML file. Registered projects live in /projects/.toml, where 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: /projects//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 - logbook: dict = field(default_factory=dict) # [logbook]: dir, auto, decomp, stats_repo (harbor.logbook_auto) - reverse: dict = field(default_factory=dict) # [reverse]: image, base, facts, imports, annotations, src (harbor.reverse) - - @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 host_of(self, agent): - return next((a["host"] for a in self.agents if a["name"] == agent), "") - - def session_identity(self, pinned, cwd): - """Who a session is when one app runs several of the project's agents (Codex and Codex2): the agent its - working folder names in [identities], if that agent runs in the same app as the installed (pinned) one; - otherwise the pinned agent. Without a pinned agent, the folder alone decides.""" - found = self.identity_for(cwd) if cwd else None - if found and found in self.ai_names and (not pinned or self.host_of(found) == self.host_of(pinned)): - return found - return pinned or found - - 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", [])), - logbook=dict(data.get("logbook", {})), - reverse=dict(data.get("reverse", {}))) - - -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 +"""Projects: which team, roster and database a Harbor command talks to. + +A project is a TOML file. Registered projects live in /projects/.toml, where 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: /projects//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 + logbook: dict = field(default_factory=dict) # [logbook]: dir, auto, decomp, stats_repo (harbor.logbook_auto) + reverse: dict = field(default_factory=dict) # [reverse]: image, base, facts, imports, annotations, src (harbor.reverse) + server: dict = field(default_factory=dict) # [server]: port, start (a command that starts the web server) + + @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 host_of(self, agent): + return next((a["host"] for a in self.agents if a["name"] == agent), "") + + def session_identity(self, pinned, cwd): + """Who a session is when one app runs several of the project's agents (Codex and Codex2): the agent its + working folder names in [identities], if that agent runs in the same app as the installed (pinned) one; + otherwise the pinned agent. Without a pinned agent, the folder alone decides.""" + found = self.identity_for(cwd) if cwd else None + if found and found in self.ai_names and (not pinned or self.host_of(found) == self.host_of(pinned)): + return found + return pinned or found + + 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", [])), + logbook=dict(data.get("logbook", {})), + reverse=dict(data.get("reverse", {})), + server=dict(data.get("server", {}))) + + +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