"""Getting a team online and watching it work: activity log the last tool calls of each agent (tool and target: a file path, a command, a Harbor tool) roll call the human asks every agent "are you there?" with a one-tap answer; shows who answered and how fast launch status per agent: app connected, brief read, roll call answered, stopped / on a trial run / running expert finder who has worked on a scope (notes, reviews, claims, tasks, messages) summaries a short summary pinned to a long thread Timed resumes ("trial runs") live in core.set_flag / stopped(): a resume with an end time stops the agent again when the time is up, with no process needed. """ import json import re from .util import HarborError, first_line, iso, parse_iso, parse_scope, scopes_overlap, utc LAUNCH_SCHEMA = """ CREATE TABLE IF NOT EXISTS activity_log(id INTEGER PRIMARY KEY, agent TEXT NOT NULL, at TEXT NOT NULL, tool TEXT, target TEXT); CREATE INDEX IF NOT EXISTS activity_agent ON activity_log(agent, id); CREATE TABLE IF NOT EXISTS summaries(id INTEGER PRIMARY KEY, thread_id INTEGER NOT NULL, body TEXT NOT NULL, author TEXT NOT NULL, at TEXT NOT NULL); CREATE INDEX IF NOT EXISTS summaries_thread ON summaries(thread_id, id); """ KEEP_ACTIVITY = 300 SECRETISH = re.compile(r"(?i)(token|secret|password|passwd|api[_-]?key|bearer)\S*\s*[=:]?\s*\S+") def describe_target(tool, tool_input, paths): """A short, safe description of what a tool call touches.""" if paths: return ", ".join(paths)[:240] if isinstance(tool_input, dict): command = tool_input.get("command") or tool_input.get("cmd") if isinstance(command, list): command = command[-1] if command else "" if isinstance(command, str) and command.strip(): return SECRETISH.sub(r"\1=…", " ".join(command.split()))[:240] for key in ("file_path", "path", "url", "pattern", "query"): if isinstance(tool_input.get(key), str): return tool_input[key][:240] return None class LaunchMixin: # ----------------------------------------------------------------- activity log def _log_activity(self, c, agent, tool, target): c.execute("INSERT INTO activity_log(agent, at, tool, target) VALUES (?,?,?,?)", (agent, self.now(), str(tool)[:60], target)) c.execute("DELETE FROM activity_log WHERE agent=? AND id <= (SELECT id FROM activity_log WHERE agent=? " "ORDER BY id DESC LIMIT 1 OFFSET ?)", (agent, agent, KEEP_ACTIVITY)) def activity_feed(self, limit=60, agent=None): query, args = "SELECT * FROM activity_log", [] if agent: query += " WHERE agent=?" args.append(agent) rows = [dict(r) for r in self.conn.execute(query + " ORDER BY id DESC LIMIT ?", (*args, int(limit)))] for r in rows: r["seconds"] = int((utc(self.clock()) - parse_iso(r["at"])).total_seconds()) return rows # ----------------------------------------------------------------- roll call def rollcall(self, by, origin, agents=None): """The human asks each agent (default: all) to answer 'Here'. Works while agents are stopped: answering is one of the things a stopped agent may still do.""" if by != self.human or origin != "web": raise HarborError(f"only {self.human} starts a roll call, from the board") targets = agents or self.project.ai_names for name in targets: self._check_agent(name, allow_human=False) stamp = self.now() row = self.conn.execute("SELECT value FROM meta WHERE key='rollcall'").fetchone() latest = json.loads(row["value"]).get("by_agent", {}) if row else {} for name in targets: # a new roll call replaces the agent's unanswered one old = latest.get(name) if old and self.question(old)["answer"] is None: self._write(lambda c, q=old: c.execute( "UPDATE questions SET answer='(replaced by a newer roll call)', answered_by=?, answered_at=? " "WHERE id=? AND answer IS NULL", (self.human, stamp, q))) msg = self.question(old)["message_id"] self._write(lambda c, m=msg, a=name: c.execute( "UPDATE deliveries SET read_at=COALESCE(read_at, ?), acked_at=COALESCE(acked_at, ?) WHERE " "message_id=? AND agent=?", (stamp, stamp, m, a))) made = [] for name in targets: q = self.ask(by, f"Roll call ({stamp}): answer 'Here' so {self.human} knows you are online and reading " "Harbor - with the Harbor answer tool, or `harbor answer Here`.", ["Here"], to=name, origin="web", thread="rollcall") made.append(q["id"]) latest[name] = q["id"] self._write(lambda c: c.execute("INSERT INTO meta(key, value) VALUES ('rollcall', ?) ON CONFLICT(key) DO UPDATE " "SET value=excluded.value", (json.dumps({"at": stamp, "by_agent": latest}),))) return self.rollcall_status() def rollcall_status(self): row = self.conn.execute("SELECT value FROM meta WHERE key='rollcall'").fetchone() if not row: return None info = json.loads(row["value"]) out = [] for qid in info.get("by_agent", {}).values() or info.get("questions", []): q = self.question(qid) d = self.conn.execute("SELECT read_at FROM deliveries WHERE message_id=? AND agent=?", (q["message_id"], q["to_name"])).fetchone() seconds = None if q["answered_at"]: seconds = int((parse_iso(q["answered_at"]) - parse_iso(q["created_at"])).total_seconds()) out.append({"agent": q["to_name"], "question": qid, "read_at": d["read_at"] if d else None, "answered_at": q["answered_at"], "answer": q["answer"], "seconds": seconds}) return {"at": info["at"], "agents": out} # ----------------------------------------------------------------- launch status def note_brief(self, agent): self._write(lambda c: c.execute("INSERT INTO meta(key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET " "value=excluded.value", (f"brief:{agent}", self.now()))) def launch_status(self): roll = {r["agent"]: r for r in (self.rollcall_status() or {}).get("agents", [])} links = self.connections() out = [] for a in self.agents(): if a["kind"] == "human": continue name = a["name"] brief = self.conn.execute("SELECT value FROM meta WHERE key=?", (f"brief:{name}",)).fetchone() stops = self.stopped(name) trial = self.trial(name) seen = [c for c in links.get(name, []) if c["minutes"] is not None and c["minutes"] <= 15] out.append({ "agent": name, "host": next((x.get("host") for x in self.project.agents if x["name"] == name), ""), "connected": bool([c for c in seen if c["channel"] in ("mcp", "hook")]), "hooks": next((c["last_at"] for c in links.get(name, []) if c["channel"] == "hook" and c["minutes"] is not None and c["minutes"] <= 24 * 60), None), "connected_via": sorted({f"{c['channel']}{'/' + c['host'] if c['host'] else ''}" for c in seen}), "brief_at": brief["value"] if brief else None, "rollcall": roll.get(name), "state": "stopped" if stops else ("trial" if trial else "running"), "stop_reason": stops[0]["reason"] if stops else None, "trial_until": trial, "working": self.working(a), "last_tool": a.get("last_tool"), }) return out # ----------------------------------------------------------------- autopilot AUTOPILOT_REST_AFTER = 3 def continuation(self, agent): """What a running agent should do next instead of ending its turn, or None to let it rest. Stopped agents always rest: the human's stop is the off switch. An agent that ends AUTOPILOT_REST_AFTER turns in a row without a single tool call in between is let rest as well (attention shows it as resting), so a stuck agent does not loop. Any tool call resets the count.""" if not self.project.autopilot or agent == self.human or self.stopped(agent): return None last = self.conn.execute("SELECT MAX(rowid) FROM activity_log WHERE agent=?", (agent,)).fetchone()[0] or 0 key = f"autopilot:{agent}" row = self.conn.execute("SELECT value FROM meta WHERE key=?", (key,)).fetchone() state = json.loads(row["value"]) if row else {} streak = state.get("streak", 0) + 1 if state.get("last_tool") == last else 0 resting = streak >= self.AUTOPILOT_REST_AFTER value = json.dumps({"streak": streak, "last_tool": last, "at": self.now(), "resting": resting}) self._write(lambda c: c.execute("INSERT INTO meta(key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET " "value=excluded.value", (key, value))) if resting: return None mine = [t for t in self.tasks(states=("in_progress",), limit=100000) if t["assignee"] == agent] band = self.conn.execute("SELECT 1 FROM claims WHERE agent=? AND kind='band' AND released_at IS NULL", (agent,)).fetchone() if mine: t = mine[0] step = (f"Continue task #{t['id']}: {t['title']}. When it is finished, submit it to the review queue with " "its task number, then take the next one with next.") elif (nxt := self.next_task(agent, take=False)) is not None: step = f"Take your next task with next (the best one now is #{nxt['id']}: {nxt['title']})." elif band: step = ("The task queue has nothing for you: take the next eligible game function in your own band " "(claim it first), make it readable, and submit it to the review queue.") else: waiting = [i for i in self.items(states=("submitted", "in_review")) if i["author"] != agent] spec = next((x for x in self.project.agents if x["name"] == agent), {}) if waiting: step = (f"Review the next item in the review queue (#{waiting[0]['id']}: {waiting[0]['title']}) and move " "it to approved or changes_requested with the evidence.") else: step = ("Continue the work your instructions describe" + (f": {spec['instructions']}" if spec.get("instructions") else ".") + " When there is nothing at all to do right now, wait for " "new work with wait_for_messages instead of ending your turn.") return ("Harbor autopilot: you are running, so keep working instead of ending your turn. First read any new " f"Harbor messages (inbox). Then: {step} Keep doing current, and write a handoff at natural checkpoints, " "not every turn. Only the user's stop ends your work.") def resting_agents(self): out = [] for r in self.conn.execute("SELECT key, value FROM meta WHERE key LIKE 'autopilot:%'"): agent = r["key"].split(":", 1)[1] info = json.loads(r["value"]) if info.get("resting") and not self.stopped(agent): out.append({"agent": agent, "since": info.get("at")}) return out # ----------------------------------------------------------------- expert finder def experts(self, scope, limit=5): """Agents ranked by how much they have worked on a scope.""" target = parse_scope(scope) score, detail = {}, {} def add(agent, kind, weight): if agent == self.human: return score[agent] = score.get(agent, 0) + weight detail.setdefault(agent, {}).setdefault(kind, 0) detail[agent][kind] += 1 for n in self.notes(target[0]): add(n["author"], "notes", 3) for r in self.conn.execute("SELECT * FROM claims ORDER BY id DESC LIMIT 5000"): if scopes_overlap(target, (r["scope"], r["lo"], r["hi"])) and r["kind"] != "band": add(r["agent"], "claims", 2) for r in self.conn.execute("SELECT i.scope, i.author, e.agent, e.to_state FROM item_events e JOIN items i " "ON i.id=e.item_id WHERE i.scope IS NOT NULL"): try: if scopes_overlap(target, parse_scope(r["scope"])): add(r["agent"], "drafts" if r["agent"] == r["author"] else "reviews", 3) except HarborError: pass ctx = self.context(scope, limit=40) for m in ctx["messages"]: add(m["sender"], "messages", 1) ranked = sorted(score, key=lambda a: (-score[a], -len(detail[a]), a))[:limit] return [{"agent": a, "score": score[a], "detail": detail[a]} for a in ranked] # ----------------------------------------------------------------- thread summaries def summarize(self, author, thread, body): self._check_agent(author) info, _ = self.thread(thread) body = (body or "").strip() if not body: raise HarborError("empty summary") if len(body) > 4000: raise HarborError("a summary is at most 4,000 characters - it is meant to be short") now = self.now() sid = self._write(lambda c: c.execute("INSERT INTO summaries(thread_id, body, author, at) VALUES (?,?,?,?)", (info["id"], body, author, now)).lastrowid) return {"id": sid, "thread": info["topic"]} def summary(self, thread_id): row = self.conn.execute("SELECT * FROM summaries WHERE thread_id=? ORDER BY id DESC LIMIT 1", (int(thread_id),)).fetchone() return dict(row) if row else None def threads_needing_summary(self, agent, min_messages=20): rows = self.conn.execute( "SELECT t.id, t.topic, COUNT(m.id) n, MAX(m.created_at) last FROM threads t JOIN messages m ON m.thread_id=t.id " "WHERE t.topic NOT LIKE 'legacy:%' AND t.state='open' AND t.topic NOT IN ('general', 'control', 'questions', " "'rollcall', 'wins') GROUP BY t.id HAVING n>=? AND EXISTS (SELECT 1 FROM messages x WHERE x.thread_id=t.id " "AND x.sender=?)", (min_messages, agent)).fetchall() out = [] for r in rows: s = self.summary(r["id"]) if not s or s["at"] < r["last"] and self.conn.execute( "SELECT COUNT(*) FROM messages WHERE thread_id=? AND created_at>?", (r["id"], s["at"])).fetchone()[0] >= 10: out.append({"thread": r["topic"], "messages": r["n"]}) return out