From 7705d0648f33b6419bc07574f9249c533fe6ae43 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Thu, 22 Oct 2026 16:25:56 +0800 Subject: [PATCH 1/2] fix(manager): observe peer routes on authorized SSH sources Signed-off-by: huangruiteng --- .../manager_context/inspection.py | 12 +-- .../manager_context/ssh_evidence.py | 48 +++++++++- loopx/codex_app_thread_activity.py | 6 +- tests/control_plane/test_peer_host_route.py | 5 +- tests/test_host_thread_activity.py | 10 ++- tests/test_manager_agent_discovery.py | 6 +- tests/test_manager_ssh_evidence.py | 88 ++++++++++++++++++- 7 files changed, 154 insertions(+), 21 deletions(-) diff --git a/loopx/capabilities/manager_context/inspection.py b/loopx/capabilities/manager_context/inspection.py index 7cb9052f9a..b30acdd74e 100644 --- a/loopx/capabilities/manager_context/inspection.py +++ b/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) diff --git a/loopx/capabilities/manager_context/ssh_evidence.py b/loopx/capabilities/manager_context/ssh_evidence.py index de9ba74b3b..bec15cb775 100644 --- a/loopx/capabilities/manager_context/ssh_evidence.py +++ b/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" diff --git a/loopx/codex_app_thread_activity.py b/loopx/codex_app_thread_activity.py index f1717d5c74..92bc39d4a6 100644 --- a/loopx/codex_app_thread_activity.py +++ b/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. diff --git a/tests/control_plane/test_peer_host_route.py b/tests/control_plane/test_peer_host_route.py index 21df7d0fcd..0290931181 100644 --- a/tests/control_plane/test_peer_host_route.py +++ b/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" diff --git a/tests/test_host_thread_activity.py b/tests/test_host_thread_activity.py index df36535199..389f67800e 100644 --- a/tests/test_host_thread_activity.py +++ b/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: diff --git a/tests/test_manager_agent_discovery.py b/tests/test_manager_agent_discovery.py index ca6ba47203..c7068404ba 100644 --- a/tests/test_manager_agent_discovery.py +++ b/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] diff --git a/tests/test_manager_ssh_evidence.py b/tests/test_manager_ssh_evidence.py index 6067bb21a1..c68444ec62 100644 --- a/tests/test_manager_ssh_evidence.py +++ b/tests/test_manager_ssh_evidence.py @@ -89,6 +89,86 @@ def test_remote_source_is_discovered_and_read_without_local_host_metadata(remote assert records[-1] == result +def _route_packet(**changes): + return {"ok": True, "schema_version": "loopx_peer_host_route_v0", + "goal_id": "remote-goal", "agent_id": "reviewer", "status": "resolved", + "reason": None, "candidate_count": 1, + "candidates": [{"host_surface": "codex-app-ssh", "thread_id": "original-task"}], + "selected_route": {"host_surface": "codex-app-ssh", "thread_id": "original-task"}, + "host_delivery": "not_attempted", "authority": "locator_only", **changes} + + +def test_remote_peer_route_uses_the_authorized_source_host(remote): + tool, calls, records, _, _ = remote + def run(argv, **kwargs): + calls.append((argv, kwargs)) + return SimpleNamespace(returncode=0, stdout=json.dumps(_route_packet())) + tool.remote_runner = run + query = {"view": "agent_route", "source_id": "ssh:research-host", + "goal_id": "remote-goal", "agent_id": "reviewer", "host_surface": "codex-app-ssh"} + result = tool.read(TOOL_NAME, query) + assert result["status"] == "resolved" and result["source_host"] == "research-host" + assert result["host_delivery"] == "not_attempted" + command = shlex.split(calls[0][0][-1]) + assert command[-7:] == ["resolve-peer-route", "--goal-id", "remote-goal", "--agent-id", "reviewer", + "--host-surface", "codex-app-ssh"] + assert records[-1] == result + + +@pytest.mark.parametrize("changes", [ + {"goal_id": "another-goal"}, {"agent_id": "another-agent"}, + {"host_delivery": "delivered"}, {"authority": "execution"}, + {"selected_route": {"host_surface": "codex-app", "thread_id": "different-task"}}, + {"status": "ambiguous"}, + {"status": "resolved", "selected_route": None}, + {"candidates": [{"host_surface": "codex-app-ssh", "thread_id": "/home/private/task"}]}, +]) +def test_remote_route_mismatch_never_becomes_delivery_or_another_target(remote, changes): + tool, calls, *_ = remote + def run(argv, **kwargs): + calls.append(argv) + return SimpleNamespace(returncode=0, stdout=json.dumps(_route_packet(**changes))) + tool.remote_runner = run + result = tool.read(TOOL_NAME, {"view": "agent_route", "source_id": "ssh:research-host", + "goal_id": "remote-goal", "agent_id": "reviewer", "host_surface": "codex-app-ssh"}) + assert len(calls) == 1 + assert result["reason_code"] == "remote_protocol_unavailable" + assert not result["ok"] and not result.get("selected_route") + + +def test_unavailable_remote_route_preserves_the_source_diagnostic(remote): + tool, *_ = remote + tool.remote_runner = lambda *a, **k: SimpleNamespace(returncode=0, stdout=json.dumps( + _route_packet(status="unavailable", reason="thread_not_found", selected_route=None))) + result = tool.read(TOOL_NAME, {"view": "agent_route", "source_id": "ssh:research-host", + "goal_id": "remote-goal", "agent_id": "reviewer"}) + assert result["status"] == "unavailable" and result["reason"] == "thread_not_found" + assert result["host_delivery"] == "not_attempted" and result["selected_route"] is None + + +def test_failed_remote_route_cannot_reuse_cached_portfolio(remote, monkeypatch): + tool, *_ = remote + def forbidden(*a, **k): + pytest.fail("route failure must not read historical portfolio cache") + monkeypatch.setattr("loopx.capabilities.manager_context.ssh_evidence._cached_entry", forbidden) + tool.remote_runner = lambda *a, **k: SimpleNamespace(returncode=255, stdout="", stderr="") + result = tool.read(TOOL_NAME, {"view": "agent_route", "source_id": "ssh:research-host", + "goal_id": "remote-goal", "agent_id": "reviewer"}) + assert result["ok"] is False and not result.get("selected_route") + assert result["rows"] == [] and not result.get("last_success_at") + + +def test_remote_route_revocation_discards_observed_identity(remote): + tool, _, _, channel, config = remote + def run(*args, **kwargs): + configure(tool.runtime_root, channel=channel, host="research-host", goal_ids=[], execute=True, config_path=config) + return SimpleNamespace(returncode=0, stdout=json.dumps(_route_packet())) + tool.remote_runner = run + result = tool.read(TOOL_NAME, {"view": "agent_route", "source_id": "ssh:research-host", + "goal_id": "remote-goal", "agent_id": "reviewer"}) + assert result == {"ok": False, "error": "authorization_changed"} + + @pytest.mark.parametrize( "query", [ @@ -706,12 +786,12 @@ def test_scope_change_between_reads_refuses_the_packet(tmp_path): @pytest.mark.parametrize( "stderr,returncode,expected", [ - ("huangruiteng@203.0.113.7: Permission denied (gssapi-with-mic).", 255, "ssh_auth_required"), + ("operator@203.0.113.7: Permission denied (gssapi-with-mic).", 255, "ssh_auth_required"), # A remote command reporting a bare permission error about its own # files is not this machine's rejected credential. ("loopx: /home/x/.config: Permission denied", 1, "remote_evidence_unavailable"), ("ssh: connect to host 203.0.113.7 port 22: Operation timed out", 255, "ssh_host_unreachable"), - ("ssh: Could not resolve hostname ark-devbox: Name or service not known", 255, "ssh_host_unreachable"), + ("ssh: Could not resolve hostname research-host: Name or service not known", 255, "ssh_host_unreachable"), ("bash: line 1: /home/x/.local/bin/loopx: No such file or directory", 127, "remote_cli_missing"), ("some other ssh failure", 255, "remote_evidence_unavailable"), ], @@ -739,7 +819,7 @@ def runner(argv, **kwargs): argv, 255, stdout="", - stderr="huangruiteng@203.0.113.7: Permission denied (gssapi-with-mic).", + stderr="operator@203.0.113.7: Permission denied (gssapi-with-mic).", ) packet = remote_evidence( @@ -769,7 +849,7 @@ def runner(argv, **kwargs): "stderr,returncode,expected_limitation", [ ( - "huangruiteng@203.0.113.7: Permission denied (gssapi-with-mic).", + "operator@203.0.113.7: Permission denied (gssapi-with-mic).", 255, "remote_source_ssh_auth_required", ), From 953e20eda13713c19d148cfdcac0f9bc51e6de38 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Thu, 22 Oct 2026 16:26:23 +0800 Subject: [PATCH 2/2] docs(manager): explain source-host observation and delivery boundaries Signed-off-by: huangruiteng --- docs/integrations/attached-agent-session-broker.md | 9 ++++++--- .../protocols/peer-agent-directory-and-observation-v0.md | 9 +++++++-- .../manager_context/skills/loopx-manager/SKILL.md | 8 +++++--- 3 files changed, 18 insertions(+), 8 deletions(-) diff --git a/docs/integrations/attached-agent-session-broker.md b/docs/integrations/attached-agent-session-broker.md index 1e41380617..ea9a4b0b2a 100644 --- a/docs/integrations/attached-agent-session-broker.md +++ b/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 `/state_.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. 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 3d22d4f324..56e0960018 100644 --- a/docs/reference/protocols/peer-agent-directory-and-observation-v0.md +++ b/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. diff --git a/loopx/capabilities/manager_context/skills/loopx-manager/SKILL.md b/loopx/capabilities/manager_context/skills/loopx-manager/SKILL.md index 6cd2183c9f..08aae64dcb 100644 --- a/loopx/capabilities/manager_context/skills/loopx-manager/SKILL.md +++ b/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