feat(notifications): relay-side identity bookkeeping + SUPERSEDED verdict

The relay persists its last-assumed identity (name, generation, session
id) to NOTIFICATIONS_IDENTITY_FILE (default ~/.claude/agent-identity.json,
pod-lifecycle storage) on every effective registration. Two detections:

- Normal succession: registering a name whose stored record shows a
  lower generation under a different session notes the succession in the
  reply (the successor already inherited the name's threads and backlog
  via name-keyed membership).
- Stand-down: a register that fails "already taken" while the stored
  identity proves THIS session held the name returns an explicit
  SUPERSEDED verdict telling the agent not to retry and how to hand off.
  Durable across daemon restarts, unlike the in-memory heir notice.

Bookkeeping never fails a registration; the file is written atomically.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: wlritchi

notifications/mcp/notifications-server.py

@@ -749,6 +749,47 @@ def _format_last_seen(last_seen: float) -> str:
     return f"~{int(round(hours / 24.0))}d ago"
 
 
+def _identity_path() -> Path:
+    """Where this host persists its last-assumed agent identity (name, generation,
+    session id) — pod-lifecycle state that survives daemon restarts, so a session
+    can durably tell 'my name moved on without me' (SUPERSEDED) from 'the daemon
+    forgot me' (docs/specs/2026-07-09 slice b, relay side)."""
+    override = os.environ.get("NOTIFICATIONS_IDENTITY_FILE")
+    if override:
+        return Path(override)
+    return Path.home() / ".claude" / "agent-identity.json"
+
+
+def _load_identity() -> dict | None:
+    try:
+        data = json.loads(_identity_path().read_text())
+    except (OSError, ValueError):
+        return None
+    return data if isinstance(data, dict) else None
+
+
+def _store_identity(name: str, generation: int, session_id: str) -> None:
+    path = _identity_path()
+    try:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        tmp = path.with_name(path.name + ".tmp")
+        tmp.write_text(
+            json.dumps(
+                {
+                    "name": name,
+                    "generation": generation,
+                    "session_id": session_id,
+                    "updated_at": time.time(),
+                }
+            )
+        )
+        tmp.replace(path)
+    except OSError as exc:  # bookkeeping must never fail a registration
+        print(
+            f"notifications: could not persist agent identity: {exc}", file=sys.stderr
+        )
+
+
 @mcp.tool()
 async def register_agent(
     name: str,
@@ -821,12 +862,64 @@ async def register_agent(
     if isinstance(reply, str):
         return reply
     if reply.get("type") == wsproto.ERROR:
-        return f"Could not register as '{name}': {reply.get('error')}"
+        error = str(reply.get("error") or "")
+        stored = _load_identity()
+        if (
+            "already taken" in error
+            and stored is not None
+            and stored.get("name") == name
+            and stored.get("session_id") == session_id
+        ):
+            # The durable half of supersession detection: our own bookkeeping says
+            # this session held the name, but another session holds it now. The
+            # daemon's heir notice covers the fresh case; this file survives daemon
+            # restarts. Tell the agent to stand down rather than retry-register.
+            held_gen = stored.get("generation")
+            print(
+                f"notifications: SUPERSEDED — '{name}' is held by another session; "
+                f"this session held gen {held_gen}",
+                file=sys.stderr,
+            )
+            return (
+                f"SUPERSEDED: this session previously held '{name}' (gen {held_gen}), "
+                "but the name is now held by another live session. Do not retry — "
+                "your successor has inherited the name's threads and messages. To "
+                f"hand off in-flight work, register under a different name and dm "
+                f"'{name}'."
+            )
+        return f"Could not register as '{name}': {error}"
     agent = reply.get("agent") or {}
     threshold = agent.get("default_threshold", "direct")
+    reg_name = str(agent.get("name") or name)
+    generation = int(agent.get("generation") or 1)
+    stored = _load_identity()
+    succession_note = ""
+    if (
+        stored is not None
+        and stored.get("name") == reg_name
+        and stored.get("session_id") != session_id
+        and int(stored.get("generation") or 1) < generation
+    ):
+        # Normal succession on this host: a predecessor session registered here
+        # before us. Membership is name-keyed, so its threads and any unsurfaced
+        # backlog are already ours.
+        succession_note = (
+            f" You are gen {generation}, succeeding a previous session of this name."
+        )
+        print(
+            f"notifications: assumed '{reg_name}' gen {generation}, superseding "
+            f"session {stored.get('session_id')}",
+            file=sys.stderr,
+        )
+    if stored is None or (
+        stored.get("name"),
+        stored.get("generation"),
+        stored.get("session_id"),
+    ) != (reg_name, generation, session_id):
+        _store_identity(reg_name, generation, session_id)
     return (
-        f"Registered as '{agent.get('name', name)}' (wake threshold: {threshold}). "
-        "Other agents can find you with list_agents."
+        f"Registered as '{reg_name}' (wake threshold: {threshold})."
+        f"{succession_note} Other agents can find you with list_agents."
     )
 
 

notifications/tests/_harness.py

@@ -123,6 +123,12 @@ def relay_env(
         env["NOTIFICATIONS_MCP_LOG_CACHE_DIR"] = str(xdg_dir)
     if project_dir is not None:
         env["CLAUDE_PROJECT_DIR"] = project_dir
+    # Per-session identity file (in prod: one per pod at ~/.claude/). Keyed by
+    # session id so each harness session models its own pod's persistent storage,
+    # and a re-opened session with the same id sees its earlier registrations.
+    env["NOTIFICATIONS_IDENTITY_FILE"] = str(
+        Path(xdg_dir) / f"agent-identity-{session_id}.json"
+    )
     return env
 
 

notifications/tests/test_e2e_agents.py

@@ -573,3 +573,73 @@ async def scenario():
                         assert "'worker' is now gen 2" in notice.params["content"]
 
         anyio.run(scenario)
+
+
+def test_superseded_session_told_to_stand_down(tmp_path):
+    """Relay-side supersession detection (docs/specs/2026-07-09, relay half): the
+    relay persists (name, generation, session_id) on successful registration; when
+    a register later fails with 'already taken' and the stored identity proves THIS
+    session held the name, the agent gets an explicit SUPERSEDED verdict instead of
+    a generic failure — durable across daemon restarts, unlike the heir notice."""
+    store, xdg = tmp_path / "store", tmp_path / "xdg"
+    store.mkdir()
+    xdg.mkdir()
+    ws = h.free_port()
+
+    with h.daemon_process(h.daemon_env(ws, store, settle="0")):
+
+        async def scenario():
+            async with h.agent_session(tmp_path, ws, store, xdg, "sid-old") as (
+                read_a,
+                write_a,
+            ):
+                text, _ = await h.mcp_call(
+                    read_a,
+                    write_a,
+                    2,
+                    "register_agent",
+                    {"name": "worker", "reclaim_key": "pod-1"},
+                )
+                assert "Registered as 'worker'" in text
+            # sid-old's identity file now records it held 'worker' gen 1.
+            import json as _json
+
+            identity = _json.loads((xdg / "agent-identity-sid-old.json").read_text())
+            assert identity["name"] == "worker"
+            assert identity["generation"] == 1
+            assert identity["session_id"] == "sid-old"
+
+            async with h.agent_session(tmp_path, ws, store, xdg, "sid-new") as (
+                read_b,
+                write_b,
+            ):
+                text, next_id = await _list_until(
+                    read_b, write_b, 2, lambda t: "offline" in t
+                )
+                text, _ = await h.mcp_call(
+                    read_b,
+                    write_b,
+                    next_id,
+                    "register_agent",
+                    {"name": "worker", "reclaim_key": "pod-1"},
+                )
+                assert "Registered as 'worker'" in text
+
+                # The displaced session comes back while the successor is LIVE and
+                # tries to retake its name: it must get the stand-down verdict.
+                async with h.agent_session(tmp_path, ws, store, xdg, "sid-old") as (
+                    read_back,
+                    write_back,
+                ):
+                    text, _ = await h.mcp_call(
+                        read_back,
+                        write_back,
+                        2,
+                        "register_agent",
+                        {"name": "worker", "reclaim_key": "pod-1"},
+                    )
+                    assert "SUPERSEDED" in text
+                    assert "gen 1" in text
+                    assert "dm 'worker'" in text
+
+        anyio.run(scenario)