feat(agents): reverse-lookup peer route candidates before substituting a worker

docs/reference/protocols/peer-agent-directory-and-observation-v0.md

@@ -131,6 +131,23 @@ an identity.
 Rules:
 
 - a row exists per registered Agent of the Goal, whether or not it is running;
+- `peer_route` summarizes the thread bindings already published for that Agent as
+  `{schema_version, agent_id, outcome, address_shared, candidate_count, candidates,
+  scope, provenance}`. `outcome` describes only what this walk can prove:
+  `single_candidate`, `multiple_candidates` or `no_candidate` — it is never a claim
+  that an exact link has been resolved, which stays the forward resolver's job.
+  `address_shared: true` means one of these addresses also names a different
+  registered Agent, the same registry conflict the forward resolver answers
+  `conflict` for, so a lone candidate is not thereby unique. `scope` is
+  `goals_supplied`: the bindings read for this Goal, which does not bound
+  project-level uniqueness. `candidates` carries at most three entries that
+  survived the public-output boundary, and any entry that did not is witheld:
+  `withheld_candidate_count` and the packet-level `route_candidate_withheld`
+  limitation name it, while `candidate_count` keeps counting it, so a filtered
+  route is visibly filtered and never reads as a binding that does not exist.
+  Withholding never truncates an identifier to make it pass. In every case a
+  route is a locator, not a claim, lease, capability or cross-host resume
+  authority;
 - `presence` is optional and must carry `provider`, `observed_at` and `basis`,
   so a reader can tell "not running" from "this machine cannot see it";
 - `provider_session_ref` is an opaque handle **inside one provider session**.

loopx/control_plane/agents/directory.py

@@ -26,6 +26,8 @@
 
 from ..runtime.public_safety import public_safe_compact_text
 from ..runtime.time import now_utc_iso
+from ...thread_agent_binding import summarize_agent_binding_routes
+from ..runtime.public_safety import validate_public_safe_value
 from ..todos.contract import normalize_todo_id
 from .management_projection import build_agent_management_projection
 
@@ -35,6 +37,7 @@
 MAX_DIRECTORY_ROWS = 24
 MAX_OBSERVATION_REFS = 1
 MAX_DELIVERY_REFS = 1
+MAX_ROUTE_CANDIDATES = 3
 MAX_ROLLUP_AGENTS = 8
 
 # Limitation codes are contract vocabulary, not prose: a reader switches on them.
@@ -43,6 +46,7 @@
 LIMITATION_LEASE_STATE_NOT_PROJECTED = "lease_state_not_projected"
 LIMITATION_CALLER_IDENTITY_NOT_SUPPLIED = "caller_identity_not_supplied"
 LIMITATION_ROWS_TRUNCATED = "rows_truncated_at_cap"
+LIMITATION_ROUTE_CANDIDATE_WITHHELD = "route_candidate_withheld"
 
 GAP_AUDIENCE_NOT_AUTHORIZED = "audience_not_authorized"
 
@@ -74,6 +78,31 @@ def _compact_refs(value: Any, *, limit: int) -> list[str]:
     return refs
 
 
+def _publishable_route(route: dict[str, Any]) -> tuple[dict[str, Any], int]:
+    """Cap the candidate list and withhold entries the public boundary rejects.
+
+    Withholding is reported, never silent: `candidate_count` keeps the withheld
+    entries, so a filtered candidate cannot be read as a binding that does not
+    exist. Values are never truncated to make them pass, because a half a thread
+    identifier is a locator that no longer locates.
+    """
+
+    visible: list[dict[str, str]] = []
+    withheld = 0
+    for candidate in route["candidates"]:
+        try:
+            validate_public_safe_value(candidate, path="peer_route.candidate")
+        except ValueError:
+            withheld += 1
+            continue
+        visible.append(candidate)
+    published = {key: value for key, value in route.items() if key != "candidates"}
+    published["candidates"] = visible[:MAX_ROUTE_CANDIDATES]
+    if withheld:
+        published["withheld_candidate_count"] = withheld
+    return published, withheld
+
+
 def _work_block(agent_row: Mapping[str, Any]) -> dict[str, Any] | None:
     """Project one Agent's bounded work facts, or nothing when it holds none."""
 
@@ -187,8 +216,12 @@ def build_peer_agent_directory(
     if not caller:
         limitations.append(LIMITATION_CALLER_IDENTITY_NOT_SUPPLIED)
 
+    # Route candidates are read from the same published bindings the management
+    # projection walks, but only the owning resolver module interprets them.
+    binding_goals = _as_list(_as_mapping(payload.get("run_history")).get("goals"))
     rows: list[dict[str, Any]] = []
     dropped_at_cap = 0
+    withheld_route_candidates = 0
     for row in agent_rows:
         agent_id = _compact(row.get("agent_id"), limit=120)
         if not agent_id:
@@ -199,6 +232,10 @@ def build_peer_agent_directory(
             dropped_at_cap += 1
             continue
         work = _work_block(row)
+        peer_route, withheld_candidates = _publishable_route(
+            summarize_agent_binding_routes(binding_goals, agent_id=agent_id)
+        )
+        withheld_route_candidates += withheld_candidates
         directory_row: dict[str, Any] = {
             "agent_id": agent_id,
             "registered": True,
@@ -210,6 +247,7 @@ def build_peer_agent_directory(
             "delivery_refs": _compact_refs(
                 row.get("handoff_refs"), limit=MAX_DELIVERY_REFS
             ),
+            "peer_route": peer_route,
         }
         rows.append(
             {
@@ -230,6 +268,8 @@ def build_peer_agent_directory(
     )
     if omitted:
         limitations.append(LIMITATION_ROWS_TRUNCATED)
+    if withheld_route_candidates:
+        limitations.append(LIMITATION_ROUTE_CANDIDATE_WITHHELD)
 
     packet: dict[str, Any] = {
         "ok": True,

loopx/thread_agent_binding.py

@@ -5,7 +5,7 @@
 import re
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any
+from typing import Any, Iterable
 from urllib.parse import urlsplit
 
 from .control_plane.projects.registry_codec import mutate_project_registry
@@ -27,6 +27,11 @@
 )
 _CODEX_THREAD_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
 
+AGENT_BINDING_ROUTE_SCHEMA_VERSION = "loopx_agent_binding_route_v0"
+ROUTE_SINGLE_CANDIDATE = "single_candidate"
+ROUTE_MULTIPLE_CANDIDATES = "multiple_candidates"
+ROUTE_NO_CANDIDATE = "no_candidate"
+
 
 class ThreadBindingRequestError(ValueError):
     """The caller supplied an invalid host-thread identity."""
@@ -168,6 +173,84 @@ def resolve_thread_agent_binding(
     return base
 
 
+def collect_accepted_bindings(goals: Iterable[Any]) -> list[dict[str, str]]:
+    """Return every thread binding the owner accepts, deduplicated, first seen first.
+
+    `_bindings_for_goal` is the single normalisation rule, so an entry the owner
+    cannot name is dropped here rather than counted or published downstream.
+    """
+
+    accepted: list[dict[str, str]] = []
+    for raw_goal in goals:
+        if not isinstance(raw_goal, dict):
+            continue
+        for binding in _bindings_for_goal(raw_goal):
+            candidate = {
+                "thread_id": binding["thread_id"],
+                "host_surface": binding["host_surface"],
+                "agent_id": binding["agent_id"],
+            }
+            if candidate not in accepted:
+                accepted.append(candidate)
+    return accepted
+
+
+def summarize_agent_binding_routes(
+    goals: Iterable[Any],
+    *,
+    agent_id: Any,
+) -> dict[str, Any]:
+    """Classify which published bindings address one Agent, keeping full identity.
+
+    This is the reverse of `resolve_thread_agent_binding`: that resolver answers
+    "which Agent does this exact link belong to", while a coordinator holding a
+    peer name and no link needs to know how many addresses the bindings on record
+    offer for that peer. The vocabulary says only what this walk can prove. The
+    bindings were read from the goals supplied, so `scope` records that; a
+    `single_candidate` result is not a project-level uniqueness claim, and
+    `address_shared` reports the one cross-identity fact that *is* visible here —
+    the same host thread also addresses a different Agent, which the forward
+    resolver would answer `conflict`.
+
+    Nothing selects a route, opens a session or transfers claim, lease or
+    capability. Candidates keep their full accepted identity because this is the
+    internal view: the projection that publishes them owns visibility and size.
+    """
+
+    wanted = normalize_todo_claimed_by(agent_id)
+    accepted = collect_accepted_bindings(goals)
+    candidates = [
+        {"thread_id": item["thread_id"], "host_surface": item["host_surface"]}
+        for item in accepted
+        if wanted and item["agent_id"] == wanted
+    ]
+    other_addresses = {
+        (item["host_surface"], item["thread_id"])
+        for item in accepted
+        if item["agent_id"] != wanted
+    }
+    address_shared = any(
+        (item["host_surface"], item["thread_id"]) in other_addresses
+        for item in candidates
+    )
+    if not candidates:
+        outcome = ROUTE_NO_CANDIDATE
+    elif len(candidates) == 1:
+        outcome = ROUTE_SINGLE_CANDIDATE
+    else:
+        outcome = ROUTE_MULTIPLE_CANDIDATES
+    return {
+        "schema_version": AGENT_BINDING_ROUTE_SCHEMA_VERSION,
+        "agent_id": wanted or "",
+        "outcome": outcome,
+        "address_shared": address_shared,
+        "candidate_count": len(candidates),
+        "candidates": candidates,
+        "scope": "goals_supplied",
+        "provenance": "run_history.goals[].coordination.thread_agent_bindings",
+    }
+
+
 def _registry_thread_binding_request(
     *,
     host_surface: str | None,

tests/control_plane/test_peer_agent_directory.py

@@ -7,11 +7,16 @@
     LIMITATION_CALLER_IDENTITY_NOT_SUPPLIED,
     LIMITATION_LEASE_STATE_NOT_PROJECTED,
     LIMITATION_PRESENCE_PROVIDER_UNAVAILABLE,
+    LIMITATION_ROUTE_CANDIDATE_WITHHELD,
     LIMITATION_ROWS_TRUNCATED,
     MAX_DIRECTORY_ROWS,
+    MAX_ROUTE_CANDIDATES,
     PEER_AGENT_DIRECTORY_SCHEMA_VERSION,
     build_peer_agent_directory,
 )
+from loopx.control_plane.runtime.public_safety import (
+    validate_public_safe_value,
+)
 
 
 GOAL_ID = "peer-directory-fixture"
@@ -150,3 +155,88 @@ def test_truncated_directory_declares_the_rows_it_omitted() -> None:
     )
     assert packet["omitted_row_count"] > 0
     assert LIMITATION_ROWS_TRUNCATED in packet["limitations"]  # type: ignore[operator]
+
+
+def test_rows_report_a_peer_route_without_selecting_one() -> None:
+    payload = _status_payload()
+    coordination = payload["run_history"]["goals"][0]["coordination"]
+    coordination["thread_agent_bindings"] = [
+        {
+            "agent_id": WORKING_AGENT,
+            "thread_id": "thread-app",
+            "host_surface": "codex-app",
+        },
+        {
+            "agent_id": WORKING_AGENT,
+            "thread_id": "thread-cli",
+            "host_surface": "codex-cli",
+        },
+    ]
+
+    packet = build_peer_agent_directory(payload, caller_agent_id=WORKING_AGENT)
+    rows = _rows_by_agent(packet)
+
+    assert rows[WORKING_AGENT]["peer_route"]["outcome"] == "multiple_candidates"
+    assert rows[WORKING_AGENT]["peer_route"]["candidate_count"] == 2
+    assert rows[WORKING_AGENT]["peer_route"]["candidates"] == [
+        {"thread_id": "thread-app", "host_surface": "codex-app"},
+        {"thread_id": "thread-cli", "host_surface": "codex-cli"},
+    ]
+    assert rows[IDLE_AGENT]["peer_route"]["outcome"] == "no_candidate"
+    assert rows[IDLE_AGENT]["peer_route"]["candidates"] == []
+    # A route is a locator: it must not read as membership, presence or a lease.
+    assert packet["scope"]["caller_membership"] == "registered_agent"
+    assert LIMITATION_LEASE_STATE_NOT_PROJECTED in packet["limitations"]
+
+
+def test_a_credential_shaped_candidate_is_withheld_not_erased() -> None:
+    """Publication filters, it does not delete: the count keeps the truth."""
+
+    payload = _status_payload()
+    payload["run_history"]["goals"][0]["coordination"]["thread_agent_bindings"] = [
+        {
+            "agent_id": WORKING_AGENT,
+            "thread_id": "ghp_" + "1234567890abcdefghijklmnopqrstuvwxyz1234",
+            "host_surface": "codex-app",
+        },
+        {
+            "agent_id": WORKING_AGENT,
+            "thread_id": "thread-visible",
+            "host_surface": "codex-app",
+        },
+    ]
+
+    packet = build_peer_agent_directory(payload, caller_agent_id=WORKING_AGENT)
+    route = _rows_by_agent(packet)[WORKING_AGENT]["peer_route"]
+
+    assert route["candidate_count"] == 2
+    assert route["candidates"] == [
+        {"thread_id": "thread-visible", "host_surface": "codex-app"}
+    ]
+    assert route["withheld_candidate_count"] == 1
+    assert LIMITATION_ROUTE_CANDIDATE_WITHHELD in packet["limitations"]
+    validate_public_safe_value(packet, path="peer_agent_directory")
+
+
+def test_the_published_candidate_list_respects_its_budget() -> None:
+    payload = _status_payload()
+    payload["run_history"]["goals"][0]["coordination"]["thread_agent_bindings"] = [
+        {
+            "agent_id": WORKING_AGENT,
+            "thread_id": f"thread-{index}",
+            "host_surface": "codex-app",
+        }
+        for index in range(5)
+    ]
+
+    packet = build_peer_agent_directory(payload, caller_agent_id=WORKING_AGENT)
+    route = _rows_by_agent(packet)[WORKING_AGENT]["peer_route"]
+
+    assert route["candidate_count"] == 5
+    assert len(route["candidates"]) == MAX_ROUTE_CANDIDATES
+    assert [item["thread_id"] for item in route["candidates"]] == [
+        "thread-0",
+        "thread-1",
+        "thread-2",
+    ]
+    assert "withheld_candidate_count" not in route