"""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 ?" args.append(int(after_id)) # urgent, then high, then the rest; within each, people before system notices, newest first, so a backlog # of old automatic notices never hides what arrived since query += (" ORDER BY CASE m.importance WHEN 'urgent' THEN 0 WHEN 'high' THEN 1 ELSE 2 END," " (m.origin = 'system'), m.id DESC" if unread_only else " ORDER BY m.id DESC") rows = [self._message_dict(r) for r in self.conn.execute(query + " LIMIT ?", (*args, int(limit)))] if mark_read and rows: self.mark_read(agent, [r["id"] for r in rows]) return rows def mark_read(self, agent, ids=None): now = self.now() def mark(c): if ids is None: c.execute("UPDATE deliveries SET read_at=? WHERE agent=? AND read_at IS NULL", (now, agent)) for mid in ids or (): c.execute("UPDATE deliveries SET read_at=COALESCE(read_at, ?) WHERE message_id=? AND agent=?", (now, int(mid), agent)) self._touch(c, agent) self._write(mark) def last_delivered_id(self, agent): row = self.conn.execute("SELECT MAX(message_id) m FROM deliveries WHERE agent=?", (agent,)).fetchone() return row["m"] or 0 def unread_count_after(self, agent, after_id): """Unread messages worth waking for: anything that arrived after after_id, and any unread message from a person or agent however old (only an old backlog of automatic system notices is ignored).""" row = self.conn.execute("SELECT COUNT(*) n FROM deliveries d JOIN messages m ON m.id=d.message_id " "WHERE d.agent=? AND d.read_at IS NULL AND (d.message_id > ? OR m.origin <> 'system')", (agent, int(after_id))).fetchone() return row["n"] def unread_count(self, agent): row = self.conn.execute("SELECT COUNT(*) n, SUM(m.importance='urgent') u FROM deliveries d JOIN messages m " "ON m.id=d.message_id WHERE d.agent=? AND d.read_at IS NULL", (agent,)).fetchone() return row["n"], row["u"] or 0 def wait(self, agent, timeout=300, poll=1.0, mark_read=True): """Block until a message for agent arrives, a stop applies to it, or timeout seconds pass.""" self._check_agent(agent, allow_human=False) timeout = max(0.0, min(float(timeout), 3600.0)) deadline = time.monotonic() + timeout self.touch(agent) since = self.last_delivered_id(agent) # wake on what arrives from now on, not on an old backlog while True: stops = self.stopped(agent) n = self.unread_count_after(agent, since) if n or stops or time.monotonic() >= 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"): if by != self.human: return False # the human resuming one agent under a later 'stop everyone': record the time, so this resume is # the most recent decision for that agent (decision() compares set_at) c.execute("UPDATE flags SET set_by=?, origin=?, relay_of=NULL, set_at=? WHERE key=?", (by, origin, now, key)) return True 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