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..3391d3a301 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,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**. diff --git a/loopx/control_plane/agents/directory.py b/loopx/control_plane/agents/directory.py index cb99162f89..80379e6548 100644 --- a/loopx/control_plane/agents/directory.py +++ b/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, diff --git a/loopx/thread_agent_binding.py b/loopx/thread_agent_binding.py index e965f249db..89e2cc1957 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,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, diff --git a/tests/control_plane/test_peer_agent_directory.py b/tests/control_plane/test_peer_agent_directory.py index 05dad9739a..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" @@ -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 diff --git a/tests/test_thread_agent_binding.py b/tests/test_thread_agent_binding.py index 17065623a7..c3a36a9418 100644 --- a/tests/test_thread_agent_binding.py +++ b/tests/test_thread_agent_binding.py @@ -9,13 +9,18 @@ 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 ( + ROUTE_MULTIPLE_CANDIDATES, + ROUTE_NO_CANDIDATE, + ROUTE_SINGLE_CANDIDATE, 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,187 @@ 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 _routes(bindings: list[dict[str, str]], agent_id: str) -> dict[str, object]: + return summarize_agent_binding_routes([_goal(*bindings)], agent_id=agent_id) + + +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_several_candidates_rather_than_choosing_one() -> None: + summary = _routes( + [ + _binding("peer", "thread-old"), + _binding("peer", "thread-new", host_surface="codex-cli"), + ], + "peer", + ) + + assert summary["outcome"] == ROUTE_MULTIPLE_CANDIDATES + assert summary["candidate_count"] == 2 + 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_MULTIPLE_CANDIDATES + assert [item["thread_id"] for item in summary["candidates"]] == [ + "thread-shared", + "thread-extra", + ] + + +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"]) == 5 + + +def test_route_summary_keeps_other_agents_out_of_the_candidates() -> None: + summary = _routes( + [_binding("peer", "thread-peer"), _binding("reviewer", "thread-reviewer")], + "peer", + ) + + assert summary["candidate_count"] == 1 + assert summary["outcome"] == ROUTE_SINGLE_CANDIDATE + + +@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-peer"))], agent_id=agent_id + ) + + 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 + + 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: + """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 + ) + + 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")