Merge feature/orchestrator/s53-agent-presence into master

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

# Conflicts:
#	docs/requirements/001-agent-orchestrator.md
#	orchestrator/templates/page-board.js
#	tests/e2e/test_orchestrator_e2e.py

tests/e2e/test_orchestrator_e2e.py

@@ -751,6 +751,62 @@ RADIO_JUMP = """(async () => {@HELPERS@
     header: all("header.top button").map((button) => [button.id, name(button), text(button)]), radioTab: Boolean(document.getElementById("tab-radio")),
   };
 })()""".replace("@HELPERS@", HELPERS)
+PRESENCE_HELPERS = """
+  const states = () => all("#main .presence:not(label)").map((node) => node.dataset.state).sort();
+  const statusMenu = () => {
+    const menu = all('#main [role="toolbar"] details.menu').find((node) => name(node.querySelector("summary")) === "Status");
+    menu.open = true;
+    return menu;
+  };
+  const looks = (menu) => all("input[data-look]", menu).map((input) => [input.dataset.look, input.checked]);
+  const graph3d = async (faces) => {
+    const box = await until(() => {
+      const found = document.querySelector("#main .graph3d");
+      return found && found.dataset.faces !== undefined && (faces === null || found.dataset.faces === faces) ? found : null;
+    }, 15000);
+    return box ? [Number(box.dataset.faces), Number(box.dataset.badges)] : null;
+  };
+"""
+PRESENCE_WALK = """(async () => {@HELPERS@@PRESENCE@
+  const views = {};
+  for (const view of ["orbit", "chord", "swimlane", "tree", "hierarchy"]) {
+    location.hash = `#agents/${view}`;
+    await until(() => all("#main .presence:not(label)").length, 6000);
+    await sleep(300);
+    views[view] = states();
+  }
+  location.hash = "#agents/orbit";
+  await sleep(700);
+  const node = document.querySelector('#main svg .node[data-pick="s1"]');
+  const motion = getComputedStyle(node.querySelector(".presence .glyph")).animationName;
+  node.dispatchEvent(new MouseEvent("click", {bubbles: true}));
+  const mark = await until(() => document.querySelector("#popup .presence"), 4000);
+  const profile = mark ? [mark.dataset.state, text(mark), mark.getAttribute("aria-label")] : null;
+  (document.activeElement || document.body).dispatchEvent(new KeyboardEvent("keydown", {key: "Escape", bubbles: true}));
+  await sleep(300);
+  location.hash = "#agents/3d";
+  const shown = await graph3d(null);
+  const menu = statusMenu();
+  const before = looks(menu);
+  menu.querySelector('input[data-look="working"]').click();
+  menu.querySelector('input[data-look="faces"]').click();
+  const hidden = await graph3d("0");
+  location.hash = "#agents/orbit";
+  await sleep(700);
+  return {views, motion, profile, shown, before, hidden, orbit: states(), @FITS@};
+})()""".replace("@HELPERS@", HELPERS).replace("@PRESENCE@", PRESENCE_HELPERS).replace("@FITS@", FITS)
+PRESENCE_READ = """(async () => {@HELPERS@@PRESENCE@
+  await until(() => document.querySelector('#main svg .node[data-pick="s2"]'), 8000);
+  await sleep(300);
+  const menu = statusMenu();
+  const kept = {boxes: looks(menu), orbit: states()};
+  menu.querySelector('input[data-look="working"]').click();
+  menu.querySelector('input[data-look="faces"]').click();
+  await sleep(300);
+  const back = states();
+  location.hash = "#agents/3d";
+  return {...kept, back, restored: await graph3d(null)};
+})()""".replace("@HELPERS@", HELPERS).replace("@PRESENCE@", PRESENCE_HELPERS)
 GEAR_SAVE = """(async () => {@HELPERS@@CLASHES@
   const spot = document.getElementById("radio-spot");
   const gear = all(".spot-head button", spot).find((button) => name(button) === "Radio settings");
@@ -1642,7 +1698,7 @@ class TestOrchestratorE2E(HubCase):
                 self.assertTrue(by_hash[f"#{tab}/3d"]["canvas"], by_hash[f"#{tab}/3d"]["note"])
                 self.assertRegex(by_hash[f"#{tab}/3d"]["note"], r"^\d+ nodes · \d+ edges · Force$")
         agents = by_hash["#agents/orbit"]
-        self.assertEqual(agents["toolbar"], ["Priority", "Day/Night"])
+        self.assertEqual(agents["toolbar"], ["Priority", "Day/Night", "Status"])
         self.assertEqual(agents["spot"], {"hidden": True, "channels": CHOICE_CHANNELS})
         hierarchy = by_hash["#agents/hierarchy"]
         self.assertEqual([office for office, _ in hierarchy["org"]], ["chair", "identity/oidc", "work/s1", "work/s2"])
@@ -1799,7 +1855,7 @@ class TestOrchestratorE2E(HubCase):
         for result, (width, _) in zip(data["results"], WIDTHS, strict=True):
             with self.subTest(width=width):
                 self.assertNotIn("error", result)
-                self.assertEqual(result["toolbar"], ["Priority", "Day/Night", "Drain", "Suspend", "Resume"])
+                self.assertEqual(result["toolbar"], ["Priority", "Day/Night", "Drain", "Suspend", "Resume", "Status"])
                 self.assertEqual((result["title"], result["shown"]), ("Cancel mayday: lift the radio silence so the agents write again; your passkey signs it", ""))
                 self.assertEqual(result["declined"], {"question": [question], "posted": [], "toasts": []})
                 self.assertEqual(result["accepted"], {"question": [question], "posted": ["/api/board/sign", "/api/board/clear-mayday"]})
@@ -1836,7 +1892,7 @@ class TestOrchestratorE2E(HubCase):
                     self.assertNotIn("error", probe)
                 self.assertEqual((sent["closed"], sent["open"]), ([True, "false"], [False, "true", True]))
                 self.assertEqual(sent["inline"], ["static", True, True])
-                self.assertEqual((sent["toolbar"], sent["channels"], sent["record"]), (["Priority", "Day/Night"], CHOICE_CHANNELS, ["Record"]))
+                self.assertEqual((sent["toolbar"], sent["channels"], sent["record"]), (["Priority", "Day/Night", "Status"], CHOICE_CHANNELS, ["Record"]))
                 self.assertEqual(sent["heads"], [["Radio settings", "Radio settings: tones, squelch and read aloud"], ["Cancel mayday", "Cancel mayday: lift the radio silence so the agents write again; your passkey signs it"]])
                 self.assertEqual((sent["tall"], sent["wide"]), (False, False))
                 self.assertEqual([run["cls"] for run in sent["runs"]], CHOICE_CHANNELS)
@@ -2509,6 +2565,46 @@ class TestOrchestratorE2E(HubCase):
     def test_firefox_draws_the_trend_forecast_point_tips_eta_and_tasks_3d_at_three_widths(self):
         self.check_trend_eta_3d("firefox", "bidi.mjs", os.environ.get("FIREFOX", "firefox"))
 
+    def check_presence(self, browser: str, driver: str, binary: str) -> None:
+        """Args:
+            browser: the file name prefix of the screenshots and the result.
+            driver: the browser driver script next to this file.
+            binary: the browser to drive.
+        """
+        record = {"status": "busy", "statusUpdatedAt": int(time.time() * 1000), "name": "Skills session", "running": True}
+        self.httpd.hub.agent_sessions.write_text(json.dumps({"written": time.time(), "sessions": {"s1": record}}), encoding="utf-8")
+        steps = []
+        for width, height in WIDTHS:
+            steps += [
+                {"width": width, "height": height, "url": f"{self.base}?{width}#agents/orbit", "wait": 1200, "probe": PRESENCE_WALK, "shot": str(OUT / f"{browser}-{width}-agents-presence.png")},
+                {"width": width, "height": height, "url": f"{self.base}?{width}-reloaded#agents/orbit", "wait": 1200, "probe": PRESENCE_READ, "shot": str(OUT / f"{browser}-{width}-agents-presence-3d.png")},
+            ]
+        data = self.drive(steps, None, driver, binary)
+        (OUT / f"{browser}-presence.json").write_text(json.dumps(data, indent=1), encoding="utf-8")
+        self.assertEqual([error for error in data["errors"] if "(style-src-elem)" not in error and "/favicon.ico" not in error], [])
+        pair = ["dead", "working"]
+        order = ["working", "alive", "sleeping", "suspending", "unreachable", "dead", "faces"]
+        for index, (width, _) in enumerate(WIDTHS):
+            walk, read = data["results"][2 * index : 2 * index + 2]
+            with self.subTest(width=width):
+                self.assertNotIn("error", walk)
+                self.assertNotIn("error", read)
+                self.assertEqual(walk["views"], {"orbit": pair, "chord": pair, "swimlane": pair, "tree": pair, "hierarchy": ["dead", "dead", "working", "working"]})
+                self.assertEqual((walk["motion"], walk["profile"]), ("presence-spin", ["working", "Working", "Working"]))
+                self.assertGreaterEqual(walk["shown"][0], 3, walk["shown"])
+                self.assertEqual(walk["shown"][1], 2)
+                self.assertEqual(walk["before"], [[look, True] for look in order])
+                self.assertEqual((walk["hidden"], walk["orbit"], walk["wide"]), ([0, 1], ["dead"], False))
+                self.assertEqual(read["boxes"], [[look, look not in ("working", "faces")] for look in order])
+                self.assertEqual((read["orbit"], read["back"], read["restored"]), (["dead"], pair, walk["shown"]))
+
+    def test_chromium_shows_agent_states_and_3d_faces_switchable_through_a_reload_at_three_widths(self):
+        self.check_presence("chromium", "cdp.mjs", os.environ.get("CHROMIUM", "chromium"))
+
+    @unittest.skipUnless(shutil.which(os.environ.get("FIREFOX", "firefox")), "no Firefox")
+    def test_firefox_shows_agent_states_and_3d_faces_switchable_through_a_reload_at_three_widths(self):
+        self.check_presence("firefox", "bidi.mjs", os.environ.get("FIREFOX", "firefox"))
+
     def test_chromium_pins_a_task_priority_after_signature_and_the_runner_applies_it(self):
         signer = self.pin()
         private = "".join(line for line in signer.pem.read_text(encoding="utf-8").splitlines() if not line.startswith("-----"))

tests/test_hub_presence.py

@@ -0,0 +1,147 @@
+"""Pin decision 61: each agent state from the host's session snapshot, the night state, radio receipts and claim beats."""
+
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+import unittest
+from pathlib import Path
+
+import hub_support  # noqa: F401
+import sessions as client
+from kvb_orchestrator.presence import SNAPSHOT_STALE, presence, read_snapshot
+
+NOW = 1_800_000_000.0
+FRESH = int((NOW - 60) * 1000)
+OLD = int((NOW - 1800) * 1000)
+
+
+def record(sid: str, status: str, updated: int, name: str) -> dict:
+    return {"status": status, "statusUpdatedAt": updated, "name": name, "running": sid != "gone"}
+
+
+def claim(session: str, name: str, beat: float | None) -> dict:
+    return {"session": session, "agent": name, "beat": beat}
+
+
+class TestPresence(unittest.TestCase):
+    def states(self, sessions: dict[str, dict], parts: dict[str, str], mode: str, claims: list[dict], news: list[dict]) -> dict[str, str]:
+        return presence(sessions, parts, mode, claims, news, NOW)
+
+    def test_each_state_by_day_and_by_night(self):
+        sessions = {sid: record(sid, status, updated, name) for sid, status, updated, name in (
+            ("work", "busy", FRESH, "Worker"),
+            ("idle", "idle", OLD, "Idler"),
+            ("gone", "idle", FRESH, "Ghost"),
+            ("stuck", "busy", OLD, "Stuck"),
+            ("called", "idle", OLD, "Callee"),
+            ("answered", "idle", OLD, "Answerer"),
+            ("sleep", "idle", FRESH, "Sleeper"),
+            ("park", "busy", FRESH, "Parker"),
+        )}
+        call = NOW - 700
+        news = [
+            {"ts": call, "from": "Operator", "to": ["Callee"], "class": "call"},
+            {"ts": call, "from": "Operator", "to": ["Answerer"], "class": "call"},
+            {"ts": call + 2, "from": "Answerer", "to": ["Operator"], "class": "receipt", "call": call},
+            {"ts": NOW - 60, "from": "Operator", "to": ["Idler"], "class": "call"},
+        ]
+        claims = [claim("codex-on", "Codex", NOW - 60), claim("codex-off", "Codex old", NOW - 3600), claim("-", "Nobody", NOW), claim("op", "Operator", NOW)]
+        day = self.states(sessions, {"sleep": "suspend", "park": "drain"}, "day", claims, news)
+        self.assertEqual(day, {
+            "work": "working", "idle": "alive", "gone": "dead", "stuck": "unreachable", "called": "unreachable",
+            "answered": "alive", "sleep": "sleeping", "park": "suspending", "codex-on": "alive", "codex-off": "dead",
+        })
+        night = self.states(sessions, {"idle": "day", "work": "suspend"}, "drain", [], [])
+        self.assertEqual({sid: night[sid] for sid in ("work", "idle", "called", "sleep", "park")}, {
+            "work": "suspending", "idle": "alive", "called": "sleeping", "sleep": "sleeping", "park": "suspending",
+        })
+
+    def test_a_status_update_after_the_call_counts_as_reached(self):
+        news = [{"ts": NOW - 900, "from": "Operator", "to": ["Alias"], "class": "call"}]
+        claims = [claim("later", "Alias", NOW)]
+        self.assertEqual(self.states({"later": record("later", "idle", FRESH, "Later")}, {}, "day", claims, news), {"later": "alive"})
+        self.assertEqual(self.states({"later": record("later", "idle", OLD, "Later")}, {}, "day", claims, news), {"later": "unreachable"})
+
+    def test_a_record_without_running_true_is_dead(self):
+        for value in (False, None, "yes", 1):
+            with self.subTest(value=value):
+                entry = {**record("s", "busy", FRESH, "S"), "running": value}
+                self.assertEqual(self.states({"s": entry}, {}, "day", [], []), {"s": "dead"})
+
+
+class TestSnapshot(unittest.TestCase):
+    def test_the_host_client_keeps_name_status_and_whether_the_process_runs(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            base = Path(tmp)
+            proc, folder = base / "proc", base / "sessions"
+            (proc / "4242").mkdir(parents=True)
+            folder.mkdir()
+            (proc / "4242" / "stat").write_text("4242 (claude (main)) S " + " ".join(str(n) for n in range(1, 19)) + " 777 0 0\n", encoding="utf-8")
+            live = {"sessionId": "s1", "pid": 4242, "procStart": "777", "status": "busy", "statusUpdatedAt": FRESH, "name": "One", "cwd": "/secret"}
+            (folder / "1.json").write_text(json.dumps(live), encoding="utf-8")
+            (folder / "2.json").write_text(json.dumps({**live, "sessionId": "s2", "procStart": "778"}), encoding="utf-8")
+            (folder / "3.json").write_text(json.dumps({"status": "idle"}), encoding="utf-8")
+            (folder / "4.json").write_text("{broken", encoding="utf-8")
+            (folder / "5.json").write_text(json.dumps(["s5"]), encoding="utf-8")
+            output = base / "agent-sessions.json"
+            client.write(output, client.snapshot(folder, proc, NOW))
+            kept = {"status": "busy", "statusUpdatedAt": FRESH, "name": "One"}
+            self.assertEqual(json.loads(output.read_text(encoding="utf-8")), {
+                "written": NOW, "sessions": {"s1": {**kept, "running": True}, "s2": {**kept, "running": False}},
+            })
+            self.assertEqual(sorted(path.name for path in base.iterdir()), ["agent-sessions.json", "proc", "sessions"])
+            self.assertEqual(client.snapshot(folder / "missing", proc, NOW), {"written": NOW, "sessions": {}})
+
+    def test_the_host_client_skips_hostile_records_and_leaves_no_temp_file(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            base = Path(tmp)
+            proc, folder = base / "proc", base / "sessions"
+            proc.mkdir()
+            folder.mkdir()
+            good = {"sessionId": "s1", "pid": 1, "procStart": "1", "status": "busy", "statusUpdatedAt": FRESH, "name": "One"}
+            (base / "outside.json").write_text(json.dumps({**good, "sessionId": "s9"}), encoding="utf-8")
+            (folder / "1.json").write_text(json.dumps({**good, "name": {"x": 1}, "status": "y" * 201, "statusUpdatedAt": True}), encoding="utf-8")
+            (folder / "2.json").symlink_to(base / "outside.json")
+            os.mkfifo(folder / "3.json")
+            (folder / "4.json").write_text("[" * 30000 + "]" * 30000, encoding="utf-8")
+            (folder / "5.json").write_text(json.dumps({**good, "sessionId": "s5", "name": "z" * client.RECORD_BYTES}), encoding="utf-8")
+            self.assertEqual(client.snapshot(folder, proc, NOW), {"written": NOW, "sessions": {"s1": {"running": False}}})
+            output = base / "agent-sessions.json"
+            with self.assertRaises(TypeError):
+                client.write(output, {"written": object()})
+            self.assertEqual(sorted(path.name for path in base.iterdir()), ["outside.json", "proc", "sessions"])
+
+    def test_running_needs_an_int_pid_and_the_same_process_start(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            proc = Path(tmp)
+            (proc / "4242").mkdir()
+            (proc / "4242" / "stat").write_text("4242 (claude (main)) S " + " ".join(str(n) for n in range(1, 19)) + " 777 0 0\n", encoding="utf-8")
+            entry = {"pid": 4242, "procStart": "777"}
+            self.assertTrue(client.running(entry, proc))
+            for bad in ({**entry, "procStart": "778"}, {**entry, "pid": 4243}, {**entry, "pid": True}, {**entry, "pid": "4242"}, {"pid": 4242}):
+                with self.subTest(bad=bad):
+                    self.assertFalse(client.running(bad, proc))
+
+    def test_the_hub_reads_only_a_fresh_well_formed_snapshot(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "agent-sessions.json"
+            self.assertEqual(read_snapshot(path, NOW), {})
+            entry = record("s1", "busy", FRESH, "One")
+            for data, expected in (
+                ({"written": NOW - SNAPSHOT_STALE, "sessions": {"s1": entry, "s2": "x"}}, {"s1": entry}),
+                ({"written": NOW - SNAPSHOT_STALE - 1, "sessions": {"s1": entry}}, {}),
+                ({"written": True, "sessions": {"s1": entry}}, {}),
+                ({"written": NOW + 1, "sessions": {"s1": entry}}, {}),
+                ({"written": float("nan"), "sessions": {"s1": entry}}, {}),
+                ({"written": NOW, "sessions": ["s1"]}, {}),
+                (["s1"], {}),
+            ):
+                with self.subTest(data=data):
+                    path.write_text(json.dumps(data), encoding="utf-8")
+                    self.assertEqual(read_snapshot(path, NOW), expected)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_hub_state.py

@@ -151,7 +151,7 @@ class TestHubState(HubCase):
             set(state),
             {
                 "now", "since", "until", "first", "alarm_since", "reviews", "asks", "decisions", "messages", "operator",
-                "register", "tasks", "task_edges", "agent_order", "radio", "claims", "grants", "night", "silence", "graph",
+                "register", "tasks", "task_edges", "agent_order", "radio", "claims", "grants", "night", "night_parts", "presence", "silence", "graph",
                 "inbox", "pending", "rules", "approvers", "jobs", "views", "llm", "people", "settings", "media", "speech",
                 "badges", "kinds",
             },

tests/test_orchestrator_migration.py

@@ -122,6 +122,7 @@ class TestOrchestratorMigration(unittest.TestCase):
         self.assertEqual((local / f"{NEW}-approvals" / "runner.json").read_text(encoding="utf-8"), '{"used": {}}\n')
         self.assertEqual(((config / NEW).stat().st_mode & 0o777, (config / NEW / "approvers.json").stat().st_mode & 0o777), (0o700, 0o600))
         self.assertTrue((units / f"{NEW}-runner.service").is_file())
+        self.assertTrue((units / f"{NEW}-sessions.timer").is_file())
         self.assertIn(f"KVB_ORCHESTRATOR_STATE={local / NEW}", (share / NEW / ".env").read_text(encoding="utf-8"))
         settings = self.home / ".claude" / "settings.json"
         data = json.loads(settings.read_text(encoding="utf-8"))
@@ -130,7 +131,7 @@ class TestOrchestratorMigration(unittest.TestCase):
         self.assertIn(f"{share / NEW}/client/permission_hook.py", hooks[1]["hooks"][0]["command"])
         self.assertEqual(settings.stat().st_mode & 0o777, 0o640)
         self.assertIn(f"{OLD}/client", (self.home / ".claude" / f"settings.json.before-{NEW}-migration").read_text(encoding="utf-8"))
-        self.assertEqual((first["images"], first["mcp"], first["enabled"]), ([f"{NEW}:local"], {NEW: "http://127.0.0.1:8737/mcp"}, [f"{NEW}-offices.path", f"{NEW}-runner.service"]))
+        self.assertEqual((first["images"], first["mcp"], first["enabled"]), ([f"{NEW}:local"], {NEW: "http://127.0.0.1:8737/mcp"}, [f"{NEW}-offices.path", f"{NEW}-runner.service", f"{NEW}-sessions.timer"]))
         self.assertIn(["docker", "compose", "--file", str(share / OLD / "compose.yml"), "down", "--remove-orphans"], first["calls"])
 
         before = snapshot(self.home)