diff --git a/harbor/core.py b/harbor/core.py index b925221..df81669 100644 --- a/harbor/core.py +++ b/harbor/core.py @@ -447,15 +447,21 @@ class Harbor(WorkMixin, ExtrasMixin, FlowMixin, LaunchMixin): out.setdefault(r["corrects"], []).append(r["id"]) return out - def inbox(self, agent, unread_only=True, limit=30, mark_read=True): + 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" - query += (" ORDER BY CASE m.importance WHEN 'urgent' THEN 0 WHEN 'high' THEN 1 ELSE 2 END, m.id" - if unread_only else " ORDER BY m.id DESC") - rows = [self._message_dict(r) for r in self.conn.execute(query + " LIMIT ?", (agent, int(limit)))] + 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 @@ -472,6 +478,18 @@ class Harbor(WorkMixin, ExtrasMixin, FlowMixin, LaunchMixin): 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() @@ -483,9 +501,10 @@ class Harbor(WorkMixin, ExtrasMixin, FlowMixin, LaunchMixin): 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(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], diff --git a/harbor/mcp_server.py b/harbor/mcp_server.py index 679af1e..428058c 100644 --- a/harbor/mcp_server.py +++ b/harbor/mcp_server.py @@ -141,6 +141,28 @@ def inbox(limit: int = 20, include_read: bool = False, peek: bool = False) -> li mark_read=not peek)]) +@mcp.tool +def mark_read(ids: list[int] | None = None, before_id: int | None = None, system_only: bool = False) -> dict: + """Mark messages to you read without loading them: the given ids, or everything unread up to before_id + (system_only=True limits that to Harbor's automatic notices). Use it to clear a stale backlog so new messages + come up first.""" + def do(h): + agent = me() + if ids: + chosen = [int(i) for i in ids] + else: + if before_id is None: + raise HarborError("give ids or before_id") + query = ("SELECT d.message_id 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 < ?") + if system_only: + query += " AND m.origin='system'" + chosen = [r["message_id"] for r in h.conn.execute(query, (agent, int(before_id)))] + h.mark_read(agent, chosen) + return {"marked_read": len(chosen), "unread_now": h.unread_count(agent)[0]} + return run(do) + + @mcp.tool async def wait_for_messages(timeout_seconds: int = 300) -> dict: """Wait (up to timeout_seconds, max 1800) until a message for you arrives or a stop applies, then return the @@ -148,8 +170,9 @@ async def wait_for_messages(timeout_seconds: int = 300) -> dict: agent = run(lambda h: me()) timeout = max(1, min(int(timeout_seconds), 1800)) deadline = asyncio.get_running_loop().time() + timeout + since = run(lambda h: h.last_delivered_id(agent)) # wake on new messages, not on an old unread backlog while True: - state = run(lambda h: (h.unread_count(agent)[0], [f["key"] for f in h.stopped(agent)])) + state = run(lambda h: (h.unread_count_after(agent, since), [f["key"] for f in h.stopped(agent)])) if state[0] or state[1] or asyncio.get_running_loop().time() >= deadline: messages = run(lambda h: [compact(m) for m in h.inbox(agent, limit=30)]) if state[0] else [] run(lambda h: h.touch(agent)) diff --git a/tests/test_core.py b/tests/test_core.py index da534a1..0fa6fc1 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -209,7 +209,7 @@ class Review(HarborCase): self.h.transition("DeepSeek", item["id"], "approved") self.h.transition("Claude", item["id"], "in_review") self.h.transition("Claude", item["id"], "changes_requested", note="movsx: signed read") - self.assertEqual(self.h.inbox("DeepSeek")[-1]["kind"], "request") + self.assertEqual(self.h.inbox("DeepSeek")[0]["kind"], "request") self.h.transition("DeepSeek", item["id"], "submitted", commit="def5678") self.h.transition("Kimi", item["id"], "approved") with self.assertRaises(HarborError): diff --git a/tests/test_launch.py b/tests/test_launch.py index 971b71d..b98e4d7 100644 --- a/tests/test_launch.py +++ b/tests/test_launch.py @@ -22,7 +22,7 @@ class TrialRuns(HarborCase): self.assertEqual(self.h.stopped("Codex"), []) self.assertIsNotNone(self.h.trial("Codex")) self.assertIn("Trial run: you may work until", self.h.brief("Codex")) - self.assertIn("trial run until", self.h.inbox("Codex")[-1]["body"]) + self.assertIn("trial run until", self.h.inbox("Codex")[0]["body"]) self.clock.t += 31 * 60 stops = self.h.stopped("Codex") self.assertEqual(len(stops), 1)