Inbox: people before system notices and newest first; wait wakes on new or human messages, not an old notice backlog; mark_read MCP tool

DeepSeek's inbox was buried under 78 automatic baseline notices read oldest
first, so it missed direct answers.

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],

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))

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):

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)