forked from Keelrleek/Harbor
KeelrLeek
dbfd1720c5
Messages, threads, acknowledgements, claims, bands, tasks, review items and stop flags for a team of AI agents; the chat page and board; host hooks and MCP server; autopilot; tools/watch.py. Put under version control after the weekend's changes so they have history and can be rolled back. Local only.
403 lines
22 KiB
Python
403 lines
22 KiB
Python
"""Experimental team features on top of messages and tasks:
|
|
|
|
questions an agent asks the human (or an agent) a question with answer options; the answer comes back as a reply
|
|
notes team memory attached to a scope (fn:0x..., file:..., name:...); `context` gathers everything known
|
|
touches which files each agent is editing (from host hooks); two agents on the same file raise a collision
|
|
pulse per-agent activity in 5-minute buckets (tool calls and messages) for the board's timeline
|
|
metrics numbers agents report (readable functions, tests passing, ...) charted on the board
|
|
wins a landed review item is announced to the human
|
|
|
|
Mixed into Harbor (core.py); every method uses Harbor's connection, clock and write helper.
|
|
"""
|
|
import datetime
|
|
import json
|
|
import math
|
|
import re
|
|
from pathlib import PureWindowsPath
|
|
|
|
from .util import HarborError, first_line, iso, parse_scope, scopes_overlap, utc
|
|
|
|
EXTRA_SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS questions(
|
|
id INTEGER PRIMARY KEY, message_id INTEGER NOT NULL, asker TEXT NOT NULL, to_name TEXT NOT NULL,
|
|
question TEXT NOT NULL, options TEXT NOT NULL, fallback TEXT, answer TEXT, answered_by TEXT,
|
|
answer_message_id INTEGER, created_at TEXT NOT NULL, answered_at TEXT);
|
|
CREATE INDEX IF NOT EXISTS questions_message ON questions(message_id);
|
|
CREATE TABLE IF NOT EXISTS notes(
|
|
id INTEGER PRIMARY KEY, scope TEXT NOT NULL, lo INTEGER, hi INTEGER, body TEXT NOT NULL, author TEXT NOT NULL,
|
|
created_at TEXT NOT NULL, replaced_by INTEGER);
|
|
CREATE TABLE IF NOT EXISTS touches(agent TEXT NOT NULL, path TEXT NOT NULL, tool TEXT, at TEXT NOT NULL,
|
|
PRIMARY KEY(agent, path));
|
|
CREATE TABLE IF NOT EXISTS pulse(agent TEXT NOT NULL, bucket TEXT NOT NULL, tools INTEGER NOT NULL DEFAULT 0,
|
|
messages INTEGER NOT NULL DEFAULT 0, PRIMARY KEY(agent, bucket));
|
|
CREATE TABLE IF NOT EXISTS metrics(id INTEGER PRIMARY KEY, name TEXT NOT NULL, value REAL NOT NULL, note TEXT,
|
|
by_agent TEXT NOT NULL, at TEXT NOT NULL);
|
|
CREATE INDEX IF NOT EXISTS metrics_name ON metrics(name, id);
|
|
"""
|
|
EDIT_TOOL = re.compile(r"edit|write|patch|apply|create|replace|insert|str_replace|multiedit|notebook", re.I)
|
|
PATCH_FILE = re.compile(r"^\*\*\* (?:Update|Add|Delete) File: (.+)$", re.M)
|
|
|
|
|
|
def bucket_of(dt):
|
|
return iso(dt.replace(minute=dt.minute - dt.minute % 5, second=0))
|
|
|
|
|
|
class ExtrasMixin:
|
|
# ----------------------------------------------------------------- questions with answer options
|
|
def ask(self, asker, question, options=(), to=None, due=None, fallback=None, origin="cli", thread=None):
|
|
"""Ask `to` (default: the human) a question; options are the answer buttons shown on the chat page."""
|
|
self._check_agent(asker, allow_human=(origin == "web"))
|
|
to = to or self.human
|
|
self._check_agent(to)
|
|
options = [" ".join(str(o).split())[:60] for o in options if str(o).strip()][:6]
|
|
if fallback:
|
|
self._check_agent(fallback, allow_human=False)
|
|
if fallback not in self.coordinators:
|
|
raise HarborError("the fallback answerer must be a coordinator")
|
|
if fallback == asker:
|
|
raise HarborError("the fallback answerer cannot be the one asking")
|
|
text = question.strip() + ("\n\nOptions: " + " / ".join(options) if options else "")
|
|
msg = self.send(asker, [to], text, thread=thread or "questions", subject="Questions", kind="question",
|
|
importance="high", due=due, origin=origin)
|
|
now = self.now()
|
|
qid = self._write(lambda c: c.execute(
|
|
"INSERT INTO questions(message_id, asker, to_name, question, options, fallback, created_at) "
|
|
"VALUES (?,?,?,?,?,?,?)", (msg["id"], asker, to, question.strip(), json.dumps(options), fallback, now)).lastrowid)
|
|
self._write(lambda c: c.execute("UPDATE messages SET refs=? WHERE id=?",
|
|
(json.dumps({**(self.message(msg["id"])["refs"] or {}), "question": qid}), msg["id"])))
|
|
return {**self.question(qid), "warnings": msg["warnings"]}
|
|
|
|
def question(self, qid):
|
|
row = self.conn.execute("SELECT * FROM questions WHERE id=?", (int(qid),)).fetchone()
|
|
if not row:
|
|
raise HarborError(f"no question #{qid}")
|
|
q = dict(row)
|
|
q["options"] = json.loads(q["options"])
|
|
return q
|
|
|
|
def questions(self, open_only=True, to=None, asker=None, limit=50):
|
|
query, args = "SELECT * FROM questions WHERE 1=1", []
|
|
if open_only:
|
|
query += " AND answer IS NULL"
|
|
if to:
|
|
query += " AND to_name=?"
|
|
args.append(to)
|
|
if asker:
|
|
query += " AND asker=?"
|
|
args.append(asker)
|
|
rows = self.conn.execute(query + " ORDER BY id DESC LIMIT ?", (*args, int(limit)))
|
|
return [self.question(r["id"]) for r in rows]
|
|
|
|
def questions_by_message(self, message_ids):
|
|
ids = [int(i) for i in message_ids]
|
|
out = {}
|
|
for start in range(0, len(ids), 500):
|
|
chunk = ids[start:start + 500]
|
|
for r in self.conn.execute(f"SELECT id FROM questions WHERE message_id IN ({','.join('?' * len(chunk))})", chunk):
|
|
q = self.question(r["id"])
|
|
out[q["message_id"]] = q
|
|
return out
|
|
|
|
def answer(self, by, qid, answer, origin="cli"):
|
|
q = self.question(qid)
|
|
if q["answer"] is not None:
|
|
raise HarborError(f"question #{qid} was already answered by {q['answered_by']}: {q['answer']}")
|
|
if by == self.human and origin != "web":
|
|
raise HarborError(f"only the chat page answers as {self.human}")
|
|
allowed = by == q["to_name"] or (q["to_name"] == self.human and by == q["fallback"] and
|
|
self.presence()["status"] != "here")
|
|
if not allowed:
|
|
raise HarborError(f"question #{qid} is for {q['to_name']}" + (
|
|
f"; {q['fallback']} may answer while {self.human} is away" if q["fallback"] else ""))
|
|
answer = " ".join(str(answer).split())
|
|
if not answer:
|
|
raise HarborError("empty answer")
|
|
if q["options"] and answer not in q["options"] and not answer.lower().startswith("other:"):
|
|
raise HarborError(f"answer with one of: {' / '.join(q['options'])} (or 'other: ...')")
|
|
now = self.now()
|
|
|
|
def claim_answer(c):
|
|
cur = c.execute("UPDATE questions SET answer=?, answered_by=?, answered_at=? WHERE id=? AND answer IS NULL",
|
|
(answer, by, now, int(qid)))
|
|
if cur.rowcount == 0:
|
|
raise HarborError(f"question #{qid} was answered meanwhile")
|
|
c.execute("UPDATE deliveries SET acked_at=COALESCE(acked_at, ?), read_at=COALESCE(read_at, ?) "
|
|
"WHERE message_id=? AND agent=?", (now, now, q["message_id"], q["to_name"]))
|
|
self._write(claim_answer)
|
|
people = [q["asker"]] + ([q["to_name"]] if by != q["to_name"] else []) # a fallback answer reaches the human
|
|
reply = self.send(by, people, f"Answer to question #{qid}: {answer}" + (
|
|
f" (answered by {by} while {q['to_name']} was away)" if by != q["to_name"] else ""),
|
|
reply_to=q["message_id"], kind="answer", importance="high", trusted_refs={"question": int(qid)},
|
|
origin="system")
|
|
self._write(lambda c: c.execute("UPDATE questions SET answer_message_id=? WHERE id=?", (reply["id"], int(qid))))
|
|
return self.question(qid)
|
|
|
|
# ----------------------------------------------------------------- notes and scope context
|
|
def note(self, author, scope, body, replaces=None):
|
|
self._check_agent(author)
|
|
canonical, lo, hi = parse_scope(scope)
|
|
body = (body or "").strip()
|
|
if not body:
|
|
raise HarborError("empty note")
|
|
now = self.now()
|
|
|
|
if replaces:
|
|
old = self.conn.execute("SELECT author FROM notes WHERE id=?", (int(replaces),)).fetchone()
|
|
if not old:
|
|
raise HarborError(f"no note #{replaces}")
|
|
if old["author"] != author and author not in self.coordinators:
|
|
raise HarborError(f"note #{replaces} is {old['author']}'s; add your own note instead")
|
|
|
|
def do(c):
|
|
nid = c.execute("INSERT INTO notes(scope, lo, hi, body, author, created_at) VALUES (?,?,?,?,?,?)",
|
|
(canonical, lo, hi, body, author, now)).lastrowid
|
|
if replaces:
|
|
c.execute("UPDATE notes SET replaced_by=? WHERE id=?", (nid, int(replaces)))
|
|
self._touch(c, author)
|
|
return nid
|
|
return {"id": self._write(do), "scope": canonical}
|
|
|
|
def notes(self, scope=None, include_replaced=False, limit=50):
|
|
rows = [dict(r) for r in self.conn.execute(
|
|
"SELECT * FROM notes" + ("" if include_replaced else " WHERE replaced_by IS NULL") + " ORDER BY id DESC")]
|
|
if scope:
|
|
target = parse_scope(scope)
|
|
rows = [r for r in rows if scopes_overlap(target, (r["scope"], r["lo"], r["hi"]))]
|
|
return rows[:limit]
|
|
|
|
def context(self, scope, limit=12):
|
|
"""Everything Harbor knows about a scope: notes, threads, messages that mention it, tasks, review items,
|
|
and who has claimed it before."""
|
|
canonical, lo, hi = parse_scope(scope)
|
|
target = (canonical, lo, hi)
|
|
needles = []
|
|
if lo is not None and lo == hi:
|
|
needles = [f"{lo:X}", f"0x{lo:X}"]
|
|
elif lo is None:
|
|
needles = [canonical.split(":", 1)[1]]
|
|
seen, messages = set(), []
|
|
for needle in needles:
|
|
for m in self.search(needle, limit=limit):
|
|
if m["id"] not in seen:
|
|
seen.add(m["id"])
|
|
messages.append(m)
|
|
messages.sort(key=lambda m: m["id"], reverse=True)
|
|
threads = [t for t in self.threads(limit=500, include_legacy=False) if t["topic"] == canonical]
|
|
tasks = [t for t in self.tasks(states=("open", "in_progress", "blocked", "review", "done", "cancelled"))
|
|
if t["scope"] and scopes_overlap(target, (t["scope"], t["lo"], t["hi"]))]
|
|
items = []
|
|
for row in self.conn.execute("SELECT * FROM items WHERE scope IS NOT NULL ORDER BY id DESC"):
|
|
try:
|
|
if scopes_overlap(target, parse_scope(row["scope"])):
|
|
items.append(dict(row))
|
|
except HarborError:
|
|
pass
|
|
claims = [dict(r) for r in self.conn.execute("SELECT * FROM claims ORDER BY id DESC LIMIT 2000")
|
|
if scopes_overlap(target, (r["scope"], r["lo"], r["hi"]))]
|
|
return {"scope": canonical, "notes": self.notes(canonical), "threads": threads, "messages": messages[:limit],
|
|
"tasks": tasks, "items": items, "claims": claims[:limit]}
|
|
|
|
def context_text(self, scope, limit=8):
|
|
ctx = self.context(scope, limit=limit)
|
|
lines = [f"What the team knows about {ctx['scope']}:"]
|
|
for n in ctx["notes"]:
|
|
lines.append(f" note #{n['id']} by {n['author']} ({n['created_at'][:10]}): {first_line(n['body'], 200)}")
|
|
for i in ctx["items"][:5]:
|
|
lines.append(f" review item #{i['id']} {i['state']} by {i['author']}: {i['title']}")
|
|
for t in ctx["tasks"][:5]:
|
|
lines.append(f" task #{t['id']} {t['state']} ({t['assignee'] or 'unassigned'}): {t['title']}")
|
|
if ctx["claims"]:
|
|
who = sorted({c["agent"] for c in ctx["claims"]})
|
|
lines.append(f" claimed before by: {', '.join(who)}")
|
|
for t in ctx["threads"][:3]:
|
|
lines.append(f" thread {t['topic']}: {t['messages']} messages (harbor thread {t['topic']})")
|
|
for m in ctx["messages"][:limit]:
|
|
lines.append(f" #{m['id']} {m['created_at'][:10]} {m['sender']}: {first_line(m['body'], 160)}")
|
|
for t in ctx["threads"][:3]:
|
|
s = self.summary(t["id"])
|
|
if s:
|
|
lines.append(f" summary of {t['topic']} by {s['author']}: {first_line(s['body'], 200)}")
|
|
experts = self.experts(ctx["scope"])
|
|
if experts:
|
|
lines.append(" who knows this: " + "; ".join(
|
|
f"{e['agent']} ({', '.join(f'{v} {k}' for k, v in e['detail'].items())})" for e in experts))
|
|
if len(lines) == 1:
|
|
lines.append(" nothing yet - add what you learn with `note`.")
|
|
return "\n".join(lines)
|
|
|
|
# ----------------------------------------------------------------- file touches and collisions
|
|
def _relative(self, agent, path, cwd=None):
|
|
"""A path as the team sees it: relative to the agent's workdir (worktrees of one repository compare equal)."""
|
|
raw = str(path).strip().strip('"').strip("'").replace("\\", "/")
|
|
p = PureWindowsPath(raw)
|
|
if not p.is_absolute() and cwd:
|
|
p = PureWindowsPath(cwd) / p
|
|
raw = str(p).replace("\\", "/")
|
|
roots = [a.get("workdir") for a in self.project.agents if a.get("workdir")]
|
|
spec = next((a for a in self.project.agents if a["name"] == agent), {})
|
|
ordered = ([spec.get("workdir")] if spec.get("workdir") else []) + roots + ([cwd] if cwd else [])
|
|
if p.is_absolute():
|
|
for root in ordered:
|
|
try:
|
|
return str(p.relative_to(PureWindowsPath(root))).replace("\\", "/").lower()
|
|
except ValueError:
|
|
continue
|
|
return raw.lower()
|
|
while raw.startswith("./"):
|
|
raw = raw[2:]
|
|
return raw.lower()
|
|
|
|
@staticmethod
|
|
def edited_paths(tool, tool_input):
|
|
"""Paths an edit-like tool call writes, from the common shapes of hook payloads."""
|
|
def strings(value):
|
|
if isinstance(value, str):
|
|
yield value
|
|
elif isinstance(value, dict):
|
|
for v in value.values():
|
|
yield from strings(v)
|
|
elif isinstance(value, list):
|
|
for v in value:
|
|
yield from strings(v)
|
|
if not EDIT_TOOL.search(str(tool or "")) or not isinstance(tool_input, dict):
|
|
return [m.strip() for s in strings(tool_input) for m in PATCH_FILE.findall(s)]
|
|
paths = []
|
|
for key in ("file_path", "filePath", "path", "notebook_path", "target_file"):
|
|
if isinstance(tool_input.get(key), str):
|
|
paths.append(tool_input[key])
|
|
for key in ("paths", "files"):
|
|
if isinstance(tool_input.get(key), list):
|
|
paths += [x for x in tool_input[key] if isinstance(x, str)]
|
|
for key in ("patch", "input", "command"):
|
|
value = tool_input.get(key)
|
|
if isinstance(value, list):
|
|
value = " ".join(map(str, value))
|
|
if isinstance(value, str):
|
|
paths += [m.strip() for m in PATCH_FILE.findall(value)]
|
|
return paths
|
|
|
|
def record_touches(self, agent, paths, tool="", cwd=None, window=None):
|
|
"""Remember that agent is editing paths; returns collisions [{path, other, minutes}] (and announces new ones)."""
|
|
if not paths:
|
|
return []
|
|
window = window or self.project.threshold("collision_window", 45)
|
|
now_dt = utc(self.clock())
|
|
now = iso(now_dt)
|
|
since = iso(now_dt - datetime.timedelta(minutes=window))
|
|
rel = sorted({self._relative(agent, p, cwd) for p in paths})
|
|
collisions = []
|
|
shared = self.is_shared_file
|
|
|
|
def do(c):
|
|
for path in rel:
|
|
c.execute("INSERT INTO touches(agent, path, tool, at) VALUES (?,?,?,?) ON CONFLICT(agent, path) DO UPDATE "
|
|
"SET tool=excluded.tool, at=excluded.at", (agent, path, str(tool)[:60], now))
|
|
if shared(path):
|
|
continue # append-only file the integrator merges: not a collision
|
|
for r in c.execute("SELECT agent, at FROM touches WHERE path=? AND agent!=? AND at>=?", (path, agent, since)):
|
|
collisions.append({"path": path, "other": r["agent"], "minutes": self.minutes_since(r["at"])})
|
|
self._write(do)
|
|
fresh = []
|
|
for col in collisions:
|
|
pair = "|".join(sorted([agent, col["other"]]))
|
|
key = f"collision:{pair}:{col['path']}"
|
|
row = self.conn.execute("SELECT value FROM meta WHERE key=?", (key,)).fetchone()
|
|
if row and self.minutes_since(row["value"]) < 120:
|
|
continue
|
|
self._write(lambda c, k=key: c.execute("INSERT INTO meta(key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE "
|
|
"SET value=excluded.value", (k, now)))
|
|
fresh.append(col)
|
|
people = sorted({col["other"], *self.project.coordinators} - {agent})
|
|
self.send(agent, people, f"Heads-up from Harbor: {agent} is editing {col['path']}, which {col['other']} "
|
|
f"edited {col['minutes']} min ago. Agree in this thread who changes what before either of you "
|
|
"commits, so the merge does not conflict.", thread=f"file:{col['path']}",
|
|
subject=f"Both editing {col['path']}", kind="finding", importance="high",
|
|
refs={"file": col["path"]}, origin="hook")
|
|
return collisions
|
|
|
|
def is_shared_file(self, path):
|
|
"""True for files the project lists in shared_files (e.g. annotations/*.csv): every agent appends to them on its
|
|
own branch and the integrator merges them, so two agents editing one is normal, not a collision."""
|
|
import fnmatch
|
|
p = str(path).replace("\\", "/")
|
|
return any(fnmatch.fnmatch(p, pattern) for pattern in self.project.shared_files)
|
|
|
|
def recent_collisions(self, minutes=120):
|
|
since = iso(utc(self.clock() - minutes * 60))
|
|
rows = self.conn.execute("SELECT a.path, a.agent a1, b.agent a2, MAX(a.at, b.at) at FROM touches a JOIN touches b "
|
|
"ON a.path=b.path AND a.agent<b.agent WHERE a.at>=? AND b.at>=?", (since, since))
|
|
return [dict(r) for r in rows if not self.is_shared_file(r["path"])]
|
|
|
|
# ----------------------------------------------------------------- pulse (activity timeline)
|
|
def _pulse(self, c, agent, tools=0, messages=0):
|
|
c.execute("INSERT INTO pulse(agent, bucket, tools, messages) VALUES (?,?,?,?) ON CONFLICT(agent, bucket) DO UPDATE "
|
|
"SET tools=tools+excluded.tools, messages=messages+excluded.messages",
|
|
(agent, bucket_of(utc(self.clock())), tools, messages))
|
|
|
|
def pulse(self, hours=12):
|
|
since = bucket_of(utc(self.clock() - hours * 3600))
|
|
out = {}
|
|
for r in self.conn.execute("SELECT * FROM pulse WHERE bucket>=? ORDER BY bucket", (since,)):
|
|
out.setdefault(r["agent"], []).append({"bucket": r["bucket"], "tools": r["tools"], "messages": r["messages"]})
|
|
return {"since": since, "now": self.now(), "agents": out}
|
|
|
|
def idle_minutes(self, agent):
|
|
row = self.conn.execute("SELECT MAX(bucket) b FROM pulse WHERE agent=?", (agent,)).fetchone()
|
|
return self.minutes_since(row["b"]) if row and row["b"] else None
|
|
|
|
# ----------------------------------------------------------------- metrics and wins
|
|
def metric(self, by, name, value, note=None):
|
|
self._check_agent(by)
|
|
name = re.sub(r"[^a-z0-9_.-]+", "-", str(name).strip().lower())[:40]
|
|
if not name:
|
|
raise HarborError("a metric needs a name")
|
|
try:
|
|
value = float(value)
|
|
except (TypeError, ValueError):
|
|
raise HarborError(f"metric value {value!r} is not a number")
|
|
if not math.isfinite(value):
|
|
raise HarborError("a metric value must be a finite number")
|
|
now = self.now()
|
|
mid = self._write(lambda c: c.execute("INSERT INTO metrics(name, value, note, by_agent, at) VALUES (?,?,?,?,?)",
|
|
(name, value, note, by, now)).lastrowid)
|
|
return {"id": mid, "name": name, "value": value}
|
|
|
|
def metrics(self, points=60):
|
|
out = {}
|
|
for r in self.conn.execute("SELECT DISTINCT name FROM metrics ORDER BY name"):
|
|
rows = [dict(x) for x in self.conn.execute("SELECT * FROM metrics WHERE name=? ORDER BY id DESC LIMIT ?",
|
|
(r["name"], int(points)))]
|
|
rows.reverse()
|
|
out[r["name"]] = {"latest": rows[-1], "series": [{"at": x["at"], "value": x["value"]} for x in rows]}
|
|
return out
|
|
|
|
def announce_win(self, item, agent, commit):
|
|
"""Tell the human, in the 'wins' thread, that an item landed."""
|
|
if agent == self.human:
|
|
return
|
|
text = f"Landed: {item['title']} by {item['author']}" + (f" ({item['scope']})" if item.get("scope") else "") + \
|
|
(f" in {commit}" if commit else "") + "."
|
|
self.send(agent, [self.human], text, thread="wins", subject="Wins", kind="status", refs={"item": item["id"]},
|
|
origin="system")
|
|
|
|
def export_thread(self, thread):
|
|
"""A thread as Markdown, for records and reports."""
|
|
info, messages = self.thread(thread)
|
|
lines = [f"# {info['subject'] or info['topic']}", "", f"Thread `{info['topic']}` in {self.project.title}, "
|
|
f"{len(messages)} messages, exported {self.now()}.", ""]
|
|
for m in messages:
|
|
to = ", ".join(m["recipients"])
|
|
tags = [t for t in (m["kind"] if m["kind"] != "note" else "", m["importance"] if m["importance"] != "normal"
|
|
else "", f"corrects #{m['corrects']}" if m.get("corrects") else "") if t]
|
|
lines.append(f"## #{m['id']} {m['sender']} → {to} · {m['created_at'].replace('T', ' ')[:16]} UTC"
|
|
+ (f" ({', '.join(tags)})" if tags else ""))
|
|
lines += ["", m["body"], ""]
|
|
return "\n".join(lines)
|
|
|
|
def wins(self, since="24h"):
|
|
from .util import parse_since
|
|
start = parse_since(since, utc(self.clock()))
|
|
return [dict(r) for r in self.conn.execute(
|
|
"SELECT e.*, i.title, i.author, i.scope FROM item_events e JOIN items i ON i.id=e.item_id "
|
|
"WHERE e.to_state='landed' AND e.at>=? ORDER BY e.id DESC", (start,))]
|