Keelrleek/Harbor
1
1 Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source
This commit is contained in:
KeelrLeek committed 2026-10-07 08:43:49 -05:00
1 parent 940cacc2d2
commit 9b6ea3bc57
2 files changed
+354 -315
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

harbor/hooks.py
+37
View File
@@ -200,6 +200,42 @@ def _context(host, event, text):
print(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"): def run(event, project_spec, agent, host="generic"):
payload = _payload() payload = _payload()
try: 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 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: if not agent or agent not in project.ai_names:
return 0 return 0
ensure_server(project)
harbor = Harbor(project, busy_ms=1500, write_attempts=2) harbor = Harbor(project, busy_ms=1500, write_attempts=2)
try: try:
return _run(event, harbor, agent, host, payload) return _run(event, harbor, agent, host, payload)
harbor/projects.py
+317 -315
View File
@@ -1,315 +1,317 @@
"""Projects: which team, roster and database a Harbor command talks to. """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 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 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); `.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 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. `boundary = true` ends the walk: nothing below it belongs to a project marked further up.
name = "open-seas" name = "open-seas"
title = "Open Sea's decomp" title = "Open Sea's decomp"
database = 'R:\\Data\\open-seas\\harbor.db' # default: <home>/projects/<name>/harbor.db 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 mirror = 'R:\\OpenSeas\\agents-chat.md' # optional plain-text copy of every message
human = "User" # the person; only the web page speaks as them human = "User" # the person; only the web page speaks as them
coordinators = ["Claude"] # may set bands, stop others and land review items coordinators = ["Claude"] # may set bands, stop others and land review items
reviewers = ["Claude"] # default reviewers of submitted items reviewers = ["Claude"] # default reviewers of submitted items
[identities] # optional: working-directory name -> agent [identities] # optional: working-directory name -> agent
"open-seas-decomp-deepseek" = "DeepSeek" "open-seas-decomp-deepseek" = "DeepSeek"
[git] # optional: the branch work lands on [git] # optional: the branch work lands on
main = "main" main = "main"
[policy] # optional rules Harbor enforces [policy] # optional rules Harbor enforces
land_requires = ["Kimi"] # approvals needed (since the last submit) before landing land_requires = ["Kimi"] # approvals needed (since the last submit) before landing
all_per_hour = 12 # cap on messages to everyone per agent per hour all_per_hour = 12 # cap on messages to everyone per agent per hour
[templates.decomp] # optional task pipelines: harbor task template decomp fn:0x... [templates.decomp] # optional task pipelines: harbor task template decomp fn:0x...
steps = [ steps = [
{name = "draft", title = "Make {scope} readable", assignee = "{owner}"}, {name = "draft", title = "Make {scope} readable", assignee = "{owner}"},
{name = "names", title = "Name the helpers {scope} calls", assignee = "{owner}", after = ["draft"]}, {name = "names", title = "Name the helpers {scope} calls", assignee = "{owner}", after = ["draft"]},
] ]
[attention] # optional thresholds in minutes [attention] # optional thresholds in minutes
ack_overdue = 30 ack_overdue = 30
item_stale = 120 item_stale = 120
agent_quiet = 60 agent_quiet = 60
[[agents]] [[agents]]
name = "Claude" name = "Claude"
role = "Integration and review" role = "Integration and review"
initials = "CL" initials = "CL"
color = "#cc8a65" color = "#cc8a65"
host = "claude" # optional: codex, opencode, deepseek, kimi, claude host = "claude" # optional: codex, opencode, deepseek, kimi, claude
workdir = 'R:/code/project' # optional: shown in restart prompts workdir = 'R:/code/project' # optional: shown in restart prompts
instructions = "Owns main; reviews everything." # optional: standing instructions for restart prompts instructions = "Owns main; reviews everything." # optional: standing instructions for restart prompts
[[agents]] [[agents]]
name = "User" name = "User"
kind = "human" kind = "human"
role = "Project owner" role = "Project owner"
initials = "You" initials = "You"
""" """
import os import os
import re import re
import tomllib import tomllib
from dataclasses import dataclass, field from dataclasses import dataclass, field
from pathlib import Path from pathlib import Path
PACKAGE_ROOT = Path(__file__).resolve().parents[1] PACKAGE_ROOT = Path(__file__).resolve().parents[1]
MARKER = ".harbor.toml" MARKER = ".harbor.toml"
NAME = re.compile(r"[a-z0-9][a-z0-9._-]{0,63}") NAME = re.compile(r"[a-z0-9][a-z0-9._-]{0,63}")
PALETTE = ("#cc8a65", "#4688e6", "#8071ce", "#39a48d", "#b45b9b", "#d0913c", "#5f8f3a", "#c0564f", "#3d8fa8") PALETTE = ("#cc8a65", "#4688e6", "#8071ce", "#39a48d", "#b45b9b", "#d0913c", "#5f8f3a", "#c0564f", "#3d8fa8")
class ProjectError(Exception): class ProjectError(Exception):
pass pass
def harbor_home(): def harbor_home():
if os.environ.get("HARBOR_HOME"): if os.environ.get("HARBOR_HOME"):
return Path(os.environ["HARBOR_HOME"]) return Path(os.environ["HARBOR_HOME"])
local = PACKAGE_ROOT / "local.toml" local = PACKAGE_ROOT / "local.toml"
if local.exists(): if local.exists():
home = tomllib.loads(local.read_text(encoding="utf-8")).get("home") home = tomllib.loads(local.read_text(encoding="utf-8")).get("home")
if home: if home:
return Path(home) return Path(home)
return Path.home() / ".harbor" return Path.home() / ".harbor"
@dataclass @dataclass
class Project: class Project:
name: str name: str
title: str title: str
db: Path db: Path
mirror: Path | None mirror: Path | None
agents: list = field(default_factory=list) agents: list = field(default_factory=list)
human: str = "User" human: str = "User"
coordinators: tuple = () coordinators: tuple = ()
reviewers: tuple = () reviewers: tuple = ()
identities: dict = field(default_factory=dict) identities: dict = field(default_factory=dict)
attention: dict = field(default_factory=dict) attention: dict = field(default_factory=dict)
source: Path | None = None source: Path | None = None
git_main: str = "main" git_main: str = "main"
templates: dict = field(default_factory=dict) templates: dict = field(default_factory=dict)
policy: dict = field(default_factory=dict) policy: dict = field(default_factory=dict)
autopilot: bool = False # running agents keep working instead of ending their turns (hooks) 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 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) 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) 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): @property
return [a["name"] for a in self.agents] def names(self):
return [a["name"] for a in self.agents]
@property
def ai_names(self): @property
return [a["name"] for a in self.agents if a["kind"] != "human"] def ai_names(self):
return [a["name"] for a in self.agents if a["kind"] != "human"]
@property
def attachments(self): @property
return self.db.parent / "attachments" def attachments(self):
return self.db.parent / "attachments"
@property
def backups(self): @property
return self.db.parent / "backups" def backups(self):
return self.db.parent / "backups"
def identity_for(self, cwd):
"""The agent whose working directory this is (from [identities]), or None.""" def identity_for(self, cwd):
parts = [p.lower() for p in Path(cwd).resolve().parts] """The agent whose working directory this is (from [identities]), or None."""
for key, agent in self.identities.items(): parts = [p.lower() for p in Path(cwd).resolve().parts]
if key.lower() in parts: for key, agent in self.identities.items():
return agent if key.lower() in parts:
return None return agent
return None
def host_of(self, agent):
return next((a["host"] for a in self.agents if a["name"] == agent), "") 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 def session_identity(self, pinned, cwd):
working folder names in [identities], if that agent runs in the same app as the installed (pinned) one; """Who a session is when one app runs several of the project's agents (Codex and Codex2): the agent its
otherwise the pinned agent. Without a pinned agent, the folder alone decides.""" working folder names in [identities], if that agent runs in the same app as the installed (pinned) one;
found = self.identity_for(cwd) if cwd else None otherwise the pinned agent. Without a pinned agent, the folder alone decides."""
if found and found in self.ai_names and (not pinned or self.host_of(found) == self.host_of(pinned)): found = self.identity_for(cwd) if cwd else None
return found if found and found in self.ai_names and (not pinned or self.host_of(found) == self.host_of(pinned)):
return pinned or found return found
return pinned or found
def threshold(self, key, default):
return int(self.attention.get(key, default)) 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, def public(self):
"coordinators": list(self.coordinators)} 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() def _normalise_agents(raw, human):
for i, a in enumerate(raw): agents, seen = [], set()
name = str(a.get("name", "")).strip() for i, a in enumerate(raw):
if not re.fullmatch(r"[A-Za-z][A-Za-z0-9_-]{0,31}", name): name = str(a.get("name", "")).strip()
raise ProjectError(f"agent name {name!r}: letters, digits, _ or -, starting with a letter") if not re.fullmatch(r"[A-Za-z][A-Za-z0-9_-]{0,31}", name):
if name.lower() == "all" or name in seen: raise ProjectError(f"agent name {name!r}: letters, digits, _ or -, starting with a letter")
raise ProjectError(f"agent name {name!r} is reserved or repeated") if name.lower() == "all" or name in seen:
seen.add(name) raise ProjectError(f"agent name {name!r} is reserved or repeated")
kind = "human" if (a.get("kind") == "human" or name == human) else "ai" seen.add(name)
agents.append({"name": name, "role": str(a.get("role", "")), "kind": kind, kind = "human" if (a.get("kind") == "human" or name == human) else "ai"
"initials": str(a.get("initials") or ("You" if kind == "human" else name[:2].upper())), agents.append({"name": name, "role": str(a.get("role", "")), "kind": kind,
"color": str(a.get("color") or ("#536070" if kind == "human" else PALETTE[i % len(PALETTE)])), "initials": str(a.get("initials") or ("You" if kind == "human" else name[:2].upper())),
"workdir": str(a.get("workdir", "")), "instructions": str(a.get("instructions", "")), "color": str(a.get("color") or ("#536070" if kind == "human" else PALETTE[i % len(PALETTE)])),
"host": str(a.get("host", "")), "git_submits": a.get("git_submits", True)}) "workdir": str(a.get("workdir", "")), "instructions": str(a.get("instructions", "")),
if human not in seen: "host": str(a.get("host", "")), "git_submits": a.get("git_submits", True)})
agents.append({"name": human, "role": "Project owner", "kind": "human", "initials": "You", "color": "#536070"}) if human not in seen:
return agents agents.append({"name": human, "role": "Project owner", "kind": "human", "initials": "You", "color": "#536070"})
return agents
def _templates(raw, names):
out = {} def _templates(raw, names):
for name, spec in (raw or {}).items(): out = {}
steps = spec.get("steps") if isinstance(spec, dict) else None for name, spec in (raw or {}).items():
if not isinstance(steps, list) or not steps: steps = spec.get("steps") if isinstance(spec, dict) else None
raise ProjectError(f"template {name!r} needs a list of steps") if not isinstance(steps, list) or not steps:
seen = set() raise ProjectError(f"template {name!r} needs a list of steps")
for step in steps: seen = set()
if not isinstance(step, dict) or not step.get("title"): for step in steps:
raise ProjectError(f"template {name!r}: every step needs a title") if not isinstance(step, dict) or not step.get("title"):
for dep in step.get("after", []): raise ProjectError(f"template {name!r}: every step needs a title")
if dep not in seen: for dep in step.get("after", []):
raise ProjectError(f"template {name!r}: step {step.get('name')!r} comes after unknown step {dep!r}") if dep not in seen:
assignee = step.get("assignee", "") raise ProjectError(f"template {name!r}: step {step.get('name')!r} comes after unknown step {dep!r}")
if assignee and "{" not in assignee and assignee not in names: assignee = step.get("assignee", "")
raise ProjectError(f"template {name!r}: {assignee!r} is not on the roster") if assignee and "{" not in assignee and assignee not in names:
seen.add(step.get("name", "")) raise ProjectError(f"template {name!r}: {assignee!r} is not on the roster")
out[name] = {"steps": steps} seen.add(step.get("name", ""))
return out out[name] = {"steps": steps}
return out
def _policy(raw, names):
policy = dict(raw or {}) def _policy(raw, names):
if "land_requires" in policy: policy = dict(raw or {})
value = policy["land_requires"] if "land_requires" in policy:
value = [value] if isinstance(value, str) else value value = policy["land_requires"]
if not isinstance(value, list) or not all(v in names for v in value): value = [value] if isinstance(value, str) else value
raise ProjectError("policy land_requires must list agents on the roster") if not isinstance(value, list) or not all(v in names for v in value):
policy["land_requires"] = value raise ProjectError("policy land_requires must list agents on the roster")
if "all_per_hour" in policy: policy["land_requires"] = value
try: if "all_per_hour" in policy:
policy["all_per_hour"] = int(policy["all_per_hour"]) try:
except (TypeError, ValueError): policy["all_per_hour"] = int(policy["all_per_hour"])
raise ProjectError("policy all_per_hour must be a whole number") except (TypeError, ValueError):
return policy 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() def project_from_dict(data, source=None, home=None):
name = str(data.get("name", "")).strip() home = Path(home) if home else harbor_home()
if not NAME.fullmatch(name): name = str(data.get("name", "")).strip()
raise ProjectError(f"project name {name!r}: lower-case letters, digits, '.', '_' or '-'") if not NAME.fullmatch(name):
human = str(data.get("human", "User")) raise ProjectError(f"project name {name!r}: lower-case letters, digits, '.', '_' or '-'")
agents = _normalise_agents(data.get("agents", []), human) human = str(data.get("human", "User"))
names = {a["name"] for a in agents} agents = _normalise_agents(data.get("agents", []), human)
coordinators = tuple(data.get("coordinators", ())) names = {a["name"] for a in agents}
reviewers = tuple(data.get("reviewers", coordinators[:1])) coordinators = tuple(data.get("coordinators", ()))
for who in (*coordinators, *reviewers, *data.get("identities", {}).values()): reviewers = tuple(data.get("reviewers", coordinators[:1]))
if who not in names: for who in (*coordinators, *reviewers, *data.get("identities", {}).values()):
raise ProjectError(f"{who!r} is named in the project but is not in its agents") if who not in names:
base = source.parent if source else home raise ProjectError(f"{who!r} is named in the project but is not in its agents")
db = Path(data["database"]) if data.get("database") else home / "projects" / name / "harbor.db" base = source.parent if source else home
mirror = data.get("mirror") db = Path(data["database"]) if data.get("database") else home / "projects" / name / "harbor.db"
return Project(name=name, title=str(data.get("title", name)), db=db if db.is_absolute() else base / db, mirror = data.get("mirror")
mirror=(Path(mirror) if Path(mirror).is_absolute() else base / mirror) if mirror else None, return Project(name=name, title=str(data.get("title", name)), db=db if db.is_absolute() else base / db,
agents=agents, human=human, coordinators=coordinators, reviewers=reviewers, mirror=(Path(mirror) if Path(mirror).is_absolute() else base / mirror) if mirror else None,
identities=dict(data.get("identities", {})), attention=dict(data.get("attention", {})), agents=agents, human=human, coordinators=coordinators, reviewers=reviewers,
source=source, git_main=str(data.get("git", {}).get("main", "main")), identities=dict(data.get("identities", {})), attention=dict(data.get("attention", {})),
templates=_templates(data.get("templates", {}), names), policy=_policy(data.get("policy"), names), source=source, git_main=str(data.get("git", {}).get("main", "main")),
autopilot=bool(data.get("autopilot", False)), templates=_templates(data.get("templates", {}), names), policy=_policy(data.get("policy"), names),
shared_files=tuple(str(p).replace("\\", "/") for p in data.get("shared_files", [])), autopilot=bool(data.get("autopilot", False)),
logbook=dict(data.get("logbook", {})), shared_files=tuple(str(p).replace("\\", "/") for p in data.get("shared_files", [])),
reverse=dict(data.get("reverse", {}))) 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: def load_file(path, home=None):
data = tomllib.loads(path.read_text(encoding="utf-8")) path = Path(path)
except (OSError, tomllib.TOMLDecodeError) as error: try:
raise ProjectError(f"{path}: {error}") data = tomllib.loads(path.read_text(encoding="utf-8"))
if "project" in data and "name" not in data: # a marker that points at a registered project except (OSError, tomllib.TOMLDecodeError) as error:
return load_named(data["project"], home) raise ProjectError(f"{path}: {error}")
return project_from_dict(data, source=path, home=home) 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" def load_named(name, home=None):
if not path.exists(): home = Path(home) if home else harbor_home()
known = ", ".join(list_names(home)) or "none" path = home / "projects" / f"{name}.toml"
raise ProjectError(f"no project {name!r} in {home / 'projects'} (registered: {known})") if not path.exists():
return load_file(path, home) 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 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() def find_marker(cwd=None):
for folder in (here, *here.parents): """The nearest .harbor.toml at or above cwd, or None (also when the nearest one is a boundary)."""
candidate = folder / MARKER here = Path(cwd or os.getcwd()).resolve()
if candidate.is_file(): for folder in (here, *here.parents):
try: candidate = folder / MARKER
boundary = tomllib.loads(candidate.read_text(encoding="utf-8")).get("boundary") is True if candidate.is_file():
except (OSError, tomllib.TOMLDecodeError): try:
boundary = False # load_file reports the broken marker boundary = tomllib.loads(candidate.read_text(encoding="utf-8")).get("boundary") is True
return None if boundary else candidate except (OSError, tomllib.TOMLDecodeError):
return None 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(): def default_name(home=None):
return tomllib.loads(config.read_text(encoding="utf-8")).get("default_project") config = (Path(home) if home else harbor_home()) / "config.toml"
return None 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.""" def resolve(spec=None, cwd=None, home=None, use_default=True):
spec = spec or os.environ.get("HARBOR_PROJECT") """--project value, else $HARBOR_PROJECT, else the nearest .harbor.toml, else (when use_default) the home's
if spec: default project. Host hooks pass use_default=False so they stay silent outside Harbor projects."""
return load_file(spec, home) if spec.endswith(".toml") else load_named(spec, home) spec = spec or os.environ.get("HARBOR_PROJECT")
marker = find_marker(cwd) if spec:
if marker: return load_file(spec, home) if spec.endswith(".toml") else load_named(spec, home)
return load_file(marker, home) marker = find_marker(cwd)
name = default_name(home) if use_default else None if marker:
if name: return load_file(marker, home)
return load_named(name, home) name = default_name(home) if use_default else None
raise ProjectError("no Harbor project here: pass --project, set HARBOR_PROJECT, add a .harbor.toml, " if name:
"or create one with `harbor init`") 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).""" def write_project(path, name, title, agents, human="User", coordinators=(), database=None, mirror=None,
path = Path(path) identities=None):
if path.exists(): """Write a new project file (refuses to overwrite)."""
raise ProjectError(f"{path} already exists") path = Path(path)
q = lambda s: "'" + str(s).replace("'", "") + "'" if path.exists():
lines = [f'name = "{name}"', f"title = {q(title)}"] raise ProjectError(f"{path} already exists")
if database: q = lambda s: "'" + str(s).replace("'", "") + "'"
lines.append(f"database = {q(database)}") lines = [f'name = "{name}"', f"title = {q(title)}"]
if mirror: if database:
lines.append(f"mirror = {q(mirror)}") lines.append(f"database = {q(database)}")
lines.append(f'human = "{human}"') if mirror:
lines.append("coordinators = [" + ", ".join(f'"{c}"' for c in coordinators) + "]") lines.append(f"mirror = {q(mirror)}")
if identities: lines.append(f'human = "{human}"')
lines.append("\n[identities]") lines.append("coordinators = [" + ", ".join(f'"{c}"' for c in coordinators) + "]")
lines += [f"{q(k)} = \"{v}\"" for k, v in identities.items()] if identities:
for a in agents: lines.append("\n[identities]")
lines.append("\n[[agents]]") lines += [f"{q(k)} = \"{v}\"" for k, v in identities.items()]
lines += [f"{k} = {q(v)}" for k, v in a.items() if v] for a in agents:
path.parent.mkdir(parents=True, exist_ok=True) lines.append("\n[[agents]]")
text = "\n".join(lines) + "\n" lines += [f"{k} = {q(v)}" for k, v in a.items() if v]
project_from_dict(tomllib.loads(text), source=path) # validate before writing path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(text, encoding="utf-8") text = "\n".join(lines) + "\n"
return path project_from_dict(tomllib.loads(text), source=path) # validate before writing
path.write_text(text, encoding="utf-8")
return path