feat(manager-context): expose bound host surfaces and route execution identity

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,

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,

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:

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

loopx/control_plane/runtime/execution_identity.ts

@@ -11,15 +11,15 @@ export type ExecutionIdentity = UnknownIdentity | RecordedIdentity;
 const unavailable = (reason: string): UnknownIdentity => ({status: "unavailable", reason});
 const object = (value: unknown): value is Record<string, unknown> =>
   value !== null && typeof value === "object" && !Array.isArray(value);
-const providers = new Map<string, string>([["openai", "OpenAI"], ["anthropic", "Anthropic"], ["google", "Google"]]);
+const providers = new Map<string, string>([["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"
   && /^[A-Za-z0-9][A-Za-z0-9 ._-]{0,95}$/.test(value);
 
 /** Read its latest recorded Turn identity, without claiming execution/authority. */
 export async function readCodexExecutionIdentity(params: Record<string, unknown>): Promise<ExecutionIdentity> {
-  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<string, unknown>
         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<string, unknown>
         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.

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);

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)]