279 lines
16 KiB
Python
279 lines
16 KiB
Python
"""How work flows: git radar, bulk tasks and templates, the stuck-agent watchdog, and team policies.
|
|
|
|
git radar read-only git facts per agent workdir: branch, last commit, uncommitted files, commits behind the main
|
|
branch, and commits made since the agent last submitted work (work not yet in the review queue)
|
|
templates multi-step task pipelines defined in the project file, created in one call with dependencies
|
|
bulk many tasks from a pasted list ("0x445310 RouteCellCost"), routed to the band owner
|
|
watchdog an agent busy with tools for hours without any visible progress
|
|
policies rules from the project file that Harbor enforces (approvals before landing, message caps)
|
|
|
|
Mixed into Harbor (core.py).
|
|
"""
|
|
import re
|
|
import subprocess
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from .util import HarborError, first_line, iso, parse_iso, parse_scope, scopes_overlap, utc
|
|
|
|
GIT_TIMEOUT = 10
|
|
_git_cache = {}
|
|
|
|
|
|
class _Values(dict):
|
|
"""Template placeholders; an unknown {name} stays as written instead of failing."""
|
|
def __missing__(self, key):
|
|
return "{" + key + "}"
|
|
|
|
|
|
def _git(workdir, *args):
|
|
"""A read-only git command (never takes locks, never writes the index)."""
|
|
# The worktree's own config is agent-controlled: never let it start programs (fsmonitor, pagers) from here.
|
|
r = subprocess.run(["git", "--no-pager", "--no-optional-locks", "-c", "core.fsmonitor=false",
|
|
"-c", "core.untrackedCache=false", "-C", str(workdir), *args], capture_output=True, text=True,
|
|
encoding="utf-8", errors="replace", timeout=GIT_TIMEOUT,
|
|
env={**__import__("os").environ, "GIT_TERMINAL_PROMPT": "0", "GIT_OPTIONAL_LOCKS": "0"})
|
|
if r.returncode != 0:
|
|
raise RuntimeError(r.stderr.strip() or f"git {' '.join(args)} failed")
|
|
return r.stdout
|
|
|
|
|
|
class FlowMixin:
|
|
def resolve_commit(self, agent, commit):
|
|
"""The full hash of a commit an item names, checked in agent's workdir (or the Logbook's decomp checkout).
|
|
A placeholder ('HEAD', 'PLACEHOLDER') or a hash git does not know is refused; without any git checkout to
|
|
ask, the value is kept as given."""
|
|
if not commit:
|
|
return commit
|
|
commit = str(commit).strip()
|
|
spec = next((a for a in self.project.agents if a["name"] == agent), {})
|
|
places = [spec.get("workdir"), (self.project.logbook or {}).get("decomp")]
|
|
repos = [p for p in places if p and (Path(p) / ".git").exists()]
|
|
if not repos:
|
|
return commit
|
|
if not re.fullmatch(r"[0-9a-fA-F]{7,40}", commit):
|
|
raise HarborError(f"commit {commit!r} is not a hash: give the commit's hash (git rev-parse HEAD), "
|
|
"not a branch name, HEAD or a placeholder")
|
|
for repo in repos:
|
|
try:
|
|
return _git(repo, "rev-parse", "--verify", "--quiet", f"{commit}^{{commit}}").strip()
|
|
except (RuntimeError, subprocess.TimeoutExpired):
|
|
continue
|
|
raise HarborError(f"commit {commit} is not in the repository: commit your work first and give the hash "
|
|
"git rev-parse HEAD prints")
|
|
|
|
def worktree_change(self, agent):
|
|
"""(mtime, path) of the most recently modified uncommitted file in agent's worktree, or None."""
|
|
spec = next((a for a in self.project.agents if a["name"] == agent), {})
|
|
workdir = spec.get("workdir")
|
|
if not workdir or not Path(workdir).is_dir():
|
|
return None
|
|
newest = None
|
|
for line in _git(workdir, "status", "--porcelain", "-uall").splitlines()[:3000]:
|
|
path = line[3:].split(" -> ")[-1].strip().strip('"')
|
|
try:
|
|
mtime = (Path(workdir) / path).stat().st_mtime
|
|
except OSError:
|
|
continue
|
|
if newest is None or mtime > newest[0]:
|
|
newest = (mtime, path)
|
|
return newest
|
|
|
|
# ----------------------------------------------------------------- git radar
|
|
def git_status(self, agent, max_age=60):
|
|
"""Git facts for agent's workdir (cached for max_age seconds)."""
|
|
spec = next((a for a in self.project.agents if a["name"] == agent), None)
|
|
workdir = spec and spec.get("workdir")
|
|
if not workdir or not Path(workdir).is_dir():
|
|
return None
|
|
key = (str(workdir), self.project.name)
|
|
hit = _git_cache.get(key)
|
|
if hit and time.monotonic() - hit[0] < max_age:
|
|
return hit[1]
|
|
main = self.project.git_main
|
|
out = {"agent": agent, "workdir": workdir}
|
|
try:
|
|
out["branch"] = _git(workdir, "rev-parse", "--abbrev-ref", "HEAD").strip()
|
|
sha, stamp, subject = (_git(workdir, "log", "-1", "--format=%h%x1f%ct%x1f%s").strip().split("\x1f") + ["", "", ""])[:3]
|
|
out["last_commit"] = {"sha": sha, "at": iso(utc(int(stamp))) if stamp else None, "subject": subject}
|
|
out["dirty"] = len([l for l in _git(workdir, "status", "--porcelain").splitlines() if l.strip()])
|
|
out["behind_main"] = out["ahead_of_main"] = 0
|
|
if out["branch"] != main:
|
|
try:
|
|
behind, ahead = _git(workdir, "rev-list", "--left-right", "--count", f"{main}...HEAD").split()
|
|
out["behind_main"], out["ahead_of_main"] = int(behind), int(ahead)
|
|
except (RuntimeError, ValueError):
|
|
out["main_missing"] = True
|
|
since = self._last_submit_at(agent)
|
|
# Author dates survive a rebase (commit dates do not), so rebased work does not look new again.
|
|
cutoff = since or iso(utc(self.clock() - 86400))
|
|
args = ["log", "--format=%h%x1f%at%x1f%s", "-200"]
|
|
if out["branch"] != main and not out.get("main_missing"):
|
|
args.append(f"{main}..HEAD")
|
|
commits = []
|
|
for line in _git(workdir, *args).splitlines():
|
|
parts = line.split("\x1f")
|
|
if len(parts) == 3 and iso(utc(int(parts[1]))) > cutoff:
|
|
commits.append({"sha": parts[0], "at": iso(utc(int(parts[1]))), "subject": parts[2]})
|
|
commits = commits[:50]
|
|
known = {r[0][:7] for r in self.conn.execute("SELECT commit_hash FROM items WHERE commit_hash IS NOT NULL")
|
|
if r[0]}
|
|
submits = out["branch"] != main and str(spec.get("git_submits", True)).lower() not in ("false", "0", "no")
|
|
out["unsubmitted"] = [c for c in commits if c["sha"][:7] not in known] if submits else []
|
|
out["since"] = since
|
|
except (RuntimeError, OSError, subprocess.SubprocessError, ValueError) as error:
|
|
out["error"] = str(error)[:200]
|
|
_git_cache[key] = (time.monotonic(), out)
|
|
return out
|
|
|
|
def _last_submit_at(self, agent):
|
|
row = self.conn.execute("SELECT MAX(at) a FROM item_events WHERE agent=? AND to_state='submitted'",
|
|
(agent,)).fetchone()
|
|
return row["a"] if row and row["a"] else None
|
|
|
|
def git_radar(self):
|
|
return {a["name"]: self.git_status(a["name"]) for a in self.project.agents
|
|
if a["kind"] != "human" and a.get("workdir")}
|
|
|
|
# ----------------------------------------------------------------- bulk tasks and templates
|
|
def band_owner(self, scope):
|
|
canonical, lo, hi = parse_scope(scope)
|
|
if lo is None:
|
|
return None
|
|
now = self.now()
|
|
for c in self.conn.execute("SELECT * FROM claims WHERE kind='band' AND released_at IS NULL AND "
|
|
"(expires_at IS NULL OR expires_at > ?)", (now,)):
|
|
if scopes_overlap((canonical, lo, hi), (c["scope"], c["lo"], c["hi"])):
|
|
return c["agent"]
|
|
return None
|
|
|
|
def run_template(self, by, template, scope=None, title=None, owner=None, priority="normal", origin="cli"):
|
|
"""Create the steps of a template as linked tasks. Placeholders: {scope}, {title}, {owner}."""
|
|
spec = self.project.templates.get(template)
|
|
if not spec:
|
|
raise HarborError(f"no template {template!r}; the project file defines: "
|
|
+ (", ".join(self.project.templates) or "none"))
|
|
owner = owner or (self._route(by, scope) if scope else None)
|
|
values = _Values(scope=scope or "", title=title or scope or template, owner=owner or "")
|
|
for step in spec.get("steps", []): # check every step before creating any
|
|
assignee = (step.get("assignee") or "").format_map(values) or None
|
|
if assignee:
|
|
self._check_agent(assignee, allow_human=False)
|
|
if assignee != by and by not in self.coordinators:
|
|
raise HarborError(f"template {template!r} assigns a step to {assignee}; only a coordinator does that")
|
|
created, by_name = [], {}
|
|
for step in spec.get("steps", []):
|
|
assignee = (step.get("assignee") or "").format_map(values) or None
|
|
after = [by_name[n] for n in step.get("after", []) if n in by_name]
|
|
task = self.add_task(by, step.get("title", "{title}").format_map(values), body=step.get("body"),
|
|
scope=scope if step.get("scoped", True) else None, assignee=assignee,
|
|
priority=step.get("priority", priority), after=after, origin=origin)
|
|
by_name[step.get("name", str(len(by_name)))] = task["id"]
|
|
created.append(task)
|
|
return created
|
|
|
|
LINE = re.compile(r"^\s*(?:fn:)?(0x[0-9A-Fa-f]+)\b[\s:|,-]*(.*)$")
|
|
|
|
def bulk_tasks(self, by, text, template=None, priority="normal", assignee=None, origin="cli"):
|
|
"""One task (or template pipeline) per line: '0x445310 RouteCellCost' or 'fn:0x445310 | Make it readable'.
|
|
Lines without an address become plain tasks. Each address goes to its band owner unless assignee is set."""
|
|
if assignee and assignee != by and by not in self.coordinators:
|
|
raise HarborError("only a coordinator assigns tasks to someone else")
|
|
created, skipped = [], []
|
|
open_scopes = {t["scope"] for t in self.tasks(states=("open", "in_progress", "blocked", "review"), limit=100000)}
|
|
for raw in str(text).splitlines(): # validate everything first: no half-created batches
|
|
m = self.LINE.match(raw.strip()) if raw.strip() and not raw.strip().startswith("#") else None
|
|
if m:
|
|
parse_scope(f"fn:{m.group(1)}")
|
|
for raw in str(text).splitlines():
|
|
line = raw.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
m = self.LINE.match(line)
|
|
scope = f"fn:{m.group(1)}" if m else None
|
|
name = (m.group(2).strip() if m else line) or None
|
|
if scope:
|
|
canonical = parse_scope(scope)[0]
|
|
if canonical in open_scopes:
|
|
skipped.append(f"{canonical} already has an open task")
|
|
continue
|
|
open_scopes.add(canonical)
|
|
if template:
|
|
created += self.run_template(by, template, scope=scope, title=name, owner=assignee, priority=priority,
|
|
origin=origin)
|
|
else:
|
|
title = (f"Make {scope.split(':', 1)[1]} readable" + (f" ({name})" if name else "")) if scope else name
|
|
created.append(self.add_task(by, title, scope=scope,
|
|
assignee=assignee or (self._route(by, scope) if scope else None),
|
|
priority=priority, origin=origin, notify=template is not None))
|
|
if not template: # one notice per agent, not one thread per task
|
|
mine = {}
|
|
for t in created:
|
|
if t.get("assignee") and t["assignee"] != by:
|
|
mine.setdefault(t["assignee"], []).append(t)
|
|
for agent, items in mine.items():
|
|
lines = "\n".join(f"#{t['id']} {t['title']}" for t in items)
|
|
self.send(by, [agent], f"{len(items)} new task{'s' if len(items) > 1 else ''} for you; take them "
|
|
f"with next:\n{lines}", thread="tasks", subject="New tasks", kind="request",
|
|
origin="system")
|
|
return {"created": created, "skipped": skipped}
|
|
|
|
def _route(self, by, scope):
|
|
"""The band owner of scope - when the caller may assign to them (a coordinator, or the owner itself)."""
|
|
owner = self.band_owner(scope)
|
|
return owner if owner and (by in self.coordinators or owner == by) else None
|
|
|
|
# ----------------------------------------------------------------- stuck-agent watchdog
|
|
def busy_without_progress(self, hours=2, min_tools=150):
|
|
"""Agents with many tool calls in the window but no message, submit, review move, task change, handoff or
|
|
new `doing` line - possibly stuck in a loop."""
|
|
since = iso(utc(self.clock() - hours * 3600))
|
|
out = []
|
|
for a in self.agents():
|
|
if a["kind"] == "human" or self.stopped(a["name"]):
|
|
continue
|
|
tools = self.conn.execute("SELECT COALESCE(SUM(tools), 0) FROM pulse WHERE agent=? AND bucket>=?",
|
|
(a["name"], since)).fetchone()[0]
|
|
if tools < min_tools:
|
|
continue
|
|
progress = any(self.conn.execute(sql, (a["name"], since)).fetchone()[0] for sql in (
|
|
"SELECT COUNT(*) FROM messages WHERE sender=? AND created_at>=? AND origin NOT IN ('hook', 'system')",
|
|
"SELECT COUNT(*) FROM item_events WHERE agent=? AND at>=?",
|
|
"SELECT COUNT(*) FROM task_events WHERE agent=? AND at>=?",
|
|
"SELECT COUNT(*) FROM handoffs WHERE agent=? AND created_at>=?",
|
|
"SELECT COUNT(*) FROM agents WHERE name=? AND doing_at>=?"))
|
|
if not progress:
|
|
out.append({"agent": a["name"], "tools": tools, "hours": hours})
|
|
return out
|
|
|
|
# ----------------------------------------------------------------- policies
|
|
def policy(self, key, default=None):
|
|
return self.project.policy.get(key, default)
|
|
|
|
def check_landing_policy(self, item_id):
|
|
required = list(self.policy("land_requires") or [])
|
|
author = self.conn.execute("SELECT author FROM items WHERE id=?", (int(item_id),)).fetchone()
|
|
required = [r for r in required if not author or r != author["author"]] # authors cannot approve
|
|
if not required:
|
|
return
|
|
last_submit = self.conn.execute("SELECT MAX(id) FROM item_events WHERE item_id=? AND to_state='submitted'",
|
|
(int(item_id),)).fetchone()[0] or 0
|
|
approvers = {r["agent"] for r in self.conn.execute(
|
|
"SELECT agent FROM item_events WHERE item_id=? AND to_state='approved' AND id>?", (int(item_id), last_submit))}
|
|
missing = [r for r in required if r not in approvers]
|
|
if missing:
|
|
raise HarborError(f"policy: item #{item_id} needs approval from {', '.join(missing)} before it lands "
|
|
"(since its last submission)")
|
|
|
|
def check_message_policy(self, sender, to_all):
|
|
cap = self.policy("all_per_hour")
|
|
if not cap or not to_all or sender in self.coordinators:
|
|
return
|
|
since = iso(utc(self.clock() - 3600))
|
|
count = self.conn.execute("SELECT COUNT(*) FROM messages WHERE sender=? AND created_at>=? AND "
|
|
"recipients='[\"All\"]' AND origin NOT IN ('legacy', 'system', 'hook')",
|
|
(sender, since)).fetchone()[0]
|
|
if count >= int(cap):
|
|
raise HarborError(f"policy: at most {cap} messages to everyone per hour - use `doing` for progress, or "
|
|
"a thread with the people concerned")
|