1390 lines
82 KiB
Python
1390 lines
82 KiB
Python
"""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<? ORDER BY m.id DESC LIMIT ?",
|
|
(int(before_id), int(limit))).fetchall()
|
|
return [self._message_dict(r) for r in reversed(rows)]
|
|
|
|
def corrections_for(self, ids):
|
|
"""{original id: [ids of messages that correct it]} for the given ids."""
|
|
ids = [int(i) for i in ids]
|
|
out = {}
|
|
for start in range(0, len(ids), 500):
|
|
chunk = ids[start:start + 500]
|
|
for r in self.conn.execute(f"SELECT id, corrects FROM messages WHERE corrects IN ({','.join('?' * len(chunk))})",
|
|
chunk):
|
|
out.setdefault(r["corrects"], []).append(r["id"])
|
|
return out
|
|
|
|
def inbox(self, agent, unread_only=True, limit=30, mark_read=True, after_id=None):
|
|
self._check_agent(agent)
|
|
query = ("SELECT m.*, t.topic, t.subject, d.read_at, d.acked_at FROM deliveries d "
|
|
"JOIN messages m ON m.id=d.message_id JOIN threads t ON t.id=m.thread_id WHERE d.agent=?")
|
|
args = [agent]
|
|
if unread_only:
|
|
query += " AND d.read_at IS NULL"
|
|
if after_id is not None:
|
|
query += " AND 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:<ISO>."""
|
|
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.append(parse_since(filters["before"], utc(self.clock())))
|
|
if "pinned" in filters:
|
|
where.append("EXISTS (SELECT 1 FROM pins p WHERE p.message_id=m.id)")
|
|
if not terms and not where:
|
|
return []
|
|
clause = (" AND " + " AND ".join(where)) if where else ""
|
|
if terms:
|
|
rows = self.conn.execute(
|
|
"SELECT m.*, t.topic, t.subject FROM messages_fts f JOIN messages m ON m.id=f.rowid "
|
|
"JOIN threads t ON t.id=m.thread_id WHERE messages_fts MATCH ?" + clause +
|
|
" ORDER BY bm25(messages_fts) LIMIT ?", (terms, *args, int(limit))).fetchall()
|
|
else:
|
|
rows = self.conn.execute(self.MESSAGE_SELECT + " WHERE 1=1" + clause + " ORDER BY m.id DESC LIMIT ?",
|
|
(*args, int(limit))).fetchall()
|
|
return [self._message_dict(r) for r in rows]
|
|
|
|
# ----------------------------------------------------------------- claims
|
|
def active_claims(self, agent=None, kind=None):
|
|
now = self.now()
|
|
query = "SELECT * FROM claims WHERE released_at IS NULL AND (expires_at IS NULL OR expires_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)
|
|
self.check_submit_policy(author)
|
|
|
|
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)
|
|
if is_author and to_state == "submitted":
|
|
self.check_submit_policy(agent)
|
|
|
|
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:<Agent>', 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:<Agent>")
|
|
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=<id>")
|
|
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:<agent>, 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')")}
|
|
retired = {r[0] for r in self.conn.execute("SELECT name FROM agents WHERE active=0")}
|
|
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
|
|
if r["agent"] in retired or r["sender"] in retired:
|
|
continue # a retired agent will never acknowledge, nor follow up
|
|
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"]) or r["agent"] in retired:
|
|
continue # a stopped or retired agent cannot answer
|
|
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 <question id> <option>`): " + "; ".join(
|
|
f"#{q['id']} from {q['asker']}: {first_line(q['question'], 90)}"
|
|
+ (f" [{' / '.join(q['options'])}]" if q["options"] else "") for q in asked[:5]))
|
|
mine_open = self.questions(open_only=True, asker=agent)
|
|
if mine_open:
|
|
out.append("Your open questions: " + ", ".join(f"#{q['id']} to {q['to_name']}" for q in mine_open[:8]))
|
|
clashes = [c for c in self.recent_collisions(minutes=self.project.threshold("collision_window", 45))
|
|
if agent in (c["a1"], c["a2"])]
|
|
if clashes:
|
|
out.append("Files you share with someone right now: " + "; ".join(
|
|
f"{c['path']} with {c['a2'] if c['a1'] == agent else c['a1']}" for c in clashes[:6]))
|
|
handoff = self.last_handoff(agent)
|
|
if handoff and not stops:
|
|
body = handoff["body"].split("\n")
|
|
out.append(f"Your last handoff ({age(handoff['created_at'])} ago) - continue from it:")
|
|
out += [" " + line for line in body[:12]] + ([" ..."] if len(body) > 12 else [])
|
|
my_tasks = self.tasks(assignee=agent)
|
|
if my_tasks:
|
|
out.append("Your tasks:")
|
|
for task in my_tasks:
|
|
extra = f" - blocked: {task['block_reason']}" if task["state"] == "blocked" else ""
|
|
out.append(f" #{task['id']} {task['state']} {task['priority_name']} {task['scope'] or ''} "
|
|
f"{task['title']}{extra}".replace(" ", " "))
|
|
if not stops and not any(task["state"] == "in_progress" for task in my_tasks):
|
|
nxt = self.next_task(agent, take=False)
|
|
if nxt:
|
|
out.append(f"Next task you can take: #{nxt['id']} {nxt['priority_name']} {nxt['title']} (run `next`)")
|
|
mine = self.items(author=agent)
|
|
if mine:
|
|
out.append("Your review items:")
|
|
for i in mine:
|
|
out.append(f" #{i['id']} {i['state']} {i['scope'] or ''} {i['title']} (updated {age(i['updated_at'])} ago)")
|
|
reviewing = [i for i in self.items(states=("submitted", "in_review", "approved"))
|
|
if i["author"] != agent and (agent in self.coordinators or agent in self.project.reviewers
|
|
or i["assignee"] == agent)]
|
|
if reviewing:
|
|
oldest = max(self.minutes_since(i["created_at"]) for i in reviewing)
|
|
out.append(f"Review queue for you: {len(reviewing)} open, oldest {oldest} min")
|
|
claims = self.active_claims(agent=agent)
|
|
if claims:
|
|
out.append("Your claims:")
|
|
for c in claims:
|
|
out.append(f" #{c['id']} {c['kind']} {c['scope']}" + (f" until {c['expires_at']}" if c['expires_at'] else "")
|
|
+ (f" - {c['note']}" if c['note'] else ""))
|
|
waiting = self.pending_acks(sender=agent)
|
|
if waiting:
|
|
out.append("Your messages still unacknowledged: " + ", ".join(f"#{r['id']}->{r['agent']}" for r in waiting[:15]))
|
|
pinned = self.pins(limit=5)
|
|
if pinned:
|
|
out.append("Pinned: " + " | ".join(f"#{m['id']} {first_line(m.get('pin_note') or m['body'], 80)}"
|
|
for m in pinned))
|
|
team = [f"{a['name']}: {a['doing']}" for a in self.agents() if a["doing"] and a["name"] != agent]
|
|
if team:
|
|
out.append("Team now: " + "; ".join(team))
|
|
long_threads = self.threads_needing_summary(agent)
|
|
if long_threads:
|
|
out.append("Long threads without a recent summary (write one with `summarize <thread> \"...\"`): "
|
|
+ ", ".join(f"{t['thread']} ({t['messages']} messages)" for t in long_threads[:4]))
|
|
if agent in self.coordinators:
|
|
for a in self.attention():
|
|
out.append(f"Attention: {a['text']}")
|
|
self.touch(agent)
|
|
if agent != self.human:
|
|
self.note_brief(agent)
|
|
return "\n".join(out)
|
|
|
|
def nudge_candidates(self, agent, urgent_only=False):
|
|
"""Unread messages that justify interrupting agent (urgent or acknowledgement requested, or any unread when
|
|
urgent_only is False) and that were not already used for a nudge. Recording the nudge keeps hosts whose
|
|
stop hooks have no loop guard from nudging forever about the same messages."""
|
|
row = self.conn.execute("SELECT value FROM meta WHERE key=?", (f"nudged:{agent}",)).fetchone()
|
|
last = int(row["value"]) if row else 0
|
|
rows = [m for m in self.inbox(agent, unread_only=True, limit=50, mark_read=False) if m["id"] > last]
|
|
if urgent_only:
|
|
rows = [m for m in rows if m["importance"] == "urgent" or m["ack_required"]]
|
|
return rows
|
|
|
|
def record_nudge(self, agent, message_ids):
|
|
if not message_ids:
|
|
return
|
|
top = max(int(i) for i in message_ids)
|
|
self._write(lambda c: c.execute(
|
|
"INSERT INTO meta(key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value=MAX(CAST(value AS INTEGER), "
|
|
"CAST(excluded.value AS INTEGER))", (f"nudged:{agent}", str(top))))
|
|
|
|
# ----------------------------------------------------------------- maintenance
|
|
def backup(self, keep=14):
|
|
"""Copy the database (consistent snapshot) into backups/; keeps the newest `keep` copies."""
|
|
folder = self.project.backups
|
|
folder.mkdir(parents=True, exist_ok=True)
|
|
target = folder / f"harbor-{utc(self.clock()).strftime('%Y%m%d-%H%M%S')}.db"
|
|
with sqlite3.connect(str(target)) as copy:
|
|
self.conn.backup(copy)
|
|
copy.close()
|
|
snapshots = sorted(folder.glob("harbor-*.db"))
|
|
for old in snapshots[:-keep] if keep else []:
|
|
old.unlink()
|
|
return target
|
|
|
|
LEGACY_LINE = re.compile(r"^\[([^\]]+)\]\s+([A-Za-z]+)\s*(?:→|->)\s*([A-Za-z]+):\s?(.*)$")
|
|
|
|
def import_legacy(self, path):
|
|
"""Import a plain `[time] From → To: text` chat history once (idempotent): one archive thread per day,
|
|
everything marked read."""
|
|
text = Path(path).read_text(encoding="utf-8", errors="replace")
|
|
body = text.split("\n---\n", 1)[1] if "\n---\n" in text else text
|
|
messages = []
|
|
for raw in body.split("\n"):
|
|
line = raw.strip()
|
|
m = self.LEGACY_LINE.match(line)
|
|
if m:
|
|
messages.append([m.group(1), m.group(2), m.group(3), m.group(4)])
|
|
elif line and messages:
|
|
messages[-1][3] += "\n" + (line[2:] if line.startswith("| ") else line)
|
|
names = set(self.agent_names())
|
|
seen = {}
|
|
|
|
def do(c):
|
|
added = 0
|
|
for stamp, sender, to, text in messages:
|
|
if sender not in names or (to not in names and to != "All"):
|
|
continue
|
|
if re.match(r"\[#\d+ · thread ", text): # Harbor's own mirror lines are already in the database
|
|
continue
|
|
key = hashlib.sha1(f"{stamp}|{sender}|{to}|{text}".encode("utf-8")).hexdigest()
|
|
seen[key] = seen.get(key, 0) + 1
|
|
idem = f"legacy:{key}:{seen[key]}"
|
|
if c.execute("SELECT 1 FROM messages WHERE idem=?", (idem,)).fetchone():
|
|
continue
|
|
m = re.match(r"(\d{4}-\d\d-\d\d) (\d\d:\d\d)Z", stamp)
|
|
created = f"{m.group(1)}T{m.group(2)}:00Z" if m else self.now()
|
|
day = created[:10]
|
|
topic = f"legacy:{day}"
|
|
row = c.execute("SELECT id FROM threads WHERE topic=?", (topic,)).fetchone()
|
|
if row:
|
|
thread_id = row["id"]
|
|
else:
|
|
thread_id = c.execute("INSERT INTO threads(topic, subject, state, created_by, created_at, updated_at) "
|
|
"VALUES (?, ?, 'archived', 'system', ?, ?)",
|
|
(topic, f"Chat archive {day}", created, created)).lastrowid
|
|
recipients = ["All"] if to == "All" else [to]
|
|
mid = c.execute("INSERT INTO messages(thread_id, sender, recipients, kind, importance, ack_required, "
|
|
"body, origin, idem, created_at) VALUES (?,?,?,'note','normal',0,?,'legacy',?,?)",
|
|
(thread_id, sender, json.dumps(recipients), text, idem, created)).lastrowid
|
|
targets = [n for n in names if n != sender] if to == "All" else ([to] if to != sender else [])
|
|
for name in targets:
|
|
c.execute("INSERT INTO deliveries(message_id, agent, read_at, acked_at) VALUES (?,?,?,?)",
|
|
(mid, name, created, created))
|
|
added += 1
|
|
return added
|
|
return {"parsed": len(messages), "added": self._write(do)}
|