From 9a1ad2d881bc7f9a6c0e8c7234ee2a5bbb75764c Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 3 Nov 2026 02:04:41 +0800 Subject: [PATCH] feat(manager-context): expose bound host surfaces and route execution identity Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/capabilities/manager_context/README.md | 8 +++- .../capabilities/manager_context/discovery.py | 36 ++++++++++---- loopx/codex_app_thread_activity.py | 48 +++++++++++++++++++ .../collaboration/peer_host_route.py | 12 ++++- .../runtime/execution_identity.ts | 16 +++++-- .../execution_identity.test.ts | 18 +++++++ tests/test_manager_agent_discovery.py | 18 +++++-- 7 files changed, 135 insertions(+), 21 deletions(-) diff --git a/loopx/capabilities/manager_context/README.md b/loopx/capabilities/manager_context/README.md index 8dbb6c0f50..9c82956359 100644 --- a/loopx/capabilities/manager_context/README.md +++ b/loopx/capabilities/manager_context/README.md @@ -457,10 +457,14 @@ not a missing Agent; delivery still rechecks the existing authority. Stopped identities are available with `include_stopped=true` for historical questions. Unreadable/ambiguous inventory remains unknown, not an empty successful search. -Each included row also exposes `registered_host_binding`: the existing binding +Each included row also exposes `host_surfaces` (the sorted public-safe host +surface kinds bound in that Goal, without thread addresses) and +`registered_host_binding`: the existing binding owner's `outcome`, `candidate_count`, `address_shared` and `scope` for that Goal. The page does not expose host addresses or read host stores; `view=agent_route` -remains the exact-identity, current-observation path. A registration or binding +remains the exact-identity, current-observation path (which also reports +`execution_identity` from the selected Codex rollout's latest recorded Turn +when available, without reading mutable store preferences). A registration or binding does not prove presence, execution readiness or delivery authority. Repeated role names alone do not require an owner question. Resolve the subject, diff --git a/loopx/capabilities/manager_context/discovery.py b/loopx/capabilities/manager_context/discovery.py index 83f69a863e..3ef85b4ae4 100644 --- a/loopx/capabilities/manager_context/discovery.py +++ b/loopx/capabilities/manager_context/discovery.py @@ -72,6 +72,14 @@ def agent_page( continue for aid in sorted(registered_agent_ids_for_goal(goal)): profile = agent_profile_for_goal(goal, aid) or {} + bindings = summarize_agent_binding_routes([goal], agent_id=aid) + host_surfaces = sorted( + { + c["host_surface"] + for c in bindings["candidates"] + if isinstance(c.get("host_surface"), str) and c["host_surface"] + } + ) row = { "goal_id": gid, "agent_id": aid, "registered": True, "goal_description": public_safe_compact_text( @@ -87,22 +95,30 @@ def agent_page( ContextDelivery.ALLOWED if delivery_known and (gid, aid) in allowed else ContextDelivery.NOT_GRANTED if delivery_known else ContextDelivery.NOT_CHECKED ), + "host_surfaces": host_surfaces, + "registered_host_binding": { + key: bindings[key] for key in + ("outcome", "candidate_count", "address_shared", "scope") + }, } if needle and needle not in " ".join( - str(row[k] or "") for k in - ("goal_id", "agent_id", "goal_description", "profile_role", "scope_summary") + [ + *( + str(row[k] or "") + for k in ( + "goal_id", + "agent_id", + "goal_description", + "profile_role", + "scope_summary", + ) + ), + *host_surfaces, + ] ).casefold(): continue rows.append(row) page = rows[offset:offset + limit] - by_goal = {goal["id"]: goal for goal in visible} - for row in page: - bindings = summarize_agent_binding_routes( - [by_goal[row["goal_id"]]], agent_id=row["agent_id"]) - row["registered_host_binding"] = { - key: bindings[key] for key in - ("outcome", "candidate_count", "address_shared", "scope") - } end = offset + len(page) return { "ok": True, "view": "agents", "rows": page, "offset": offset, diff --git a/loopx/codex_app_thread_activity.py b/loopx/codex_app_thread_activity.py index 565fe659f0..c81cfa11c4 100644 --- a/loopx/codex_app_thread_activity.py +++ b/loopx/codex_app_thread_activity.py @@ -159,6 +159,54 @@ def current_codex_execution_identity( return {"status": "unavailable", "reason": "host_runtime_unavailable"} +def observe_codex_thread_execution_identity( + thread_id: str, + *, + homes: list[Path] | None = None, +) -> dict[str, Any]: + """Read the latest recorded Turn identity for a bound Codex thread across homes.""" + from .control_plane.effect_runtime import effect_runtime_result + + normalized = str(thread_id or "").strip() + if not normalized: + return {"status": "unavailable", "reason": "session_not_bound"} + search_homes = codex_homes() if homes is None else homes + fallback = "session_not_found" + for home in search_homes: + try: + db = _state_db(home) + if db is None: + continue + rows = _thread_rows(db, [normalized]) + except _UnrecognizedStore: + fallback = "host_store_unavailable" + continue + except (OSError, sqlite3.Error): + if fallback == "session_not_found": + fallback = "host_store_unavailable" + continue + row = rows.get(normalized) + if row is None: + continue + path, archived = row + if archived != 0 or not isinstance(path, str) or not path: + return {"status": "unavailable", "reason": "session_record_unavailable"} + try: + result: dict[str, Any] = effect_runtime_result( + "runtime.execution_identity.codex", + { + "home": str(home.expanduser().resolve()), + "path": path, + "thread_id": normalized, + "allow_completed_turn": True, + }, + ) + return result + except RuntimeError: + return {"status": "unavailable", "reason": "host_runtime_unavailable"} + return {"status": "unavailable", "reason": fallback} + + def _rollout_activity( rollout_path: Any, home: Path, *, thread_id: str, cache_root: Path | None = None, ) -> HostThreadActivity: diff --git a/loopx/control_plane/collaboration/peer_host_route.py b/loopx/control_plane/collaboration/peer_host_route.py index 81af813625..596f71b22e 100644 --- a/loopx/control_plane/collaboration/peer_host_route.py +++ b/loopx/control_plane/collaboration/peer_host_route.py @@ -12,7 +12,10 @@ from typing import Any from ...agent_registry import registered_agent_ids_for_goal -from ...codex_app_thread_activity import codex_thread_observers +from ...codex_app_thread_activity import ( + codex_thread_observers, + observe_codex_thread_execution_identity, +) from ...control_plane.agents.host_thread_activity import ( HostThreadActivity, HostThreadObserver, @@ -196,6 +199,11 @@ def resolve_peer_host_route( return result index = selection["selected_index"] selected = matching[index] + execution_identity = ( + observe_codex_thread_execution_identity(selected["thread_id"]) + if observers is None and selected["host_surface"] in CODEX_THREAD_HOST_SURFACES + else None + ) # Host I/O must not hide a newly bound alternative or a revoked registration. try: latest_registry = _read_route_registry(registry_path) @@ -230,5 +238,7 @@ def resolve_peer_host_route( selected_activity = activity[index] assert selected_activity is not None # The typed selector requires a readable observation. result["host_observation"] = selected_activity.to_payload() + if execution_identity is not None: + result["execution_identity"] = execution_identity result.update(status="resolved", reason=None, selected_route=selected) return result diff --git a/loopx/control_plane/runtime/execution_identity.ts b/loopx/control_plane/runtime/execution_identity.ts index 2bd0c25bd5..6d2985de65 100644 --- a/loopx/control_plane/runtime/execution_identity.ts +++ b/loopx/control_plane/runtime/execution_identity.ts @@ -11,7 +11,7 @@ export type ExecutionIdentity = UnknownIdentity | RecordedIdentity; const unavailable = (reason: string): UnknownIdentity => ({status: "unavailable", reason}); const object = (value: unknown): value is Record => value !== null && typeof value === "object" && !Array.isArray(value); -const providers = new Map([["openai", "OpenAI"], ["anthropic", "Anthropic"], ["google", "Google"]]); +const providers = new Map([["openai", "OpenAI"], ["anthropic", "Anthropic"], ["google", "Google"], ["deepseek", "DeepSeek"]]); // Only a provider's ordinary model name is publishable. Endpoint/router labels // remain unavailable; do not guess a provider from a model prefix or configuration. const publicModel = (value: unknown): value is string => typeof value === "string" @@ -19,7 +19,7 @@ const publicModel = (value: unknown): value is string => typeof value === "strin /** Read its latest recorded Turn identity, without claiming execution/authority. */ export async function readCodexExecutionIdentity(params: Record): Promise { - const {home, path, thread_id: thread} = params; + const {home, path, thread_id: thread, allow_completed_turn: allowCompletedTurn} = params; if (typeof home !== "string" || typeof path !== "string" || typeof thread !== "string" || !thread) return unavailable("session_not_bound"); try { @@ -66,6 +66,7 @@ export async function readCodexExecutionIdentity(params: Record return {status: "runtime_reported", model: context.model, provider, reasoning_effort: effort, observation_id, active_turn_verified}; }; + let finished = false; for (const line of lines.reverse()) { if (!line.trim()) continue; // Parse only metadata envelopes, never project conversational content. @@ -74,14 +75,19 @@ export async function readCodexExecutionIdentity(params: Record const payload = record.payload; if (record.type === "turn_context" && !context) context = payload; if (record.type !== "event_msg") continue; - if (["task_complete", "task_completed", "turn_aborted"].includes(String(payload.type))) - return unavailable("no_active_turn"); + if (["task_complete", "task_completed", "turn_aborted"].includes(String(payload.type))) { + if (allowCompletedTurn === true && !context) { + finished = true; + continue; + } + return unavailable(allowCompletedTurn === true ? "turn_identity_unavailable" : "no_active_turn"); + } if (payload.type !== "task_started") continue; if (!context || typeof payload.turn_id !== "string" || !payload.turn_id || context.turn_id !== payload.turn_id) return unavailable("turn_identity_unavailable"); const end = await file.stat(); if (end.ino !== start.ino || end.size < start.size) return unavailable("source_changed_during_read"); - return project(true); + return project(!finished); } // Long Turns can push their opening event outside this bounded read. // The latest context still records its model, but it cannot prove liveness. diff --git a/tests/control_plane_ts/execution_identity.test.ts b/tests/control_plane_ts/execution_identity.test.ts index 222f4a90ba..1e2ffbebf7 100644 --- a/tests/control_plane_ts/execution_identity.test.ts +++ b/tests/control_plane_ts/execution_identity.test.ts @@ -61,6 +61,24 @@ test("a Turn's recorded provider supersedes the opening provider", async t => { if (observed.status === "runtime_reported") assert.equal(observed.provider, "Anthropic"); }); +test("completed Turn identity is reported with active_turn_verified=false only when allow_completed_turn is set", async t => { + const input = await fixture(t, [ + record("session_meta", {id: "current", model_provider: "deepseek"}), + start("done"), + context("done", "deepseek-flash", "high"), + record("event_msg", {type: "task_complete", turn_id: "done"}), + ]); + assert.deepEqual(await readCodexExecutionIdentity(input), {status: "unavailable", reason: "no_active_turn"}); + const observed = await readCodexExecutionIdentity({...input, allow_completed_turn: true}); + assert.equal(observed.status, "runtime_reported"); + if (observed.status === "runtime_reported") { + assert.equal(observed.model, "deepseek-flash"); + assert.equal(observed.provider, "DeepSeek"); + assert.equal(observed.reasoning_effort, "high"); + assert.equal(observed.active_turn_verified, false); + } +}); + test("missing effort remains null; source stays in its selected home", async t => { const input = await fixture(t, [meta(), start(), context("new", "example-model-2", null)]); const observed = await readCodexExecutionIdentity(input); diff --git a/tests/test_manager_agent_discovery.py b/tests/test_manager_agent_discovery.py index c7068404ba..265206f9d2 100644 --- a/tests/test_manager_agent_discovery.py +++ b/tests/test_manager_agent_discovery.py @@ -77,17 +77,20 @@ def no_host_access(*args, **kwargs): result = reader.read(TOOL_NAME, {"view": "agents", "query": "operations"}) assert result["matched"] == 2 rows = {row["goal_id"]: row for row in result["rows"]} + assert rows["research"]["host_surfaces"] == ["codex-app"] assert rows["research"]["registered_host_binding"] == { "outcome": "single_candidate", "candidate_count": 1, "address_shared": False, "scope": "goals_supplied", } + assert rows["new-work"]["host_surfaces"] == [] assert rows["new-work"]["registered_host_binding"]["outcome"] == "no_candidate" assert rows["new-work"]["registered_host_binding"]["candidate_count"] == 0 + assert reader.read(TOOL_NAME, {"view": "agents", "query": "codex-app"})["matched"] == 1 assert all(row["execution_readiness"] == "not_checked" for row in rows.values()) assert all(row["context_delivery"] == "not_granted" for row in rows.values()) assert "private-host-address" not in json.dumps(result) assert "hidden-worker" not in json.dumps(result) - assert records == [result] + assert records[0] == result assert not (tmp_path / "manager-context").exists() @@ -183,8 +186,12 @@ def test_conversation_resolves_existing_peer_from_real_read_only_host_store(tmp_ home = tmp_path / "host" home.mkdir() rollout = home / "current.jsonl" - rollout.write_text(json.dumps({"type": "session_meta", "payload": {"id": "current"}}) + "\n" + json.dumps({"timestamp": "2026-01-01T00:00:00Z", "type": "event_msg", - "payload": {"type": "task_complete"}}) + "\n") + rollout.write_text( + json.dumps({"type": "session_meta", "payload": {"id": "current", "model_provider": "openai"}}) + "\n" + + json.dumps({"timestamp": "2026-01-01T00:00:00Z", "type": "event_msg", "payload": {"type": "task_started", "turn_id": "turn-1"}}) + "\n" + + json.dumps({"timestamp": "2026-01-01T00:00:00Z", "type": "turn_context", "payload": {"turn_id": "turn-1", "model": "gpt-6.1-sol", "effort": "max"}}) + "\n" + + json.dumps({"timestamp": "2026-01-01T00:00:00Z", "type": "event_msg", "payload": {"type": "task_complete", "turn_id": "turn-1"}}) + "\n" + ) db = home / "state_5.sqlite" with sqlite3.connect(db) as connection: connection.execute("CREATE TABLE threads(id TEXT, rollout_path TEXT, archived INTEGER)") @@ -195,6 +202,11 @@ def test_conversation_resolves_existing_peer_from_real_read_only_host_store(tmp_ assert result["status"] == "resolved" assert result["selected_route"] == {"host_surface": "codex-app", "thread_id": "current"} assert result["host_observation"]["state"] == "idle" + assert result["execution_identity"]["status"] == "runtime_reported" + assert result["execution_identity"]["model"] == "gpt-6.1-sol" + assert result["execution_identity"]["provider"] == "OpenAI" + assert result["execution_identity"]["reasoning_effort"] == "max" + assert result["execution_identity"]["active_turn_verified"] is False assert result["host_delivery"] == "not_attempted" and result["authority"] == "locator_only" assert records == [result] assert before == [p.read_bytes() for p in (registry, db, rollout)]