"""Harbor's local web front: the team chat page and the coordination board, for every registered project. harbor serve --port 8766 (or: py -3.12 -B harbor_server.py 8766) / chat for the default project /board its board /p// chat for another project /p//board Only this computer can reach it (127.0.0.1, Host header checked). Messages written here are the only ones Harbor accepts from the human. A browser POST must carry the page's per-start token and the browser's Sec-Fetch-Site header, so a script on this machine cannot speak as the human by accident. A program that forges all of that on purpose is breaking the team's rules; this is not a security boundary against software running as the same user. """ import argparse import http.server import json import re import secrets import sys import threading import time import traceback from pathlib import Path from urllib.parse import parse_qs, unquote, urlsplit from . import projects from .core import Harbor, HarborError WEB = Path(__file__).resolve().parent / "web" # The files pages load from static/: content type and how long a browser may keep them (seconds). STATIC = {"harbor.css": ("text/css; charset=utf-8", 60), "themes.css": ("text/css; charset=utf-8", 60), "themes.js": ("text/javascript; charset=utf-8", 60), "icons.svg": ("image/svg+xml", 60), "InterVariable.woff2": ("font/woff2", 86400)} TOKEN = secrets.token_urlsafe(24) LEGACY_LIMIT = 800 BACKUP_HOURS = 6 class Registry: """One Harbor (shared connection + lock) per project.""" def __init__(self, primary): self.primary = primary self.lock = threading.Lock() self.open = {} def names(self): names = projects.list_names() return [self.primary.name] + [n for n in names if n != self.primary.name] def get(self, name=None): name = name or self.primary.name with self.lock: if name not in self.open: project = self.primary if name == self.primary.name else projects.load_named(name) self.open[name] = (Harbor(project, shared=True), threading.Lock()) h, lock = self.open[name] with lock: h.refresh_project() # project-file edits apply without a restart return h, lock REGISTRY = None PORT = 8766 def chat_shape(m): """A message as the pages read it (the old {time, from, to, text} plus Harbor's fields).""" recipients = m["recipients"] return {"id": m["id"], "time": m["created_at"].replace("T", " ")[:16] + "Z", "at": m["created_at"], "from": m["sender"], "to": recipients[0] if len(recipients) == 1 else ", ".join(recipients), "recipients": recipients, "text": m["body"], "topic": m["topic"], "subject": m["subject"], "kind": m["kind"], "importance": m["importance"], "ack": m["ack_required"], "origin": m["origin"], "reply_to": m.get("reply_to"), "attachments": safe_attachments(m.get("refs", {}).get("attachments")), "corrects": m["corrects"] if isinstance(m.get("corrects"), int) else None, "due_at": m.get("due_at")} def safe_attachments(value): """Only stored attachment names reach the page (anything else in old data is dropped).""" if not isinstance(value, list): return [] return [v for v in value if isinstance(v, str) and re.fullmatch(r"[0-9a-f]{12}-[A-Za-z0-9._-]{1,80}", v)] def shaped(h, rows): """chat_shape for a batch, plus which of them were corrected later and which are pinned.""" out = [chat_shape(m) for m in rows] ids = [m["id"] for m in out] corrected = h.corrections_for(ids) if ids else {} pinned = {m["id"] for m in h.pins(limit=500)} asked = h.questions_by_message(ids) if ids else {} for m in out: if m["id"] in corrected: m["corrected_by"] = corrected[m["id"]] if m["id"] in pinned: m["pinned"] = True q = asked.get(m["id"]) if q: m["question"] = {"id": q["id"], "to": q["to_name"], "options": q["options"], "answer": q["answer"], "answered_by": q["answered_by"]} return out GIT = {} # project name -> {"at": monotonic, "radar": {...}} RUNNING = {} # agent name -> number of Harbor MCP server processes seen running (refreshed every minute) SEEN_WORK = {} def note_worktree_work(h, project, agent, git): """Record activity when the agent changed a file or committed in the last 3 minutes (once per change).""" now = time.time() found = [] change = h.worktree_change(agent) if change and now - change[0] < 180: found.append((change[0], "edit", change[1])) commit = (git or {}).get("last_commit") or {} if commit.get("at"): from .util import parse_iso at = parse_iso(commit["at"]).timestamp() if now - at < 180: found.append((at, "commit", f"{commit.get('sha', '')} {commit.get('subject', '')}"[:120])) for stamp, tool, target in found: key = (project, agent, tool) if SEEN_WORK.get(key, 0) < stamp: SEEN_WORK[key] = stamp h.activity(agent, tool, "", target=target, channel="files") def git_loop(): """Refresh every project's git radar in the background (git can be slow; requests only read the cache).""" while True: try: from .doctor import running_mcp_servers fresh = running_mcp_servers() # compute first, then swap: no empty moment RUNNING.clear() RUNNING.update(fresh) except Exception: traceback.print_exc() for name in REGISTRY.names(): try: shared, _ = REGISTRY.get(name) agents = [a for a in shared.project.agents if a["kind"] != "human" and a.get("workdir")] if not agents: continue own = Harbor(shared.project) # its own connection: git never holds the request lock try: radar = {a["name"]: own.git_status(a["name"], max_age=50) for a in agents} GIT[name] = {"at": time.monotonic(), "radar": radar} for a in agents: # fresh edits and commits are work, hooks or not try: note_worktree_work(own, name, a["name"], radar.get(a["name"]) or {}) except Exception: # noqa: BLE001 - a broken worktree must not stop the loop traceback.print_exc() finally: own.close() except Exception: traceback.print_exc() time.sleep(60) def toast_loop(): """Windows notifications for the human (when switched on for a project).""" from . import notify while True: time.sleep(10) for name in REGISTRY.names(): try: h, lock = REGISTRY.get(name) with lock: if not notify.enabled(h): continue pending, top = notify.pending_for_human(h) notify.mark(h, top) for m in pending: title = (f"{m['sender']} asks you" if m["kind"] == "question" else f"{m['sender']} (urgent)" if m["importance"] == "urgent" else m["sender"]) notify.toast(f"{title} ยท {h.project.title}", m["body"].split("\n", 1)[0]) except Exception: traceback.print_exc() def git_attention(h, radar): out = [] for agent, g in (radar or {}).items(): if not g or g.get("error") or h.stopped(agent): continue waiting = g.get("unsubmitted") or [] if len(waiting) >= 3 and h.minutes_since(waiting[-1]["at"]) >= 120: out.append({"kind": "unsubmitted", "agent": agent, "minutes": h.minutes_since(waiting[-1]["at"]), "text": f"{agent} has {len(waiting)} commits on {g['branch']} since its last submit " f"(oldest {h.minutes_since(waiting[-1]['at']) // 60} h ago) - nothing in the review queue"}) if (g.get("behind_main") or 0) >= 100: out.append({"kind": "old_base", "agent": agent, "minutes": 0, "text": f"{agent}'s branch {g['branch']} is {g['behind_main']} commits behind main"}) return out def overview(h): agents = [] links = h.connections() hosts = {a["name"]: a.get("host", "") for a in h.project.agents} for a in h.agents(): unread, urgent = h.unread_count(a["name"]) handoff = h.last_handoff(a["name"]) if a["kind"] != "human" else None agents.append({**a, "unread": unread, "urgent": urgent, "idle_minutes": h.minutes_since(a["last_seen"]), "doing_minutes": h.minutes_since(a.get("doing_at")), "host": hosts.get(a["name"], ""), "working": h.working(a), "active_minutes": h.minutes_since(a.get("active_at")), "last_target": (h.activity_feed(limit=1, agent=a["name"]) or [{}])[0].get("target"), "trial_until": h.trial(a["name"]) if a["kind"] != "human" else None, "connections": links.get(a["name"], []), "handoff": {"at": handoff["created_at"], "minutes": h.minutes_since(handoff["created_at"]), "text": handoff["body"][:600]} if handoff else None, "stopped": [f["key"] for f in h.stopped(a["name"])] if a["kind"] != "human" else []}) queue = [{**i, "age_minutes": h.minutes_since(i["created_at"]), "idle_minutes": h.minutes_since(i["updated_at"])} for i in h.items()] claims = [{**c, "age_minutes": h.minutes_since(c["created_at"])} for c in h.active_claims()] acks = [{**r, "body": r["body"].split("\n", 1)[0][:160], "age_minutes": h.minutes_since(r["created_at"])} for r in h.pending_acks()] attention = h.attention() + git_attention(h, (GIT.get(h.project.name) or {}).get("radar")) return {"now": h.now(), "project": h.project.public(), "flags": list(h.flags().values()), "agents": agents, "queue": queue, "claims": claims, "threads": h.threads(limit=30, agent=h.human), "pending_acks": acks, "attention": attention, "projects": REGISTRY.names(), "presence": h.presence(), "tasks": h.tasks(), "pins": [chat_shape(m) for m in h.pins(limit=10)], "questions": h.questions(open_only=True, to=h.human), "collisions": h.recent_collisions(), "metrics": h.metrics(points=40), "wins": h.wins("24h"), "git": (GIT.get(h.project.name) or {}).get("radar") or {}, "templates": list(h.project.templates), "toasts": __import__("harbor.notify", fromlist=["x"]).enabled(h)} def address_mentions(h, to, text): """'@Codex @Kimi please look' sent to All goes to Codex and Kimi.""" if to not in (None, "", "All"): return to names = [] for word in text.split(): m = re.fullmatch(r"@([A-Za-z][A-Za-z0-9_-]*)[,:]?", word) if not m: break match = next((n for n in h.project.names if n.lower() == m.group(1).lower()), None) if not match: break names.append(match) return names or to class Handler(http.server.BaseHTTPRequestHandler): server_version = "Harbor/2" # ------------------------------------------------------------------ plumbing def _reply(self, code, body, kind="application/json; charset=utf-8", extra=None): data = body if isinstance(body, bytes) else ( body.encode("utf-8") if isinstance(body, str) else json.dumps(body, ensure_ascii=False, default=str).encode("utf-8")) self.send_response(code) self.send_header("Content-Type", kind) self.send_header("Content-Length", str(len(data))) extra = dict(extra or {}) self.send_header("Cache-Control", extra.pop("Cache-Control", "no-store")) self.send_header("X-Content-Type-Options", "nosniff") self.send_header("Referrer-Policy", "no-referrer") for k, v in extra.items(): self.send_header(k, v) self.end_headers() self.wfile.write(data) def _host_ok(self): return self.headers.get("Host", "") in (f"127.0.0.1:{PORT}", f"localhost:{PORT}") def _route(self): """-> (project name or None, path inside the project, query dict).""" url = urlsplit(self.path) path = url.path name = None m = re.fullmatch(r"/p/([a-z0-9][a-z0-9._-]{0,63})(/.*)?", path) if m: name, path = m.group(1), m.group(2) or "/" return name, path, {k: v[-1] for k, v in parse_qs(url.query).items()} def _page(self, filename, project_name): text = (WEB / filename).read_text(encoding="utf-8") base = f"/p/{project_name}/" if project_name else "/" text = text.replace("", f'\n\n', 1) self._reply(200, text, "text/html; charset=utf-8") def _body(self): length = int(self.headers.get("Content-Length") or 0) if length > 100_000: raise HarborError("request too large") try: return json.loads(self.rfile.read(length) or b"{}") except ValueError: raise HarborError("the request body is not JSON") def _from_page(self): origin = self.headers.get("Origin") if origin and origin not in (f"http://127.0.0.1:{PORT}", f"http://localhost:{PORT}"): return "wrong origin" if self.headers.get("Sec-Fetch-Site") != "same-origin": return "not sent by the chat page in a browser" if not secrets.compare_digest(self.headers.get("X-Harbor-Token", ""), TOKEN): return "stale page; reload it" if not (self.headers.get("Content-Type") or "").startswith("application/json"): return "JSON only" return None def log_message(self, *args): pass # ------------------------------------------------------------------ GET def do_GET(self): if not self._host_ok(): return self._reply(403, {"error": "unknown host"}) name, path, query = self._route() try: if name and name not in REGISTRY.names(): return self._reply(404, {"error": f"no project {name}"}) if path.startswith("/static/") and path[8:] in STATIC: kind, age = STATIC[path[8:]] return self._reply(200, (WEB / path[8:]).read_bytes(), kind, {"Cache-Control": f"max-age={age}"}) if path == "/": return self._page("app.html", name) if path == "/board": return self._page("app.html", name) if path == "/api/projects": out = [] for n in REGISTRY.names(): h, lock = REGISTRY.get(n) with lock: out.append({"name": n, "title": h.project.title, "unread": h.unread_count(h.human)[0], "stopped": any(h.stopped(a) for a in h.project.ai_names)}) return self._reply(200, out) h, lock = REGISTRY.get(name) if path == "/api/logbook": from . import logbook return self._reply(200, logbook.read_all(h.project, log_limit=min(int(query.get("limit", 80)), 500))) if path.startswith("/api/decomp/"): # its own connection and lock: these answers take seconds and must not hold up the chat from . import decomp_api status, body = decomp_api.handle(h.project, path, query) return self._reply(status, body) if path.startswith("/files/"): filename = unquote(path[len("/files/"):]) if not re.fullmatch(r"[0-9a-f]{12}-[A-Za-z0-9._-]{1,80}", filename): return self._reply(404, {"error": "no such file"}) target = h.project.attachments / filename if not target.is_file(): return self._reply(404, {"error": "no such file"}) return self._reply(200, target.read_bytes(), "application/octet-stream", {"Content-Disposition": f'attachment; filename="{filename[13:]}"'}) with lock: if path == "/participants": return self._reply(200, [{"name": a["name"], "initials": a["initials"], "color": a["color"], "role": a["role"], "kind": a["kind"], "doing": a.get("doing"), "last_seen": a["last_seen"]} for a in h.agents()]) if path == "/messages": after = int(query.get("after", 0)) before = int(query.get("before", 0)) rows = (h.messages_after(after) if after else h.messages_before(before) if before else h.recent(LEGACY_LIMIT)) return self._reply(200, shaped(h, rows)) if path == "/api/receipts": ids = [int(i) for i in query.get("ids", "").split(",") if i.strip().isdigit()][:200] return self._reply(200, h.receipts(ids)) if path == "/api/overview": return self._reply(200, overview(h)) if path == "/api/thread": thread, messages = h.thread(query.get("id", "")) return self._reply(200, {"thread": thread, "messages": shaped(h, messages), "summary": h.summary(thread["id"])}) if path == "/api/tasks": from .work import OPEN_TASK_STATES, TASK_STATES return self._reply(200, h.tasks(states=TASK_STATES if query.get("all") == "1" else OPEN_TASK_STATES)) if path == "/api/task": task, events = h.task(query.get("id", "0")) return self._reply(200, {"task": task, "events": events}) if path == "/api/prompt": return self._reply(200, {"agent": query.get("agent"), "prompt": h.restart_prompt(query.get("agent", ""))}) if path == "/api/handoffs": return self._reply(200, h.handoffs(agent=query.get("agent") or None, limit=int(query.get("limit", 10)))) if path == "/api/pins": return self._reply(200, {"pins": shaped(h, h.pins()), "decisions": shaped(h, h.decisions())}) if path == "/api/launch": agents = h.launch_status() for a in agents: if RUNNING.get(a["agent"]) and not a["connected"]: a["connected"] = True a["connected_via"] = a["connected_via"] + ["Harbor tools running in its app"] return self._reply(200, {"agents": agents, "rollcall": h.rollcall_status(), "presence": h.presence()}) if path == "/api/feed": return self._reply(200, h.activity_feed(limit=min(int(query.get("limit", 60)), 300), agent=query.get("agent") or None)) if path == "/api/pulse": return self._reply(200, h.pulse(hours=min(int(query.get("hours", 12)), 72))) if path == "/api/context": return self._reply(200, {"text": h.context_text(query.get("scope", ""), limit=20), **{k: v for k, v in h.context(query.get("scope", "")).items() if k in ("scope", "notes", "tasks", "items")}}) if path == "/api/export": text = h.export_thread(query.get("thread", "")) name = re.sub(r"[^A-Za-z0-9._-]+", "_", query.get("thread", "thread"))[:60] or "thread" return self._reply(200, text, "text/markdown; charset=utf-8", {"Content-Disposition": f'attachment; filename="{name}.md"'}) if path == "/api/doctor": from . import doctor return self._reply(200, doctor.run(h.project, h, port=PORT, server_known=True)) if path == "/api/threads": return self._reply(200, h.threads(limit=int(query.get("limit", 60)), agent=h.human, include_legacy=query.get("archive") == "1")) if path == "/api/search": return self._reply(200, shaped(h, h.search(query.get("q", ""), limit=60))) if path == "/api/item": item, events = h.item(query.get("id", "0")) return self._reply(200, {"item": item, "events": events}) if path == "/api/inbox": return self._reply(200, [chat_shape(m) for m in h.inbox(h.human, unread_only=True, limit=100, mark_read=False)]) if path == "/api/digest": d = h.digest(query.get("since", "8h")) d["to_human"] = [chat_shape(m) for m in d["to_human"]] return self._reply(200, d) except (HarborError, projects.ProjectError) as error: return self._reply(400, {"error": str(error)}) except Exception: traceback.print_exc() return self._reply(500, {"error": "server error; see the server log"}) self._reply(404, {"error": "not found"}) # ------------------------------------------------------------------ POST def do_POST(self): if not self._host_ok(): return self._reply(403, {"error": "unknown host"}) refused = self._from_page() if refused: return self._reply(403, {"error": refused}) name, path, _ = self._route() try: if name and name not in REGISTRY.names(): return self._reply(404, {"error": f"no project {name}"}) payload = self._body() h, lock = REGISTRY.get(name) with lock: if path in ("/send", "/api/send"): text = str(payload.get("text", "")) to = payload.get("to") or (None if payload.get("reply_to") else "All") if isinstance(to, str) and "," in to: to = [n.strip() for n in to.split(",") if n.strip()] to = address_mentions(h, to, text) result = h.send(h.human, to, text, subject=payload.get("subject"), thread=payload.get("thread") or None, kind=payload.get("kind") or "note", importance=payload.get("importance") or "normal", ack_required=bool(payload.get("ack")), reply_to=payload.get("reply_to"), origin="web") return self._reply(200, {"ok": True, **result}) if path == "/api/flag": target = str(payload.get("target", "")) key = "stop:all" if target in ("all", "All") else f"stop:{target}" flags = h.set_flag(key, str(payload.get("value", "")), h.human, "web", reason=payload.get("reason") or None, until=payload.get("until") or None) return self._reply(200, {"ok": True, "flags": list(flags.values())}) if path == "/api/ack": done = h.ack(h.human, [int(i) for i in payload.get("ids", [])], note=payload.get("note")) return self._reply(200, {"ok": True, "acked": done}) if path == "/api/read": h.mark_read(h.human, payload.get("ids")) return self._reply(200, {"ok": True}) if path == "/api/rollcall": return self._reply(200, {"ok": True, "rollcall": h.rollcall(h.human, "web", payload.get("agents") or None)}) if path == "/api/bulk": r = h.bulk_tasks(h.human, str(payload.get("text", "")), template=payload.get("template") or None, priority=payload.get("priority") or "normal", assignee=payload.get("assignee") or None, origin="web") return self._reply(200, {"ok": True, "created": len(r["created"]), "skipped": r["skipped"]}) if path == "/api/toasts": from . import notify notify.set_enabled(h, bool(payload.get("on"))) if payload.get("on"): threading.Thread(target=notify.toast, args=("Harbor", f"Notifications are on for " f"{h.project.title}: questions, urgent messages and messages to you."), daemon=True).start() return self._reply(200, {"ok": True, "on": notify.enabled(h)}) if path == "/api/answer": q = h.answer(h.human, int(payload.get("question", 0)), str(payload.get("answer", "")), origin="web") return self._reply(200, {"ok": True, "question": q}) if path == "/api/presence": return self._reply(200, {"ok": True, "presence": h.set_presence( h.human, "web", str(payload.get("status", "here")), payload.get("until") or None, payload.get("note") or None)}) if path == "/api/ping": agent = str(payload.get("agent", "")) if agent not in h.project.ai_names: raise HarborError(f"no agent {agent!r}") r = h.send(h.human, [agent], "Are you there? Please acknowledge this message so I know you are " "running and reading Harbor.", thread="ping", subject="Ping", kind="question", importance="urgent", ack_required=True, origin="web") return self._reply(200, {"ok": True, **r}) if path == "/api/pin": pins = (h.unpin(h.human, int(payload["id"])) if payload.get("unpin") else h.pin(h.human, int(payload["id"]), payload.get("note") or None)) return self._reply(200, {"ok": True, "pins": len(pins)}) if path == "/api/task": action = payload.get("action") if action == "add": task = h.add_task(h.human, str(payload.get("title", "")), body=payload.get("body") or None, scope=payload.get("scope") or None, assignee=payload.get("assignee") or None, priority=payload.get("priority") or "normal", after=[int(x) for x in payload.get("after") or []], due=payload.get("due") or None, origin="web") else: task = h.update_task(h.human, int(payload.get("id", 0)), str(payload.get("state", "")), note=payload.get("note") or None, origin="web") return self._reply(200, {"ok": True, "task": task}) if path == "/api/thread-state": t = h.set_thread_state(h.human, payload.get("thread"), payload.get("state", "closed")) return self._reply(200, {"ok": True, "thread": t}) except (HarborError, projects.ProjectError) as error: return self._reply(400, {"ok": False, "error": str(error)}) except Exception: traceback.print_exc() return self._reply(500, {"ok": False, "error": "server error; see the server log"}) self._reply(404, {"error": "not found"}) def backup_loop(): """Snapshot every project's database at start and every BACKUP_HOURS hours.""" while True: for name in REGISTRY.names(): try: h, lock = REGISTRY.get(name) newest = max(h.project.backups.glob("harbor-*.db"), default=None) if newest is None or time.time() - newest.stat().st_mtime > BACKUP_HOURS * 3600: with lock: h.backup() except Exception: traceback.print_exc() time.sleep(1800) def main(argv=None): global REGISTRY, PORT parser = argparse.ArgumentParser(prog="harbor serve") parser.add_argument("port", nargs="?", type=int) parser.add_argument("--port", dest="port_option", type=int) parser.add_argument("--project") args = parser.parse_args(argv) PORT = args.port_option or args.port or 8766 REGISTRY = Registry(projects.resolve(args.project)) REGISTRY.get() server = http.server.ThreadingHTTPServer(("127.0.0.1", PORT), Handler) server.daemon_threads = True threading.Thread(target=backup_loop, daemon=True).start() threading.Thread(target=git_loop, daemon=True).start() threading.Thread(target=toast_loop, daemon=True).start() from . import logbook_auto threading.Thread(target=logbook_auto.loop, args=(REGISTRY,), daemon=True).start() print(f"Harbor on http://127.0.0.1:{PORT} - projects: {', '.join(REGISTRY.names())}", flush=True) server.serve_forever() if __name__ == "__main__": try: main() except (HarborError, projects.ProjectError) as error: sys.exit(f"harbor serve: {error}")