From 940cacc2d296ab820876127000db4acbd3a86cfa Mon Sep 17 00:00:00 2001 From: KeelrLeek <156356974+KeelrLeek@users.noreply.github.com> Date: Wed, 7 Oct 2026 08:38:27 -0500 Subject: [PATCH] attention: acknowledgements and unread questions involving a retired agent (active=0) are no longer listed; they can never be answered --- harbor/core.py | 2769 ++++++++++++++++++++++++------------------------ 1 file changed, 1386 insertions(+), 1383 deletions(-) diff --git a/harbor/core.py b/harbor/core.py index ddc3673..106975f 100644 --- a/harbor/core.py +++ b/harbor/core.py @@ -1,1383 +1,1386 @@ -"""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 - commit = self.resolve_commit(author, commit) - - 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) - if is_author and commit: - commit = self.resolve_commit(agent, commit) - - 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