From c77ed1a4dbecc4ab7f553c9924aafd270d70f1c0 Mon Sep 17 00:00:00 2001 From: KeelrLeek <156356974+KeelrLeek@users.noreply.github.com> Date: Wed, 7 Oct 2026 11:04:54 -0500 Subject: [PATCH] policy submits_per_hour: an agent can be capped at N submissions or resubmissions per hour (the User asked the DeepSeeks to slow down and check their work) --- harbor/core.py | 3 + harbor/flow.py | 570 +++++++++++++++++++++++---------------------- harbor/projects.py | 8 + 3 files changed, 303 insertions(+), 278 deletions(-) diff --git a/harbor/core.py b/harbor/core.py index 106975f..b3eb268 100644 --- a/harbor/core.py +++ b/harbor/core.py @@ -762,6 +762,7 @@ class Harbor(WorkMixin, ExtrasMixin, FlowMixin, LaunchMixin): 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() @@ -825,6 +826,8 @@ class Harbor(WorkMixin, ExtrasMixin, FlowMixin, LaunchMixin): 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() diff --git a/harbor/flow.py b/harbor/flow.py index 524343a..e6372e8 100644 --- a/harbor/flow.py +++ b/harbor/flow.py @@ -1,278 +1,292 @@ -"""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") +"""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_submit_policy(self, author): + """policy submits_per_hour: an agent capped at N submissions (new or resubmitted) per hour must slow down + and check its work before sending more.""" + cap = (self.policy("submits_per_hour") or {}).get(author) + if not cap: + return + since = iso(utc(self.clock() - 3600)) + count = self.conn.execute("SELECT COUNT(*) FROM item_events WHERE agent=? AND to_state='submitted' AND at>=?", + (author, since)).fetchone()[0] + if count >= int(cap): + raise HarborError(f"policy: at most {cap} submissions per hour for {author}. Slow down: re-read the " + "disassembly against your C line by line, run quick_readable and re_behaviour, list " + "your changed annotation rows, and submit only work you have checked twice") + + 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") diff --git a/harbor/projects.py b/harbor/projects.py index 30d8db3..69e7584 100644 --- a/harbor/projects.py +++ b/harbor/projects.py @@ -196,6 +196,14 @@ def _policy(raw, names): policy["all_per_hour"] = int(policy["all_per_hour"]) except (TypeError, ValueError): raise ProjectError("policy all_per_hour must be a whole number") + if "submits_per_hour" in policy: # {agent: n}: submissions and resubmissions per hour + caps = policy["submits_per_hour"] + if not isinstance(caps, dict) or not all(k in names for k in caps): + raise ProjectError("policy submits_per_hour must map agents on the roster to whole numbers") + try: + policy["submits_per_hour"] = {k: int(v) for k, v in caps.items()} + except (TypeError, ValueError): + raise ProjectError("policy submits_per_hour must map agents on the roster to whole numbers") return policy