commit dbfd1720c54a1c29f8a4cd4a69eb4f541f2e5463 Author: KeelrLeek <156356974+KeelrLeek@users.noreply.github.com> Date: Mon Oct 5 10:44:23 2026 -0500 Harbor as of 2026-10-05: the coordination hub for the Open Sea's decomp team 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. diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..7a60b85 --- /dev/null +++ b/.gitignore @@ -0,0 +1,2 @@ +__pycache__/ +*.pyc diff --git a/README.md b/README.md new file mode 100644 index 0000000..b480d05 --- /dev/null +++ b/README.md @@ -0,0 +1,186 @@ +# Harbor + +Harbor is a local coordination hub for a team of AI coding agents and the person running them. It works with Codex, Claude Code, OpenCode, Kimi, DeepSeek Harness, or anything that can run a command or speak MCP. + +**Talking** +- **Threads.** Messages are grouped by subject (`fn:0x529350`, `commit:b4b44f0`, `task:5`, `item:12`), each with explicit recipients. +- **Read and acknowledged state for every recipient.** Agents acknowledge with a note instead of sending "received" messages. +- **Due times** when an answer is needed by a certain time. +- **Corrections linked to the message they fix**: the original shows "corrected by #N". + +**Work** +- **A task queue.** Tasks have priorities, dependencies, scopes and due times. An idle agent runs `next` to take the best task it may do. Submitting the work puts the task in review, landing it closes the task, and a change request reopens it. +- **Claims.** These are advisory locks on functions, address ranges, names, tasks or files, and they expire. Exclusive claims are refused with the holder's name. Bands are standing claims a coordinator sets. +- **A review queue.** Work moves submitted → in review → approved or changes requested → landed, and only a coordinator lands it. + +**Continuity** +- **Handoffs.** An agent records "where I left off", and it opens that agent's next brief. +- **Restart prompts** for any agent. Each is generated from the agent's role, worktree, standing instructions, the rules and the live state, ready to paste into a fresh session. +- **Pins and a decisions list** for the things nobody should have to scroll for. + +**Awareness** +- **Presence.** Each agent has a one-line "doing" status, a last-seen time, and a "working now" indicator driven by its tool use through the hooks. +- **The human's availability.** You can be Here, Away until a time, or Do not disturb, with a note. Agents see it in every brief, and questions waiting for you get flagged to the coordinator. +- **Attention alerts**: + - overdue acknowledgements and answers; + - stale review items; + - blocked or overdue tasks; + - quiet agents. +- **A digest** of what happened over any window, and **search** with filters (`from:` `to:` `kind:` `thread:` `since:` `before:`). +- **`harbor doctor`** checks the project, the database, the backups, the server and every agent app's integration. It also reports when each agent last reached Harbor and through what. + +**Team intelligence** +- **Questions with answer buttons.** An agent runs `ask "Rename it?" --options "Yes|No|Later"`, and the human taps an answer in the chat. The agent's `wait` returns with the reply. While the human is away, an optional coordinator fallback may answer instead. +- **Team memory.** `note fn:0x52BD10 "stdcall, ret 4"` attaches knowledge to a scope. `context ` gathers the notes, threads, messages that mention it, tasks, review items and earlier claims. `next` hands this to the agent along with its task. +- **Collision radar.** Hooks see which files each agent edits. When two agents edit the same repository-relative path within 45 minutes, even in different worktrees, both are told, and so is the coordinator, before anyone commits. +- **Team pulse.** Activity per agent in 5-minute steps, from tool calls and messages, drawn as a timeline. Idle gaps show at a glance. +- **Scoreboard and wins.** Agents report numbers (`metric readable 779`), and the board charts them. A landed item is announced to the human in the `wins` thread. +- **Handoff reminders.** After 4 hours of real work without a handoff, the Stop hook asks for one. Every stop message asks for a handoff before stopping. + +**Flow** +- **Git radar** (read-only, `git --no-optional-locks`). For each agent's worktree: + - branch and last commit; + - uncommitted files; + - commits behind the main branch; + - commits made since the agent's last submit, i.e. work not yet in the review queue. + + Unsubmitted work over 2 hours old, and very old bases, become attention alerts. +- **Bulk tasks and templates.** + - Paste one address per line; each becomes a task routed to its band owner. + - `[templates.]` in the project file defines multi-step pipelines that are created with dependencies in one go. +- **Watchdog.** An agent with 150+ tool calls in 2 hours and no message, submit, task change, handoff or `doing` update is flagged as possibly stuck. +- **Policies** in the project file, enforced by Harbor: + - `land_requires = ["Kimi"]`: approvals needed since the last submit before an item can land. + - `all_per_hour = 12`: a cap on messages to everyone. + +**Launch** +- **The Launch panel.** Each agent goes through a checklist: app connected, then brief read, then roll call answered, then running. The actions sit on the same row: restart prompt, roll call, trial run, resume. +- **Roll call.** The human asks every agent to answer "Here". The panel shows who read it and how fast they answered. It works while agents are stopped. +- **Trial runs.** A resume with an end time (15 minutes, 30 minutes, 1 hour). When the time is up the stop comes back by itself, with no process needed. +- **Live feed.** Each agent's latest tool calls and what they touched, from the hooks. Secrets that look like tokens are masked. +- **Who knows this.** `who ` ranks agents by their notes, reviews, drafts, claims and messages on a scope, and the same ranking appears in `context`. +- **Thread summaries.** `summarize "..."` pins a short summary on top of a thread. Briefs remind participants when a long thread has none. + +**Control** +- **One stop switch.** It applies to everyone or to one agent. An agent can stop itself and a coordinator can stop anyone, but only the human resumes. Host hooks enforce it on every tool call. +- **Waiting without polling.** `wait` returns the moment a message arrives. Hooks nudge an agent about new urgent messages at the end of its turn, and OpenCode can be woken while idle. + +**Pages** +- **A texting-style chat page**, with: + - answer buttons on questions; + - slash commands (`/stop`, `/resume`, `/away 8h note`, `/here`, `/ping`, `/task title @Agent !high fn:0x...`, `/pin`, `/status`, `/find`, `/help`); + - sound or read-aloud for messages addressed to you; + - Good night / Good morning / Status? buttons (away until 8 AM with a message to the team, back with a catch-up screen, an instant summary) and a microphone for dictation; + - live "working…" indicators; + - read receipts; + - pins and @mentions; + - "load earlier"; + - desktop notifications; + - your availability. +- **A board**, with: + - stop and resume; + - Ping and Copy-restart-prompt for each agent; + - tasks, the review queue and claims; + - pins and decisions; + - a health panel, the digest and search; + - questions for you, the team pulse, the scoreboard; + - git status on every agent card, bulk task entry, Windows notifications that work with the page closed; + - thread export to Markdown; + - searching for a scope shows its context. + +Each project is one SQLite database (WAL mode), so the CLI, the web server and many MCP servers can write at once safely. A project can also mirror every message into a plain `[time] From → To: text` file for older tools. + +## Quick start + +```bash +py -3.12 -B run.py init garden --title "Garden app" --agent Lead:"Builds it" --agent Helper --coordinator Lead --marker D:/code/garden +py -3.12 -B run.py serve --port 8766 # chat at http://127.0.0.1:8766, board at /board +py -3.12 -B run.py install codex --agent Lead # or opencode, deepseek, kimi, claude +py -3.12 -B run.py doctor +``` + +- `init` writes the project file into the Harbor home: `$HARBOR_HOME`, else the `home` set in `local.toml`, else `~/.harbor`. With `--marker` it also writes a `.harbor.toml` file in the code folder. Commands and hooks run anywhere below that folder then find the project, the way git finds a repository. +- In the project file, each `[[agents]]` entry may carry: + - `host`: which app it runs in, for the doctor; + - `workdir` and `instructions`: for restart prompts. +- An `[identities]` table maps folder names to agents, so `harbor brief` inside an agent's worktree needs no `--as`. + +## The protocol agents follow + +1. **Start every session with `brief`.** If it reports a STOP, do no work; only the human can resume. If it shows your last handoff, continue from it. +2. **Take work with `next`** instead of waiting to be told. Finish with `task done`, or with `submit --task N`. +3. **Keep `doing` current** instead of posting progress to everyone. Harbor warns when an agent sends more than 8 messages to everyone in an hour. +4. **Talk in threads named after the subject**, and answer with `reply`. + - Write plain sentences. There is no length limit to squeeze into (20,000 characters), and Harbor warns about run-together text such as "Confirmed652". + - Fix a mistake with `--corrects `. +5. **Ask for an acknowledgement, or set `--due`, only when it matters.** Acknowledge with `ack --note "..."`. +6. **Claim what you work on outside tasks**, and release it when done. +7. **Wait with `wait` / `wait_for_messages`**, not with sleep loops. +8. **Write a `handoff` before your session or allowance ends.** + +## Command line + +```bash +harbor brief | next | doing "drafting 0x529350" | wait --timeout 600 | handoff "done X; Y in flight; next Z" +harbor send Kimi "text" --thread fn:0x529350 --kind question --ack --due 2h --attach report.txt +harbor reply 123 "text" | send Kimi "fixed wording" --corrects 120 | ack 123 --note "done in a2d7525" +harbor thread fn:0x529350 | close fn:0x529350 | search "glyph key from:Kimi since:2d" | pin 123 --note "precision rule" +harbor task add "Make 0x445310 readable" --scope fn:0x445310 --priority high --after 3 --for DeepSeek +harbor tasks | task show 5 | task take 5 | task done 5 --note "..." | task block 5 --note "needs 0x4E2680" | task release 5 +harbor claim fn:0x529350 | release --scope fn:0x529350 | claims +harbor submit "AudioManagerTick readable" --commit a2d7525 --task 5 | review 7 approved | land 7 --commit b4b44f0 | queue +harbor status | digest --since 8h | attention | doctor | prompt Kimi | handoffs | pins | decisions +harbor ask "Rename 0x445310?" --options "Yes|No" --fallback Claude | answer 3 Yes | questions +harbor note fn:0x52BD10 "stdcall, ret 4" | notes | context fn:0x445310 | metric readable 779 | metrics | pulse | wins +harbor export item:12 --out review-12.md +harbor git | templates | task bulk --file held.txt [--template decomp] | task template decomp fn:0x445310 +harbor who fn:0x52BD10 | feed [--agent Codex] | summarize fn:0x52BD10 "decided ...; open ...; next ..." +harbor stop | resume all --relay +harbor whoami | projects | init | backup | import-legacy chat.md | serve | mcp | install | hook +``` + +The command is `py -3.12 -B /run.py ...`; with `pip install -e .` it becomes `harbor`. Every command takes `--json`. + +**Choosing a project:** `--project`, then `$HARBOR_PROJECT`, then the nearest `.harbor.toml`, then the home's default project. + +**Choosing an identity:** `--as`, then `$HARBOR_AGENT`, then `[identities]`. + +## MCP + +`harbor --as mcp` serves these tools over stdio: + +| Area | Tools | +|---|---| +| Session | `brief`, `doing`, `handoff`, `status`, `digest` | +| Messages | `inbox`, `wait_for_messages`, `send` (with `corrects`, `due`, `attachments`), `reply`, `ack`, `pin` | +| Threads | `thread`, `threads`, `close_thread`, `search` | +| Tasks | `next_task`, `take_task`, `tasks`, `add_task`, `update_task` | +| Claims | `claim`, `release`, `claims` | +| Review | `submit` (with `task`), `review`, `withdraw`, `land`, `queue` | +| Team | `ask`, `answer`, `questions`, `note`, `context`, `metric` | +| Stop | `stop`, `resume` | + +The identity is fixed by the host's configuration (`HARBOR_AGENT`). A tool call cannot change it. + +## Hosts + +`harbor install ` wires a host up. Each run backs up every file it edits, writes only between marker lines, validates the result, and is undone with `--uninstall`. `--dry-run` shows what would change. [integrations/README.md](integrations/README.md) lists what each host gets and the evidence for each format. + +Hooks resolve the project from the agent's working folder and stay silent anywhere that is not a Harbor project. That makes installing them in a host you also use for other work harmless. + +## Web + +`harbor serve` binds 127.0.0.1 only and checks the Host header. + +- `/` is the default project's chat; `/p//` reaches any registered project. +- Stop flags: an agent may stop itself, a coordinator may stop anyone, and an agent may relay the human's typed request to stop someone. **Only the human resumes**, from the page; a coordinator may lift a stop it set itself, unless it was relayed for the human. +- POSTs must come from the page itself: a per-start token, the browser's `Sec-Fetch-Site: same-origin` header, and JSON. That makes speaking as the human a deliberate forgery rather than an accident. It is not a boundary against software running as the same Windows user. +- The server backs up every project database every 6 hours and keeps 14 copies. + +## Tests + +```bash +py -3.12 -B -m unittest discover -s tests +``` + +The tests use temporary homes under `R:\Workspace\Data\openseas-chat\test-tmp`, or `$HARBOR_TEST_TMP`. diff --git a/harbor/__init__.py b/harbor/__init__.py new file mode 100644 index 0000000..0fa7b11 --- /dev/null +++ b/harbor/__init__.py @@ -0,0 +1,6 @@ +"""Harbor: a local coordination hub for teams of AI coding agents and the person running them.""" +from .core import Harbor, HarborError +from .projects import Project, ProjectError, resolve + +__version__ = "2.0.0" +__all__ = ["Harbor", "HarborError", "Project", "ProjectError", "resolve", "__version__"] diff --git a/harbor/__main__.py b/harbor/__main__.py new file mode 100644 index 0000000..27cfa48 --- /dev/null +++ b/harbor/__main__.py @@ -0,0 +1,3 @@ +from .cli import run + +run() diff --git a/harbor/cli.py b/harbor/cli.py new file mode 100644 index 0000000..1408831 --- /dev/null +++ b/harbor/cli.py @@ -0,0 +1,721 @@ +"""harbor: messages, threads, acknowledgements, claims, review items and stop flags for a team of AI agents. + +Project: --project , $HARBOR_PROJECT, or the nearest .harbor.toml above the working directory. +Identity: --as , $HARBOR_AGENT, or the project's [identities] entry for the working directory. +The human speaks from the chat page only, never from this command line. + +Start of every session: harbor brief +Take work: harbor next (the best task you may do; claims its scope) +While working: harbor doing "drafting 0x529350" (one line the whole team sees) +Waiting for answers: harbor wait --timeout 600 (returns as soon as a message arrives) +Before the session ends: harbor handoff "done X; Y in flight; next Z" +Tasks: harbor task add "Name 0x445310" --scope fn:0x445310 --priority high --after 3 + harbor tasks harbor task done 5 --note "..." harbor task block 5 --note "needs 0x4E2680" +Messages: + harbor send Claude "text" --thread fn:0x529350 --subject "AudioManagerTick" --kind question + harbor reply 123 "text" harbor inbox harbor ack 123 --note "fixed in a2d7525" + harbor thread fn:0x529350 harbor search "AudioManagerTick" harbor threads +Claims: harbor claim fn:0x529350 --note "writing the body" harbor release --scope fn:0x529350 +Review queue: harbor submit "AudioManagerTick readable" --scope fn:0x529350 --commit a2d7525 + harbor review 7 approved --note "..." harbor land 7 --commit b4b44f0 harbor queue +Overview: harbor status harbor digest --since 8h harbor attention +Stop: harbor stop (a coordinator may stop anyone; only the human resumes, from the page) +Long bodies: --file path (or --file - for stdin); attachments: --attach path (repeatable). +""" +import argparse +import json +import os +import sys +from pathlib import Path + +from . import projects +from .core import CLAIM_KINDS, IMPORTANCE, ITEM_STATES, KINDS, OPEN_ITEM_STATES, TASK_STATES, Harbor, HarborError + + +def fmt_message(m): + tags = [f"thread {m['thread_id']} {m.get('topic', '')}".strip()] + if m["kind"] != "note": + tags.append(m["kind"]) + if m["importance"] != "normal": + tags.append(m["importance"]) + if m.get("ack_required"): + tags.append("ack") + if m.get("reply_to"): + tags.append(f"re #{m['reply_to']}") + if m.get("corrects"): + tags.append(f"corrects #{m['corrects']}") + if m.get("due_at"): + tags.append(f"due {m['due_at']}") + to = ",".join(m["recipients"]) + head = f"#{m['id']} {m['created_at'].replace('T', ' ')[:16]}Z {m['sender']} → {to} [{' · '.join(tags)}]" + body = "\n".join(" " + line for line in m["body"].split("\n")) + files = m.get("refs", {}).get("attachments") + return head + "\n" + body + (f"\n attachments: {', '.join(files)}" if files else "") + + +def fmt_task(task): + who = task["assignee"] or "unassigned" + extra = [] + if task["scope"]: + extra.append(task["scope"]) + if task["waiting_on"]: + extra.append("after " + ",".join(f"#{d}" for d in task["waiting_on"])) + if task["due_at"]: + extra.append(f"due {task['due_at']}") + if task["item_id"]: + extra.append(f"item #{task['item_id']}") + if task["state"] == "blocked": + extra.append(f"blocked: {task['block_reason']}") + return (f"#{task['id']:>4} {task['state']:<11} {task['priority_name']:<6} {who:<9} {task['title']}" + + (f" ({'; '.join(extra)})" if extra else "")) + + +def body_from(args): + if getattr(args, "file", None): + return sys.stdin.read() if args.file == "-" else Path(args.file).read_text(encoding="utf-8") + return " ".join(args.text) + + +def refs_from(pairs): + refs = {} + for pair in pairs or []: + if "=" not in pair: + raise HarborError(f"--ref {pair!r}: use key=value, e.g. commit=b4b44f0 or fn=0x529350") + key, value = pair.split("=", 1) + refs[key.strip()] = value.strip() + return refs + + +def recipients(text): + return None if not text else ("All" if text == "All" else [t.strip() for t in text.split(",") if t.strip()]) + + +def build_parser(): + parser = argparse.ArgumentParser(prog="harbor", description=__doc__.split("\n")[0], + formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__) + parser.add_argument("--as", dest="agent", default=None) + parser.add_argument("--project", default=None, help="project name or path to its .toml") + parser.add_argument("--json", action="store_true", help="machine-readable output") + sub = parser.add_subparsers(dest="cmd", required=True) + + p = sub.add_parser("brief", help="what you need to know at the start of a session") + p.add_argument("--mark-read", action="store_true") + p = sub.add_parser("inbox", help="unread messages (urgent first); marks them read") + p.add_argument("--all", action="store_true", help="recent messages, read or not") + p.add_argument("--peek", action="store_true", help="do not mark read") + p.add_argument("--limit", type=int, default=30) + p = sub.add_parser("wait", help="block until a message for you arrives or a stop applies (then print it)") + p.add_argument("--timeout", type=float, default=300, help="seconds (max 3600)") + p.add_argument("--peek", action="store_true") + for name in ("send", "reply"): + p = sub.add_parser(name) + if name == "send": + p.add_argument("to", help="Agent, Agent,Agent or All") + p.add_argument("--subject") + p.add_argument("--thread", help="thread id or topic, e.g. fn:0x529350, commit:b4b44f0, general") + else: + p.add_argument("message_id", type=int) + p.add_argument("--to", help="override recipients (Agent,Agent or All)") + p.add_argument("text", nargs="*") + p.add_argument("--file") + p.add_argument("--attach", action="append", help="copy a file into Harbor and link it (repeatable)") + p.add_argument("--kind", choices=KINDS, default="answer" if name == "reply" else "note") + p.add_argument("--importance", choices=IMPORTANCE, default="normal") + p.add_argument("--ack", action="store_true", help="ask the recipients to acknowledge") + p.add_argument("--ref", action="append", help="key=value (commit=..., fn=0x..., file=...)") + p.add_argument("--corrects", type=int, help="id of your earlier message this corrects") + p.add_argument("--due", help="answer needed within 30m / 2h / 1d, or by an ISO time") + p = sub.add_parser("ack", help="acknowledge messages addressed to you") + p.add_argument("ids", type=int, nargs="+") + p.add_argument("--note") + p = sub.add_parser("doing", help="set the one line the team sees about what you are doing (no text clears it)") + p.add_argument("text", nargs="*") + p = sub.add_parser("thread", help="show a whole thread") + p.add_argument("thread") + for name in ("close", "reopen"): + p = sub.add_parser(name, help=f"{name} a thread") + p.add_argument("thread") + p = sub.add_parser("threads", help="recent threads") + p.add_argument("--limit", type=int, default=25) + p.add_argument("--archive", action="store_true", help="include imported chat archives") + p.add_argument("--open", action="store_true", help="hide closed threads") + p = sub.add_parser("search", help="full-text search over every message") + p.add_argument("query", nargs="+") + p.add_argument("--limit", type=int, default=15) + p = sub.add_parser("tail", help="the most recent messages") + p.add_argument("count", type=int, nargs="?", default=20) + p = sub.add_parser("claim", help="claim a function, range, name, task or file") + p.add_argument("scope", help="fn:0xADDR, range:0xLO-0xHI, name:Symbol, task:text, file:path") + p.add_argument("--kind", choices=sorted(CLAIM_KINDS), default="draft") + p.add_argument("--ttl", type=float, help="hours (default by kind)") + p.add_argument("--note") + p.add_argument("--for", dest="for_agent", help="coordinator only: record the claim for another agent") + p.add_argument("--override", action="store_true", help="coordinator only") + p = sub.add_parser("release", help="release a claim") + p.add_argument("claim_id", type=int, nargs="?") + p.add_argument("--scope") + p = sub.add_parser("claims", help="active claims") + p.add_argument("--agent", dest="filter_agent") + p.add_argument("--kind") + p = sub.add_parser("submit", help="put a commit in the review queue") + p.add_argument("title") + p.add_argument("--scope") + p.add_argument("--commit") + p.add_argument("--branch") + p.add_argument("--note") + p.add_argument("--reviewer", action="append") + p.add_argument("--task", type=int, help="the task this work completes (landing closes it)") + p = sub.add_parser("review", help="move a review item") + p.add_argument("item", type=int) + p.add_argument("state", choices=[s for s in ITEM_STATES if s not in ("landed", "withdrawn")]) + p.add_argument("--note") + p.add_argument("--commit") + p = sub.add_parser("withdraw", help="author withdraws an item") + p.add_argument("item", type=int) + p.add_argument("--note") + p = sub.add_parser("land", help="coordinator: the items are on main (several items: one message per author)") + p.add_argument("item", type=int, nargs="+") + p.add_argument("--commit", required=True) + p.add_argument("--note") + p = sub.add_parser("item", help="one review item with its history") + p.add_argument("item", type=int) + p = sub.add_parser("queue", help="open review items") + p.add_argument("--all", action="store_true") + sub.add_parser("status", help="stop flags, presence, unread counts and the queue at a glance") + p = sub.add_parser("tasks", help="the task list (open work by default)") + p.add_argument("--all", action="store_true", help="include done and cancelled") + p.add_argument("--mine", action="store_true") + p.add_argument("--agent", dest="filter_agent") + p = sub.add_parser("task", help="add, show or move a task") + p.add_argument("action", choices=("add", "show", "take", "done", "block", "release", "cancel", "reopen", "bulk", + "template")) + p.add_argument("args", nargs="*", help="add: the title; template: name and scope; bulk: none (--file); " + "others: the task id") + p.add_argument("--template", help="bulk: run this template for every line") + p.add_argument("--body") + p.add_argument("--file") + p.add_argument("--scope") + p.add_argument("--for", dest="for_agent", help="assign to an agent (coordinators; or yourself)") + p.add_argument("--priority", default="normal", choices=("urgent", "high", "normal", "low")) + p.add_argument("--after", type=int, action="append", help="a task that must be done first (repeatable)") + p.add_argument("--due", help="30m / 2h / 1d or an ISO time") + p.add_argument("--note") + p = sub.add_parser("next", help="take the best task you may do now (claims its scope)") + p.add_argument("--peek", action="store_true", help="show it without taking it") + p = sub.add_parser("handoff", help="record where you left off; shown at the start of your next session") + p.add_argument("text", nargs="*") + p.add_argument("--file") + p = sub.add_parser("handoffs", help="recent handoffs") + p.add_argument("--agent", dest="filter_agent") + p.add_argument("--limit", type=int, default=5) + for name in ("pin", "unpin"): + p = sub.add_parser(name, help=f"{name} a message for the whole team") + p.add_argument("message_id", type=int) + if name == "pin": + p.add_argument("--note", help="a one-line label") + sub.add_parser("pins", help="pinned messages") + sub.add_parser("decisions", help="recent messages of kind decision") + p = sub.add_parser("prompt", help="a restart prompt for an agent (paste it into a fresh session)") + p.add_argument("agent_name") + sub.add_parser("doctor", help="check the project, the database, the server and every host integration") + p = sub.add_parser("ask", help="ask the human (or an agent) a question with answer options") + p.add_argument("question", nargs="+") + p.add_argument("--options", help='answer buttons, separated by | e.g. "Yes|No|Later"') + p.add_argument("--to", help="who answers (default: the human)") + p.add_argument("--due", help="30m / 2h / an ISO time") + p.add_argument("--fallback", help="a coordinator who may answer while the human is away") + p.add_argument("--thread") + p = sub.add_parser("answer", help="answer a question addressed to you") + p.add_argument("question_id", type=int) + p.add_argument("text", nargs="+") + p = sub.add_parser("questions", help="open questions (to you and from you)") + p.add_argument("--all", action="store_true", help="include answered ones") + p = sub.add_parser("note", help="add to the team's memory about a scope (fn:0x..., file:..., name:...)") + p.add_argument("scope") + p.add_argument("text", nargs="+") + p.add_argument("--replaces", type=int, help="the id of an older note this one corrects") + p = sub.add_parser("notes", help="notes, all or for one scope") + p.add_argument("scope", nargs="?") + p = sub.add_parser("context", help="everything the team knows about a scope") + p.add_argument("scope") + p = sub.add_parser("metric", help="report a number the board charts (e.g. readable 779)") + p.add_argument("name") + p.add_argument("value") + p.add_argument("--note") + sub.add_parser("metrics", help="the latest value of every metric") + p = sub.add_parser("pulse", help="who was active when (tool calls and messages per hour)") + p.add_argument("--hours", type=int, default=12) + p = sub.add_parser("wins", help="landed review items") + p.add_argument("--since", default="24h") + sub.add_parser("git", help="each agent's branch, last commit, uncommitted files and unsubmitted commits") + p = sub.add_parser("summarize", help="pin a short summary to a thread (shown on top of it and in context)") + p.add_argument("thread") + p.add_argument("text", nargs="*") + p.add_argument("--file") + p = sub.add_parser("who", help="who has worked on a scope (notes, reviews, claims, messages)") + p.add_argument("scope") + p = sub.add_parser("feed", help="the latest tool calls of the team (or one agent), from the hooks") + p.add_argument("--agent", dest="filter_agent") + p.add_argument("--limit", type=int, default=30) + sub.add_parser("templates", help="task templates defined in the project file") + p = sub.add_parser("export", help="a thread as Markdown") + p.add_argument("thread") + p.add_argument("--out", help="write to this file instead of printing") + p = sub.add_parser("digest", help="what happened in a window (default the last 8 hours)") + p.add_argument("--since", default="8h", help="8h, 90m, 2d or an ISO time") + sub.add_parser("attention", help="overdue acknowledgements, stale review items, quiet agents") + for name in ("stop", "resume"): + p = sub.add_parser(name, help=f"{name} everyone or one agent") + p.add_argument("target", help="all or an agent name") + p.add_argument("--relay", type=int, help="id of the human's chat-page message that asked for this") + p.add_argument("--reason") + p = sub.add_parser("pending-acks", help="messages still waiting for acknowledgement") + p.add_argument("--overdue", type=int, default=0, help="minutes") + p = sub.add_parser("import-legacy", help="import a plain `[time] From → To: text` chat history (idempotent)") + p.add_argument("path") + p = sub.add_parser("backup", help="snapshot the database into the project's backups folder") + p.add_argument("--keep", type=int, default=14) + sub.add_parser("whoami", help="which project and identity this directory resolves to") + sub.add_parser("projects", help="registered projects") + p = sub.add_parser("init", help="create a project") + p.add_argument("name") + p.add_argument("--title") + p.add_argument("--agent", dest="roster", action="append", required=True, + help="Name or Name:role (repeat for each agent; the human is added automatically)") + p.add_argument("--human", default="User") + p.add_argument("--coordinator", action="append", default=[]) + p.add_argument("--database") + p.add_argument("--mirror") + p.add_argument("--marker", help="also write a .harbor.toml in this folder pointing at the project") + p = sub.add_parser("hook", help="entry point for an agent host's hooks (see harbor.hooks)") + p.add_argument("event") + p.add_argument("--host", default="generic") + p = sub.add_parser("install", help="connect an agent host (codex, opencode, deepseek, kimi, claude) to Harbor") + p.add_argument("host", choices=("codex", "opencode", "deepseek", "kimi", "claude")) + p.add_argument("--agent", dest="host_agent", + help="the host's Harbor identity (default: Codex, OpenCode, DeepSeek, Kimi, Claude)") + p.add_argument("--bind", action="store_true", help="bind the host to --project instead of the working folder") + p.add_argument("--dry-run", action="store_true", help="show what would change") + p.add_argument("--uninstall", action="store_true", help="remove Harbor's entries again") + p.add_argument("--python", help='interpreter command for hosts to run, e.g. "py -3.12" (default: detected)') + p.add_argument("--no-register", action="store_true", help="kimi: build and validate the plugin only") + p = sub.add_parser("serve", help="run the chat page and board") + p.add_argument("--port", type=int, default=8766) + sub.add_parser("mcp", help="run the MCP server on stdio (identity from --as or $HARBOR_AGENT)") + return parser + + +def main(argv=None): + if hasattr(sys.stdout, "reconfigure"): + sys.stdout.reconfigure(encoding="utf-8") + args = build_parser().parse_args(argv) + cmd = args.cmd + + if cmd == "projects": + home = projects.harbor_home() + names = projects.list_names(home) + marker = projects.find_marker() + default = projects.default_name(home) + rows = [{"name": n, "default": n == default} for n in names] + text = f"Harbor home: {home}\n" + ("\n".join(f" {n}{' (default)' if n == default else ''}" for n in names) + or " no registered projects") + if marker: + text += f"\nThis directory belongs to: {marker}" + print(json.dumps(rows) if args.json else text) + return + if cmd == "init": + home = projects.harbor_home() + agents = [] + for spec in args.roster: + name, _, role = spec.partition(":") + agents.append({"name": name.strip(), "role": role.strip()}) + path = projects.write_project(home / "projects" / f"{args.name}.toml", args.name, args.title or args.name, + agents, human=args.human, coordinators=args.coordinator, + database=args.database, mirror=args.mirror) + print(f"created {path}") + if not projects.default_name(home): + (home / "config.toml").write_text(f'default_project = "{args.name}"\n', encoding="utf-8") + print(f"{args.name} is the default project") + if args.marker: + marker = Path(args.marker) / projects.MARKER + if marker.exists(): + raise HarborError(f"{marker} already exists") + marker.write_text(f'project = "{args.name}"\n', encoding="utf-8") + print(f"wrote {marker}") + return + + if cmd == "install": + from . import install + bound = args.project if (args.bind or args.host == "kimi") else None + try: + report = install.install(args.host, agent=args.host_agent, project=bound, dry_run=args.dry_run, + remove=args.uninstall, python=args.python, register=not args.no_register) + except install.InstallError as error: + raise HarborError(str(error)) + print(("Dry run - nothing written.\n" if args.dry_run else "") + "\n".join(report)) + return + if cmd == "hook": # resolves the project from the host's payload, never raises + from . import hooks + sys.exit(hooks.run(args.event, args.project, args.agent, host=args.host)) + project = projects.resolve(args.project) + if cmd == "serve": + from . import server + server.main(["--port", str(args.port)] + (["--project", args.project] if args.project else [])) + return + if cmd == "mcp": + if args.agent: + os.environ["HARBOR_AGENT"] = args.agent + from . import mcp_server + mcp_server.main(project) + return + + harbor = Harbor(project) + source = "--as" if args.agent else None + agent = args.agent or os.environ.get("HARBOR_AGENT") + if agent and not source: + source = "HARBOR_AGENT" + if not agent: + agent = project.identity_for(os.getcwd()) + source = "working directory" if agent else None + + recorded = [] + + def who(): + if not agent: + raise HarborError("say who you are: --as , set HARBOR_AGENT, or work inside a directory listed " + "under [identities] in the project file") + if agent == project.human: + raise HarborError(f"{project.human} speaks from the chat page, not the command line") + if agent not in project.ai_names: + raise HarborError(f"unknown agent {agent!r}; this project has {', '.join(project.ai_names)}") + if not recorded and cmd not in ("whoami", "prompt"): + recorded.append(True) + harbor.activity(agent, f"harbor {cmd}", os.environ.get("HARBOR_HOST", ""), channel="cli") + return agent + + def out(value, text): + print(json.dumps(value, ensure_ascii=False, indent=1, default=str) if args.json else text) + + if cmd == "whoami": + out({"project": project.name, "source": str(project.source), "agent": agent, "identity_from": source}, + f"project {project.name} ({project.title}) from {project.source}\n" + f"agent {agent or '(none)'}" + (f" from {source}" if source else "")) + elif cmd == "brief": + text = harbor.brief(who(), mark_read=args.mark_read) + out({"brief": text}, text) + elif cmd == "inbox": + rows = harbor.inbox(who(), unread_only=not args.all, limit=args.limit, mark_read=not args.peek) + out(rows, "\n\n".join(fmt_message(m) for m in rows) or "No unread messages.") + elif cmd == "wait": + r = harbor.wait(who(), timeout=args.timeout, mark_read=not args.peek) + if r["stopped"]: + text = "STOP in force (" + ", ".join(r["stopped"]) + "): do not start or continue work." + elif r["timed_out"]: + text = "No new messages." + else: + text = "\n\n".join(fmt_message(m) for m in r["messages"]) + out(r, text) + elif cmd in ("send", "reply"): + text = body_from(args) + if cmd == "send": + r = harbor.send(who(), recipients(args.to), text, subject=args.subject, thread=args.thread, kind=args.kind, + importance=args.importance, ack_required=args.ack, refs=refs_from(args.ref), + attachments=args.attach, corrects=args.corrects, due=args.due) + else: + r = harbor.send(who(), recipients(args.to), text, reply_to=args.message_id, kind=args.kind, + importance=args.importance, ack_required=args.ack, refs=refs_from(args.ref), + attachments=args.attach, corrects=args.corrects, due=args.due) + lines = [f"sent #{r['id']} in thread {r['thread_id']} ({r['topic']}) to {', '.join(r['targets'])}"] + lines += [f"warning: {w}" for w in r["warnings"]] + out(r, "\n".join(lines)) + elif cmd == "ack": + out(harbor.ack(who(), args.ids, args.note), "acknowledged " + ", ".join(f"#{i}" for i in args.ids)) + elif cmd == "doing": + r = harbor.doing(who(), " ".join(args.text)) + out(r, f"{r['agent']}: {r['doing'] or '(cleared)'}") + elif cmd == "thread": + thread, messages = harbor.thread(args.thread) + out({"thread": thread, "messages": messages}, + f"Thread {thread['id']} {thread['topic']}: {thread['subject']} ({thread['state']})\n\n" + + "\n\n".join(fmt_message(m) for m in messages)) + elif cmd in ("close", "reopen"): + t = harbor.set_thread_state(who(), args.thread, "closed" if cmd == "close" else "open") + out(t, f"thread {t['id']} {t['topic']} is {t['state']}") + elif cmd == "threads": + rows = harbor.threads(limit=args.limit, agent=agent if agent and agent != project.human else None, + include_legacy=args.archive, include_closed=not args.open) + out(rows, "\n".join(f"{r['id']:>5} {r['updated_at'][:16]}Z {r['messages']:>4} msgs" + + (f" {r['unread']} unread" if r.get('unread') else "") + + (f" [{r['state']}]" if r['state'] != 'open' else "") + + f" {r['topic']}: {r['subject']}" for r in rows) or "No threads.") + elif cmd == "search": + rows = harbor.search(" ".join(args.query), limit=args.limit) + out(rows, "\n\n".join(fmt_message(m) for m in rows) or "No matches.") + elif cmd == "tail": + rows = harbor.recent(args.count) + out(rows, "\n\n".join(fmt_message(m) for m in rows)) + elif cmd == "claim": + r = harbor.claim(who(), args.scope, kind=args.kind, ttl_hours=args.ttl, note=args.note, + override=args.override, on_behalf_of=args.for_agent) + out(r, f"{'renewed' if r['renewed'] else 'claimed'} #{r['id']} {r['kind']} {r['scope']}" + + (f" until {r['expires_at']}" if r["expires_at"] else " (standing)") + + (f" overriding {r['conflicts']}" if r["conflicts"] else "")) + elif cmd == "release": + if args.claim_id is None and not args.scope: + raise HarborError("give a claim id or --scope") + out(harbor.release(who(), args.claim_id, args.scope), "released") + elif cmd == "claims": + rows = harbor.active_claims(agent=args.filter_agent, kind=args.kind) + out(rows, "\n".join(f"#{c['id']:>4} {c['agent']:<9} {c['kind']:<7} {c['scope']:<32}" + + (f" until {c['expires_at']}" if c['expires_at'] else " standing") + + (f" {c['note']}" if c['note'] else "") for c in rows) or "No active claims.") + elif cmd == "submit": + r = harbor.submit(who(), args.title, scope=args.scope, commit=args.commit, branch=args.branch, note=args.note, + reviewers=args.reviewer, task=args.task) + out(r, f"review item #{r['id']} submitted (thread {r['thread_id']})") + elif cmd == "review": + r = harbor.transition(who(), args.item, args.state, note=args.note, commit=args.commit) + out(r, f"item #{r['id']} is now {r['state']}") + elif cmd == "withdraw": + r = harbor.transition(who(), args.item, "withdrawn", note=args.note) + out(r, f"item #{r['id']} withdrawn") + elif cmd == "land": + if len(args.item) == 1 and harbor.item(args.item[0])[0]["state"] == "approved": + r = harbor.transition(who(), args.item[0], "landed", note=args.note, commit=args.commit) + out(r, f"item #{r['id']} landed in {args.commit}") + else: # approves items still under review on the way + r = harbor.land_many(who(), args.item, args.commit, note=args.note) + out(r, f"{len(r)} items landed in {args.commit}: " + ", ".join(f"#{i['id']}" for i in r)) + elif cmd == "item": + item, events = harbor.item(args.item) + out({"item": item, "events": events}, + f"#{item['id']} {item['state']} {item['scope'] or ''} {item['title']} (author {item['author']}, " + f"commit {item['commit_hash'] or '-'})\n" + "\n".join( + f" {e['at']} {e['agent']}: {e['from_state'] or '-'} -> {e['to_state']}" + + (f" ({e['commit_hash']})" if e['commit_hash'] else "") + (f" {e['note']}" if e['note'] else "") + for e in events)) + elif cmd == "queue": + rows = harbor.items(states=ITEM_STATES if args.all else OPEN_ITEM_STATES) + out(rows, "\n".join(f"#{i['id']:>4} {i['state']:<17} {i['author']:<9} {i['scope'] or '':<18} {i['title']}" + f" (since {i['created_at'][:16]}Z)" for i in rows) or "The review queue is empty.") + elif cmd == "status": + flags = harbor.flags() + stopped = [a for a in project.ai_names if harbor.stopped(a)] + trials = {a: harbor.trial(a) for a in project.ai_names if harbor.trial(a)} + lines = [f"{project.title}: " + (f"stopped: {', '.join(stopped)}" if stopped else "nobody is stopped") + + ("; trial runs: " + ", ".join(f"{a} until {u}" for a, u in trials.items()) if trials else "") + "."] + agents = harbor.agents() + for a in agents: + n, u = harbor.unread_count(a["name"]) + seen = harbor.minutes_since(a["last_seen"]) + lines.append(f" {a['name']:<9} unread {n:>3}" + (f" ({u} urgent)" if u else " ") + + f" seen {'never' if seen is None else str(seen) + ' min ago':<14}" + + (f" doing: {a['doing']}" if a.get("doing") else "")) + lines.append(f"Review queue: {len(harbor.items())} open. Active claims: {len(harbor.active_claims())}.") + for a in harbor.attention(): + lines.append(f"Attention: {a['text']}") + out({"flags": flags, "agents": agents, "queue": len(harbor.items()), "attention": harbor.attention()}, + "\n".join(lines)) + elif cmd == "digest": + d = harbor.digest(args.since) + lines = [f"From {d['since']} to {d['until']}:"] + for a in d["agents"]: + if any(a[k] for k in ("messages", "submitted", "reviewed", "landed", "claims")): + lines.append(f" {a['name']:<9} {a['messages']:>3} messages in {a['threads']} threads, " + f"{a['submitted']} submitted, {a['reviewed']} reviewed, {a['landed']} landed, " + f"{a['claims']} claims" + (f"; now: {a['doing']}" if a["doing"] else "")) + for e in d["item_events"]: + lines.append(f" item #{e['item_id']} {e['title']}: {e['from_state'] or '-'} -> {e['to_state']} by {e['agent']}") + for f in d["flags"]: + lines.append(f" {f['key']} {f['value']} by {f['set_by']} at {f['at']}" + (f": {f['reason']}" if f['reason'] else "")) + if d["to_human"]: + lines.append(f"Messages to {project.human}: " + ", ".join(f"#{m['id']} from {m['sender']}" for m in d["to_human"])) + out(d, "\n".join(lines) if len(lines) > 1 else lines[0] + " nothing happened.") + elif cmd == "attention": + rows = harbor.attention() + out(rows, "\n".join(a["text"] for a in rows) or "Nothing needs attention.") + elif cmd in ("stop", "resume"): + key = "stop:all" if args.target.lower() == "all" else f"stop:{args.target}" + flags = harbor.set_flag(key, "on" if cmd == "stop" else "off", who(), "cli", reason=args.reason, + relay_of=args.relay) + out(flags, f"{key} is now {flags[key]['value']}") + elif cmd == "pending-acks": + rows = harbor.pending_acks(sender=args.agent, overdue_minutes=args.overdue) + out(rows, "\n".join(f"#{r['id']} from {r['sender']} to {r['agent']} at {r['created_at']}" for r in rows) + or "Nothing waiting for acknowledgement.") + elif cmd == "import-legacy": + if who() not in project.coordinators: + raise HarborError("only a coordinator imports chat history") + r = harbor.import_legacy(args.path) + out(r, f"parsed {r['parsed']} messages, imported {r['added']} new") + elif cmd == "tasks": + states = TASK_STATES if args.all else ("open", "in_progress", "blocked", "review") + rows = harbor.tasks(states=states, assignee=(who() if args.mine else args.filter_agent)) + out(rows, "\n".join(fmt_task(x) for x in rows) or "No tasks.") + elif cmd == "task": + if args.action == "bulk": + text = sys.stdin.read() if not args.file or args.file == "-" else Path(args.file).read_text(encoding="utf-8") + r = harbor.bulk_tasks(who(), text, template=args.template, priority=args.priority, assignee=args.for_agent) + out(r, "\n".join(["added " + fmt_task(x).strip() for x in r["created"]] + [f"skipped: {s}" for s in r["skipped"]]) + or "nothing to add") + elif args.action == "template": + if not args.args: + raise HarborError("task template [scope]") + rows = harbor.run_template(who(), args.args[0], scope=args.args[1] if len(args.args) > 1 else None, + title=args.body, owner=args.for_agent, priority=args.priority) + out(rows, "\n".join("added " + fmt_task(x).strip() for x in rows)) + elif args.action == "add": + title = " ".join(args.args) + body = Path(args.file).read_text(encoding="utf-8") if args.file else args.body + r = harbor.add_task(who(), title, body=body, scope=args.scope, assignee=args.for_agent, + priority=args.priority, after=args.after, due=args.due) + out(r, "added " + fmt_task(r).strip()) + else: + if len(args.args) != 1 or not args.args[0].isdigit(): + raise HarborError(f"task {args.action} needs one task id") + tid = int(args.args[0]) + if args.action == "show": + task, events = harbor.task(tid) + out({"task": task, "events": events}, fmt_task(task) + (f"\n{task['body']}" if task["body"] else "") + + "\n" + "\n".join(f" {e['at']} {e['agent']}: {e['from_state'] or '-'} -> {e['to_state']}" + + (f" ({e['note']})" if e['note'] else "") for e in events)) + elif args.action == "take": + r = harbor.take_task(who(), tid, args.note) + out(r, "took " + fmt_task(r).strip()) + else: + state = {"done": "done", "block": "blocked", "release": "open", "cancel": "cancelled", + "reopen": "open"}[args.action] + r = harbor.update_task(who(), tid, state, note=args.note) + out(r, fmt_task(r).strip()) + elif cmd == "next": + r = harbor.next_task(who(), take=not args.peek) + if r and r["scope"]: + r["context"] = harbor.context_text(r["scope"]) + out(r, ("next: " if args.peek else "took ") + fmt_task(r).strip() + (f"\n\n{r['context']}" if r.get("context") + else "") if r else + "No task you can take right now (all assigned, waiting on other tasks, inside someone else's claim, " + "or you are stopped).") + elif cmd == "ask": + q = harbor.ask(who(), " ".join(args.question), options=(args.options or "").split("|") if args.options else (), + to=args.to, due=args.due, fallback=args.fallback, thread=args.thread) + out(q, f"question #{q['id']} sent to {q['to_name']} (message #{q['message_id']}); `wait` returns when it is " + "answered") + elif cmd == "answer": + q = harbor.answer(who(), args.question_id, " ".join(args.text)) + out(q, f"answered #{q['id']}: {q['answer']}") + elif cmd == "questions": + me = who() + rows = [q for q in harbor.questions(open_only=not args.all, limit=100) if me in (q["to_name"], q["asker"])] + out(rows, "\n".join(f"#{q['id']} {q['asker']} -> {q['to_name']}: {q['question']}" + + (f" [{' / '.join(q['options'])}]" if q["options"] else "") + + (f" => {q['answer']} ({q['answered_by']})" if q["answer"] else "") for q in rows) + or "No open questions.") + elif cmd == "note": + r = harbor.note(who(), args.scope, " ".join(args.text), replaces=args.replaces) + out(r, f"note #{r['id']} on {r['scope']}") + elif cmd == "notes": + rows = harbor.notes(args.scope) + out(rows, "\n".join(f"#{n['id']} {n['scope']} {n['author']} {n['created_at'][:10]}: {n['body']}" for n in rows) + or "No notes.") + elif cmd == "context": + out(harbor.context(args.scope), harbor.context_text(args.scope, limit=15)) + elif cmd == "metric": + r = harbor.metric(who(), args.name, args.value, args.note) + out(r, f"{r['name']} = {r['value']:g}") + elif cmd == "metrics": + rows = harbor.metrics() + out(rows, "\n".join(f"{k:<20} {v['latest']['value']:g} ({v['latest']['by_agent']}, {v['latest']['at']})" + + (f" {v['latest']['note']}" if v['latest']['note'] else "") for k, v in rows.items()) + or "No metrics yet.") + elif cmd == "pulse": + data = harbor.pulse(hours=args.hours) + hours = {} + for agent, buckets in data["agents"].items(): + for b in buckets: + hours.setdefault(agent, {}).setdefault(b["bucket"][:13], 0) + hours[agent][b["bucket"][:13]] += b["tools"] + b["messages"] + marks = " .:-=+*#%@" + stamps = sorted({h for row in hours.values() for h in row}) + lines = [f"Activity per hour since {data['since']} (UTC hours {stamps[0][11:13] if stamps else '-'}" + f"..{stamps[-1][11:13] if stamps else '-'}):"] + for agent in harbor.agent_names(): + row = hours.get(agent, {}) + peak = max(row.values() or [1]) + lines.append(f" {agent:<9} " + "".join(marks[min(9, int(9 * row.get(h, 0) / peak + 0.99))] if row.get(h) else " " + for h in stamps)) + out(data, "\n".join(lines)) + elif cmd == "wins": + rows = harbor.wins(args.since) + out(rows, "\n".join(f"{w['at']} #{w['item_id']} {w['title']} by {w['author']}" for w in rows) or "No wins yet.") + elif cmd == "git": + radar = harbor.git_radar() + lines = [] + for agent, g in radar.items(): + if not g: + continue + if g.get("error"): + lines.append(f"{agent:<9} {g['workdir']}: {g['error']}") + continue + last = g["last_commit"] + lines.append(f"{agent:<9} {g['branch']:<18} last {last['sha']} {harbor.minutes_since(last['at'])} min ago: " + f"{last['subject'][:60]}") + lines.append(f"{'':<9} {g['dirty']} uncommitted, {g['behind_main']} behind main, " + f"{len(g['unsubmitted'])} commit(s) since the last submit") + out(radar, "\n".join(lines) or "No agent has a workdir in the project file.") + elif cmd == "summarize": + text = Path(args.file).read_text(encoding="utf-8") if args.file else " ".join(args.text) + r = harbor.summarize(who(), args.thread, text) + out(r, f"summary #{r['id']} pinned to {r['thread']}") + elif cmd == "who": + rows = harbor.experts(args.scope) + out(rows, "\n".join(f"{e['agent']:<9} " + ", ".join(f"{v} {k}" for k, v in e["detail"].items()) for e in rows) + or "Nobody has worked on it yet.") + elif cmd == "feed": + rows = harbor.activity_feed(limit=args.limit, agent=args.filter_agent) + out(rows, "\n".join(f"{r['at'][11:19]} {r['agent']:<9} {r['tool'] or '':<14} {r['target'] or ''}" for r in reversed(rows)) + or "No tool activity recorded yet.") + elif cmd == "templates": + out(project.templates, "\n".join(f"{name}: " + " -> ".join(s.get("name", s["title"]) for s in spec["steps"]) + for name, spec in project.templates.items()) or "No templates defined.") + elif cmd == "export": + text = harbor.export_thread(args.thread) + if args.out: + Path(args.out).write_text(text, encoding="utf-8") + print(f"wrote {args.out}") + else: + print(text) + elif cmd == "handoff": + text = Path(args.file).read_text(encoding="utf-8") if args.file else " ".join(args.text) + r = harbor.handoff(who(), text) + out(r, f"handoff #{r['id']} saved; it opens your next brief") + elif cmd == "handoffs": + rows = harbor.handoffs(agent=args.filter_agent, limit=args.limit) + out(rows, "\n\n".join(f"#{h['id']} {h['agent']} at {h['created_at']}\n" + "\n".join( + " " + line for line in h["body"].split("\n")) for h in rows) or "No handoffs.") + elif cmd in ("pin", "unpin"): + rows = harbor.pin(who(), args.message_id, args.note) if cmd == "pin" else harbor.unpin(who(), args.message_id) + out(rows, f"{len(rows)} pinned message(s)") + elif cmd in ("pins", "decisions"): + rows = harbor.pins() if cmd == "pins" else harbor.decisions() + out(rows, "\n\n".join(fmt_message(m) for m in rows) or f"No {cmd}.") + elif cmd == "prompt": + text = harbor.restart_prompt(args.agent_name) + out({"prompt": text}, text) + elif cmd == "doctor": + from . import doctor + checks = doctor.run(project, harbor) + out(checks, doctor.render(checks)) + elif cmd == "backup": + path = harbor.backup(keep=args.keep) + out({"backup": str(path)}, f"backed up to {path}") + + +def run(): + try: + main() + except (HarborError, projects.ProjectError) as error: + sys.exit(f"harbor: {error}") + + +if __name__ == "__main__": + run() diff --git a/harbor/core.py b/harbor/core.py new file mode 100644 index 0000000..b925221 --- /dev/null +++ b/harbor/core.py @@ -0,0 +1,1355 @@ +"""Harbor's store: messages in threads, per-recipient read and acknowledgement state, claims, review items and stop +flags for one project's team of agents and its human. + +One SQLite database per project (WAL mode). Writes use BEGIN IMMEDIATE with a busy timeout, so the command line, the +web server and any number of MCP servers can write at once. When the project has a mirror file, every message is +also appended to it in the plain `[time] From → To: text` line format, so tools that read that file keep working; +the database is the source of truth. +""" +import hashlib +import json +import re +import shutil +import sqlite3 +import time +from pathlib import Path + +from .util import (HarborError, first_line, iso, parse_iso, parse_scope, parse_since, parse_until, # noqa: F401 + scopes_overlap, squashed_tokens, utc) +from .work import OPEN_TASK_STATES, TASK_STATES, WORK_SCHEMA, WorkMixin # noqa: F401 +from .extras import EXTRA_SCHEMA, ExtrasMixin # noqa: F401 +from .flow import FlowMixin # noqa: F401 +from .launch import LAUNCH_SCHEMA, LaunchMixin, describe_target # noqa: F401 + +KINDS = ("note", "question", "answer", "request", "finding", "retraction", "decision", "status", "handoff", "stop", + "resume") +IMPORTANCE = ("normal", "high", "urgent") +ORIGINS = ("cli", "cli-legacy", "mcp", "web", "legacy", "system", "hook") +MAX_BODY = 20000 +CLAIM_KINDS = {"band": None, "draft": 8, "naming": 24, "edit": 2, "review": 4} # default TTL in hours +EXCLUSIVE = ("band", "draft", "naming", "edit") +ITEM_STATES = ("submitted", "in_review", "changes_requested", "approved", "landed", "held", "withdrawn", "rejected") +TRANSITIONS = { + "submitted": {"in_review", "approved", "changes_requested", "held", "withdrawn", "rejected"}, + "in_review": {"changes_requested", "approved", "held", "rejected", "submitted"}, + "changes_requested": {"submitted", "withdrawn", "held"}, + "approved": {"landed", "held", "changes_requested"}, + "held": {"submitted", "in_review", "withdrawn", "rejected"}, + "landed": set(), "withdrawn": set(), "rejected": set(), +} +AUTHOR_STATES = {"submitted", "withdrawn"} +OPEN_ITEM_STATES = ("submitted", "in_review", "changes_requested", "approved", "held") +FLOOD_PER_HOUR = 8 +RESERVED_REFS = {"corrects", "attachments", "question", "item", "task"} # set by Harbor itself, never by callers # messages to everyone per sender per hour before Harbor suggests a status line instead +SCHEMA_VERSION = "5" + +SCHEMA = """ +CREATE TABLE IF NOT EXISTS meta(key TEXT PRIMARY KEY, value TEXT); +CREATE TABLE IF NOT EXISTS agents(name TEXT PRIMARY KEY, role TEXT, kind TEXT, last_seen TEXT); +CREATE TABLE IF NOT EXISTS threads( + id INTEGER PRIMARY KEY, topic TEXT UNIQUE NOT NULL, subject TEXT NOT NULL, state TEXT NOT NULL DEFAULT 'open', + created_by TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL); +CREATE TABLE IF NOT EXISTS messages( + id INTEGER PRIMARY KEY, thread_id INTEGER NOT NULL REFERENCES threads(id), sender TEXT NOT NULL, + recipients TEXT NOT NULL, kind TEXT NOT NULL, importance TEXT NOT NULL, ack_required INTEGER NOT NULL DEFAULT 0, + reply_to INTEGER, refs TEXT, body TEXT NOT NULL, origin TEXT NOT NULL, idem TEXT UNIQUE, created_at TEXT NOT NULL); +CREATE INDEX IF NOT EXISTS messages_thread ON messages(thread_id, id); +CREATE INDEX IF NOT EXISTS messages_sender ON messages(sender, created_at); +CREATE TABLE IF NOT EXISTS deliveries( + message_id INTEGER NOT NULL REFERENCES messages(id), agent TEXT NOT NULL, read_at TEXT, acked_at TEXT, + ack_note TEXT, PRIMARY KEY(message_id, agent)); +CREATE INDEX IF NOT EXISTS deliveries_agent ON deliveries(agent, read_at); +CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5(body, content='messages', content_rowid='id'); +CREATE TRIGGER IF NOT EXISTS messages_fts_insert AFTER INSERT ON messages BEGIN + INSERT INTO messages_fts(rowid, body) VALUES (new.id, new.body); +END; +CREATE TABLE IF NOT EXISTS claims( + id INTEGER PRIMARY KEY, agent TEXT NOT NULL, scope TEXT NOT NULL, lo INTEGER, hi INTEGER, kind TEXT NOT NULL, + note TEXT, created_at TEXT NOT NULL, expires_at TEXT, released_at TEXT); +CREATE TABLE IF NOT EXISTS items( + id INTEGER PRIMARY KEY, title TEXT NOT NULL, scope TEXT, commit_hash TEXT, branch TEXT, author TEXT NOT NULL, + assignee TEXT, state TEXT NOT NULL, thread_id INTEGER, note TEXT, created_at TEXT NOT NULL, + updated_at TEXT NOT NULL); +CREATE TABLE IF NOT EXISTS item_events( + id INTEGER PRIMARY KEY, item_id INTEGER NOT NULL, agent TEXT NOT NULL, from_state TEXT, to_state TEXT NOT NULL, + note TEXT, commit_hash TEXT, at TEXT NOT NULL); +CREATE TABLE IF NOT EXISTS flags(key TEXT PRIMARY KEY, value TEXT NOT NULL, set_by TEXT NOT NULL, origin TEXT NOT NULL, + relay_of INTEGER, reason TEXT, set_at TEXT NOT NULL); +CREATE TABLE IF NOT EXISTS flag_events(id INTEGER PRIMARY KEY, key TEXT NOT NULL, value TEXT NOT NULL, + set_by TEXT NOT NULL, origin TEXT NOT NULL, relay_of INTEGER, reason TEXT, at TEXT NOT NULL); +""" +MIGRATIONS = (("agents", "doing", "TEXT"), ("agents", "doing_at", "TEXT"), ("agents", "active", "INTEGER DEFAULT 1"), + ("agents", "active_at", "TEXT"), ("agents", "last_tool", "TEXT"), + ("messages", "corrects", "INTEGER"), ("messages", "due_at", "TEXT"), ("flags", "until", "TEXT")) + +class Harbor(WorkMixin, ExtrasMixin, FlowMixin, LaunchMixin): + def __init__(self, project, clock=None, shared=False, busy_ms=30000, write_attempts=60): + """shared=True lets several threads use this object; the caller must serialize the calls (the web server + holds a lock per project). Hooks pass a short busy_ms and few write_attempts so a busy database can never + stall an agent's tool call.""" + self.write_attempts = write_attempts + self.project = project + self._project_mtime = self._source_mtime() + self.human = project.human + self.coordinators = set(project.coordinators) | {project.human} + self.db_path = Path(project.db) + self.mirror = Path(project.mirror) if project.mirror else None + self.clock = clock or time.time + self.db_path.parent.mkdir(parents=True, exist_ok=True) + self.conn = sqlite3.connect(str(self.db_path), timeout=busy_ms / 1000, isolation_level=None, + check_same_thread=not shared) + self.conn.row_factory = sqlite3.Row + self.conn.execute("PRAGMA journal_mode=WAL") + self.conn.execute(f"PRAGMA busy_timeout={int(busy_ms)}") + self.conn.execute("PRAGMA foreign_keys=ON") + self.conn.executescript(SCHEMA + WORK_SCHEMA + EXTRA_SCHEMA + LAUNCH_SCHEMA) + try: + self._migrate_and_seed() + except (HarborError, sqlite3.OperationalError): + if write_attempts > 5: + raise # a quick (hook) open may skip seeding; the next normal open does it + + def close(self): + self.conn.close() + + # ----------------------------------------------------------------- plumbing + def now(self): + return iso(utc(self.clock())) + + def minutes_since(self, stamp): + return None if not stamp else int((utc(self.clock()) - parse_iso(stamp)).total_seconds() // 60) + + def _write(self, fn): + """Run fn(conn) inside BEGIN IMMEDIATE ... COMMIT; retries briefly if another writer holds the lock.""" + for attempt in range(self.write_attempts): + try: + self.conn.execute("BEGIN IMMEDIATE") + break + except sqlite3.OperationalError as error: + if "locked" not in str(error) and "busy" not in str(error): + raise + time.sleep(0.05 * (attempt + 1)) + else: + raise HarborError("database is busy; try again") + try: + result = fn(self.conn) + self.conn.execute("COMMIT") + return result + except BaseException: + self.conn.execute("ROLLBACK") + raise + + def _migrate_and_seed(self): + signature = json.dumps([SCHEMA_VERSION, self.project.name, self.project.agents], sort_keys=True) + row = self.conn.execute("SELECT value FROM meta WHERE key='signature'").fetchone() + if row and row["value"] == signature: + return # nothing to do: hooks open Harbor on every tool call, keep that read-only + + def do(c): + for table, column, decl in MIGRATIONS: + if column not in {r["name"] for r in c.execute(f"PRAGMA table_info({table})")}: + c.execute(f"ALTER TABLE {table} ADD COLUMN {column} {decl}") + c.execute("CREATE INDEX IF NOT EXISTS messages_corrects ON messages(corrects)") + roster = {a["name"]: a for a in self.project.agents} + for a in self.project.agents: + c.execute("INSERT INTO agents(name, role, kind, active) VALUES (?, ?, ?, 1) ON CONFLICT(name) DO " + "UPDATE SET role=excluded.role, kind=excluded.kind, active=1", (a["name"], a["role"], a["kind"])) + for r in c.execute("SELECT name FROM agents").fetchall(): + if r["name"] not in roster: # left the roster: keep their history, stop delivering + c.execute("UPDATE agents SET active=0 WHERE name=?", (r["name"],)) + c.execute("INSERT INTO meta(key, value) VALUES ('schema', ?) ON CONFLICT(key) DO UPDATE SET value=excluded.value", + (SCHEMA_VERSION,)) + c.execute("INSERT OR IGNORE INTO meta(key, value) VALUES ('project', ?)", (self.project.name,)) + owner = c.execute("SELECT value FROM meta WHERE key='project'").fetchone()["value"] + if owner == self.project.name: + c.execute("INSERT INTO meta(key, value) VALUES ('signature', ?) ON CONFLICT(key) DO UPDATE SET " + "value=excluded.value", (signature,)) + self._write(do) + owner = self.conn.execute("SELECT value FROM meta WHERE key='project'").fetchone()["value"] + if owner != self.project.name: + raise HarborError(f"{self.db_path} belongs to project {owner!r}, not {self.project.name!r}") + + def agents(self): + rows = [dict(r) for r in self.conn.execute("SELECT * FROM agents WHERE active=1")] + order = {name: i for i, name in enumerate(self.project.names)} + style = {a["name"]: a for a in self.project.agents} + for r in rows: + spec = style.get(r["name"], {}) + r["initials"] = spec.get("initials") or r["name"][:2].upper() + r["color"] = spec.get("color") or "#7b8797" + return sorted(rows, key=lambda r: order.get(r["name"], 999)) + + def agent_names(self): + return list(self.project.names) + + def _check_agent(self, name, allow_human=True): + if name not in self.project.names: + raise HarborError(f"unknown agent {name!r}; known: {', '.join(self.project.names)}") + if name == self.human and not allow_human: + raise HarborError(f"only the chat page can speak as {self.human}") + + def _touch(self, c, agent): + c.execute("UPDATE agents SET last_seen=? WHERE name=?", (self.now(), agent)) + + def touch(self, agent, channel=None, host=""): + self._check_agent(agent) + if channel: + self.connect(agent, channel, host) + else: + self._write(lambda c: self._touch(c, agent)) + + def doing(self, agent, text): + """Set (or clear, with empty text) the one-line 'what I am doing' shown to the team.""" + self._check_agent(agent, allow_human=False) + text = " ".join((text or "").split())[:200] or None + now = self.now() + + def do(c): + c.execute("UPDATE agents SET doing=?, doing_at=?, last_seen=? WHERE name=?", (text, now, now, agent)) + self._write(do) + return {"agent": agent, "doing": text, "at": now} + + # ----------------------------------------------------------------- threads and messages + def _thread_id(self, c, thread, subject, sender): + if isinstance(thread, int) or (isinstance(thread, str) and thread.isdigit()): + row = c.execute("SELECT id FROM threads WHERE id=?", (int(thread),)).fetchone() + if not row: + raise HarborError(f"no thread #{thread}") + return row["id"] + topic = (thread or "general").strip() + row = c.execute("SELECT id FROM threads WHERE topic=?", (topic,)).fetchone() + if row: + return row["id"] + now = self.now() + cur = c.execute("INSERT INTO threads(topic, subject, created_by, created_at, updated_at) VALUES (?,?,?,?,?)", + (topic, (subject or topic).strip()[:200], sender, now, now)) + return cur.lastrowid + + def lint(self, sender, body, kind, refs, to_all): + warnings = [] + squashed = len(squashed_tokens(body)) + if squashed >= 2 and squashed * 15 > len(body.split()): + warnings.append("the text looks compressed (words and numbers run together); write plain sentences - " + "there is no length limit to squeeze into") + if kind in ("finding", "retraction") and not refs: + warnings.append(f"a {kind} should name what it is about (refs: commit, address or file, or --corrects)") + if to_all and sender != self.human: + since = iso(utc(self.clock() - 3600)) + count = self.conn.execute("SELECT COUNT(*) FROM messages WHERE sender=? AND created_at>=? AND " + "recipients='[\"All\"]' AND origin!='legacy'", (sender, since)).fetchone()[0] + if count >= FLOOD_PER_HOUR: + warnings.append(f"that is {count + 1} messages to everyone in the last hour; use `doing` for progress " + "and a thread with the people concerned for details") + return warnings + + def attach(self, path): + """Copy a file into the project's attachment store; returns its stored name.""" + source = Path(path) + if not source.is_file(): + raise HarborError(f"attachment {path}: not a file") + if source.stat().st_size > 25 * 1024 * 1024: + raise HarborError(f"attachment {path}: larger than 25 MB; reference its path instead") + digest = hashlib.sha256(source.read_bytes()).hexdigest()[:12] + safe = re.sub(r"[^A-Za-z0-9._-]+", "_", source.name)[-80:] + name = f"{digest}-{safe}" + self.project.attachments.mkdir(parents=True, exist_ok=True) + target = self.project.attachments / name + if not target.exists(): + shutil.copyfile(source, target) + return name + + def send(self, sender, to, body, subject=None, thread=None, kind="note", importance="normal", + ack_required=False, reply_to=None, refs=None, origin="cli", idem=None, attachments=None, corrects=None, + due=None, trusted_refs=None): + """Post a message. to: list of agent names, or 'All'. corrects: id of the message this corrects (it goes + to the same thread and people). due: '30m', '2h' or an ISO time by which an answer is needed.""" + if origin not in ORIGINS: + raise HarborError(f"unknown origin {origin}") + self._check_agent(sender, allow_human=(origin in ("web", "legacy", "system"))) + body = (body or "").replace("\r\n", "\n").strip() + if not body: + raise HarborError("empty message") + if len(body) > MAX_BODY: + raise HarborError(f"message is {len(body)} characters; the limit is {MAX_BODY} - attach a file instead") + if kind not in KINDS: + raise HarborError(f"kind must be one of {', '.join(KINDS)}") + if importance not in IMPORTANCE: + raise HarborError(f"importance must be one of {', '.join(IMPORTANCE)}") + refs = {str(k)[:40]: str(v)[:300] for k, v in (refs or {}).items() + if str(k) not in RESERVED_REFS and v is not None} + refs.update(trusted_refs or {}) + due_at = parse_until(due, utc(self.clock())) if due else None + if corrects is not None: + original = self.message(corrects) + if not original: + raise HarborError(f"no message #{corrects} to correct") + if original["sender"] != sender and sender not in self.coordinators: + raise HarborError(f"#{corrects} is {original['sender']}'s message; reply to it instead") + corrects = int(corrects) + refs["corrects"] = corrects + if kind == "note": + kind = "retraction" + if reply_to is None and thread is None: + reply_to = int(corrects) + if attachments: + refs["attachments"] = [self.attach(p) for p in attachments] + to_all = to in (None, "", "All", ["All"]) and reply_to is None + if origin not in ("system", "legacy", "hook"): + self.check_message_policy(sender, to_all) + warnings = self.lint(sender, body, kind, refs, to_all) + + def do(c): + if idem: + row = c.execute("SELECT id, thread_id, recipients FROM messages WHERE idem=?", (idem,)).fetchone() + if row: + return {"id": row["id"], "thread_id": row["thread_id"], "recipients": json.loads(row["recipients"]), + "duplicate": True} + nonlocal to + thread_id = None + if reply_to is not None: + parent = c.execute("SELECT * FROM messages WHERE id=?", (int(reply_to),)).fetchone() + if not parent: + raise HarborError(f"no message #{reply_to}") + thread_id = parent["thread_id"] + if not to: + previous = json.loads(parent["recipients"]) + if "All" in previous: + to = "All" + else: + to = sorted({parent["sender"], *previous} - {sender}) or [parent["sender"]] + if thread_id is None: + thread_id = self._thread_id(c, thread, subject, sender) + names = self.agent_names() + if to in (None, "", "All", ["All"]): + recipients = ["All"] + targets = [n for n in names if n != sender] + else: + recipients = list(dict.fromkeys([to] if isinstance(to, str) else list(to))) + for name in recipients: + if name not in names: + raise HarborError(f"unknown recipient {name!r}; known: {', '.join(names)} or All") + targets = [n for n in recipients if n != sender] + now = self.now() + cur = c.execute( + "INSERT INTO messages(thread_id, sender, recipients, kind, importance, ack_required, reply_to, refs, " + "body, origin, idem, created_at, corrects, due_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)", + (thread_id, sender, json.dumps(recipients), kind, importance, int(bool(ack_required)), reply_to, + json.dumps(refs) if refs else None, body, origin, idem, now, corrects, due_at)) + message_id = cur.lastrowid + for name in targets: + c.execute("INSERT INTO deliveries(message_id, agent) VALUES (?, ?)", (message_id, name)) + if reply_to is not None: + c.execute("UPDATE deliveries SET acked_at=COALESCE(acked_at, ?), read_at=COALESCE(read_at, ?), " + "ack_note=COALESCE(ack_note, ?) WHERE message_id=? AND agent=?", + (now, now, f"replied #{message_id}", int(reply_to), sender)) + c.execute("UPDATE threads SET updated_at=?, state=CASE state WHEN 'closed' THEN 'open' ELSE state END " + "WHERE id=?", (now, thread_id)) + self._touch(c, sender) + if origin != "legacy": + self._pulse(c, sender, messages=1) + topic = c.execute("SELECT topic FROM threads WHERE id=?", (thread_id,)).fetchone()["topic"] + return {"id": message_id, "thread_id": thread_id, "topic": topic, "recipients": recipients, + "targets": targets, "created_at": now} + + result = self._write(do) + result["warnings"] = warnings + if not result.get("duplicate") and origin != "legacy": + try: + self._mirror(result, sender, kind, importance, ack_required, body, refs) + except Exception as error: # the database already holds the message + result["warnings"].append(f"mirror to {self.mirror} failed: {error}") + return result + + def _mirror(self, result, sender, kind, importance, ack_required, body, refs): + if not self.mirror: + return + import msvcrt + stamp = result["created_at"].replace("T", " ")[:16] + "Z" + single = len(result["recipients"]) == 1 + to = result["recipients"][0] if single else "All" # old readers parse a single name only + tags = [f"#{result['id']}", f"thread {result['thread_id']} {result['topic']}"] + if not single: + tags.append("to " + ",".join(result["recipients"])) + if kind != "note": + tags.append(kind) + if importance != "normal": + tags.append(importance) + if ack_required: + tags.append("ack") + if refs.get("corrects"): + tags.append(f"corrects #{refs['corrects']}") + lines = body.split("\n") + if refs.get("attachments"): + lines.append("Attachments: " + ", ".join(refs["attachments"])) + text = f"[{stamp}] {sender} → {to}: [{' · '.join(tags)}] {lines[0]}" + rest = "".join(f"\n | {line}" for line in lines[1:]) + self.mirror.parent.mkdir(parents=True, exist_ok=True) + lock_path = self.mirror.with_suffix(".lock") + with lock_path.open("a+b") as lock: + lock.seek(0, 2) + if lock.tell() == 0: + lock.write(b"\0") + lock.flush() + deadline = time.monotonic() + 10 + while True: + lock.seek(0) + try: + msvcrt.locking(lock.fileno(), msvcrt.LK_NBLCK, 1) + break + except OSError: + if time.monotonic() >= deadline: + raise TimeoutError("chat mirror lock is busy") + time.sleep(0.02) + try: + needs_newline = False + if self.mirror.exists(): + with self.mirror.open("rb") as previous: + previous.seek(0, 2) + if previous.tell(): + previous.seek(-1, 2) + needs_newline = previous.read(1) != b"\n" + with self.mirror.open("ab") as stream: + stream.write((("\n" if needs_newline else "") + "\n" + text + rest + "\n").encode("utf-8")) + finally: + lock.seek(0) + msvcrt.locking(lock.fileno(), msvcrt.LK_UNLCK, 1) + + MESSAGE_SELECT = "SELECT m.*, t.topic, t.subject FROM messages m JOIN threads t ON t.id=m.thread_id" + + def message(self, message_id): + row = self.conn.execute(self.MESSAGE_SELECT + " WHERE m.id=?", (int(message_id),)).fetchone() + return self._message_dict(row) if row else None + + def _message_dict(self, row): + d = dict(row) + d["recipients"] = json.loads(d["recipients"]) + d["refs"] = json.loads(d["refs"]) if d.get("refs") else {} + d["ack_required"] = bool(d["ack_required"]) + return d + + def messages_after(self, after_id, limit=500): + return [self._message_dict(r) for r in self.conn.execute( + self.MESSAGE_SELECT + " WHERE m.id>? ORDER BY m.id LIMIT ?", (int(after_id), int(limit)))] + + def messages_before(self, before_id, limit=300): + rows = self.conn.execute(self.MESSAGE_SELECT + " WHERE m.id= deadline: + messages = self.inbox(agent, unread_only=True, limit=30, mark_read=mark_read) if n else [] + return {"messages": messages, "stopped": [f["key"] for f in stops], + "timed_out": not n and not stops} + time.sleep(min(poll, max(0.05, deadline - time.monotonic()))) + + def ack(self, agent, message_ids, note=None): + self._check_agent(agent) + now = self.now() + + def do(c): + done = [] + for mid in message_ids: + cur = c.execute("UPDATE deliveries SET acked_at=COALESCE(acked_at, ?), read_at=COALESCE(read_at, ?), " + "ack_note=COALESCE(?, ack_note) WHERE message_id=? AND agent=?", + (now, now, note, int(mid), agent)) + if cur.rowcount == 0: + raise HarborError(f"message #{mid} was not addressed to {agent}") + done.append(int(mid)) + self._touch(c, agent) + return done + return self._write(do) + + def receipts(self, message_ids): + """{message id: {"read": [...], "acked": [...], "waiting": [...]}} for the given messages.""" + ids = [int(i) for i in message_ids] + out = {i: {"read": [], "acked": [], "waiting": []} for i in ids} + for start in range(0, len(ids), 500): + chunk = ids[start:start + 500] + for r in self.conn.execute(f"SELECT message_id, agent, read_at, acked_at FROM deliveries WHERE message_id IN " + f"({','.join('?' * len(chunk))})", chunk): + entry = out[r["message_id"]] + if r["acked_at"]: + entry["acked"].append(r["agent"]) + elif r["read_at"]: + entry["read"].append(r["agent"]) + else: + entry["waiting"].append(r["agent"]) + return out + + def pending_acks(self, sender=None, overdue_minutes=0): + """Messages that asked for an acknowledgement and still lack it, per recipient.""" + query = ("SELECT m.id, m.sender, m.created_at, m.body, d.agent FROM messages m JOIN deliveries d " + "ON d.message_id=m.id WHERE m.ack_required=1 AND d.acked_at IS NULL") + args = [] + if sender: + query += " AND m.sender=?" + args.append(sender) + rows = [dict(r) for r in self.conn.execute(query + " ORDER BY m.id", args)] + if overdue_minutes: + cutoff = utc(self.clock() - overdue_minutes * 60) + rows = [r for r in rows if parse_iso(r["created_at"]) <= cutoff] + return rows + + def thread(self, thread): + if isinstance(thread, int) or str(thread).isdigit(): + row = self.conn.execute("SELECT * FROM threads WHERE id=?", (int(thread),)).fetchone() + else: + row = self.conn.execute("SELECT * FROM threads WHERE topic=?", (str(thread),)).fetchone() + if not row: + raise HarborError(f"no thread {thread!r}") + messages = [self._message_dict(r) for r in self.conn.execute( + "SELECT m.*, ? topic, ? subject FROM messages m WHERE thread_id=? ORDER BY id", + (row["topic"], row["subject"], row["id"]))] + return dict(row), messages + + def set_thread_state(self, agent, thread, state): + if state not in ("open", "closed"): + raise HarborError("thread state is open or closed") + self._check_agent(agent) + info, _ = self.thread(thread) + + def do(c): + c.execute("UPDATE threads SET state=?, updated_at=? WHERE id=?", (state, self.now(), info["id"])) + self._touch(c, agent) + self._write(do) + return self.thread(info["id"])[0] + + def threads(self, limit=30, agent=None, include_legacy=False, include_closed=True): + query = ("SELECT t.*, (SELECT COUNT(*) FROM messages m WHERE m.thread_id=t.id) messages, " + "(SELECT MAX(id) FROM messages m WHERE m.thread_id=t.id) last_id") + args = [] + if agent: + query += (", (SELECT COUNT(*) FROM deliveries d JOIN messages m ON m.id=d.message_id WHERE " + "m.thread_id=t.id AND d.agent=? AND d.read_at IS NULL) unread") + args.append(agent) + query += " FROM threads t WHERE 1=1" + if not include_legacy: + query += " AND t.topic NOT LIKE 'legacy:%'" + if not include_closed: + query += " AND t.state != 'closed'" + query += " ORDER BY t.updated_at DESC, t.id DESC LIMIT ?" + args.append(int(limit)) + return [dict(r) for r in self.conn.execute(query, args)] + + def recent(self, limit=500): + rows = self.conn.execute(self.MESSAGE_SELECT + " ORDER BY m.id DESC LIMIT ?", (int(limit),)).fetchall() + return [self._message_dict(r) for r in reversed(rows)] + + SEARCH_FILTER = re.compile(r"\b(from|to|kind|thread|since|before|pinned):(\S+)") + + def search(self, query, limit=20): + """Full-text search. Filters: from:Agent to:Agent kind:decision thread:fn:0x... since:2d before:.""" + filters = {k.lower(): v for k, v in self.SEARCH_FILTER.findall(query)} + text = self.SEARCH_FILTER.sub(" ", query) + terms = " ".join(f'"{t}"' for t in re.findall(r"[\w\-\.]+", text)) + where, args = [], [] + if "from" in filters: + where.append("m.sender=?") + args.append(next((n for n in self.project.names if n.lower() == filters["from"].lower()), filters["from"])) + if "to" in filters: + name = next((n for n in self.project.names if n.lower() == filters["to"].lower()), filters["to"]) + where.append("EXISTS (SELECT 1 FROM deliveries d WHERE d.message_id=m.id AND d.agent=?)") + args.append(name) + if "kind" in filters: + where.append("m.kind=?") + args.append(filters["kind"].lower()) + if "thread" in filters: + where.append("t.topic=?") + args.append(filters["thread"]) + if "since" in filters: + where.append("m.created_at>=?") + args.append(parse_since(filters["since"], utc(self.clock()))) + if "before" in filters: + where.append("m.created_at ?)" + args = [now] + if agent: + query += " AND agent=?" + args.append(agent) + if kind: + query += " AND kind=?" + args.append(kind) + return [dict(r) for r in self.conn.execute(query + " ORDER BY COALESCE(lo, 0), id", args)] + + def require_running(self, agent, what): + """Refuse work while a STOP applies to agent (messages, acknowledgements and handoffs stay allowed).""" + if agent != self.human and self.stopped(agent): + raise HarborError(f"STOP in force for {agent}: you cannot {what} until {self.human} resumes you") + + def claim(self, agent, scope, kind="draft", ttl_hours=None, note=None, override=False, on_behalf_of=None): + """Record a claim. Exclusive kinds are refused when another agent holds an overlapping one; a draft is also + refused inside another agent's band. Coordinators may override (the override is recorded in the note).""" + if not (on_behalf_of and agent in self.coordinators and kind in ("band", "naming")): + self.require_running(on_behalf_of or agent, "claim work") # recording an allocation starts no work + if on_behalf_of: + if agent not in self.coordinators: + raise HarborError("only a coordinator records claims for another agent") + self._check_agent(on_behalf_of, allow_human=False) + note = ((note or "") + f" (recorded by {agent})").strip() + agent = on_behalf_of + coordinator = True + else: + self._check_agent(agent, allow_human=False) + coordinator = agent in self.coordinators + if kind not in CLAIM_KINDS: + raise HarborError(f"claim kind must be one of {', '.join(CLAIM_KINDS)}") + if kind == "band" and not coordinator: + raise HarborError("bands (standing ownership) are set by a coordinator") + canonical, lo, hi = parse_scope(scope) + if kind == "edit" and not canonical.startswith("file:"): + raise HarborError("edit claims are for files (file:path)") + hours = CLAIM_KINDS[kind] if ttl_hours is None else ttl_hours + expires = None if hours is None else iso(utc(self.clock() + float(hours) * 3600)) + if override and not coordinator: + raise HarborError("only a coordinator can override a conflict") + + def do(c): + now = self.now() + active = [dict(r) for r in c.execute( + "SELECT * FROM claims WHERE released_at IS NULL AND (expires_at IS NULL OR expires_at > ?)", (now,))] + mine = [a for a in active if a["agent"] == agent and a["kind"] == kind and a["scope"] == canonical] + if mine: # renewal + c.execute("UPDATE claims SET expires_at=?, note=COALESCE(?, note) WHERE id=?", + (expires, note, mine[0]["id"])) + self._touch(c, agent) + return {"id": mine[0]["id"], "scope": canonical, "kind": kind, "expires_at": expires, "renewed": True, + "conflicts": []} + conflicts = [] + if kind in EXCLUSIVE: + conflicts += [a for a in active if a["agent"] != agent and a["kind"] == kind + and scopes_overlap((canonical, lo, hi), (a["scope"], a["lo"], a["hi"]))] + if kind == "draft" and lo is not None: + bands = [a for a in active if a["kind"] == "band" + and scopes_overlap((canonical, lo, hi), (a["scope"], a["lo"], a["hi"]))] + if bands and not any(b["agent"] == agent for b in bands): + conflicts += bands + if conflicts and not override: + described = "; ".join(f"{a['agent']} holds {a['kind']} {a['scope']}" + + (f" until {a['expires_at']}" if a["expires_at"] else "") for a in conflicts) + raise HarborError(f"claim refused: {described}. Ask the holder in a thread, or a coordinator.") + text = note + if conflicts: + text = ((note or "") + f" [override of {', '.join(str(a['id']) for a in conflicts)}]").strip() + cur = c.execute("INSERT INTO claims(agent, scope, lo, hi, kind, note, created_at, expires_at) " + "VALUES (?,?,?,?,?,?,?,?)", (agent, canonical, lo, hi, kind, text, now, expires)) + self._touch(c, agent) + return {"id": cur.lastrowid, "scope": canonical, "kind": kind, "expires_at": expires, "renewed": False, + "conflicts": [a["id"] for a in conflicts]} + return self._write(do) + + def release(self, agent, claim_id=None, scope=None): + def do(c): + if claim_id is not None: + rows = c.execute("SELECT * FROM claims WHERE id=? AND released_at IS NULL", (int(claim_id),)).fetchall() + else: + canonical = parse_scope(scope)[0] + rows = c.execute("SELECT * FROM claims WHERE scope=? AND agent=? AND released_at IS NULL", + (canonical, agent)).fetchall() + if not rows: + raise HarborError("no matching active claim") + for r in rows: + if r["agent"] != agent and agent not in self.coordinators: + raise HarborError(f"claim #{r['id']} belongs to {r['agent']}") + c.execute("UPDATE claims SET released_at=? WHERE id=?", (self.now(), r["id"])) + self._touch(c, agent) + return [r["id"] for r in rows] + return self._write(do) + + # ----------------------------------------------------------------- review items + def submit(self, author, title, scope=None, commit=None, branch=None, note=None, reviewers=None, task=None): + self._check_agent(author, allow_human=False) + self.require_running(author, "submit work") + for name in reviewers or (): + self._check_agent(name) + linked = None + if task is not None: + linked, _ = self.task(task) + if linked["assignee"] not in (None, author): + raise HarborError(f"task #{task} is {linked['assignee']}'s") + if linked["state"] not in ("in_progress", "blocked", "open"): + raise HarborError(f"task #{task} is {linked['state']}") + scope = scope or linked["scope"] + reviewers = [r for r in (reviewers or self.project.reviewers or self.project.coordinators) if r != author] + if not reviewers: + raise HarborError("no reviewer: name one, or set reviewers in the project file") + canonical = parse_scope(scope)[0] if scope else None + + def do(c): + now = self.now() + cur = c.execute("INSERT INTO items(title, scope, commit_hash, branch, author, state, note, created_at, " + "updated_at) VALUES (?,?,?,?,?,'submitted',?,?,?)", + (title, canonical, commit, branch, author, note, now, now)) + item_id = cur.lastrowid + c.execute("INSERT INTO item_events(item_id, agent, from_state, to_state, note, commit_hash, at) " + "VALUES (?,?,NULL,'submitted',?,?,?)", (item_id, author, note, commit, now)) + if linked: + c.execute("UPDATE tasks SET item_id=?, assignee=COALESCE(assignee, ?) WHERE id=?", + (item_id, author, linked["id"])) + self._move_task(c, linked, author, "review", f"review item #{item_id}") + return item_id + item_id = self._write(do) + body = f"Review item #{item_id} submitted: {title}" + (f" (task #{linked['id']})" if linked else "") + if canonical: + body += f"\nScope: {canonical}" + if commit: + body += f"\nCommit: {commit}" + (f" on {branch}" if branch else "") + if note: + body += f"\n{note}" + message = self.send(author, list(reviewers), body, subject=f"Review: {title}", thread=f"item:{item_id}", + kind="request", refs={"commit": commit, "scope": canonical}, + trusted_refs={"item": item_id}, origin="system") + self._write(lambda c: c.execute("UPDATE items SET thread_id=? WHERE id=?", (message["thread_id"], item_id))) + return {"id": item_id, "thread_id": message["thread_id"]} + + def item(self, item_id): + row = self.conn.execute("SELECT * FROM items WHERE id=?", (int(item_id),)).fetchone() + if not row: + raise HarborError(f"no review item #{item_id}") + events = [dict(r) for r in self.conn.execute("SELECT * FROM item_events WHERE item_id=? ORDER BY id", + (int(item_id),))] + return dict(row), events + + def transition(self, agent, item_id, to_state, note=None, commit=None, notify=True): + self._check_agent(agent) + if to_state not in ITEM_STATES: + raise HarborError(f"state must be one of {', '.join(ITEM_STATES)}") + item, _ = self.item(item_id) + allowed = TRANSITIONS[item["state"]] + # the author may always hand a newer commit in: a submitted, in-review or even approved item goes back to + # submitted with it, so the reviewer looks at what will actually land + refresh = (agent == item["author"] and to_state == "submitted" and commit + and item["state"] in ("submitted", "in_review", "approved")) + if to_state not in allowed and not refresh: + raise HarborError(f"item #{item_id} is {item['state']}; it can move to {', '.join(sorted(allowed)) or 'nothing (final)'}") + is_author = agent == item["author"] + if is_author and to_state not in AUTHOR_STATES: + raise HarborError("the author cannot review their own item") + if not is_author and to_state in AUTHOR_STATES and to_state != "submitted": + raise HarborError("only the author can withdraw an item") + if to_state == "landed" and agent not in self.coordinators: + raise HarborError("only a coordinator (the integrator) lands items") + if to_state == "landed" and not commit: + raise HarborError("landing needs the main commit hash") + if to_state != "withdrawn": + self.require_running(agent, "move review items") + if to_state == "landed": + self.check_landing_policy(item_id) + + def do(c): + now = self.now() + current = c.execute("SELECT state FROM items WHERE id=?", (int(item_id),)).fetchone()["state"] + if current != item["state"]: + raise HarborError(f"item #{item_id} changed meanwhile (now {current}); look again") + c.execute("UPDATE items SET state=?, updated_at=?, assignee=CASE WHEN ? THEN assignee ELSE ? END, " + "commit_hash=COALESCE(?, commit_hash) WHERE id=?", + (to_state, now, int(is_author), agent, commit if is_author else None, int(item_id))) + c.execute("INSERT INTO item_events(item_id, agent, from_state, to_state, note, commit_hash, at) " + "VALUES (?,?,?,?,?,?,?)", (int(item_id), agent, item["state"], to_state, note, commit, now)) + self._write(do) + self._task_follows_item(item_id, to_state, agent) + if not notify: + return self.item(item_id)[0] + if to_state == "landed": + self.announce_win(item, agent, commit) + fallback = (self.project.reviewers or self.project.coordinators or (self.human,))[0] + people = {item["author"], item["assignee"] or fallback} - {agent} or {item["author"]} + body = f"Review item #{item_id} ({item['title']}): {item['state']} -> {to_state}" + if commit: + body += f" (commit {commit})" + if note: + body += f"\n{note}" + kind = "status" if to_state != "changes_requested" else "request" + self.send(agent, sorted(people), body, thread=f"item:{item_id}", subject=f"Review: {item['title']}", + kind=kind, refs={"commit": commit}, trusted_refs={"item": int(item_id)}, origin="system") + return self.item(item_id)[0] + + def land_many(self, agent, item_ids, commit, note=None): + """Land several items that went to main in one commit: each is approved first when still under review, the + authors get one message each and the human one wins note, instead of messages for every step of every item.""" + if agent not in self.coordinators: + raise HarborError("only a coordinator (the integrator) lands items") + items = [self.item(i)[0] for i in dict.fromkeys(int(i) for i in item_ids)] + for it in items: + if it["state"] not in ("submitted", "in_review", "approved"): + raise HarborError(f"item #{it['id']} is {it['state']}; only submitted, in-review or approved items land") + landed = [] + for it in items: + if it["state"] != "approved": + self.transition(agent, it["id"], "approved", note=note, notify=False) + landed.append(self.transition(agent, it["id"], "landed", note=note, commit=commit, notify=False)) + by_author = {} + for it in landed: + by_author.setdefault(it["author"], []).append(it) + for author, mine in by_author.items(): + lines = "\n".join(f"#{i['id']} {i['title']}" for i in mine) + self.send(agent, [author], f"Landed in {commit}: {len(mine)} item{'s' if len(mine) > 1 else ''} of yours" + + (f" ({note})" if note else "") + f"\n{lines}", thread=f"commit:{commit}", + subject=f"Landed in {commit}", kind="status", refs={"commit": commit}, origin="system") + if agent != self.human: + authors = ", ".join(sorted(by_author)) + self.send(agent, [self.human], f"Landed in {commit}: {len(landed)} item{'s' if len(landed) > 1 else ''} by " + f"{authors}.\n" + "\n".join(f"#{i['id']} {i['title']}" for i in landed), thread="wins", + subject="Wins", kind="status", refs={"commit": commit}, origin="system") + return landed + + def _source_mtime(self): + source = getattr(self.project, "source", None) + try: + return Path(source).stat().st_mtime if source else None + except OSError: + return None + + def refresh_project(self): + """Reload the project file when it changed since it was read: roster, reviewers, autopilot, policy and + templates take effect in long-running servers without a restart. A file that no longer parses, or that + names another project or database, is ignored (that needs a restart). One stat when nothing changed.""" + mtime = self._source_mtime() + if mtime is None or mtime == self._project_mtime: + return False + self._project_mtime = mtime + from . import projects + try: + fresh = projects.load_file(self.project.source) + except Exception: # noqa: BLE001 - keep serving the last good project + return False + if fresh.name != self.project.name or Path(fresh.db) != self.db_path: + return False + self.project = fresh + self.human = fresh.human + self.coordinators = set(fresh.coordinators) | {fresh.human} + self.mirror = Path(fresh.mirror) if fresh.mirror else None + return True + + def items(self, states=OPEN_ITEM_STATES, author=None): + query = f"SELECT * FROM items WHERE state IN ({','.join('?' * len(states))})" + args = list(states) + if author: + query += " AND author=?" + args.append(author) + return [dict(r) for r in self.conn.execute(query + " ORDER BY id", args)] + + # ----------------------------------------------------------------- control flags + RELAY_MINUTES = 180 + STOP_WORDS = re.compile(r"\b(stop|pause|halt)\b", re.I) + + def _check_relay(self, c, relay_of, by, key): + """Relaying is for STOP only (the safe direction): the quoted message must be one the human typed on the chat + page (kind note), recent, addressed to the relaying agent, asking to stop, naming the target, unused.""" + source = c.execute("SELECT * FROM messages WHERE id=?", (int(relay_of),)).fetchone() + if not source or source["sender"] != self.human or source["origin"] != "web" or source["kind"] != "note": + raise HarborError(f"message #{relay_of} is not one {self.human} typed on the chat page") + if not c.execute("SELECT 1 FROM deliveries WHERE message_id=? AND agent=?", (int(relay_of), by)).fetchone(): + raise HarborError(f"message #{relay_of} was not addressed to {by}") + if self.minutes_since(source["created_at"]) > self.RELAY_MINUTES: + raise HarborError(f"message #{relay_of} is older than {self.RELAY_MINUTES} minutes; ask {self.human} again") + body = source["body"] + if not self.STOP_WORDS.search(body): + raise HarborError(f"message #{relay_of} does not ask to stop") + who = key[5:] + recipients = json.loads(source["recipients"]) + named = re.search(rf"\b{re.escape(who)}\b", body, re.I) or (who == "all" and re.search( + r"\b(all|everyone|everybody|team|guys)\b", body, re.I)) + if not named and recipients != ([who] if who != "all" else ["All"]): + raise HarborError(f"message #{relay_of} does not name {who}") + if c.execute("SELECT 1 FROM flag_events WHERE relay_of=?", (int(relay_of),)).fetchone(): + raise HarborError(f"message #{relay_of} was already relayed") + + def set_flag(self, key, value, by, origin, reason=None, relay_of=None, announce=True, until=None): + """STOP flags: key 'stop:all' or 'stop:', value 'on'/'off'. + The human sets them from the chat page (origin web). Stopping is the safe direction: an agent may stop + itself, a coordinator may stop anyone, and an agent may relay the human's typed request to stop someone. + Resuming is the human's alone - except that a coordinator may lift a stop it set itself (not one it relayed) + before anyone else touched it. Stopping something already stopped changes nothing.""" + if not (key == "stop:all" or (key.startswith("stop:") and key[5:] in self.agent_names())): + raise HarborError("flag must be stop:all or stop:") + if value not in ("on", "off"): + raise HarborError("value must be on or off") + if by == self.human: + if origin != "web": + raise HarborError(f"only the chat page can act as {self.human}") + else: + self._check_agent(by, allow_human=False) + until_at = None + if until: + if by != self.human or value != "off": + raise HarborError(f"only {self.human} gives a timed resume (a trial run)") + until_at = parse_until(until, utc(self.clock())) + + def do(c): + now = self.now() + row = c.execute("SELECT * FROM flags WHERE key=?", (key,)).fetchone() + current = dict(row) if row else None + checked_relay = None + if by != self.human: + safe_stop = value == "on" and (key == f"stop:{by}" or by in self.coordinators) + own_lift = (value == "off" and by in self.coordinators and current and current["value"] == "on" + and current["set_by"] == by and not current["relay_of"]) + if value == "off" and not own_lift: + raise HarborError(f"only {self.human} resumes, from the chat page or the board") + if value == "on" and relay_of is not None and (by in self.coordinators or not safe_stop): + self._check_relay(c, relay_of, by, key) # a cited request is checked and recorded: + checked_relay = int(relay_of) # the stop is then the human's to lift + elif value == "on" and not safe_stop: + raise HarborError(f"only {self.human} or a coordinator stops another agent; relay their " + "chat-page message with relay_of=") + relay_of_recorded = checked_relay if by != self.human else None + c.execute("INSERT INTO flag_events(key, value, set_by, origin, relay_of, reason, at) VALUES (?,?,?,?,?,?,?)", + (key, value, by, origin, relay_of_recorded, reason, now)) + if current and current["value"] == value == "on" and by != self.human: + return False # already stopped: keep who stopped it + if current and current["value"] == value == "on": # the human confirms it: now it is theirs + c.execute("UPDATE flags SET set_by=?, origin=?, reason=COALESCE(?, reason), set_at=? WHERE key=?", + (by, origin, reason, now, key)) + return False + if current and current["value"] == "off" and value == "off" and not until_at and not current.get("until"): + return False + c.execute("INSERT INTO flags(key, value, set_by, origin, relay_of, reason, set_at, until) VALUES (?,?,?,?,?,?,?,?) " + "ON CONFLICT(key) DO UPDATE SET value=excluded.value, set_by=excluded.set_by, " + "origin=excluded.origin, relay_of=excluded.relay_of, reason=excluded.reason, set_at=excluded.set_at, " + "until=CASE WHEN ? THEN excluded.until ELSE flags.until END", # an agent's stop/lift keeps a trial's end + (key, value, by, origin, relay_of_recorded, reason, now, until_at, int(by == self.human))) + return True + changed = self._write(do) + if not announce or not changed: + return self.flags() + who = key[5:] + target = "All" if who == "all" else who + text = (f"{'STOP' if value == 'on' else 'RESUME'} for {'everyone' if who == 'all' else who}" + + (f": {reason}" if reason else "") + (f" (relaying #{relay_of})" if relay_of and by != self.human + else "")) + if until_at: + text += (f"\nThis is a trial run until {until_at}: at that time the stop comes back by itself. Before then, " + "write a handoff.") + if value == "on": + text += ("\nFinish the step you are in only if stopping halfway would break something, write a handoff " + "(`handoff`: what is finished, what is in flight, what comes next), acknowledge this, and stop.") + self.send(by, target if target != by else "All", text, thread="control", subject="Stop and resume", + kind="stop" if value == "on" else "resume", importance="urgent", ack_required=value == "on", + origin="system") + return self.flags() + + def flags(self): + return {r["key"]: dict(r) for r in self.conn.execute("SELECT * FROM flags")} + + def _flag_state(self, f, now): + """(effective value, time of that decision, record) for one flag row. A trial run whose time is up is a stop + decided at its end time.""" + if not f: + return None + if f["value"] == "on": + return ("on", f["set_at"], f) + if f.get("until") and f["until"] <= now: + return ("on", f["until"], {**f, "value": "on", "set_by": self.human, "set_at": f["until"], + "reason": f"the trial run ended at {f['until']}"}) + return ("off", f["set_at"], f) + + def decision(self, agent): + """The flag that decides whether agent may work: of stop:all and stop:, the most recent decision wins + (so resuming one agent while everyone is stopped works, and a later 'stop everyone' covers everyone).""" + flags, now = self.flags(), self.now() + own = self._flag_state(flags.get(f"stop:{agent}"), now) + team = self._flag_state(flags.get("stop:all"), now) + options = [(state, i) for i, state in enumerate((team, own)) if state] + return max(options, key=lambda x: (x[0][1], x[1]))[0] if options else None + + def stopped(self, agent): + """The stop that applies to agent ([] when it may work).""" + d = self.decision(agent) + return [d[2]] if d and d[0] == "on" else [] + + def trial(self, agent): + """The end of agent's trial run, if it is on one.""" + d = self.decision(agent) + return d[2]["until"] if d and d[0] == "off" and d[2].get("until") else None + + # ----------------------------------------------------------------- overviews + def attention(self): + """What needs someone's attention: overdue acknowledgements, stale review items, quiet agents.""" + ack_m = self.project.threshold("ack_overdue", 30) + item_m = self.project.threshold("item_stale", 120) + quiet_m = self.project.threshold("agent_quiet", 60) + out = [] + overdue = {} + closed = {f"item:{r['id']}" for r in self.conn.execute( + "SELECT id FROM items WHERE state IN ('landed', 'rejected', 'withdrawn')")} + for r in self.pending_acks(overdue_minutes=ack_m): + if r["agent"] != self.human and self.stopped(r["agent"]): + continue # a stopped agent cannot acknowledge; its stop shows already + topic = self.conn.execute("SELECT t.topic FROM messages m JOIN threads t ON t.id=m.thread_id WHERE m.id=?", + (r["id"],)).fetchone() + if topic and topic[0] in closed: + continue # the item it was about is landed, rejected or withdrawn + overdue.setdefault(r["id"], {"sender": r["sender"], "at": r["created_at"], "agents": []})["agents"].append(r["agent"]) + for mid, r in overdue.items(): + out.append({"kind": "ack_overdue", "message_id": mid, "minutes": self.minutes_since(r["at"]), + "text": f"#{mid} from {r['sender']} still waits for {', '.join(r['agents'])}"}) + for i in self.items(states=("submitted", "in_review", "approved")): + idle = self.minutes_since(i["updated_at"]) + if idle >= item_m: + who = i["assignee"] or "a reviewer" + out.append({"kind": "item_stale", "item_id": i["id"], "minutes": idle, + "text": f"review item #{i['id']} ({i['title']}) has been {i['state']} for {idle} min; waiting on {who}"}) + now_iso = self.now() + for r in self.conn.execute("SELECT m.id, m.sender, m.due_at, m.body, GROUP_CONCAT(d.agent) waiting FROM messages m " + "JOIN deliveries d ON d.message_id=m.id AND d.acked_at IS NULL " + "WHERE m.due_at IS NOT NULL AND m.due_at < ? GROUP BY m.id", (now_iso,)): + out.append({"kind": "due_overdue", "message_id": r["id"], "minutes": self.minutes_since(r["due_at"]), + "text": f"#{r['id']} from {r['sender']} was due at {r['due_at']}; still waiting for {r['waiting']}"}) + for t in self.tasks(states=("open", "in_progress", "blocked")): + if t["due_at"] and t["due_at"] < now_iso: + out.append({"kind": "task_overdue", "task_id": t["id"], "minutes": self.minutes_since(t["due_at"]), + "text": f"task #{t['id']} ({t['title']}) was due at {t['due_at']} and is {t['state']}"}) + elif t["state"] == "blocked": + out.append({"kind": "task_blocked", "task_id": t["id"], "minutes": self.minutes_since(t["updated_at"]), + "text": f"task #{t['id']} ({t['title']}) is blocked: {t['block_reason']}"}) + for b in self.busy_without_progress(hours=self.project.threshold("busy_hours", 2), + min_tools=self.project.threshold("busy_tools", 150)): + out.append({"kind": "busy_no_progress", "agent": b["agent"], "minutes": b["hours"] * 60, + "text": f"{b['agent']} made {b['tools']} tool calls in {b['hours']} h without a message, submit, " + "task change, handoff or new `doing` line - maybe stuck; ping it"}) + for col in self.recent_collisions(minutes=self.project.threshold("collision_window", 45)): + out.append({"kind": "collision", "minutes": self.minutes_since(col["at"]), + "text": f"{col['a1']} and {col['a2']} are both editing {col['path']}"}) + unread_after = self.project.threshold("unread_question_minutes", 20) + for r in self.conn.execute( + "SELECT d.agent, COUNT(*) n, MIN(m.created_at) first, GROUP_CONCAT(m.id) ids FROM deliveries d " + "JOIN messages m ON m.id=d.message_id WHERE d.read_at IS NULL AND d.agent != ? AND m.origin != 'system' " + "AND m.kind IN ('question', 'request') AND m.recipients != '[\"All\"]' AND m.created_at <= ? " + "GROUP BY d.agent", (self.human, iso(utc(self.clock() - unread_after * 60)))): + if self.stopped(r["agent"]): + continue # a stopped agent cannot answer; its stop is shown already + ids = r["ids"].split(",")[:5] + out.append({"kind": "unread_questions", "agent": r["agent"], "minutes": self.minutes_since(r["first"]), + "text": f"{r['agent']} has not read {r['n']} question(s) or request(s) sent to it: " + + ", ".join(f"#{i}" for i in ids) + " - the senders are waiting for an answer"}) + open_q = self.questions(open_only=True, to=self.human) + if open_q: + out.append({"kind": "questions", "minutes": self.minutes_since(open_q[-1]["created_at"]), + "text": f"{len(open_q)} question(s) wait for {self.human}'s answer: " + + ", ".join(f"#{q['message_id']} from {q['asker']}" for q in open_q[:5])}) + if self.presence()["status"] != "here": + waiting = [m for m in self.inbox(self.human, unread_only=True, limit=20, mark_read=False) + if m["recipients"] != ["All"]] + for m in waiting: + out.append({"kind": "human_away", "message_id": m["id"], "minutes": self.minutes_since(m["created_at"]), + "text": f"{self.human} is away; #{m['id']} from {m['sender']} asks them: " + f"{first_line(m['body'], 90)} - a coordinator could answer"}) + for a in self.agents(): + if a["kind"] == "human" or self.stopped(a["name"]) or not a["last_seen"]: + continue + quiet = self.minutes_since(a["last_seen"]) + if quiet >= quiet_m: + out.append({"kind": "agent_quiet", "agent": a["name"], "minutes": quiet, + "text": f"{a['name']} has not touched Harbor for {quiet} min and is not stopped"}) + for r in self.resting_agents(): + out.append({"kind": "resting", "agent": r["agent"], "minutes": self.minutes_since(r["since"]), + "text": f"{r['agent']} ended {self.AUTOPILOT_REST_AFTER} turns in a row without doing anything, " + "so autopilot let it rest. Read its last messages, then send it work or stop it"}) + links = self.connections() + for a in self.agents(): + if a["kind"] == "human" or self.stopped(a["name"]): + continue # a stopped agent runs no hooks: not a missing gate + seen = {} + for c in links.get(a["name"], []): # newest first + seen.setdefault(c["channel"], c["minutes"]) + mcp, hook = seen.get("mcp"), seen.get("hook") + if mcp is not None and mcp <= 60 and (hook is None or hook > 24 * 60): + out.append({"kind": "no_hooks", "agent": a["name"], "minutes": mcp, + "text": f"{a['name']} reaches Harbor over MCP, but its app's hooks have not reported in: " + "the stop is not enforced for it, the live feed cannot see its work and autopilot " + "cannot keep it going. If its session folder is outside the project, add a " + ".harbor.toml above it; some apps (DeepSeek Harness) do not run hooks at all"}) + return out + + def digest(self, since, until=None): + """What happened in a window: per-agent activity, review and claim changes, stops, messages to the human.""" + start = parse_since(since, utc(self.clock())) + end = until or self.now() + q = lambda sql, *a: [dict(r) for r in self.conn.execute(sql, (start, end, *a))] + per_agent = {r["sender"]: r for r in q( + "SELECT sender, COUNT(*) messages, COUNT(DISTINCT thread_id) threads FROM messages " + "WHERE created_at>=? AND created_at<=? AND origin!='legacy' GROUP BY sender")} + events = q("SELECT e.*, i.title, i.author FROM item_events e JOIN items i ON i.id=e.item_id " + "WHERE e.at>=? AND e.at<=? ORDER BY e.id") + claims = q("SELECT * FROM claims WHERE created_at>=? AND created_at<=? ORDER BY id") + flags = q("SELECT * FROM flag_events WHERE at>=? AND at<=? ORDER BY id") + to_human = [self._message_dict(r) for r in self.conn.execute( + self.MESSAGE_SELECT + " JOIN deliveries d ON d.message_id=m.id AND d.agent=? " + "WHERE m.created_at>=? AND m.created_at<=? AND m.recipients!='[\"All\"]' ORDER BY m.id", + (self.human, start, end))] + agents = [] + for a in self.agents(): + stats = per_agent.get(a["name"], {}) + agents.append({"name": a["name"], "messages": stats.get("messages", 0), "threads": stats.get("threads", 0), + "submitted": sum(1 for e in events if e["agent"] == a["name"] and e["to_state"] == "submitted"), + "reviewed": sum(1 for e in events if e["agent"] == a["name"] and e["to_state"] in + ("approved", "changes_requested", "held", "rejected")), + "landed": sum(1 for e in events if e["agent"] == a["name"] and e["to_state"] == "landed"), + "claims": sum(1 for c in claims if c["agent"] == a["name"]), + "doing": a.get("doing"), "last_seen": a["last_seen"]}) + return {"since": start, "until": end, "agents": agents, "item_events": events, "claims": claims, + "flags": flags, "to_human": to_human} + + def brief(self, agent, mark_read=False): + self._check_agent(agent) + + def age(text): + minutes = self.minutes_since(text) + return f"{minutes}m" if minutes < 120 else f"{minutes // 60}h" + out = [f"Harbor brief for {agent} in {self.project.title} at {self.now()}"] + stops = self.stopped(agent) + if stops: + for f in stops: + out.append(f"STOP ({f['key']}) since {f['set_at']} by {f['set_by']}: {f['reason'] or '-'} " + f"-> do not start or continue work; only {self.human} can resume") + else: + out.append("No stop in force.") + presence = self.presence_line() + if presence: + out.append(presence) + trial_end = self.trial(agent) if not stops else None + if trial_end: + out.append(f"Trial run: you may work until {trial_end}; then the stop comes back by itself. Write a handoff " + "before then.") + total, urgent = self.unread_count(agent) + out.append(f"Unread: {total}" + (f" ({urgent} urgent)" if urgent else "")) + for m in self.inbox(agent, unread_only=True, limit=10, mark_read=mark_read): + first = m["body"].split("\n", 1)[0][:160] + flag = " [ack needed]" if m["ack_required"] else "" + imp = f"{m['importance']} " if m["importance"] != "normal" else "" + out.append(f" #{m['id']} {imp}{m['kind']} from {m['sender']} in {m['topic']} ({age(m['created_at'])} ago)" + f"{flag}: {first}") + if total > 10: + out.append(f" ... {total - 10} more: run inbox") + needs_ack = [r for r in self.conn.execute( + "SELECT m.id, m.sender FROM deliveries d JOIN messages m ON m.id=d.message_id " + "WHERE d.agent=? AND m.ack_required=1 AND d.acked_at IS NULL ORDER BY m.id", (agent,))] + if needs_ack: + out.append("Waiting for your acknowledgement: " + ", ".join(f"#{r['id']} ({r['sender']})" for r in needs_ack)) + due = [r for r in self.conn.execute( + "SELECT m.id, m.sender, m.due_at FROM deliveries d JOIN messages m ON m.id=d.message_id WHERE d.agent=? AND " + "m.due_at IS NOT NULL AND d.acked_at IS NULL ORDER BY m.due_at", (agent,))] + if due: + out.append("Answers due from you: " + ", ".join(f"#{r['id']} ({r['sender']}) by {r['due_at']}" for r in due)) + asked = self.questions(open_only=True, to=agent) + if asked: + out.append("Questions for you (answer with `answer