fix(manager): resolve existing agent routes on authorized SSH sources

docs/integrations/attached-agent-session-broker.md

@@ -98,10 +98,13 @@ turn marker restores its observed state.
 | `archived` | The host archived the thread. |
 | `unknown` | Not observable; `reason` is `unsupported_host`, `store_unavailable`, `thread_not_found`, `record_unrecognized` or `no_turn_marker`. |
 
-Codex local surfaces (`codex-app`, `codex-cli-tui`, `codex-ide-plugin`) are read
+Codex surfaces (`codex-app`, `codex-app-ssh`, `codex-cli-tui`, `codex-ide-plugin`) are read
 from `<CODEX_HOME>/state_<n>.sqlite` and the thread's rollout JSONL. The Codex
 store is not a public contract: any shape the adapter does not recognize is
-`unknown`, never `turn_open`. Remote surfaces such as `codex-app-ssh` are
-`unsupported_host`. Homes are searched in the order `CODEX_HOME`, `~/.codex`,
+`unknown`, never `turn_open`. An SSH session's store is local on its source
+host, so `codex-app-ssh` now uses this same observer instead of always returning
+`unsupported_host`. A missing store stays unknown; the observer does not connect
+to another host, deliver a message or qualify live execution. Homes are searched
+in the order `CODEX_HOME`, `~/.codex`,
 then sibling `~/.codex-*` directories; set `LOOPX_CODEX_HOMES` (an
 `os.pathsep`-separated list) to search exactly those homes instead.

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

@@ -183,8 +183,13 @@ This changes the previous default refusal for *all* multiple bindings: archived
 history no longer requires the owner to supply a link. It applies to the local
 CLI and `manager-inbox request --require-host-route`, including trusted steward
 and peer callers. Readable records are still neither fresh runtime liveness nor
-model/capacity qualification. Restricted Chat and remote hosts do not acquire a
-new observer or execution path. Registration, the scoped candidate set and the
+model/capacity qualification. Manager `agent_route` reads now also use an
+authorized SSH evidence source and execute the same resolver on that source
+host; they no longer reject every remote source. Existing source/Goal grants
+are rechecked after the read, and cached portfolio evidence cannot qualify a
+route. Restricted Chat gains no host tools or execution permission. SSH-origin
+Codex bindings use the source host's local store when available; a missing or
+unrecognized store remains unknown. Registration, the scoped candidate set and the
 selected identity are rechecked after host reads; a changed set remains
 unresolved, while a mere reorder does not change the route. Request replay
 keeps its original exact route.

loopx/capabilities/manager_context/inspection.py

@@ -19,7 +19,7 @@
     "name": TOOL_NAME,
     "description": (
         "Read authorized LoopX Core evidence on demand: the global Goal portfolio, "
-        "registered Agent responsibilities, one Goal's current Todos, recorded deliveries, or handoff receipt status. Use view=agents to search before reporting a missing worker; delivery targets are not the discovery inventory. Use view=agent_route with exact goal_id and agent_id to observe an existing local host binding before asking the user for a task link. A readable route is not execution readiness or delivery authority. "
+        "registered Agent responsibilities, one Goal's current Todos, recorded deliveries, or handoff receipt status. Use view=agents to search before reporting a missing worker; delivery targets are not the discovery inventory. Use view=agent_route with exact goal_id and agent_id to observe a binding on the selected source host before asking the user for a task link. A readable route is not execution readiness or delivery authority. "
         "For a current-work overview, start with the index's bounded Agent Todo summaries; "
         "use view=portfolio for source and coverage, and view=todos for needed detail. "
         "Delivery history answers past results, not live activity. "
@@ -45,6 +45,8 @@
                          "description": "Agent_route only, with goal_id: exact registered Agent to locate. Search view=agents first; does not delegate or create an execution driver."},
             "thread_link": {"type": "string", "minLength": 1,
                             "description": "Agent_route only: optional exact host task link already supplied by the user. Do not guess a link to resolve ambiguous bindings."},
+            "host_surface": {"type": "string", "minLength": 1,
+                             "description": "Agent_route only: optionally narrow to an observed binding's exact host surface. This does not select another machine or grant delivery."},
             "todo_id": {"type": "string", "minLength": 1,
                         "description": "Todos only, with goal_id: read one exact record including its complete permitted text. Recover content_truncated excerpts; a completed record does not authorize new work."},
             "query": {"type": "string", "maxLength": 200, "description": "Agents only: case-insensitive text match on identity and declared responsibility. Omit to browse all permitted registrations."},
@@ -109,7 +111,7 @@ def rejected_read_arguments(arguments: dict[str, Any]) -> list[str]:
         rejected.append("view:must_be_one_of_" + ",".join(READ_VIEWS))
     if "request_id" in arguments and view != "handoffs":
         rejected.append("request_id:only_for_view_handoffs")
-    for name in ("agent_id", "thread_link"):
+    for name in ("agent_id", "thread_link", "host_surface"):
         if name in arguments:
             if view != "agent_route":
                 rejected.append(f"{name}:only_for_view_agent_route")
@@ -405,9 +407,6 @@ def read(self, tool: str, arguments: Any) -> dict[str, Any]:
             self.record(result)
             return result
         if source_id != "local":
-            if view == "agent_route":
-                return {"ok": False, "error": "remote_agent_route_not_supported",
-                        "detail": "Observe bindings on the trusted source host with resolve-peer-route; a local host observation cannot qualify a remote route."}
             if not source_id.startswith("ssh:") or view == "handoffs" or (view not in {"portfolio", "agents"} and not goal_id):
                 return {"ok": False, "error": "invalid_remote_read"}
             from .ssh_evidence import read_remote
@@ -430,7 +429,8 @@ def read(self, tool: str, arguments: Any) -> dict[str, Any]:
 
                 result = resolve_peer_host_route(self.registry_path, goal_id=goal_id,
                                                 agent_id=arguments["agent_id"],
-                                                thread_link=arguments.get("thread_link"))
+                                                thread_link=arguments.get("thread_link"),
+                                                host_surface=arguments.get("host_surface"))
                 if not self.scope_valid():
                     return {"ok": False, "error": "authorization_changed"}
                 self.record(result)

loopx/capabilities/manager_context/skills/loopx-manager/SKILL.md

@@ -141,9 +141,11 @@ When a user names an existing peer for review or other work, keep that Agent
 identity. The directory's `peer_route` is a candidate list, not a selected or
 verified host task. In App/Goal conversations use the existing read tool with
 `view=agent_route`, exact `goal_id` and `agent_id` after discovery. This reuses
-the shared local route observation and audience scope; it is not executor
-readiness or a message submission. Remote sources remain explicitly unsupported
-by this read view. On a trusted host with CLI access, use `resolve-peer-route`
+the shared route observation and audience scope; it is not executor
+readiness or a message submission. For an authorized SSH source, pass its
+`source_id` so observation runs on that source host. Use `host_surface` only
+to narrow a known binding, never to choose a machine or infer authority.
+On a trusted host with CLI access, use `resolve-peer-route`
 for the exact Goal and Agent. It may select one readable task when every other
 binding is explicitly archived. If it still reports ambiguity, use the user's
 exact task link or report the gap; never treat unknown as archived. If a host message tool is

loopx/capabilities/manager_context/ssh_evidence.py

@@ -169,6 +169,7 @@ def read_remote(
     ):
         return {"ok": False, "error": "source_outside_available_scope"}
     goal = args.get("goal_id")
+    route_read = args.get("view") == "agent_route"
     if goal and not owner and goal not in before[host]:
         return {"ok": False, "error": "goal_outside_available_scope"}
 
@@ -181,6 +182,9 @@ def unavailable(code: str, reason: str) -> dict[str, Any]:
             or host not in configured_ssh_host_aliases(config_path)
         ):
             return {"ok": False, "error": "authorization_changed"}
+        # A historical portfolio cannot establish a current host route.
+        if route_read:
+            return _unavailable_read(args, code, reason)
         return _unavailable_read_with_cached_portfolio(
             root,
             channel,
@@ -210,6 +214,11 @@ def unavailable(code: str, reason: str) -> dict[str, Any]:
         argv += ["--todo-id", args["todo_id"]]
     if args.get("include_stopped"):
         argv += ["--include-stopped"]
+    if route_read:
+        argv = ["resolve-peer-route", "--goal-id", goal, "--agent-id", args["agent_id"]]
+        for field, flag in (("thread_link", "--thread-link"), ("host_surface", "--host-surface")):
+            if args.get(field):
+                argv += [flag, args[field]]
     command = (
         'exec "$HOME/.local/bin/loopx" --registry @host-global --format json '
         + shlex.join(argv)
@@ -248,7 +257,7 @@ def unavailable(code: str, reason: str) -> dict[str, Any]:
             )
             return unavailable(code, reason)
         packet = json.loads(result.stdout)
-        if (
+        if not route_read and (
             not isinstance(packet, dict)
             or packet.get("schema_version") != "manager_evidence_page_v1"
             or not isinstance(packet.get("rows"), list)
@@ -264,6 +273,13 @@ def unavailable(code: str, reason: str) -> dict[str, Any]:
             or host not in configured_ssh_host_aliases(config_path)
         ):
             return {"ok": False, "error": "authorization_changed"}
+        if route_read:
+            # Wire validation only. The source host's existing typed route
+            # owner selects the task; this adapter cannot infer a new route or
+            # promote a locator to delivery/execution authority.
+            if not _valid_peer_route_packet(packet, args):
+                return unavailable(*_remote_read_failure(returncode=0, stderr="", protocol=True))
+            return {**packet, "source_id": args["source_id"], "source_host": host}
         if not owner and any(
             r.get("goal_id") not in before[host] for r in packet["rows"]
         ):
@@ -283,6 +299,36 @@ def unavailable(code: str, reason: str) -> dict[str, Any]:
         return unavailable(code, reason)
 
 
+def _valid_peer_route_packet(packet: Any, args: Mapping[str, Any]) -> bool:
+    from ...control_plane.collaboration.peer_host_route import (
+        MAX_PUBLISHED_CANDIDATES, PEER_HOST_ROUTE_SCHEMA_VERSION,
+    )
+    if (not isinstance(packet, dict)
+            or packet.get("schema_version") != PEER_HOST_ROUTE_SCHEMA_VERSION
+            or packet.get("ok") is not True
+            or any(packet.get(key) != args[key] for key in ("goal_id", "agent_id"))
+            or packet.get("authority") != "locator_only"
+            or packet.get("host_delivery") != "not_attempted"
+            or not isinstance(packet.get("candidates"), list)
+            or len(packet["candidates"]) > MAX_PUBLISHED_CANDIDATES):
+        return False
+    selected = packet.get("selected_route")
+    if packet.get("status") == "resolved":
+        if (not isinstance(selected, dict) or set(selected) != {"host_surface", "thread_id"}
+                or not all(isinstance(v, str) and v for v in selected.values())
+                or (args.get("host_surface") and selected["host_surface"] != args["host_surface"])):
+            return False
+        if args.get("thread_link"):
+            from ...thread_agent_binding import codex_thread_deep_link_locator
+            if selected["thread_id"] != codex_thread_deep_link_locator(args["thread_link"])["thread_id"]:
+                return False
+    elif packet.get("status") not in {"unavailable", "ambiguous", "not_authorized"} or selected is not None:
+        return False
+    from ...control_plane.runtime.public_safety import validate_public_safe_value
+    validate_public_safe_value(packet, path="remote_peer_route")
+    return True
+
+
 def _cache_dir(root) -> Path:
     return _root(root) / "remote-reads"
 

loopx/codex_app_thread_activity.py

@@ -29,8 +29,10 @@
     HostThreadUnknownReason,
 )
 
-# Remote surfaces such as codex-app-ssh keep their store on another machine.
-CODEX_LOCAL_STORE_SURFACES = frozenset({"codex-app", "codex-cli-tui", "codex-ide-plugin"})
+# The surface describes the session, not the observer's machine. On an SSH
+# source host a codex-app-ssh task has a local store too. Missing local records
+# remain unknown; this observer never connects to another host or starts work.
+CODEX_LOCAL_STORE_SURFACES = frozenset({"codex-app", "codex-app-ssh", "codex-cli-tui", "codex-ide-plugin"})
 
 # An installed app automation drives the app surface; the home's other local
 # surfaces keep their own loop, so the automation cadence is not theirs to carry.

tests/control_plane/test_peer_host_route.py

@@ -189,8 +189,9 @@ def forbidden(ids):
     assert result["selected_route"] is None
 
 
-def test_real_cli_uses_read_only_host_store_and_pins_one_request_after_archived_history(tmp_path):
-    registry = _registry(tmp_path, [_binding("old"), _binding("current")])
+@pytest.mark.parametrize("surface", ["codex-app", "codex-app-ssh"])
+def test_real_cli_uses_source_host_store_and_pins_one_request_after_archived_history(tmp_path, surface):
+    registry = _registry(tmp_path, [_binding("old", host_surface=surface), _binding("current", host_surface=surface)])
     home = tmp_path / "codex-home"
     home.mkdir()
     rollout = home / "current.jsonl"

tests/test_host_thread_activity.py

@@ -388,9 +388,13 @@ def _send_error(self, message: str, **kwargs: Any) -> None:
     ]
 
 
-def test_remote_codex_surfaces_have_no_local_observer() -> None:
-    assert "codex-app-ssh" not in codex_thread_observers()
-    assert {"codex-app", "codex-cli-tui", "codex-ide-plugin"} <= set(codex_thread_observers())
+def test_ssh_origin_uses_source_host_observer_and_missing_store_stays_unknown(tmp_path, monkeypatch) -> None:
+    monkeypatch.setenv("LOOPX_CODEX_HOMES", str(tmp_path / "missing-home"))
+    observers = codex_thread_observers()
+    assert {"codex-app", "codex-app-ssh", "codex-cli-tui", "codex-ide-plugin"} <= set(observers)
+    result = observers["codex-app-ssh"](["remote-task"])["remote-task"]
+    assert result.state is HostThreadState.UNKNOWN
+    assert result.reason is HostThreadUnknownReason.THREAD_NOT_FOUND
 
 
 def test_automation_rrule_interval_reads_only_minutely_intervals() -> None:

tests/test_manager_agent_discovery.py

@@ -256,9 +256,9 @@ def test_invalid_route_read_is_repairable_and_effect_free(tmp_path, arguments):
     assert not records
 
 
-def test_remote_route_read_does_not_silently_inspect_local_host(tmp_path):
+def test_unconfigured_remote_route_cannot_fall_back_to_local_host(tmp_path):
     _, reader, records = setup(tmp_path)
     result = reader.read(TOOL_NAME, {"view": "agent_route", "source_id": "ssh:remote",
         "goal_id": "research", "agent_id": "worker-34"})
-    assert result["error"] == "remote_agent_route_not_supported"
-    assert not records
+    assert result["error"] == "source_outside_available_scope"
+    assert records == [result]