README: front page explaining what Harbor is, why it helps agent teams, how it works, and a quick start

README.md

@@ -1,6 +1,85 @@
-# Harbor
+<div align="center">
+
+# ⚓ Harbor
+
+**A local coordination hub for teams of AI coding agents, and the person running them.**
+
+Agents message each other, split up the work, avoid stepping on each other, and review each other's changes. You watch it all from one page, and step in only when you want to.
+
+`Python 3.11+` · `standard library only` · `SQLite` · `MCP` · `Claude Code` · `Codex` · `OpenCode` · `Kimi` · `DeepSeek`
+
+</div>
+
+---
+
+## Why Harbor?
+
+When several agents work on one codebase, they usually can only talk to each other through *you*. You copy questions from one window into another. You notice too late that two of them rewrote the same file. You can't see who is stuck, who is idle, or who quietly stopped an hour ago.
+
+Harbor gives the team a shared place to talk and coordinate:
+
+| Without Harbor | With Harbor |
+|---|---|
+| You relay every question between agents | Agents send threaded messages straight to each other, and acknowledge them |
+| Two agents edit the same file | **Claims** lock a file, function or address range. A collision alert fires before anyone commits |
+| "What should I do next?" | A **task queue** with priorities, dependencies and owners. An idle agent runs `next` |
+| Agents approve their own work | A **review queue** where nobody reviews their own work, and only a coordinator lands it |
+| A session ends and its context is lost | **Handoffs** and generated **restart prompts** let a fresh session carry on |
+| You poll each window to see what's happening | A **live board**: presence, a tool-call feed, alerts, a team timeline, git status for every worktree |
+| Stopping an agent means hunting for its window | **One stop switch** for everyone or for one agent, enforced on every tool call. Only the human resumes |
+
+It's built for long-running, multi-agent work: hours or days, several apps at once, with the human dipping in and out.
+
+## How it works
+
+```mermaid
+flowchart LR
+    subgraph Agents
+        A1[Claude Code]
+        A2[Codex]
+        A3[OpenCode / Kimi / DeepSeek]
+    end
+    A1 -- MCP tools --> H
+    A2 -- MCP tools --> H
+    A3 -- CLI / MCP --> H
+    A1 -. hooks: activity, stop switch .-> H
+    A2 -. hooks .-> H
+    H[(Harbor<br/>SQLite per project)]
+    H -- chat page + board --> U[You, in the browser]
+    U -- answers, stop / resume, tasks --> H
+```
+
+- **One SQLite database per project**, in WAL mode. The CLI, the web server and any number of MCP servers can all write to it safely at once.
+- **Agents connect through MCP or the CLI.** `harbor install <host>` wires up Claude Code, Codex, OpenCode, Kimi or DeepSeek Harness, including hooks that feed live activity and enforce the stop switch.
+- **Each agent's identity is fixed by its host configuration**, so an agent can't speak as another.
+- **The web server binds to localhost only.** It gives you a texting-style chat page and a full coordination board.
+
+## Quick start
+
+```bash
+git clone https://git.whoagland.com/Keelrleek/Harbor.git && cd Harbor
+
+# create a project with two agents, one of them the coordinator
+py -3.12 -B run.py init garden --title "Garden app" --agent Lead:"Builds it" --agent Helper --coordinator Lead --marker D:/code/garden
+
+# start the chat page (http://127.0.0.1:8766) and the board (/board)
+py -3.12 -B run.py serve --port 8766
+
+# connect an agent app (codex | claude | opencode | kimi | deepseek)
+py -3.12 -B run.py install claude --agent Lead
+
+# check everything is wired up
+py -3.12 -B run.py doctor
+```
+
+On Linux or macOS, use `python3` in place of `py -3.12`. To get a `harbor` command, run `pip install -e .`. The optional MCP server needs `pip install -e .[mcp]`.
+
+Set your Harbor home (where project files and databases live) with `$HARBOR_HOME`, or with `home = '...'` in `local.toml`. The default is `~/.harbor`.
+
+---
+
+## Features
 
-Harbor is a local coordination hub for a team of AI coding agents and the person running them. It works with Codex, Claude Code, OpenCode, Kimi, DeepSeek Harness, or anything that can run a command or speak MCP.
 
 **Talking**
 - **Threads.** Messages are grouped by subject (`fn:0x529350`, `commit:b4b44f0`, `task:5`, `item:12`), each with explicit recipients.
@@ -90,7 +169,7 @@ Harbor is a local coordination hub for a team of AI coding agents and the person
 
 Each project is one SQLite database (WAL mode), so the CLI, the web server and many MCP servers can write at once safely. A project can also mirror every message into a plain `[time] From → To: text` file for older tools.
 
-## Quick start
+## Project setup
 
 ```bash
 py -3.12 -B run.py init garden --title "Garden app" --agent Lead:"Builds it" --agent Helper --coordinator Lead --marker D:/code/garden
@@ -183,4 +262,4 @@ Hooks resolve the project from the agent's working folder and stay silent anywhe
 py -3.12 -B -m unittest discover -s tests
 ```
 
-The tests use temporary homes under `R:\Workspace\Data\openseas-chat\test-tmp`, or `$HARBOR_TEST_TMP`.
+Set `$HARBOR_TEST_TMP` to a scratch folder first; the tests create their temporary homes there (the built-in default is the original author's path).

harbor/cli.py

@@ -255,7 +255,7 @@ def build_parser():
     p = sub.add_parser("who", help="who has worked on a scope (notes, reviews, claims, messages)")
     p.add_argument("scope")
     p = sub.add_parser("re", help="reverse-engineering lookups from the project's image and annotations")
-    p.add_argument("action", choices=["disasm", "xrefs", "bytes", "lookup", "ghidra", "ghidra-tools", "debugger-tools", "match", "similar", "struct-audit", "behaviour"])
+    p.add_argument("action", choices=["disasm", "xrefs", "bytes", "lookup", "summary", "ghidra", "ghidra-tools", "ghidra-check", "debugger-tools", "match", "similar", "struct-audit", "behaviour"])
     p.add_argument("target", nargs="?", default="", help="an address (0x4AD850) or a name; for ghidra, the tool")
     p.add_argument("arguments", nargs="?", default="{}", help="ghidra: the tool's arguments as JSON")
     p.add_argument("--decomp", help="match: the checkout or worktree to compile from")
@@ -722,6 +722,21 @@ def main(argv=None):
         elif args.action == "ghidra-tools":
             res = Ghidra(harbor.project.reverse).tools()
             out(res, "\n".join(f"{t['name']:<26} {t['description'].splitlines()[0][:90]}" for t in res))
+        elif args.action == "ghidra-check":
+            from .reverse import ghidra_check
+            res = ghidra_check(harbor.project.reverse)
+            mark = lambda ok: "ok  " if ok else ("??  " if ok is None else "FAIL")
+            out(res, "\n".join(
+                [f"{'Ghidra serves our program' if res['ok'] else 'Ghidra does NOT serve our program'}: "
+                 f"{res['program']} ({res['functions_in_ghidra']} functions); image {res['image_base']}, "
+                 f"{res['image_size']} bytes, sha256 {res['image_sha256'][:16]}..."] +
+                [f"  {mark(c['ok'])} {c['check']}: " + (f"expected {c['expected']}, Ghidra {c['ghidra']}" if "ghidra" in c
+                                                        else ("differs at " + ", ".join(c["differ_at"]) if c.get("differ_at") else "all equal"))
+                 for c in res["checks"]]))
+        elif args.action == "summary":
+            from .reverse import format_summary
+            res = r.summary(args.target)
+            out(res, format_summary(res))
         elif args.action == "ghidra":
             text = Ghidra(harbor.project.reverse).call(args.target, json.loads(args.arguments))
             out({"result": text}, text)

harbor/core.py

@@ -709,7 +709,14 @@ class Harbor(WorkMixin, ExtrasMixin, FlowMixin, LaunchMixin):
             if kind == "draft" and lo is not None:
                 bands = [a for a in active if a["kind"] == "band"
                          and scopes_overlap((canonical, lo, hi), (a["scope"], a["lo"], a["hi"]))]
-                if bands and not any(b["agent"] == agent for b in bands):
+                # a task a coordinator assigned to this agent for this scope is the exception to another's band
+                assigned = any(scopes_overlap((canonical, lo, hi), (t["scope"], *parse_scope(t["scope"])[1:]))
+                               for t in c.execute("SELECT scope FROM tasks WHERE assignee=? AND scope IS NOT NULL "
+                                                  "AND state IN ('open','in_progress') AND created_by IN (%s)"
+                                                  % ",".join("?" * len(self.coordinators)),
+                                                  (agent, *self.coordinators)).fetchall()
+                               if t["scope"].startswith(("fn:", "range:", "file:")))
+                if bands and not any(b["agent"] == agent for b in bands) and not assigned:
                     conflicts += bands
             if conflicts and not override:
                 described = "; ".join(f"{a['agent']} holds {a['kind']} {a['scope']}"
@@ -762,6 +769,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 +833,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()

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

harbor/lander_gate.py

@@ -0,0 +1,69 @@
+"""The server's background compiles (the Logbook's statistics, the whole-program check, the emulator run) give way to
+the lander: its gates (a full readable check of every file) are what land work on main, and a second compile of main
+beside them roughly quadruples their time.
+
+Two parts:
+  lander_busy(project)  while [logbook] lander_lock names a file that exists and was written in the last
+                        STALE_MINUTES, the lander is running its gates: background jobs do not start. The lander
+                        creates the file before its gates and removes it after the push; an older file is a lander
+                        that died, and is ignored.
+  run_low(...)          subprocess.run at BELOW_NORMAL priority (children such as cl.exe inherit it) that, on a
+                        timeout, ends the whole process tree instead of leaving the compiler running orphaned.
+
+Configured per project in the [logbook] table:
+    lander_lock = 'R:/Workspace/Data/lander-busy.lock'      unset or "": never wait"""
+import os
+import subprocess
+import time
+from pathlib import Path
+
+STALE_MINUTES = 60
+BELOW_NORMAL = getattr(subprocess, "BELOW_NORMAL_PRIORITY_CLASS", 0)
+IDLE = getattr(subprocess, "IDLE_PRIORITY_CLASS", 0)
+NO_WINDOW = getattr(subprocess, "CREATE_NO_WINDOW", 0)
+
+
+def lock_path(project):
+    path = ((getattr(project, "logbook", None) or {}).get("lander_lock") or "").strip()
+    return Path(path) if path else None
+
+
+def lander_busy(project, now=None):
+    """Why background work should wait (the lander's lock is fresh), or None."""
+    path = lock_path(project)
+    if path is None:
+        return None
+    try:
+        written = path.stat().st_mtime
+    except OSError:
+        return None
+    age = (time.time() if now is None else now) - written
+    if age >= STALE_MINUTES * 60:
+        return None
+    return f"the lander is running its gates ({path.name}, {max(0, int(age // 60))} min old)"
+
+
+def kill_tree(pid):
+    if os.name == "nt":
+        subprocess.run(["taskkill", "/PID", str(pid), "/T", "/F"], capture_output=True, creationflags=NO_WINDOW)
+    else:
+        try:
+            os.kill(pid, 9)
+        except OSError:
+            pass
+
+
+def run_low(argv, timeout, priority=BELOW_NORMAL, **kwargs):
+    """subprocess.run(argv, capture_output=True, text=True, ...) at a lower priority class; on a timeout the whole
+    tree (cmd.exe, cl.exe) is ended before TimeoutExpired is raised."""
+    flags = kwargs.pop("creationflags", 0) | (priority | NO_WINDOW if os.name == "nt" else 0)
+    with subprocess.Popen(argv, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+                          text=True, creationflags=flags, **kwargs) as proc:
+        try:
+            stdout, stderr = proc.communicate(timeout=timeout)
+        except subprocess.TimeoutExpired:
+            kill_tree(proc.pid)
+            proc.kill()
+            proc.communicate()
+            raise subprocess.TimeoutExpired(argv, timeout) from None
+    return subprocess.CompletedProcess(argv, proc.returncode, stdout, stderr)

harbor/logbook_auto.py

@@ -9,6 +9,8 @@ Configured per project in a [logbook] table:
     decomp = 'R:/.../open-seas-decomp'          the repository whose main branch is followed
     stats_repo = 'R:/.../stats-work/repo'       a scratch copy of open-seas holding tools/progress/*.py
     interval_minutes = 10                       how often the server looks (default 10)
+    lander_lock = 'R:/.../lander-busy.lock'     wait while the lander runs its gates (harbor.lander_gate)
+The progress tools run below normal priority.
 Milestones and narrative entries stay with people: they need judgement.
 
 State (the last commit seen) is kept in the logbook folder as auto.json."""
@@ -22,9 +24,10 @@ import time
 import traceback
 from pathlib import Path
 
-from . import logbook
+from . import lander_gate, logbook
 
 MAX_ENTRIES = 25
+STATS_TIMEOUT = 1800        # each progress tool; sync_decomp compiles every readable file, below normal priority
 
 
 def _git(repo, *args):
@@ -73,8 +76,8 @@ def log_commits(project, repo, since, head):
     for sha, when, subject, body in commits:
         files = [f for f in _git(repo, "show", "--name-only", "--format=", sha).splitlines() if f.strip()]
         lines = [l for l in body.strip().splitlines() if l.strip() and not l.lower().startswith("co-authored-by")]
-        detail = "\n".join(lines[:8]) if lines else subject
-        detail += f"\n(main {sha[:8]}, {len(files)} file{'s' if len(files) != 1 else ''})"
+        detail = "\n".join(lines[:8]) + "\n" if lines else ""          # the board shows the title already
+        detail += f"(main {sha[:8]}, {len(files)} file{'s' if len(files) != 1 else ''})"
         added.append(logbook.add_entry(project, subject, detail, kind=_kind(files), at=_utc(when)))
     return added
 
@@ -88,8 +91,10 @@ def rebuild_stats(project, repo, stats_repo, head):
     installer = r"C:\Program Files (x86)\Microsoft Visual Studio\Installer"
     if os.path.isdir(installer) and installer.lower() not in env.get("PATH", "").lower():
         env["PATH"] = installer + os.pathsep + env.get("PATH", "")
-    run = lambda *argv: subprocess.run([sys.executable, "-B", *map(str, argv)], cwd=str(stats_repo), env=env,
-                                       capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=900)
+    # below normal priority (cl.exe inherits it), so the lander's and the agents' compiles go first; a timeout ends
+    # the whole tree, not only python (an orphaned cl.exe went on compiling all of main at normal priority)
+    run = lambda *argv: lander_gate.run_low([sys.executable, "-B", *map(str, argv)], STATS_TIMEOUT,
+                                            cwd=str(stats_repo), env=env, encoding="utf-8", errors="replace")
     sync = run(tools / "sync_decomp.py", "--decomp", repo)
     if sync.returncode != 0:
         raise RuntimeError("sync_decomp: " + (sync.stdout + sync.stderr).strip()[-400:])
@@ -134,12 +139,20 @@ def refresh(project, force=False):
     state = _state(project)
     if state.get("last_head") == head and not force:
         return {"changed": False, "head": head[:8]}
+    busy = None if force else lander_gate.lander_busy(project)
+    if busy:                                             # its gates compile all of main: look again next time
+        return {"changed": False, "head": head[:8], "waiting": busy}
     out = {"changed": True, "head": head[:8]}
-    if state.get("last_head"):
+    # last_head is where the statistics are; logged_head where the ship's log is (a failed statistics run is retried,
+    # and must not log the same commits twice)
+    logged = state.get("logged_head") or state.get("last_head")
+    if logged and logged != head:
         try:
-            out["entries"] = len(log_commits(project, repo, state["last_head"], head))
+            out["entries"] = len(log_commits(project, repo, logged, head))
         except RuntimeError as error:                    # e.g. the old commit vanished after a rewrite
             out["log_error"] = str(error)
+    if logged:
+        state["logged_head"] = head
     if stats_repo:
         try:
             out["stats"] = rebuild_stats(project, snapshot_at(config, repo, head), stats_repo, head)

harbor/mcp_server.py

@@ -38,6 +38,8 @@ review queue, presence and stop flags. Protocol:
 10. Before your session ends or your allowance runs out, write handoff(): finished, in flight, next.
 11. To read the original machine code use re_disassemble, re_xrefs, re_bytes and re_lookup: they read the project's
     own image with the team's current names, so do not write your own disassembler or offset maths.
+    re_summary(address) gives the overview in one call: name, prototype, size, convention, readable status,
+    callers and callees with their status, globals, strings, vtable slots and the Mac/Xbox hints.
     For Ghidra's view (decompiler, p-code, stack frames, struct field references, types) use ghidra_tools() once,
     then ghidra(tool, arguments): one shared read-only Ghidra session holding the team's annotations.
     To check a value in the running game (x87 control word, `this`, struct fields), use debugger_tools() and
@@ -563,6 +565,14 @@ def re_lookup(target: str) -> dict:
     return _re(lambda r: r.lookup(target))
 
 
+@mcp.tool
+def re_summary(target: str) -> dict:
+    """One call before starting on a function: its current name and prototype, size and calling convention (read
+    from the bytes), whether it is readable on main, its callers and callees with their names and whether each is
+    readable on main, the globals and strings it uses, the vtable slots it fills, and the Mac/Xbox name hints."""
+    return _re(lambda r: r.summary(target))
+
+
 _ghidra = {}
 
 
@@ -692,6 +702,17 @@ def ghidra_tools() -> list:
     return _gh(lambda g: g.tools())
 
 
+@mcp.tool()
+def ghidra_check() -> dict:
+    """Is the shared Ghidra session serving our program? Compares its address range with the project image's base
+    and size, the SHA-256 of the file it imported with ours, and sample bytes; ok is false when any differs."""
+    try:
+        from .reverse import ghidra_check as check
+        return check(harbor().project.reverse)
+    except (HarborError, projects.ProjectError) as error:
+        raise ToolError(str(error)) from None
+
+
 @mcp.tool()
 def ghidra(tool: str, arguments: dict | None = None) -> str:
     """Run one tool in the shared read-only Ghidra session, e.g. ghidra('get_code', {'function': 'GetFogLevel',

harbor/progress_map.py

@@ -1,321 +1,331 @@
-"""The Logbook's progress map, in the style of decomp.dev: one unit per original source file, its functions and how
-far each has come, plus the headline share of game code decompiled and the Code and Data bars.
-
-GET /api/logbook/progress-map answers with one compact JSON document. It is computed once and kept until one of its
-inputs changes (main's HEAD, the Logbook's symbol table and statistics, the match, emulator and library results,
-facts, names, source files, globals, or a deferral note); `?have=<version>` answers {"version", "same": true} when
-the caller already holds the current one, so a page that polls does not download or re-lay-out anything.
-
-Definitions follow the Logbook exactly (open-seas tools/progress/progress.py, which writes stats.json):
-  - the symbol table is <[logbook] stats_repo>/re/symbols/pc102.csv: its category and status per function;
-  - counted ("game code") = category game and status not superseded or wontfix;
-  - decompiled = status decompiled, reimplemented or verified, so percent = decompiled game bytes / game bytes,
-    the Logbook's game_bytes_decompiled_percent.
-Decompiled functions are split further, best first: exact (byte match in the [reverse] match_results), verified
-(behaves the same in the emulator, diffrun_results, not 'weak'), readable. The rest: deferred (a live DEFERRED note),
-raw, and, not counted, unsorted, engine (Gamebryo, named only) and library (C runtime, D3DX9, platform wrappers).
-"results" says which commit the exact and verified tiers are from (harbor.results_refresh keeps them current).
-Without a stats_repo the categories come from names.csv / facts and readable from the src headers (the queue's view).
-
-Units, first that applies: the source file from the Xbox build's assert sites ([reverse] source_files); the D3DX9 block
-(platform/d3dx9) or the import-thunk stubs from library_ranges; the unit facts gives (engine/<class>,
-platform/<library>); otherwise
-<category> plus its 64 KB address block, labelled "(no source file)"."""
-import csv
-import hashlib
-import json
-import subprocess
-import threading
-import time
-from pathlib import Path
-
-from .queue import FunctionQueue
-
-STATUSES = ("exact", "verified", "readable", "deferred", "raw", "unsorted", "engine", "library")
-DECOMPILED = ("decompiled", "reimplemented", "verified")
-EXCLUDED = ("superseded", "wontfix")
-UNIT_FIELDS = ("name", "kind", "lo", "hi", "bytes", "counted", "exact", "verified", "readable", "functions")
-FUNCTION_FIELDS = ("unit", "address", "size", "status", "name")
-KEY_TTL = 15            # seconds a computed input key is trusted before the files and HEAD are looked at again
-
-
-def _int(text):
-    return int(str(text), 16) if str(text).lower().startswith("0x") else int(text)
-
-
-def _mtime(path):
-    try:
-        return Path(path).stat().st_mtime_ns if path else None
-    except OSError:
-        return None
-
-
-def _git(repo, *args):
-    try:
-        out = subprocess.run(["git", "-C", str(repo), *args], capture_output=True, text=True, timeout=20,
-                             encoding="utf-8", errors="replace")
-    except (OSError, subprocess.TimeoutExpired):
-        return ""
-    return out.stdout.strip() if out.returncode == 0 else ""
-
-
-def _utc(iso):
-    import datetime
-    try:
-        return datetime.datetime.fromisoformat(iso).astimezone(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
-    except ValueError:
-        return None
-
-
-class ProgressMap:
-    def __init__(self, project, queue_factory):
-        self.project = project
-        self.queue_factory = queue_factory        # -> FunctionQueue on a connection this thread may use
-        self.lock = threading.Lock()
-        self.value = None                         # (key, document)
-        self.checked = (0.0, None)                # (monotonic time, key)
-        self.computations = 0
-
-    # ------------------------------------------------------------------ inputs
-    def _config(self):
-        return self.project.reverse or {}, getattr(self.project, "logbook", None) or {}
-
-    def _symbols_path(self):
-        _, lb = self._config()
-        return Path(lb["stats_repo"]) / "re" / "symbols" / "pc102.csv" if lb.get("stats_repo") else None
-
-    def _decomp(self):
-        rev, lb = self._config()
-        if lb.get("decomp"):
-            return Path(lb["decomp"])
-        return Path(rev["annotations"]).parent if rev.get("annotations") else None
-
-    def _logbook_dir(self):
-        from . import logbook
-        return logbook.folder(self.project)
-
-    def key(self, q):
-        rev, _ = self._config()
-        decomp = self._decomp()
-        head = _git(decomp, "rev-parse", "HEAD") if decomp and (decomp / ".git").exists() else ""
-        paths = [self._symbols_path(), self._logbook_dir() / "stats.json", self._logbook_dir() / "history.json",
-                 self._logbook_dir() / "results.json"]
-        paths += [rev.get(k) for k in ("facts", "match_results", "diffrun_results", "library_results",
-                                       "library_ranges", "source_files")]
-        if rev.get("annotations"):
-            paths += [Path(rev["annotations"]) / "names.csv", Path(rev["annotations"]) / "globals.csv"]
-        try:
-            note = q.h.conn.execute("SELECT MAX(id) FROM notes").fetchone()[0]
-        except Exception:
-            note = None
-        raw = json.dumps([head, note, [str(p) for p in paths], [_mtime(p) for p in paths]])
-        return hashlib.sha1(raw.encode("utf-8")).hexdigest()[:16], head
-
-    # ------------------------------------------------------------------ answer
-    def get(self, have=None):
-        with self.lock:
-            q = self.queue_factory()
-            now = time.monotonic()
-            if self.value and now - self.checked[0] < KEY_TTL:
-                key = self.checked[1]
-            else:
-                key, head = self.key(q)
-                self.checked = (now, key)
-                if not self.value or self.value[0] != key:
-                    document = self.compute(q, head)
-                    if document is None and self.value:          # the symbol table is being rewritten: keep the last
-                        document = self.value[1]
-                    elif document is None:
-                        document = self.compute(q, head, use_symbols=False)
-                    document["version"] = key
-                    self.value = (key, document)
-                    self.computations += 1
-            if have and have == self.value[0]:
-                return {"version": have, "same": True}
-            return self.value[1]
-
-    # ------------------------------------------------------------------ the computation
-    def _symbols(self, expected):
-        """pc102.csv as {address: (category, status, name)}, or None while it is missing or half written."""
-        path = self._symbols_path()
-        if not path or not path.exists():
-            return None
-        try:
-            with path.open(encoding="utf-8", newline="") as stream:
-                rows = {_int(r["address"]): (r["category"], r["status"], r.get("name") or "")
-                        for r in csv.DictReader(stream)}
-        except (OSError, ValueError, KeyError):
-            return None
-        return rows if len(rows) >= 0.98 * expected else None
-
-    def _source_files(self, path):
-        out = {}
-        if path and Path(path).exists():
-            with open(path, encoding="utf-8", newline="") as stream:
-                for row in csv.DictReader(stream):
-                    if row.get("source_file"):
-                        out[_int(row["pc_address"])] = row["source_file"]
-        return out
-
-    def _all_ranges(self, path):
-        """[(lo, hi, label)] of the library_ranges csv (the D3DX9 block and the compiler's stubs)."""
-        out = []
-        if path and Path(path).exists():
-            with open(path, encoding="utf-8", newline="") as stream:
-                for r in csv.DictReader(stream):
-                    verdict = r.get("verdict", "")
-                    label = "platform/d3dx9" if verdict.startswith("D3DX") else \
-                        "Import thunks and unwind funclets" if "stubs" in verdict else None
-                    if label:
-                        out.append((_int(r["start"]), _int(r["end"]), label))
-        return out
-
-    def _data(self, facts, rev):
-        """Named globals against the distinct .data/.rdata addresses the code touches (needs sections.json)."""
-        facts_path = Path(rev["facts"]) if rev.get("facts") else None
-        sections_path = facts_path.parent / "sections.json" if facts_path else None
-        globals_path = Path(rev["annotations"]) / "globals.csv" if rev.get("annotations") else None
-        if not sections_path or not sections_path.exists() or not globals_path or not globals_path.exists():
-            return None
-        spans = [(s["start"], s["end"]) for s in json.loads(sections_path.read_text(encoding="utf-8"))
-                 if s.get("name") in (".data", ".rdata")]
-        touched = set()
-        for d in facts.values():
-            for ref in d.get("data", []):
-                a = _int(ref)
-                if any(lo <= a < hi for lo, hi in spans):
-                    touched.add(a)
-        with globals_path.open(encoding="utf-8", newline="") as stream:
-            named = {_int(r["address"]) for r in csv.DictReader(line for line in stream if not line.startswith("#"))
-                     if r.get("address") and r.get("name")}
-        if not touched:
-            return None
-        return {"addresses": len(touched), "named": len(touched & named), "globals": len(named),
-                "bytes": sum(hi - lo for lo, hi in spans)}
-
-    def compute(self, q, head, use_symbols=True):
-        rev, _ = self._config()
-        facts = q.r._facts()
-        symbols = self._symbols(len(facts)) if use_symbols else None
-        if use_symbols and self._symbols_path() and symbols is None and self._symbols_path().exists():
-            return None
-        names = q.r._csv("names.csv")
-        statuses = None if symbols else q.statuses()
-        exact = {a for a, row in q._results("match_results").items()
-                 if str(row.get("status")) in ("match", "exact", "exact-relaxed")}
-        verified = {a for a, row in q._results("diffrun_results").items()
-                    if str(row.get("verdict")) == "same" and not row.get("weak")}
-        library = set(q._results("library_results"))
-        ranges = self._all_ranges(rev.get("library_ranges"))
-        lib_ranges = [(lo, hi) for lo, hi, _ in ranges]
-        deferred = q.deferrals()
-        sources = self._source_files(rev.get("source_files"))
-
-        units, unit_index, functions = [], {}, []
-        game_functions = game_done = 0
-        for addr in sorted(facts):
-            d = facts[addr]
-            size = int(d.get("size") or 0)
-            if symbols and addr in symbols:
-                category, status, sym_name = symbols[addr]
-            else:
-                category = q._category(addr)
-                status = "decompiled" if (statuses or {}).get(addr, ("raw",))[0] == "readable" else "classified"
-                sym_name = ""
-            row = names.get(addr)
-            name = (row[1] if row and len(row) > 1 and row[1] else "") or sym_name or d.get("name") or f"FUN_{addr:06X}"
-            counted = category == "game" and status not in EXCLUDED
-            done = status in DECOMPILED
-            is_lib = q.is_library(addr, library, lib_ranges) or category in ("library", "platform", "crt")
-            if counted or (done and not is_lib) or (category == "unsorted" and not is_lib):
-                tier = "exact" if done and addr in exact else "verified" if done and addr in verified else \
-                    "readable" if done else "deferred" if addr in deferred else \
-                    "raw" if counted else "unsorted"
-            elif is_lib:
-                tier = "library"
-            else:
-                tier = "engine"
-
-            source = sources.get(addr)
-            in_range = next((label for lo, hi, label in ranges if lo <= addr < hi), None)
-            if source:
-                unit = source
-            elif in_range:
-                unit = in_range
-            elif d.get("unit"):
-                unit = d["unit"]
-            else:
-                unit = f"{category} · 0x{addr & ~0xFFFF:06X} block (no source file)"
-            if unit not in unit_index:
-                unit_index[unit] = len(units)
-                units.append({"name": unit, "lo": addr, "hi": addr + size, "bytes": 0, "counted": 0, "exact": 0,
-                              "verified": 0, "readable": 0, "functions": 0, "kinds": {}})
-            u = units[unit_index[unit]]
-            u["hi"] = max(u["hi"], addr + size)
-            u["bytes"] += size
-            u["functions"] += 1
-            kind = "game" if counted else "library" if tier == "library" else "engine" if category == "engine" \
-                else "unsorted" if category == "unsorted" else "library"
-            u["kinds"][kind] = u["kinds"].get(kind, 0) + size
-            if counted:
-                game_functions += 1
-                game_done += done
-                u["counted"] += size
-                if tier in ("exact", "verified", "readable"):
-                    u[tier] += size
-            functions.append([unit_index[unit], addr, size, STATUSES.index(tier), name])
-
-        for u in units:
-            kinds = u.pop("kinds")
-            u["kind"] = max(kinds, key=kinds.get) if kinds else "unsorted"      # what most of its bytes are
-
-        code = {k: sum(u[k] for u in units) for k in ("exact", "verified", "readable")}
-        code["bytes"] = sum(u["counted"] for u in units)
-        code["functions"], code["decompiled_functions"] = game_functions, game_done
-        percent = round(100 * (code["exact"] + code["verified"] + code["readable"]) / code["bytes"], 2) \
-            if code["bytes"] else 0.0
-
-        from . import logbook
-        stats = logbook._read(self.project, "stats.json", None) or {}
-        history = sorted(logbook._read(self.project, "history.json", []), key=lambda p: p.get("at", ""))
-        decomp = self._decomp()
-        commit = {"sha": stats.get("commit"), "at": None}
-        if commit["sha"] and decomp:
-            commit["at"] = _utc(_git(decomp, "log", "-1", "--format=%cI", commit["sha"])) or None
-        main = {"sha": head[:7] if head else None,
-                "at": _utc(_git(decomp, "log", "-1", "--format=%cI", head)) if head else None}
-        from . import results_refresh
-        return {
-            "project": self.project.title or self.project.name,
-            "results": results_refresh.freshness(self.project),
-            "source": "logbook symbol table" if symbols else "src headers and names.csv",
-            "commit": commit, "main": main, "updated_at": stats.get("updated_at"),
-            "percent": percent, "logbook_percent": stats.get("game_bytes_decompiled_percent"),
-            "code": code, "data": self._data(facts, rev),
-            "history": [[p.get("at"), p.get("decompiled"), p.get("commit"), p.get("game_bytes")]
-                        for p in history if p.get("decompiled") is not None],
-            "statuses": list(STATUSES),
-            "unit_fields": list(UNIT_FIELDS),
-            "units": [[u[k] for k in UNIT_FIELDS] for u in units],
-            "function_fields": list(FUNCTION_FIELDS),
-            "functions": functions,
-        }
-
-
-_MAPS = {}
-_LOCK = threading.Lock()
-
-
-def map_for(project):
-    with _LOCK:
-        if project.name not in _MAPS:
-            from . import decomp_api
-            api = decomp_api.api_for(project)
-            _MAPS[project.name] = ProgressMap(project, lambda: FunctionQueue(api.h))
-        return _MAPS[project.name]
-
-
-def handle(project, query):
-    if not (project.reverse or {}).get("facts"):
-        return 404, {"error": "this project has no [reverse] table"}
-    return 200, map_for(project).get(query.get("have"))
+"""The Logbook's progress map, in the style of decomp.dev: one unit per original source file, its functions and how
+far each has come, plus the headline share of game code decompiled and the Code and Data bars.
+
+GET /api/logbook/progress-map answers with one compact JSON document. It is computed once and kept until one of its
+inputs changes (main's HEAD, the Logbook's symbol table and statistics, the match, emulator and library results,
+facts, names, source files, globals, or a deferral note); `?have=<version>` answers {"version", "same": true} when
+the caller already holds the current one, so a page that polls does not download or re-lay-out anything.
+
+Definitions follow the Logbook exactly (open-seas tools/progress/progress.py, which writes stats.json):
+  - the symbol table is <[logbook] stats_repo>/re/symbols/pc102.csv: its category and status per function;
+  - counted ("game code") = category game and status not superseded or wontfix;
+  - decompiled = status decompiled, reimplemented or verified, so percent = decompiled game bytes / game bytes,
+    the Logbook's game_bytes_decompiled_percent.
+Decompiled functions are split further, best first: exact (byte match in the [reverse] match_results), verified
+(behaves the same in the emulator, diffrun_results, not 'weak'), readable. The rest: deferred (a live DEFERRED note),
+raw, and, not counted, unsorted, engine (Gamebryo, named only) and library (C runtime, D3DX9, platform wrappers).
+"results" says which commit the exact and verified tiers are from (harbor.results_refresh keeps them current).
+Without a stats_repo the categories come from names.csv / facts and readable from the src headers (the queue's view).
+
+Units, first that applies: the source file from the Xbox build's assert sites ([reverse] source_files); the D3DX9 block
+(platform/d3dx9) or the import-thunk stubs from library_ranges; the unit facts gives (engine/<class>,
+platform/<library>); otherwise
+<category> plus its 64 KB address block, labelled "(no source file)"."""
+import csv
+import hashlib
+import json
+import subprocess
+import threading
+import time
+from pathlib import Path
+
+from .queue import FunctionQueue
+
+STATUSES = ("exact", "verified", "readable", "deferred", "raw", "unsorted", "engine", "library")
+DECOMPILED = ("decompiled", "reimplemented", "verified")
+EXCLUDED = ("superseded", "wontfix")
+UNIT_FIELDS = ("name", "kind", "lo", "hi", "bytes", "counted", "exact", "verified", "readable", "functions")
+FUNCTION_FIELDS = ("unit", "address", "size", "status", "name")
+KEY_TTL = 15            # seconds a computed input key is trusted before the files and HEAD are looked at again
+
+
+def _int(text):
+    return int(str(text), 16) if str(text).lower().startswith("0x") else int(text)
+
+
+def _mtime(path):
+    try:
+        return Path(path).stat().st_mtime_ns if path else None
+    except OSError:
+        return None
+
+
+def _git(repo, *args):
+    try:
+        out = subprocess.run(["git", "-C", str(repo), *args], capture_output=True, text=True, timeout=20,
+                             encoding="utf-8", errors="replace")
+    except (OSError, subprocess.TimeoutExpired):
+        return ""
+    return out.stdout.strip() if out.returncode == 0 else ""
+
+
+def _utc(iso):
+    import datetime
+    try:
+        return datetime.datetime.fromisoformat(iso).astimezone(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+    except ValueError:
+        return None
+
+
+class ProgressMap:
+    def __init__(self, project, queue_factory):
+        self.project = project
+        self.queue_factory = queue_factory        # -> FunctionQueue on a connection this thread may use
+        self.lock = threading.Lock()
+        self.value = None                         # (key, document)
+        self.checked = (0.0, None)                # (monotonic time, key)
+        self.computations = 0
+
+    # ------------------------------------------------------------------ inputs
+    def _config(self):
+        return self.project.reverse or {}, getattr(self.project, "logbook", None) or {}
+
+    def _symbols_path(self):
+        _, lb = self._config()
+        return Path(lb["stats_repo"]) / "re" / "symbols" / "pc102.csv" if lb.get("stats_repo") else None
+
+    def _decomp(self):
+        rev, lb = self._config()
+        if lb.get("decomp"):
+            return Path(lb["decomp"])
+        return Path(rev["annotations"]).parent if rev.get("annotations") else None
+
+    def _logbook_dir(self):
+        from . import logbook
+        return logbook.folder(self.project)
+
+    def key(self, q):
+        rev, _ = self._config()
+        decomp = self._decomp()
+        head = _git(decomp, "rev-parse", "HEAD") if decomp and (decomp / ".git").exists() else ""
+        paths = [self._symbols_path(), self._logbook_dir() / "stats.json", self._logbook_dir() / "history.json",
+                 self._logbook_dir() / "results.json"]
+        paths += [rev.get(k) for k in ("facts", "match_results", "diffrun_results", "library_results",
+                                       "library_ranges", "source_files")]
+        if rev.get("annotations"):
+            paths += [Path(rev["annotations"]) / "names.csv", Path(rev["annotations"]) / "globals.csv"]
+        try:
+            note = q.h.conn.execute("SELECT MAX(id) FROM notes").fetchone()[0]
+        except Exception:
+            note = None
+        raw = json.dumps([head, note, [str(p) for p in paths], [_mtime(p) for p in paths]])
+        return hashlib.sha1(raw.encode("utf-8")).hexdigest()[:16], head
+
+    # ------------------------------------------------------------------ answer
+    def get(self, have=None):
+        with self.lock:
+            q = self.queue_factory()
+            now = time.monotonic()
+            if self.value and now - self.checked[0] < KEY_TTL:
+                key = self.checked[1]
+            else:
+                key, head = self.key(q)
+                self.checked = (now, key)
+                if not self.value or self.value[0] != key:
+                    document = self.compute(q, head)
+                    if document is None and self.value:          # the symbol table is being rewritten: keep the last
+                        document = self.value[1]
+                    elif document is None:
+                        document = self.compute(q, head, use_symbols=False)
+                    document["version"] = key
+                    self.value = (key, document)
+                    self.computations += 1
+            if have and have == self.value[0]:
+                return {"version": have, "same": True}
+            return self.value[1]
+
+    # ------------------------------------------------------------------ the computation
+    def _symbols(self, expected):
+        """pc102.csv as {address: (category, status, name)}, or None while it is missing or half written."""
+        path = self._symbols_path()
+        if not path or not path.exists():
+            return None
+        try:
+            with path.open(encoding="utf-8", newline="") as stream:
+                rows = {_int(r["address"]): (r["category"], r["status"], r.get("name") or "")
+                        for r in csv.DictReader(stream)}
+        except (OSError, ValueError, KeyError):
+            return None
+        return rows if len(rows) >= 0.98 * expected else None
+
+    def _source_files(self, path):
+        out = {}
+        if path and Path(path).exists():
+            with open(path, encoding="utf-8", newline="") as stream:
+                for row in csv.DictReader(stream):
+                    if row.get("source_file"):
+                        out[_int(row["pc_address"])] = row["source_file"]
+        return out
+
+    def _all_ranges(self, path):
+        """[(lo, hi, label)] of the library_ranges csv (the D3DX9 block and the compiler's stubs)."""
+        out = []
+        if path and Path(path).exists():
+            with open(path, encoding="utf-8", newline="") as stream:
+                for r in csv.DictReader(stream):
+                    verdict = r.get("verdict", "")
+                    label = "platform/d3dx9" if verdict.startswith("D3DX") else \
+                        "Import thunks and unwind funclets" if "stubs" in verdict else None
+                    if label:
+                        out.append((_int(r["start"]), _int(r["end"]), label))
+        return out
+
+    def _data(self, facts, rev):
+        """Named globals against the distinct .data/.rdata addresses the code touches (needs sections.json)."""
+        facts_path = Path(rev["facts"]) if rev.get("facts") else None
+        sections_path = facts_path.parent / "sections.json" if facts_path else None
+        globals_path = Path(rev["annotations"]) / "globals.csv" if rev.get("annotations") else None
+        if not sections_path or not sections_path.exists() or not globals_path or not globals_path.exists():
+            return None
+        spans = [(s["start"], s["end"]) for s in json.loads(sections_path.read_text(encoding="utf-8"))
+                 if s.get("name") in (".data", ".rdata")]
+        touched = set()
+        for d in facts.values():
+            for ref in d.get("data", []):
+                a = _int(ref)
+                if any(lo <= a < hi for lo, hi in spans):
+                    touched.add(a)
+        with globals_path.open(encoding="utf-8", newline="") as stream:
+            named = {_int(r["address"]) for r in csv.DictReader(line for line in stream if not line.startswith("#"))
+                     if r.get("address") and r.get("name")}
+        if not touched:
+            return None
+        return {"addresses": len(touched), "named": len(touched & named), "globals": len(named),
+                "bytes": sum(hi - lo for lo, hi in spans)}
+
+    def compute(self, q, head, use_symbols=True):
+        rev, _ = self._config()
+        facts = q.r._facts()
+        symbols = self._symbols(len(facts)) if use_symbols else None
+        if use_symbols and self._symbols_path() and symbols is None and self._symbols_path().exists():
+            return None
+        names = q.r._csv("names.csv")
+        statuses = None if symbols else q.statuses()
+        exact = {a for a, row in q._results("match_results").items()
+                 if str(row.get("status")) in ("match", "exact", "exact-relaxed")}
+        verified = {a for a, row in q._results("diffrun_results").items()
+                    if str(row.get("verdict")) == "same" and not row.get("weak")}
+        library = set(q._results("library_results"))
+        ranges = self._all_ranges(rev.get("library_ranges"))
+        lib_ranges = [(lo, hi) for lo, hi, _ in ranges]
+        deferred = q.deferrals()
+        sources = self._source_files(rev.get("source_files"))
+
+        units, unit_index, functions = [], {}, []
+        game_functions = game_done = 0
+        tiers = {t: {"functions": 0, "bytes": 0, "game_functions": 0, "game_bytes": 0} for t in STATUSES}
+        for addr in sorted(facts):
+            d = facts[addr]
+            size = int(d.get("size") or 0)
+            if symbols and addr in symbols:
+                category, status, sym_name = symbols[addr]
+            else:
+                category = q._category(addr)
+                status = "decompiled" if (statuses or {}).get(addr, ("raw",))[0] == "readable" else "classified"
+                sym_name = ""
+            row = names.get(addr)
+            name = (row[1] if row and len(row) > 1 and row[1] else "") or sym_name or d.get("name") or f"FUN_{addr:06X}"
+            counted = category == "game" and status not in EXCLUDED
+            done = status in DECOMPILED
+            is_lib = q.is_library(addr, library, lib_ranges) or category in ("library", "platform", "crt")
+            if counted or (done and not is_lib) or (category == "unsorted" and not is_lib):
+                tier = "exact" if done and addr in exact else "verified" if done and addr in verified else \
+                    "readable" if done else "deferred" if addr in deferred else \
+                    "raw" if counted else "unsorted"
+            elif is_lib:
+                tier = "library"
+            else:
+                tier = "engine"
+
+            source = sources.get(addr)
+            in_range = next((label for lo, hi, label in ranges if lo <= addr < hi), None)
+            if source:
+                unit = source
+            elif in_range:
+                unit = in_range
+            elif d.get("unit"):
+                unit = d["unit"]
+            else:
+                unit = f"{category} · 0x{addr & ~0xFFFF:06X} block (no source file)"
+            if unit not in unit_index:
+                unit_index[unit] = len(units)
+                units.append({"name": unit, "lo": addr, "hi": addr + size, "bytes": 0, "counted": 0, "exact": 0,
+                              "verified": 0, "readable": 0, "functions": 0, "kinds": {}})
+            u = units[unit_index[unit]]
+            u["hi"] = max(u["hi"], addr + size)
+            u["bytes"] += size
+            u["functions"] += 1
+            kind = "game" if counted else "library" if tier == "library" else "engine" if category == "engine" \
+                else "unsorted" if category == "unsorted" else "library"
+            u["kinds"][kind] = u["kinds"].get(kind, 0) + size
+            tiers[tier]["functions"] += 1
+            tiers[tier]["bytes"] += size
+            if counted:                                       # the Code bar's game functions, split by tier
+                tiers[tier]["game_functions"] += 1
+                tiers[tier]["game_bytes"] += size
+            if counted:
+                game_functions += 1
+                game_done += done
+                u["counted"] += size
+                if tier in ("exact", "verified", "readable"):
+                    u[tier] += size
+            functions.append([unit_index[unit], addr, size, STATUSES.index(tier), name])
+
+        for u in units:
+            kinds = u.pop("kinds")
+            u["kind"] = max(kinds, key=kinds.get) if kinds else "unsorted"      # what most of its bytes are
+
+        code = {k: sum(u[k] for u in units) for k in ("exact", "verified", "readable")}
+        code["bytes"] = sum(u["counted"] for u in units)
+        code["functions"], code["decompiled_functions"] = game_functions, game_done
+        percent = round(100 * (code["exact"] + code["verified"] + code["readable"]) / code["bytes"], 2) \
+            if code["bytes"] else 0.0
+
+        from . import logbook
+        stats = logbook._read(self.project, "stats.json", None) or {}
+        history = sorted(logbook._read(self.project, "history.json", []), key=lambda p: p.get("at", ""))
+        decomp = self._decomp()
+        commit = {"sha": stats.get("commit"), "at": None}
+        if commit["sha"] and decomp:
+            commit["at"] = _utc(_git(decomp, "log", "-1", "--format=%cI", commit["sha"])) or None
+        main = {"sha": head[:7] if head else None,
+                "at": _utc(_git(decomp, "log", "-1", "--format=%cI", head)) if head else None}
+        from . import results_refresh
+        return {
+            "project": self.project.title or self.project.name,
+            "results": results_refresh.freshness(self.project),
+            "source": "logbook symbol table" if symbols else "src headers and names.csv",
+            "commit": commit, "main": main, "updated_at": stats.get("updated_at"),
+            "percent": percent, "logbook_percent": stats.get("game_bytes_decompiled_percent"),
+            "code": code, "data": self._data(facts, rev), "tiers": tiers,
+            # every function in the executable (game, engine, platform, CRT), so engine work shows too
+            "all_code": {"bytes": stats.get("bytes"), "functions": stats.get("functions"),
+                         "decompiled_bytes": ((stats.get("by_status") or {}).get("decompiled") or {}).get("bytes"),
+                         "decompiled_functions": ((stats.get("by_status") or {}).get("decompiled") or {}).get("functions")},
+            "history": [[p.get("at"), p.get("decompiled"), p.get("commit"), p.get("game_bytes")]
+                        for p in history if p.get("decompiled") is not None],
+            "statuses": list(STATUSES),
+            "unit_fields": list(UNIT_FIELDS),
+            "units": [[u[k] for k in UNIT_FIELDS] for u in units],
+            "function_fields": list(FUNCTION_FIELDS),
+            "functions": functions,
+        }
+
+
+_MAPS = {}
+_LOCK = threading.Lock()
+
+
+def map_for(project):
+    with _LOCK:
+        if project.name not in _MAPS:
+            from . import decomp_api
+            api = decomp_api.api_for(project)
+            _MAPS[project.name] = ProgressMap(project, lambda: FunctionQueue(api.h))
+        return _MAPS[project.name]
+
+
+def handle(project, query):
+    if not (project.reverse or {}).get("facts"):
+        return 404, {"error": "this project has no [reverse] table"}
+    return 200, map_for(project).get(query.get("have"))

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
 
 

harbor/queue.py

@@ -207,6 +207,10 @@ class FunctionQueue:
             head = self.r.read(addr, 10) if int(facts[addr].get("size") or 0) == 10 else b""
             if head[:1] and 0xB8 <= head[0] <= 0xBF and head[5:6] == b"\xe9":
                 continue        # mov reg, imm32; jmp: an at-exit destructor thunk or an SEH handler stub (compiler's)
+            first = self.r.read(addr, 3)
+            if len(first) == 3 and first[0] in (0x8B, 0x8D, 0xFF, 0x83, 0xC7, 0x89) and (first[1] & 0xC7) == 0x45 \
+                    and first[2] >= 0x80:
+                continue        # opens on the parent's [ebp-x] with no frame: a catch or unwind funclet (ruling #12777)
             score, reasons = self.score(addr, facts, named, rejected, similar_readable, with_convention=False)
             ranked.append((score, addr, reasons))
         ranked.sort(reverse=True)

harbor/results_refresh.py

@@ -13,6 +13,7 @@ Configured per project in the [logbook] table (main is [logbook] decomp):
     behaviour_every_hours = 12
     behaviour_window = "01-06"      from 01:00 to 06:00 local time; "" for any hour
 
+No run starts while the lander runs its gates ([logbook] lander_lock, harbor.lander_gate).
 One run at a time (the server runs one thread for every project), at the lowest process priority, into a .partial file
 that replaces the results only once it is complete. The match run holds the same per-checkout lock as re_match_program,
 so it waits its turn behind any other whole-program build of main.
@@ -31,7 +32,7 @@ import time
 import traceback
 from pathlib import Path
 
-from . import logbook
+from . import lander_gate, logbook
 
 JOBS = ("match", "behaviour")
 RESULTS_KEY = {"match": "match_results", "behaviour": "diffrun_results"}
@@ -180,14 +181,7 @@ def _spawn(argv, cwd, stdout, stderr):
 
 
 def _kill_tree(pid):
-    if os.name == "nt":
-        subprocess.run(["taskkill", "/PID", str(pid), "/T", "/F"], capture_output=True,
-                       creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
-    else:
-        try:
-            os.kill(pid, 9)
-        except OSError:
-            pass
+    lander_gate.kill_tree(pid)
 
 
 def _alive(pid):
@@ -205,12 +199,12 @@ def _paths(project, job):
     return paths
 
 
-def _argv(project, job, out):
+def _argv(project, job, out, decomp=None):
     need = configured(project, job)
     if job == "match":
         script = Path(need["script"])
         return [sys.executable, "-B", str(script), "check", "--build", "--decomp",
-                str(Path(project.logbook["decomp"])), "--json", str(out)], script.parent
+                str(Path(decomp or project.logbook["decomp"])), "--json", str(out)], script.parent
     command = [str(c) for c in need["command"]]
     return command + ["--readable", "--jobs", "4", "--json", str(out)], Path(command[-1]).parent
 
@@ -260,7 +254,12 @@ def _complete(project, state, run, now, returncode=None):
 
 def run_job(project, state, job, head, now, reason=""):
     paths = _paths(project, job)
-    argv, cwd = _argv(project, job, paths["partial"])
+    decomp = None
+    if job == "match" and project.logbook.get("results_snapshot") and head:
+        # a clean checkout of exactly `head`: the live checkout is the lander's and changes under a running build
+        from .logbook_auto import snapshot_at
+        decomp = snapshot_at({"snapshot": project.logbook["results_snapshot"]}, project.logbook["decomp"], head)
+    argv, cwd = _argv(project, job, paths["partial"], decomp)
     for stale in [paths["partial"]] + ([paths["text_partial"]] if job == "behaviour" else []):
         if stale.exists():
             stale.write_bytes(b"")                     # a leftover of an interrupted run is never taken for this one
@@ -310,6 +309,9 @@ def tick(project, now=None, hour=None):
     job, why = choose(settings(project), state, head, now, hour, requested, available)
     if not job:
         return {"idle": why, "head": head[:8]}
+    busy = lander_gate.lander_busy(project, now)
+    if busy:                                             # a request is kept for the next look
+        return {"waiting": busy, "job": job}
     if job == "match":
         from .util import HarborError
         try:

harbor/reverse.py

@@ -7,6 +7,8 @@ Configured per project in a [reverse] table:
     imports = 'R:/.../imports.json'   optional: [{module, name, slot}]
     annotations = 'R:/.../annotations' names.csv, functions.csv, globals.csv, prototypes.h
     src = 'R:/.../src'                optional: the decompiled C files (<ADDR>_<Name>.c)
+    sections = 'R:/.../sections.json' optional: [{name, start, end}] (default: sections.json beside the image),
+                                      used to find the vtables and pointer tables a function sits in
 
 Every answer comes from those files, read fresh when they change, so the agents see the team's current names.
 Nothing here writes anything."""
@@ -42,6 +44,7 @@ class Reverse:
         self.conventions_path = Path(config["conventions"]) if config.get("conventions") else None
         self.source_files = Path(config["source_files"]) if config.get("source_files") else None
         self.mac_names = Path(config["mac_names"]) if config.get("mac_names") else None
+        self.sections_path = Path(config["sections"]) if config.get("sections") else None
         self._cache = {}
 
     # ----------------------------------------------------------------- loading (cached by file mtime)
@@ -491,6 +494,176 @@ class Reverse:
                 break
         return out
 
+    # ----------------------------------------------------------------- one-call summary
+    def _src_index(self):
+        """address -> path of src/**/<ADDR>_<Name>.c (one directory walk, reused for 30 seconds)."""
+        import time
+        hit = self._cache.get("src_index")
+        if hit and time.monotonic() - hit[0] < 30:
+            return hit[1]
+        out = {}
+        if self.src and self.src.exists():
+            for folder, _, files in os.walk(self.src):
+                for name in files:
+                    m = re.match(r"([0-9A-Fa-f]{6,8})_.*\.c$", name)
+                    if m:
+                        out.setdefault(int(m.group(1), 16), Path(folder) / name)
+        self._cache["src_index"] = (time.monotonic(), out)
+        return out
+
+    def status_of(self, addr):
+        """The decompiled file's status on main ('readable', 'raw', ...), or None when there is no file."""
+        path = self._src_index().get(addr)
+        if not path:
+            return None
+        try:
+            with path.open(encoding="utf-8", errors="replace") as f:
+                head = [next(f, "") for _ in range(8)]
+        except OSError:
+            return None
+        return next((l.split("status:", 1)[1].strip() for l in head if "status:" in l), "raw")
+
+    def _sections(self):
+        """[(name, start, end)] from sections.json beside the image (or [reverse] sections), else []."""
+        path = self.sections_path or self.image.with_name("sections.json")
+        return self._load("sections", path, lambda p: [
+            (s["name"], _int(s["start"]), _int(s["end"])) for s in json.loads(p.read_text(encoding="utf-8"))] if p else [])
+
+    def _code_data_refs(self):
+        facts = self._facts()
+        key = ("data_refs", id(facts))
+        if key not in self._cache:
+            self._cache[key] = {_int(a) for d in facts.values() for a in d.get("data", [])}
+        return self._cache[key]
+
+    def _mac_vtables(self):
+        return self._load("mac_vtables", self.mac_names.with_name("vtables.csv") if self.mac_names else None,
+                          lambda p: {int(r["pc_vtable"], 16): r.get("class") for r in
+                                     csv.DictReader(p.read_text(encoding="utf-8").splitlines())
+                                     if (r.get("pc_vtable") or "").startswith("0x")} if p else {})
+
+    def vtable_slots(self, addr):
+        """Where the function's address sits in a table of code pointers outside .text: the table start (walked back
+        over function addresses to the entry the code references), the slot index, and what the table is."""
+        data = self._bytes()
+        facts = self._facts()
+        sections = self._sections()
+        text = next(((lo, hi) for name, lo, hi in sections if name == ".text"), None)
+        tables = [(lo, hi) for name, lo, hi in sections if name in (".rdata", ".data")]
+        if not tables:                                     # no section list: everything past the last function
+            end = max((a + int(d.get("size") or 0) for a, d in facts.items()), default=self.base)
+            tables = [(end, self.base + len(data))]
+        refs, mac = self._code_data_refs(), self._mac_vtables()
+        is_code = (lambda v: v in facts) if facts else (lambda v: text and text[0] <= v < text[1])
+        needle = struct.pack("<I", addr)
+        out = []
+        for lo, hi in tables:
+            pos = data.find(needle, max(0, lo - self.base), hi - self.base)
+            while pos >= 0 and len(out) < 40:
+                if pos % 4 == 0:
+                    start = pos
+                    while start - 4 >= lo - self.base and (start + self.base) not in refs and \
+                            (start + self.base) not in mac and \
+                            is_code(struct.unpack_from("<I", data, start - 4)[0]) and pos - start < 4 * 400:
+                        start -= 4
+                    table = start + self.base
+                    slot = (pos - start) // 4
+                    name = self.name_of(table)
+                    kind = "vtable" if table in refs or table in mac or (name and "vtable" in name.lower()) \
+                        else "pointer table"
+                    out.append({"table": f"0x{table:X}", "slot": slot, "offset": f"0x{slot * 4:X}", "kind": kind,
+                                **({"table_name": name} if name else {}),
+                                **({"mac_class": mac[table]} if table in mac else {})})
+                pos = data.find(needle, pos + 1, hi - self.base)
+        return out
+
+    def summary(self, target):
+        """Everything a worker needs before starting on a function, in one call: name, prototype, size, convention,
+        whether it is readable on main, callers and callees (named, with their status on main), globals and strings,
+        the vtable slots it fills, and the Mac/Xbox name hints."""
+        addr = self.resolve(target)
+        func = self.function_at(addr)
+        if not func:
+            raise HarborError(f"0x{addr:X} is not inside a known function")
+        start, size, name = func
+        d = self._facts()[start]
+
+        def fn(a):
+            status = self.status_of(a)
+            return {"address": f"0x{a:X}", "name": self.name_of(a), "status": status or "no file",
+                    "readable": status == "readable"}
+
+        out = {"address": f"0x{start:X}", "name": name, "size": size, "end": f"0x{start + size:X}"}
+        if addr != start:
+            out["asked"] = f"0x{addr:X} (+0x{addr - start:X})"
+        proto = self._prototypes().get(name) if name else None
+        out["prototype"] = proto
+        conv = self.convention(start)
+        out["convention"] = conv["summary"] if conv else None
+        if conv:
+            out["registers_in"], out["ret_pops"] = conv["registers_in"], conv["ret_pops"]
+        status = self.status_of(start)
+        out["status"] = status or "no file"
+        out["readable"] = status == "readable"
+        path = self._src_index().get(start)
+        if path and self.src:
+            out["file"] = str(path.relative_to(self.src.parent)).replace(os.sep, "/")
+        row = self._csv("names.csv").get(start)
+        if row and len(row) > 3 and row[3]:
+            out["note"] = row[3] if len(row[3]) < 600 else row[3][:600] + "..."
+        callers = sorted(set(self._callers().get(start, [])))
+        callees = list(dict.fromkeys(_int(a) for a in d.get("calls", [])))
+        out["callers"] = [fn(a) for a in callers]
+        out["callees"] = [fn(a) for a in callees]
+        if d.get("imports"):
+            out["imports"] = d["imports"]
+        glob = self._csv("globals.csv")
+        out["globals"] = [{"address": f"0x{a:X}", "name": self.name_of(a),
+                           **({"type": glob[a][2]} if a in glob and len(glob[a]) > 2 and glob[a][2] else {})}
+                          for a in dict.fromkeys(_int(x) for x in d.get("data", []))]
+        out["strings"] = d.get("strings", [])
+        out["vtable_slots"] = self.vtable_slots(start)
+        out["hints"] = self.origins(start)
+        out["counts"] = {"callers": len(callers), "callers_readable": sum(c["readable"] for c in out["callers"]),
+                         "callees": len(callees), "callees_readable": sum(c["readable"] for c in out["callees"])}
+        return out
+
+
+def format_summary(s):
+    """Reverse.summary() as text for the command line."""
+    mark = lambda f: f"{f['address']} {f['name'] or '(unnamed)'} [{'readable' if f['readable'] else f['status']}]"
+    lines = [f"{s['name'] or '(unnamed)'} {s['address']}..{s['end']} ({s['size']} bytes)"
+             + (f", asked {s['asked']}" if s.get("asked") else ""),
+             f"status on main: {s['status']}" + (f" ({s['file']})" if s.get("file") else ""),
+             f"convention: {s['convention'] or 'unknown'}",
+             "prototype: " + (s["prototype"] or "none in prototypes.h")]
+    c = s["counts"]
+    lines.append(f"callers ({c['callers']}, {c['callers_readable']} readable):")
+    lines += [f"  {mark(f)}" for f in s["callers"]] or ["  none"]
+    lines.append(f"callees ({c['callees']}, {c['callees_readable']} readable):")
+    lines += [f"  {mark(f)}" for f in s["callees"]] or ["  none"]
+    if s.get("imports"):
+        lines.append("imports: " + ", ".join(s["imports"]))
+    lines.append(f"globals ({len(s['globals'])}):")
+    lines += [f"  {g['address']} {g['name'] or '(unnamed)'}" + (f" : {g['type']}" if g.get("type") else "")
+              for g in s["globals"]] or ["  none"]
+    if s["strings"]:
+        lines.append("strings: " + ", ".join(repr(x) for x in s["strings"]))
+    lines.append("vtable slots:" + ("" if s["vtable_slots"] else " none"))
+    lines += [f"  {v['kind']} {v['table']}" + (f" {v['table_name']}" if v.get("table_name") else "")
+              + (f" (Mac class {v['mac_class']})" if v.get("mac_class") else "") + f" slot {v['slot']} (+{v['offset']})"
+              for v in s["vtable_slots"]]
+    h = s["hints"]
+    if h.get("mac_name"):
+        lines.append(f"Mac name: {h['mac_name'].get('demangled')} ({h['mac_name'].get('confidence')})")
+    if h.get("source_file"):
+        sf = h["source_file"]
+        lines.append(f"Xbox source file: {sf.get('source_file')}" + (f" lines {sf['lines']}" if sf.get("lines") else "")
+                     + f" ({sf.get('confidence')})")
+    if not h:
+        lines.append("Mac/Xbox hints: none")
+    return "\n".join(lines)
+
 
 class Ghidra:
     """A small client for the shared, read-only Ghidra MCP server (GhidrAssistMCP over streamable HTTP), so every
@@ -572,6 +745,63 @@ class Ghidra:
         return re.sub(r"^\[Context\] Operating on: [^\n]*\n+", "", text)
 
 
+def _hexdump_bytes(text):
+    """The bytes of a GhidrAssistMCP get_data_at hexdump ('401000  53 55 8b ...  |SU.|')."""
+    out = bytearray()
+    for line in text.splitlines():
+        m = re.match(r"^\s*[0-9A-Fa-f]{4,16}\s{2}((?:[0-9A-Fa-f]{2}\s{1,2})+)", line + " ")
+        if m:
+            out += bytes.fromhex("".join(m.group(1).split()))
+    return bytes(out)
+
+
+def ghidra_check(config, samples=16, sample_len=128):
+    """Is the shared Ghidra server serving our program? Read-only: compares the program's address range with the
+    project image (base and size), the SHA-256 of the file Ghidra imported (when that path is readable here) with
+    the image's, and sample bytes read through Ghidra (get_data_at) with the image at spread-out addresses."""
+    import hashlib
+    reverse = Reverse(config)
+    image = reverse._bytes()
+    if not image:
+        raise HarborError(f"the project image {reverse.image} is missing")
+    base = reverse.base
+    g = Ghidra(config)
+    info = g.call("get_binary_info", {})
+    binaries = g.call("list_binaries", {})
+    field = lambda label, text: (re.search(rf"^\s*{label}:\s*(.+?)\s*$", text, re.M) or [None, None])[1]
+    lo, hi = field("Min Address", info), field("Max Address", info)
+    checks = []
+    want_lo, want_hi = base, base + len(image) - 1
+    got = (int(lo, 16), int(hi, 16)) if lo and hi else None
+    checks.append({"check": "address range", "ok": got == (want_lo, want_hi),
+                   "expected": f"0x{want_lo:X}..0x{want_hi:X} ({len(image)} bytes)",
+                   "ghidra": f"0x{got[0]:X}..0x{got[1]:X} ({got[1] - got[0] + 1} bytes)" if got else info.strip()[:200]})
+    sha = hashlib.sha256(image).hexdigest()
+    exe = field("Executable Path", binaries)
+    local = Path(exe[1:] if exe and re.match(r"^/[A-Za-z]:/", exe) else exe) if exe else None
+    if local and local.is_file():
+        theirs = hashlib.sha256(local.read_bytes()).hexdigest()
+        checks.append({"check": "sha256 of the imported file", "ok": theirs == sha, "expected": sha,
+                       "ghidra": theirs, "file": str(local)})
+    else:
+        checks.append({"check": "sha256 of the imported file", "ok": None, "expected": sha,
+                       "ghidra": f"not readable here ({exe or 'no path reported'})"})
+    step = max(sample_len, (len(image) - sample_len) // max(1, samples - 1)) & ~0xF
+    bad = []
+    for i in range(samples):
+        addr = base + min(i * step, len(image) - sample_len)
+        theirs = _hexdump_bytes(g.call("get_data_at", {"address": f"0x{addr:X}", "len": sample_len}))
+        ours = image[addr - base:addr - base + sample_len]
+        if theirs != ours:
+            bad.append(f"0x{addr:X}")
+    checks.append({"check": f"{samples} byte samples of {sample_len}", "ok": not bad,
+                   **({"differ_at": bad} if bad else {})})
+    ok = all(c["ok"] is not False for c in checks)
+    return {"ok": ok, "program": field("Name", info), "image": str(reverse.image), "image_sha256": sha,
+            "image_base": f"0x{base:X}", "image_size": len(image), "functions_in_ghidra": field("Function Count", info),
+            "checks": checks}
+
+
 class Debugger:
     """A per-agent x64dbg session through the x64dbg-automate MCP server (stdio), started on first use and kept for
     the life of this Harbor process, so breakpoints and the paused game survive between calls. Configured by

tests/test_logbook.py

@@ -58,6 +58,54 @@ class LogbookAuto(HarborCase):
         self.assertTrue(entry["at"].endswith("Z"))
         self.assertFalse(logbook_auto.refresh(self.h.project)["changed"])
 
+    def test_waits_for_the_lander_and_a_failed_stats_run_logs_nothing_twice(self):
+        import os
+        import time
+        from unittest import mock
+        logbook_auto._save_state(self.h.project, {"last_head": self.first})
+        lock = self.dir / "lander-busy.lock"
+        lock.write_text("x", encoding="utf-8")
+        self.h.project.logbook.update(lander_lock=str(lock), stats_repo=str(self.dir / "stats"))
+        out = logbook_auto.refresh(self.h.project)
+        self.assertFalse(out["changed"])
+        self.assertIn("lander", out["waiting"])
+        self.assertEqual(logbook.read_all(self.h.project)["log"], [])
+        old = time.time() - 61 * 60
+        os.utime(lock, (old, old))                                              # stale: ignored
+        with mock.patch.object(logbook_auto, "rebuild_stats", side_effect=RuntimeError("timed out")):
+            out = logbook_auto.refresh(self.h.project)
+            self.assertEqual((out["entries"], out["stats_error"]), (1, "timed out"))
+            again = logbook_auto.refresh(self.h.project)                       # the statistics are retried ...
+            self.assertIn("stats_error", again)
+            self.assertNotIn("entries", again)                                 # ... but the commit is not relogged
+        self.assertEqual(len(logbook.read_all(self.h.project)["log"]), 1)
+
+
+class LanderGate(unittest.TestCase):
+    def test_run_low_lowers_priority_and_ends_the_tree_on_timeout(self):
+        import os
+        import sys
+        import time
+        from harbor import lander_gate
+        script = "import subprocess, sys, time; subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(60)']);" \
+                 " print('ok', flush=True); time.sleep(60)"
+        from unittest import mock
+        seen = {}
+        real = subprocess.Popen
+
+        def popen(argv, **kw):
+            seen.update(kw)
+            return real([sys.executable, "-c", "pass"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+
+        with mock.patch.object(lander_gate.subprocess, "Popen", popen):
+            self.assertEqual(lander_gate.run_low(["x"], 60).returncode, 0)
+        if os.name == "nt":
+            self.assertTrue(seen["creationflags"] & subprocess.BELOW_NORMAL_PRIORITY_CLASS)
+        start = time.monotonic()
+        with self.assertRaises(subprocess.TimeoutExpired):
+            lander_gate.run_low([sys.executable, "-c", script], 3)
+        self.assertLess(time.monotonic() - start, 45)                         # the grandchild did not hold the pipes
+
 
 class ProgressMapTests(HarborCase):
     """GET /api/logbook/progress-map: units, tiers and the headline, by the Logbook's own definitions."""

tests/test_results_refresh.py

@@ -208,6 +208,19 @@ class Refresher(HarborCase):
         rr.save_state(self.h.project, st)
         self.assertIsNone(rr.freshness(self.h.project)["match"]["sha"])
 
+    def test_waits_while_the_lander_runs_its_gates(self):
+        lock = self.dir / "lander-busy.lock"
+        self.h.project.logbook["lander_lock"] = str(lock)
+        lock.write_text("lander", encoding="utf-8")
+        os.utime(lock, (T0 - 300, T0 - 300))
+        rr.request(self.h.project, "match")
+        out = rr.tick(self.h.project, now=T0, hour=12)
+        self.assertIn("lander", out["waiting"])
+        self.assertEqual(self.calls, [])
+        self.assertEqual(rr._requested(self.h.project), {"match"})             # kept for the next look
+        self.assertTrue(rr.tick(self.h.project, now=T0 + 61 * 60, hour=12)["ok"])   # a lock over an hour old is stale
+        self.assertEqual(len(self.calls), 1)
+
     def test_runs_at_idle_priority(self):
         seen = {}
 

tests/test_reverse.py

@@ -58,6 +58,105 @@ class ReverseTools(unittest.TestCase):
         with self.assertRaises(HarborError):
             Reverse({})
 
+    def test_summary_names_status_vtables_and_hints(self):
+        image = bytearray((self.d / "image.bin").read_bytes())
+        image[0xC0:0xC8] = struct.pack("<II", 0x401000, 0x401040)          # 0x4010C0: a two-slot vtable
+        (self.d / "image.bin").write_bytes(bytes(image))
+        (self.d / "sections.json").write_text(json.dumps([
+            {"name": ".text", "start": 0x401000, "end": 0x401080}, {"name": ".rdata", "start": 0x4010B0,
+                                                                     "end": 0x401100}]), encoding="utf-8")
+        facts = [json.loads(l) for l in (self.d / "facts.jsonl").read_text(encoding="utf-8").splitlines()]
+        facts[1]["data"] = ["0x4010C0"]                                    # the constructor installs it
+        facts[0]["strings"] = [".cue"]
+        (self.d / "facts.jsonl").write_text("\n".join(json.dumps(f) for f in facts), encoding="utf-8")
+        src = self.d / "src" / "game"
+        src.mkdir(parents=True)
+        (src / "401040_Helper.c").write_text("/* Helper\n * status: readable\n */\nvoid Helper(void) {}\n",
+                                             encoding="utf-8")
+        (self.d / "mac.csv").write_text("pc_address,demangled,confidence\n0x00401000,Level::Get(),high\n",
+                                        encoding="utf-8")
+        r = Reverse({"image": str(self.d / "image.bin"), "base": "0x401000", "facts": str(self.d / "facts.jsonl"),
+                     "annotations": str(self.d / "annotations"), "src": str(self.d / "src"),
+                     "mac_names": str(self.d / "mac.csv")})
+        s = r.summary("GetLevel")
+        self.assertEqual((s["name"], s["size"], s["status"], s["readable"]), ("GetLevel", 18, "no file", False))
+        self.assertIn("plain ret", s["prototype"])
+        self.assertEqual(s["callees"], [{"address": "0x401040", "name": "Helper", "status": "readable",
+                                         "readable": True}])
+        self.assertEqual(s["globals"], [{"address": "0x401080", "name": "g_root", "type": "int"}])
+        self.assertEqual(s["strings"], [".cue"])
+        self.assertEqual(s["vtable_slots"], [{"table": "0x4010C0", "slot": 0, "offset": "0x0", "kind": "vtable"}])
+        self.assertEqual(s["hints"]["mac_name"]["demangled"], "Level::Get()")
+        h = r.summary("0x401040")
+        self.assertTrue(h["readable"])
+        self.assertEqual([c["name"] for c in h["callers"]], ["GetLevel"])
+        self.assertEqual(h["vtable_slots"][0]["slot"], 1)
+        from harbor.reverse import format_summary
+        self.assertIn("vtable 0x4010C0 slot 1 (+0x4)", format_summary(h))
+
+
+class GhidraCheck(unittest.TestCase):
+    """ghidra_check compares the served program with the project image: range, SHA-256 of the file, byte samples."""
+
+    def serve(self, lo, hi, path, image):
+        import http.server
+        import threading
+
+        class Fake(http.server.BaseHTTPRequestHandler):
+            def log_message(self, *a):
+                pass
+
+            def do_POST(self):
+                body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
+                if "id" not in body:
+                    self.send_response(202)
+                    self.end_headers()
+                    return
+                params = body.get("params", {})
+                name, args = params.get("name"), params.get("arguments", {})
+                text = ""
+                if name == "get_binary_info":
+                    text = f"Program Information:\nName: image.bin\nImage Base: 00000000\nMin Address: {lo:08x}\n" \
+                           f"Max Address: {hi:08x}\nFunction Count: 2\n"
+                elif name == "list_binaries":
+                    text = f"Open Programs in Ghidra:\n\n1. image.bin [ACTIVE]\n   Executable Path: /{path}\n"
+                elif name == "get_data_at":
+                    a, n = int(args["address"], 16), int(args["len"])
+                    chunk = image[a - 0x401000:a - 0x401000 + n]
+                    rows = [f"{a + i:x}  " + " ".join(f"{b:02x}" for b in chunk[i:i + 16]) + "  |....|"
+                            for i in range(0, len(chunk), 16)]
+                    text = f"Hexdump at {a:08x} ({n} bytes):\n\n" + "\n".join(rows) + "\n"
+                result = {"content": [{"type": "text", "text": text}]} if body["method"] == "tools/call" else {}
+                data = json.dumps({"jsonrpc": "2.0", "id": body["id"], "result": result}).encode()
+                self.send_response(200)
+                self.send_header("Content-Type", "application/json")
+                self.send_header("Content-Length", str(len(data)))
+                self.end_headers()
+                self.wfile.write(data)
+
+        server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Fake)
+        threading.Thread(target=server.serve_forever, daemon=True).start()
+        self.addCleanup(server.server_close)
+        self.addCleanup(server.shutdown)
+        return f"http://127.0.0.1:{server.server_port}/mcp"
+
+    def test_match_and_mismatch(self):
+        from harbor.reverse import ghidra_check
+        d = temp_dir()
+        self.addCleanup(shutil.rmtree, d, True)
+        image = bytes(range(256)) * 4
+        (d / "image.bin").write_bytes(image)
+        config = {"image": str(d / "image.bin"), "base": "0x401000"}
+        good = ghidra_check(config | {"ghidra_url": self.serve(0x401000, 0x4013FF, (d / "image.bin").as_posix(), image)})
+        self.assertTrue(good["ok"], good)
+        self.assertEqual([c["ok"] for c in good["checks"]], [True, True, True])
+        other = d / "other.bin"
+        other.write_bytes(image[:-1] + b"\x00")
+        bad = ghidra_check(config | {"ghidra_url": self.serve(0x401000, 0x4013FE, other.as_posix(),
+                                                               image[:512] + b"\0" * 512)})
+        self.assertFalse(bad["ok"])
+        self.assertEqual([c["ok"] for c in bad["checks"]], [False, False, False])
+
 
 class GhidraForwarder(unittest.TestCase):
     """The ghidra() tool against a fake streamable-HTTP MCP server: a session, a polled async task, a refusal."""

tools/watch.py

@@ -18,6 +18,19 @@ sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
 from harbor import projects  # noqa: E402
 
 
+def cst(iso):
+    """'2026-10-07T23:21:05Z' -> '18:21 CST' (US Central, as the User reads every time)."""
+    import datetime
+    from zoneinfo import ZoneInfo
+    try:
+        when = datetime.datetime.fromisoformat(str(iso).replace("Z", "+00:00"))
+    except ValueError:
+        return str(iso)
+    if when.tzinfo is None:
+        when = when.replace(tzinfo=datetime.timezone.utc)
+    return when.astimezone(ZoneInfo("America/Chicago")).strftime("%H:%M") + " CST"
+
+
 def snapshot(conn, me):
     q = lambda sql, *a: conn.execute(sql, a).fetchall()
     return {
@@ -88,7 +101,7 @@ def main():
         for agent, last in stalled(conn, project, args.me, args.stall):
             if agent in {a.strip() for a in args.ignore.split(",") if a.strip()}:
                 continue
-            found.append(f"stalled: {agent} has made no tool call since {last[11:16]}Z (not stopped) - wake it")
+            found.append(f"stalled: {agent} has made no tool call since {cst(last)} (not stopped) - wake it")
         if found:
             print("\n".join(found))
             return 0
@@ -96,7 +109,7 @@ def main():
     tasks = conn.execute("SELECT state, COUNT(*) FROM tasks GROUP BY state").fetchall()
     last = conn.execute("SELECT agent, MAX(at) FROM activity_log GROUP BY agent").fetchall()
     print(f"quiet for {args.minutes:g} min; open review items {open_items}; tasks {dict(tasks)}; last tool call "
-          + ", ".join(f"{a} {t[11:16]}Z" for a, t in last))
+          + ", ".join(f"{a} {cst(t)}" for a, t in last))
     return 0