From 3abef5b1b6edfa9c109eee1523d98ffaf986972f Mon Sep 17 00:00:00 2001 From: JunZ-Leo <100498253+JunZ-Leo@users.noreply.github.com> Date: Fri, 25 Sep 2026 22:41:59 +0800 Subject: [PATCH 1/2] feat(agents): report peer route candidates without choosing one MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A coordinator holding a peer name and no task link had no published way to ask "does the state on record single out a route for this Agent?". The exact-link resolver answers which Agent a thread belongs to, and the directory projects identity, work and observation refs, but neither said how many bindings address a peer — so the cheapest available move was to substitute another worker, which is what #5039 reports. `summarize_agent_binding_routes` is the reverse lookup, kept in the binding owner so the accepted-candidate rule stays single-sourced: a binding whose host surface cannot be named is dropped there rather than counted here. It reports `resolved`, `ambiguous` or `unbound`, carries up to three candidates and the full distinct count, and selects nothing. `agent-directory` publishes it per row as `peer_route`, documented as a locator that grants no claim, lease, capability or cross-host resume authority. Verifying a selected route through the owning host, and separating delivery from receiver adoption and result return, stay in #5039: this adds no host call and no second registry. Signed-off-by: JunZ-Leo <100498253+JunZ-Leo@users.noreply.github.com> --- ...peer-agent-directory-and-observation-v0.md | 8 ++ loopx/control_plane/agents/directory.py | 7 + loopx/thread_agent_binding.py | 60 ++++++++- .../test_peer_agent_directory.py | 28 ++++ tests/test_thread_agent_binding.py | 124 ++++++++++++++++++ 5 files changed, 226 insertions(+), 1 deletion(-) diff --git a/docs/reference/protocols/peer-agent-directory-and-observation-v0.md b/docs/reference/protocols/peer-agent-directory-and-observation-v0.md index f88e0571ea..9ad06c90c1 100644 --- a/docs/reference/protocols/peer-agent-directory-and-observation-v0.md +++ b/docs/reference/protocols/peer-agent-directory-and-observation-v0.md @@ -131,6 +131,14 @@ 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, candidate_count, candidates[], provenance}` + with `outcome` in `resolved` (exactly one binding addresses the Agent), + `ambiguous` (more than one does) or `unbound` (none does). At most three + candidates are carried while `candidate_count` keeps the full total, so a short + list is a cap and not a disproof. This field selects nothing: `ambiguous` means + the caller must resolve an exact link before addressing the peer, and a route is + a locator, never 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**. diff --git a/loopx/control_plane/agents/directory.py b/loopx/control_plane/agents/directory.py index cb99162f89..384aa5840f 100644 --- a/loopx/control_plane/agents/directory.py +++ b/loopx/control_plane/agents/directory.py @@ -26,6 +26,7 @@ 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 ..todos.contract import normalize_todo_id from .management_projection import build_agent_management_projection @@ -187,6 +188,9 @@ 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 for row in agent_rows: @@ -210,6 +214,9 @@ def build_peer_agent_directory( "delivery_refs": _compact_refs( row.get("handoff_refs"), limit=MAX_DELIVERY_REFS ), + "peer_route": summarize_agent_binding_routes( + binding_goals, agent_id=agent_id + ), } rows.append( { diff --git a/loopx/thread_agent_binding.py b/loopx/thread_agent_binding.py index e965f249db..1d4b246042 100644 --- a/loopx/thread_agent_binding.py +++ b/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,12 @@ ) _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" +MAX_ROUTE_CANDIDATES = 3 +ROUTE_RESOLVED = "resolved" +ROUTE_AMBIGUOUS = "ambiguous" +ROUTE_UNBOUND = "unbound" + class ThreadBindingRequestError(ValueError): """The caller supplied an invalid host-thread identity.""" @@ -168,6 +174,58 @@ def resolve_thread_agent_binding( return base +def summarize_agent_binding_routes( + goals: Iterable[Any], + *, + agent_id: Any, +) -> dict[str, Any]: + """Report every published route that addresses one Agent, or say none selects. + + 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 whether the bindings on record single + one out at all. Candidates are exactly the entries the owner's own + normalisation accepts: a binding whose host surface cannot be named is + dropped there rather than counted here, so this never advertises a route the + resolver could not later match. Every accepted binding is counted, the first + `MAX_ROUTE_CANDIDATES` are carried, and `candidate_count` keeps the omitted + remainder visible. Nothing here picks a route, opens a session, or transfers + claim, lease or capability: several candidates is an answer of "ask", and the + caller still resolves an exact link before addressing the peer. + """ + + wanted = normalize_todo_claimed_by(agent_id) + candidates: list[dict[str, str]] = [] + if wanted: + for raw_goal in goals: + if not isinstance(raw_goal, dict): + continue + for binding in _bindings_for_goal(raw_goal): + if binding["agent_id"] != wanted or binding in candidates: + continue + candidates.append(binding) + if not candidates: + outcome = ROUTE_UNBOUND + elif len(candidates) == 1: + outcome = ROUTE_RESOLVED + else: + outcome = ROUTE_AMBIGUOUS + return { + "schema_version": AGENT_BINDING_ROUTE_SCHEMA_VERSION, + "agent_id": wanted or "", + "outcome": outcome, + "candidate_count": len(candidates), + "candidates": [ + { + "thread_id": item["thread_id"], + "host_surface": item["host_surface"], + } + for item in candidates[:MAX_ROUTE_CANDIDATES] + ], + "provenance": "run_history.goals[].coordination.thread_agent_bindings", + } + + def _registry_thread_binding_request( *, host_surface: str | None, diff --git a/tests/control_plane/test_peer_agent_directory.py b/tests/control_plane/test_peer_agent_directory.py index 05dad9739a..e3fc5162f7 100644 --- a/tests/control_plane/test_peer_agent_directory.py +++ b/tests/control_plane/test_peer_agent_directory.py @@ -150,3 +150,31 @@ 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"] == "ambiguous" + assert rows[WORKING_AGENT]["peer_route"]["candidate_count"] == 2 + assert rows[IDLE_AGENT]["peer_route"]["outcome"] == "unbound" + 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"] diff --git a/tests/test_thread_agent_binding.py b/tests/test_thread_agent_binding.py index 17065623a7..0a7312e081 100644 --- a/tests/test_thread_agent_binding.py +++ b/tests/test_thread_agent_binding.py @@ -10,12 +10,17 @@ from loopx.cli import main as cli_main from loopx.global_registry import global_registry_path from loopx.thread_agent_binding import ( + MAX_ROUTE_CANDIDATES, + ROUTE_AMBIGUOUS, + ROUTE_RESOLVED, + ROUTE_UNBOUND, ThreadBindingRequestError, bind_thread_agent_in_registry, codex_thread_deep_link_locator, normalize_thread_id, resolve_registry_thread_agent_binding, resolve_thread_agent_binding, + summarize_agent_binding_routes, unbind_thread_agent_in_registry, ) @@ -688,3 +693,122 @@ def test_unbind_is_idempotent_and_expected_agent_mismatch_fails_closed( assert missing["ok"] is True assert missing["changed"] is False assert path.read_bytes() == before + + +def _binding( + agent_id: str, + thread_id: str, + host_surface: str = "codex-app", +) -> dict[str, str]: + return { + "agent_id": agent_id, + "thread_id": thread_id, + "host_surface": host_surface, + } + + +def _goal(*bindings: dict[str, str]) -> dict[str, object]: + return {"coordination": {"thread_agent_bindings": [dict(b) for b in bindings]}} + + +def test_route_summary_resolves_only_when_one_binding_addresses_the_agent() -> None: + summary = summarize_agent_binding_routes( + [_goal(_binding("peer", "thread-one"))], + agent_id="peer", + ) + + assert summary["outcome"] == ROUTE_RESOLVED + assert summary["candidate_count"] == 1 + assert summary["candidates"] == [ + {"thread_id": "thread-one", "host_surface": "codex-app"} + ] + + +def test_route_summary_reports_ambiguous_instead_of_choosing_a_row() -> None: + summary = summarize_agent_binding_routes( + [ + _goal( + _binding("peer", "thread-old"), + _binding("peer", "thread-new", "codex-cli"), + ) + ], + agent_id="peer", + ) + + assert summary["outcome"] == ROUTE_AMBIGUOUS + assert [item["thread_id"] for item in summary["candidates"]] == [ + "thread-old", + "thread-new", + ] + + +def test_route_summary_counts_a_republished_binding_once() -> None: + summary = summarize_agent_binding_routes( + [ + _goal(_binding("peer", "thread-shared")), + _goal(_binding("peer", "thread-shared"), _binding("peer", "thread-extra")), + ], + agent_id="peer", + ) + + assert summary["candidate_count"] == 2 + assert summary["outcome"] == ROUTE_AMBIGUOUS + + +def test_route_summary_caps_candidates_without_losing_the_count() -> None: + summary = summarize_agent_binding_routes( + [_goal(*[_binding("peer", f"thread-{index}") for index in range(5)])], + agent_id="peer", + ) + + assert summary["candidate_count"] == 5 + assert len(summary["candidates"]) == MAX_ROUTE_CANDIDATES + assert summary["outcome"] == ROUTE_AMBIGUOUS + + +def test_route_summary_keeps_other_agents_out_of_the_candidates() -> None: + summary = summarize_agent_binding_routes( + [ + _goal( + _binding("peer", "thread-peer"), _binding("reviewer", "thread-reviewer") + ) + ], + agent_id="peer", + ) + + assert summary["candidate_count"] == 1 + assert summary["outcome"] == ROUTE_RESOLVED + + +def test_route_summary_excludes_a_binding_the_owner_cannot_normalise() -> None: + """No host surface is not a route, so it is not counted as one either.""" + + summary = summarize_agent_binding_routes( + [ + _goal( + _binding("peer", "thread-named"), + {"agent_id": "peer", "thread_id": "thread-surfaceless"}, + ) + ], + agent_id="peer", + ) + + assert summary["candidate_count"] == 1 + assert summary["outcome"] == ROUTE_RESOLVED + assert summary["candidates"] == [ + {"thread_id": "thread-named", "host_surface": "codex-app"} + ] + + +@pytest.mark.parametrize("agent_id", ["agent-absent", "", None, 42, "x" * 400]) +def test_route_summary_reports_unbound_for_any_unaddressable_agent( + agent_id: object, +) -> None: + summary = summarize_agent_binding_routes( + [_goal(_binding("peer", "thread-peer"))], + agent_id=agent_id, + ) + + assert summary["outcome"] == ROUTE_UNBOUND + assert summary["candidate_count"] == 0 + assert summary["candidates"] == [] From 57718d52f304386a4f8ee7d60d370a8d576fb33f Mon Sep 17 00:00:00 2001 From: JunZ-Leo <100498253+JunZ-Leo@users.noreply.github.com> Date: Sat, 26 Sep 2026 21:31:36 +0800 Subject: [PATCH 2/2] fix(agents): filter and qualify the published peer route MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two boundaries were wrong at exact head 3abef5b1b. The directory published `peer_route.candidates` straight from the binding owner's accepted identities. Normalisation proves a string can be matched, not that it may be shown: a `ghp_`-shaped thread identifier registered through the real write entry point reached the real `loopx agent-directory` JSON verbatim, which the projection's own public boundary rejects. Publication now applies that existing check at the directory edge, where the other row budgets already live. A candidate that does not pass is witheld rather than truncated — half a thread identifier locates nothing — and the withholding is declared through `withheld_candidate_count` plus the packet-level `route_candidate_withheld` limitation while `candidate_count` keeps counting it, so a filtered route can never be read as a binding that does not exist. The owner keeps the full internal view and no longer caps what it returns, because size is a publication choice. `resolved` also claimed more than the walk can prove. Filtering by the target Agent and then counting cannot see that one host thread may address a second registered Agent, which the forward resolver answers `conflict` for. The label is now `single_candidate` / `multiple_candidates` / `no_candidate`, paired with an explicit `address_shared` fact computed against the other identities in the same data, and `scope: goals_supplied` states that a lone candidate here is not a project-level uniqueness claim. Tests add the real write-then-CLI withholding regression, the directory to forward-resolver contradiction on a shared address, and the publication budget. Signed-off-by: JunZ-Leo <100498253+JunZ-Leo@users.noreply.github.com> --- ...peer-agent-directory-and-observation-v0.md | 23 ++- loopx/control_plane/agents/directory.py | 39 +++- loopx/thread_agent_binding.py | 91 +++++---- .../test_peer_agent_directory.py | 66 ++++++- tests/test_thread_agent_binding.py | 179 ++++++++++++------ 5 files changed, 296 insertions(+), 102 deletions(-) diff --git a/docs/reference/protocols/peer-agent-directory-and-observation-v0.md b/docs/reference/protocols/peer-agent-directory-and-observation-v0.md index 9ad06c90c1..3391d3a301 100644 --- a/docs/reference/protocols/peer-agent-directory-and-observation-v0.md +++ b/docs/reference/protocols/peer-agent-directory-and-observation-v0.md @@ -132,13 +132,22 @@ 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, candidate_count, candidates[], provenance}` - with `outcome` in `resolved` (exactly one binding addresses the Agent), - `ambiguous` (more than one does) or `unbound` (none does). At most three - candidates are carried while `candidate_count` keeps the full total, so a short - list is a cap and not a disproof. This field selects nothing: `ambiguous` means - the caller must resolve an exact link before addressing the peer, and a route is - a locator, never a claim, lease, capability or cross-host resume authority; + `{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**. diff --git a/loopx/control_plane/agents/directory.py b/loopx/control_plane/agents/directory.py index 384aa5840f..80379e6548 100644 --- a/loopx/control_plane/agents/directory.py +++ b/loopx/control_plane/agents/directory.py @@ -27,6 +27,7 @@ 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 @@ -36,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. @@ -44,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" @@ -75,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.""" @@ -193,6 +221,7 @@ def build_peer_agent_directory( 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: @@ -203,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, @@ -214,9 +247,7 @@ def build_peer_agent_directory( "delivery_refs": _compact_refs( row.get("handoff_refs"), limit=MAX_DELIVERY_REFS ), - "peer_route": summarize_agent_binding_routes( - binding_goals, agent_id=agent_id - ), + "peer_route": peer_route, } rows.append( { @@ -237,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, diff --git a/loopx/thread_agent_binding.py b/loopx/thread_agent_binding.py index 1d4b246042..89e2cc1957 100644 --- a/loopx/thread_agent_binding.py +++ b/loopx/thread_agent_binding.py @@ -28,10 +28,9 @@ _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" -MAX_ROUTE_CANDIDATES = 3 -ROUTE_RESOLVED = "resolved" -ROUTE_AMBIGUOUS = "ambiguous" -ROUTE_UNBOUND = "unbound" +ROUTE_SINGLE_CANDIDATE = "single_candidate" +ROUTE_MULTIPLE_CANDIDATES = "multiple_candidates" +ROUTE_NO_CANDIDATE = "no_candidate" class ThreadBindingRequestError(ValueError): @@ -174,54 +173,80 @@ 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]: - """Report every published route that addresses one Agent, or say none selects. + """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 whether the bindings on record single - one out at all. Candidates are exactly the entries the owner's own - normalisation accepts: a binding whose host surface cannot be named is - dropped there rather than counted here, so this never advertises a route the - resolver could not later match. Every accepted binding is counted, the first - `MAX_ROUTE_CANDIDATES` are carried, and `candidate_count` keeps the omitted - remainder visible. Nothing here picks a route, opens a session, or transfers - claim, lease or capability: several candidates is an answer of "ask", and the - caller still resolves an exact link before addressing the peer. + 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) - candidates: list[dict[str, str]] = [] - if wanted: - for raw_goal in goals: - if not isinstance(raw_goal, dict): - continue - for binding in _bindings_for_goal(raw_goal): - if binding["agent_id"] != wanted or binding in candidates: - continue - candidates.append(binding) + 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_UNBOUND + outcome = ROUTE_NO_CANDIDATE elif len(candidates) == 1: - outcome = ROUTE_RESOLVED + outcome = ROUTE_SINGLE_CANDIDATE else: - outcome = ROUTE_AMBIGUOUS + 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": [ - { - "thread_id": item["thread_id"], - "host_surface": item["host_surface"], - } - for item in candidates[:MAX_ROUTE_CANDIDATES] - ], + "candidates": candidates, + "scope": "goals_supplied", "provenance": "run_history.goals[].coordination.thread_agent_bindings", } diff --git a/tests/control_plane/test_peer_agent_directory.py b/tests/control_plane/test_peer_agent_directory.py index e3fc5162f7..b7d8af9796 100644 --- a/tests/control_plane/test_peer_agent_directory.py +++ b/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" @@ -171,10 +176,67 @@ def test_rows_report_a_peer_route_without_selecting_one() -> None: packet = build_peer_agent_directory(payload, caller_agent_id=WORKING_AGENT) rows = _rows_by_agent(packet) - assert rows[WORKING_AGENT]["peer_route"]["outcome"] == "ambiguous" + assert rows[WORKING_AGENT]["peer_route"]["outcome"] == "multiple_candidates" assert rows[WORKING_AGENT]["peer_route"]["candidate_count"] == 2 - assert rows[IDLE_AGENT]["peer_route"]["outcome"] == "unbound" + 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 diff --git a/tests/test_thread_agent_binding.py b/tests/test_thread_agent_binding.py index 0a7312e081..c3a36a9418 100644 --- a/tests/test_thread_agent_binding.py +++ b/tests/test_thread_agent_binding.py @@ -9,11 +9,11 @@ from loopx.cli import main as cli_main from loopx.global_registry import global_registry_path +from loopx.control_plane.runtime.public_safety import validate_public_safe_value from loopx.thread_agent_binding import ( - MAX_ROUTE_CANDIDATES, - ROUTE_AMBIGUOUS, - ROUTE_RESOLVED, - ROUTE_UNBOUND, + ROUTE_MULTIPLE_CANDIDATES, + ROUTE_NO_CANDIDATE, + ROUTE_SINGLE_CANDIDATE, ThreadBindingRequestError, bind_thread_agent_in_registry, codex_thread_deep_link_locator, @@ -711,31 +711,33 @@ def _goal(*bindings: dict[str, str]) -> dict[str, object]: return {"coordination": {"thread_agent_bindings": [dict(b) for b in bindings]}} -def test_route_summary_resolves_only_when_one_binding_addresses_the_agent() -> None: - summary = summarize_agent_binding_routes( - [_goal(_binding("peer", "thread-one"))], - agent_id="peer", - ) +def _routes(bindings: list[dict[str, str]], agent_id: str) -> dict[str, object]: + return summarize_agent_binding_routes([_goal(*bindings)], agent_id=agent_id) + - assert summary["outcome"] == ROUTE_RESOLVED +def test_route_summary_reports_a_single_candidate_and_keeps_full_identity() -> None: + summary = _routes([_binding("peer", "thread-one")], "peer") + + assert summary["outcome"] == ROUTE_SINGLE_CANDIDATE assert summary["candidate_count"] == 1 assert summary["candidates"] == [ {"thread_id": "thread-one", "host_surface": "codex-app"} ] + assert summary["address_shared"] is False + assert summary["scope"] == "goals_supplied" -def test_route_summary_reports_ambiguous_instead_of_choosing_a_row() -> None: - summary = summarize_agent_binding_routes( +def test_route_summary_reports_several_candidates_rather_than_choosing_one() -> None: + summary = _routes( [ - _goal( - _binding("peer", "thread-old"), - _binding("peer", "thread-new", "codex-cli"), - ) + _binding("peer", "thread-old"), + _binding("peer", "thread-new", host_surface="codex-cli"), ], - agent_id="peer", + "peer", ) - assert summary["outcome"] == ROUTE_AMBIGUOUS + assert summary["outcome"] == ROUTE_MULTIPLE_CANDIDATES + assert summary["candidate_count"] == 2 assert [item["thread_id"] for item in summary["candidates"]] == [ "thread-old", "thread-new", @@ -752,63 +754,126 @@ def test_route_summary_counts_a_republished_binding_once() -> None: ) assert summary["candidate_count"] == 2 - assert summary["outcome"] == ROUTE_AMBIGUOUS + assert summary["outcome"] == ROUTE_MULTIPLE_CANDIDATES + assert [item["thread_id"] for item in summary["candidates"]] == [ + "thread-shared", + "thread-extra", + ] -def test_route_summary_caps_candidates_without_losing_the_count() -> None: - summary = summarize_agent_binding_routes( - [_goal(*[_binding("peer", f"thread-{index}") for index in range(5)])], - agent_id="peer", - ) +def test_route_summary_never_caps_the_internal_view() -> None: + """Size is a publication budget, so the owner keeps every candidate.""" + + bindings = [_binding("peer", f"thread-{index}") for index in range(5)] + + summary = _routes(bindings, "peer") assert summary["candidate_count"] == 5 - assert len(summary["candidates"]) == MAX_ROUTE_CANDIDATES - assert summary["outcome"] == ROUTE_AMBIGUOUS + assert len(summary["candidates"]) == 5 def test_route_summary_keeps_other_agents_out_of_the_candidates() -> None: - summary = summarize_agent_binding_routes( - [ - _goal( - _binding("peer", "thread-peer"), _binding("reviewer", "thread-reviewer") - ) - ], - agent_id="peer", + summary = _routes( + [_binding("peer", "thread-peer"), _binding("reviewer", "thread-reviewer")], + "peer", ) assert summary["candidate_count"] == 1 - assert summary["outcome"] == ROUTE_RESOLVED + assert summary["outcome"] == ROUTE_SINGLE_CANDIDATE -def test_route_summary_excludes_a_binding_the_owner_cannot_normalise() -> None: - """No host surface is not a route, so it is not counted as one either.""" - +@pytest.mark.parametrize("agent_id", ["agent-absent", "", None, 42, "x" * 400]) +def test_route_summary_reports_no_candidate_for_any_unaddressable_agent( + agent_id: object, +) -> None: summary = summarize_agent_binding_routes( - [ - _goal( - _binding("peer", "thread-named"), - {"agent_id": "peer", "thread_id": "thread-surfaceless"}, - ) - ], - agent_id="peer", + [_goal(_binding("peer", "thread-peer"))], agent_id=agent_id ) - assert summary["candidate_count"] == 1 - assert summary["outcome"] == ROUTE_RESOLVED - assert summary["candidates"] == [ - {"thread_id": "thread-named", "host_surface": "codex-app"} + assert summary["outcome"] == ROUTE_NO_CANDIDATE + assert summary["candidate_count"] == 0 + assert summary["candidates"] == [] + + +def test_one_candidate_shared_by_two_agents_is_not_called_unique() -> None: + """The reverse view must not contradict the forward resolver on the same data. + + One host thread bound to two Agents is exactly the registry conflict + `resolve_thread_agent_binding` answers `conflict` for. Per-agent counts here + stay at one, so the label has to be `single_candidate` plus an explicit + shared-address fact rather than a resolved route. + """ + + bindings = [ + _binding("peer", "thread-shared"), + _binding("reviewer", "thread-shared"), ] + forward = resolve_thread_agent_binding( + _goal(*bindings), host_surface="codex-app", thread_id="thread-shared" + ) + for agent_id in ("peer", "reviewer"): + summary = _routes(bindings, agent_id) + assert summary["outcome"] == ROUTE_SINGLE_CANDIDATE + assert summary["candidate_count"] == 1 + assert summary["address_shared"] is True -@pytest.mark.parametrize("agent_id", ["agent-absent", "", None, 42, "x" * 400]) -def test_route_summary_reports_unbound_for_any_unaddressable_agent( - agent_id: object, + assert forward["status"] == "conflict" + assert sorted(item["agent_id"] for item in forward["matches"]) == [ + "peer", + "reviewer", + ] + + +def test_agent_directory_withholds_a_credential_shaped_thread_from_the_packet( + tmp_path, ) -> None: - summary = summarize_agent_binding_routes( - [_goal(_binding("peer", "thread-peer"))], - agent_id=agent_id, + """A binding the registry accepts can still be unfit to publish. + + Registered through the real write entry point, then read back through the + real CLI. The published packet must not carry the value, must not pretend + the binding is gone, and must announce the withholding. + """ + + registry = _registry(tmp_path, ["agent-a"]) + secret_thread = "ghp_" + "1234567890abcdefghijklmnopqrstuvwxyz1234" + assert ( + bind_thread_agent_in_registry( + registry_path=registry, + goal_id="goal", + host_surface="codex-app", + thread_id=secret_thread, + agent_id="agent-a", + execute=True, + )["ok"] + is True ) - assert summary["outcome"] == ROUTE_UNBOUND - assert summary["candidate_count"] == 0 - assert summary["candidates"] == [] + output = io.StringIO() + with contextlib.redirect_stdout(output): + exit_code = cli_main( + [ + "--registry", + str(registry), + "--format", + "json", + "agent-directory", + "--goal-id", + "goal", + "--agent-id", + "agent-a", + "--scan-path", + str(tmp_path), + ] + ) + + published = output.getvalue() + assert exit_code == 0 + assert secret_thread not in published + packet = json.loads(published) + row = next(row for row in packet["rows"] if row["agent_id"] == "agent-a") + assert row["peer_route"]["candidates"] == [] + assert row["peer_route"]["candidate_count"] == 1 + assert row["peer_route"]["withheld_candidate_count"] == 1 + assert "route_candidate_withheld" in packet["limitations"] + validate_public_safe_value(packet, path="peer_agent_directory")