fix(chat): bind managed conversations to registered Agents

docs/product/periodic-reports.md

@@ -454,10 +454,21 @@ postcondition cannot complete the Task or spend a delivery slot. The low-level
 request accepts no runtime profile or caller validation command. Stable Core
 operation identity retains failed or unknown requests without replay.
 
-This groundwork does **not** create an Agent-owned managed conversation or a
-finished user entry. The logical registered Agent must be bound to its actual
-managed Session independently of the executor endpoint before exposing that
-entry alongside attached sources. Reuse existing Session, native Turn,
+The existing session creation API now accepts `managed_agent_id` for a local
+Goal. It binds the registered work identity separately from the executor
+(`agent_id`) and model. Core persists that immutable identity, Goal lifetime
+and original thread, with a distinct Agent channel. Resume selects only that
+Goal/Agent/executor channel; another Agent cannot execute its native work.
+Old Goal-wide conversations remain readable and unassigned, and cannot gain a
+native work identity through a request-time selector. Existing project,
+steward and attached entrypoints retain their original identities and workers.
+Registration removal, Goal stop, a changed lifetime or a mismatched workspace
+refuses continued execution rather than transferring the conversation.
+
+This is an API/host lifecycle slice, **not** the finished Agent workspace
+entry or general periodic journey. The Agent view still needs to expose this
+binding alongside attached sources, including all original history and honest
+unavailable states. Reuse existing Session, native Turn,
 monitor/calendar and periodic-report owners; do not add a second scheduler,
 task store or model worker. General writable execution, natural-language
 configuration and recurring task composition remain unfinished.

loopx/capabilities/native_chat/codex_context.py

@@ -34,6 +34,21 @@ def home_for_project(base_home: Path, context: dict[str, str] | None) -> Path:
     return codex_home(base_home, policy)
 
 
+def checked_session_home(base_home: Path, session: Mapping[str, Any]) -> Path:
+    if session.get("agent_id") != "codex" or session.get("session_mode") == "attached_host":
+        return base_home
+    selected = session_base_home(base_home, session)
+    require_session_home(selected, session)
+    return selected
+
+
+def managed_upstream_mode(session: Mapping[str, Any]) -> str:
+    from ...chat_codex_goal import UPSTREAM_MODE
+    if session.get("agent_id") == "codex":
+        return UPSTREAM_MODE if session.get("upstream_mode") == UPSTREAM_MODE else "chat"
+    return str(session.get("upstream_mode") or "default")
+
+
 def session_base_home(base_home: Path, session: Mapping[str, Any]) -> Path:
     """Recover an ordinary thread in its independently discovered native profile.
 

loopx/capabilities/native_chat/steward_managed_mode.py

@@ -177,6 +177,49 @@ class ManagedTaskValidationUnavailable(ValueError):
     """A native selection has no usable caller-declared independent validator."""
 
 
+def managed_session_identity(controller: Any, *, goal_id: str, agent_id: str, workspace: Path) -> dict[str, Any]:
+    """Observe existing registration; the typed owner admits the exact binding."""
+    from .conversation_bindings import ChatConversationBindings
+    native_goal = load_goal_from_registry(controller.registry_path, goal_id)
+    return ChatConversationBindings._core("collaboration.session.managed_identity", {
+        "goal": StewardManagedMode(controller, None)._goal(goal_id),
+        "goal_active": bool(native_goal and native_goal.get("status") == "active" and not goal_is_stopped(native_goal)),
+        "managed_agent_id": agent_id, "workspace_path": str(workspace.resolve())})
+
+
+def prepare_managed_session_route(controller: Any, *, goal_id: str | None, executor: str, workspace: Path,
+        channel_id: str, goal_instance_id: str | None, managed_agent_id: str | None,
+        project_context: Any, steward_context: Any) -> dict[str, Any]:
+    """Compose the existing Core route; unbound channels retain their old path."""
+    from ...chat_manager import is_manager_channel
+    route = {"goal_id": goal_id, "goal_instance_id": goal_instance_id, "channel_id": channel_id}
+    if managed_agent_id is None:
+        return route
+    if (controller.closed.is_set() or project_context is not None or steward_context is not None
+            or is_manager_channel(channel_id) or executor != "codex" or goal_id is None):
+        raise ValueError("registered managed Agents require the local Goal's managed Codex host")
+    identity = managed_session_identity(controller, goal_id=goal_id, agent_id=managed_agent_id, workspace=workspace)
+    if channel_id not in (f"goal.{goal_id}", identity["channel_id"]):
+        raise ValueError("the managed Agent channel cannot be substituted")
+    if goal_instance_id is not None and goal_instance_id != identity["goal_instance_id"]:
+        raise ValueError("the original managed Agent Goal lifetime changed")
+    return {**route, **identity, "session_mode": "managed_runtime", "managed_workspace_path": str(workspace.resolve())}
+
+
+def require_managed_agent_session(controller: Any, session: dict[str, Any], workspace: Path) -> None:
+    """Reobserve registration before Core admission, resume or host recovery."""
+    if session.get("managed_agent_id") is None:
+        return
+    from ...control_plane.collaboration import conversation_scope
+    if session.get("managed_workspace_path") != str(workspace.resolve()):
+        raise ValueError("the original managed Agent workspace changed")
+    identity = managed_session_identity(controller, goal_id=session["goal_id"], agent_id=session["managed_agent_id"], workspace=workspace)
+    if (identity["goal_instance_id"] != session.get("goal_instance_id")
+            or identity["channel_id"] != session.get("channel_id")
+            or conversation_scope(session)["kind"] != "owner_goal"):
+        raise ValueError("the original registered Agent Session identity changed")
+
+
 class ManagedSessionTurn:
     """Thin local-owner entry to the existing Turn worker and task validator.
 
@@ -227,14 +270,14 @@ def inspect(self, session_id: str, *, agent_id: str | None = None,
         goal = StewardManagedMode(self.controller, None)._goal(session.get("goal_id"))
         agents = goal["registered_agents"] if goal else []
         result: dict[str, Any] = {"ok": True, "session_id": session_id, "registered_agents": agents,
-            "available": False, "turn": None}
+            "managed_agent_id": session.get("managed_agent_id"), "available": False, "turn": None}
         if operation_id is not None:
             if not re.fullmatch(r"[A-Za-z0-9._:-]{1,100}", operation_id):
                 raise ValueError("invalid operation identity")
             # Readback of an accepted/unknown request is not another execution.
             result["turn"] = self.store.turn_for_client(session_id, "managed." + operation_id)
         try:
-            self.admit(self._request(session_id, agent_id or next(iter(agents), ""), "preview"))
+            self.admit(self._request(session_id, agent_id or session.get("managed_agent_id") or "", "preview"))
             result["available"] = True
         except ValueError as exc:
             result["reason"] = str(exc)

loopx/chat_runtime.py

@@ -24,6 +24,7 @@
 from .capabilities.native_chat import codex_context, project_context as project_context_policy
 from .capabilities.native_chat.codex_adapter import CodexAppServerAdapter
 from .capabilities.native_chat.machine_defaults import codex_profile_home, load_codex_chat_defaults
+from .capabilities.native_chat.steward_managed_mode import prepare_managed_session_route, require_managed_agent_session
 from .capabilities.native_chat.turn_release import TurnReleaseWakeup
 from .control_plane.collaboration import conversation_scope
 from .capabilities.manager_runtime import (
@@ -361,12 +362,6 @@ def _observed_model_settings(adapter: ChatRuntimeAdapter, requested_effort: str
             )
         return settings
 
-    @staticmethod
-    def _managed_upstream_mode(session: dict[str, Any]) -> str:
-        if session.get("agent_id") == "codex":
-            return CODEX_GOAL_CHAT_MODE if session.get("upstream_mode") == CODEX_GOAL_CHAT_MODE else "chat"
-        return str(session.get("upstream_mode") or "default")
-
     @staticmethod
     def _session_objective(
         *,
@@ -562,6 +557,7 @@ def open_session(
         channel_id: str | None = None,
         agent_goal_id: str | None = None,
         goal_instance_id: str | None = None,
+        managed_agent_id: str | None = None,
         manager_executor_allocation: Mapping[str, Any] | None = None,
         project_ref: str | None = None,
         conversation_binding_id: str | None = None,
@@ -608,6 +604,10 @@ def open_session(
             work_dir, objective = context["project"], context["objective"]
         elif goal_id is None:
             raise ValueError("goal_id or an authorized project_ref is required")
+        selected_scope = prepare_managed_session_route(self, goal_id=goal_id, executor=agent_id, workspace=work_dir,
+            channel_id=selected_channel, goal_instance_id=goal_instance_id, managed_agent_id=managed_agent_id,
+            project_context=project_context, steward_context=steward_context)
+        selected_channel, goal_instance_id = selected_scope["channel_id"], selected_scope["goal_instance_id"]
         if project_context is not None:
             project_context_policy.validate_project_executor_scope(agent_id, project_context, self.capabilities, capability)
         manager_runtime = (
@@ -665,7 +665,8 @@ def open_session(
                 # starts a new Session while the old context and history remain intact.
                 if latest is not None and project_context is not None and latest.get("project_context") != project_context:
                     latest = None
-                if latest is not None and goal_instance_id is not None and latest.get("goal_instance_id") != goal_instance_id:
+                if latest is not None and ((goal_instance_id is not None and latest.get("goal_instance_id") != goal_instance_id)
+                        or managed_agent_id is not None and (latest.get("managed_agent_id") != managed_agent_id or latest.get("goal_instance_id") != goal_instance_id)):
                     latest = None
                 if latest is not None and (latest.get("active_turn_id") or latest.get("upstream_mode") == CODEX_GOAL_CHAT_MODE or latest.get("session_mode") == CHAT_SESSION_MODE_ATTACHED):
                     if ((agent_reasoning_effort and latest.get("agent_reasoning_effort") != agent_reasoning_effort)
@@ -684,7 +685,7 @@ def open_session(
             codex_defaults = {}
             selected_codex_home = self.codex_home
             if (agent_id == "codex" and goal_id != MANAGER_AGENT_GOAL_ID
-                and conversation_scope({"channel_id": selected_channel, "goal_id": goal_id})["kind"] == "owner_goal"
+                and conversation_scope(selected_scope)["kind"] == "owner_goal"
                 and not (latest and (latest.get("active_turn_id") or latest.get("upstream_mode") == CODEX_GOAL_CHAT_MODE))):
                 codex_defaults = load_codex_chat_defaults(self.coordination_runtime_root)
                 selected_codex_home = codex_profile_home(self.codex_home, codex_defaults)
@@ -709,7 +710,7 @@ def open_session(
                 goal_id=agent_goal_id or goal_id,
                 objective=objective,
                 execution_mode=selected_channel.startswith("task."),
-                project_coordination=conversation_scope({"channel_id": selected_channel, "goal_id": goal_id})["kind"] == "owner_goal",
+                project_coordination=conversation_scope(selected_scope)["kind"] == "owner_goal",
                 manager_runtime=manager_runtime,
                 project_context=project_context,
                 executor_model=alloc.manager_executor_model(manager_executor_allocation),
@@ -719,10 +720,12 @@ def open_session(
                 codex_home_override=selected_codex_home,
             )
             try:
+                require_managed_agent_session(self, selected_scope, work_dir)
                 observed_settings = self._observed_model_settings(adapter, agent_reasoning_effort, agent_model_id)
                 persisted = self.store.create_session(
                     goal_id=goal_id,
                     goal_instance_id=goal_instance_id,
+                    **({"managed_agent_id": managed_agent_id, "managed_workspace_path": str(work_dir.resolve())} if managed_agent_id is not None else {}),
                     agent_id=agent_id,
                     executor_endpoint_id=agent_id,
                     adapter_kind=str(capability["adapter_kind"]),
@@ -761,13 +764,6 @@ def _session_adapter_lock(self, session_id: str) -> threading.Lock:
         with self.lock:
             return self.session_adapter_locks.setdefault(session_id, threading.Lock())
 
-    def _check_codex_home(self, session: dict[str, Any]) -> Path:
-        if session.get("agent_id") != "codex" or session.get("session_mode") == CHAT_SESSION_MODE_ATTACHED:
-            return self.codex_home
-        selected = codex_context.session_base_home(self.codex_home, session)
-        codex_context.require_session_home(selected, session)
-        return selected
-
     def _ensure_adapter(
         self,
         session: dict[str, Any],
@@ -799,6 +795,7 @@ def _ensure_adapter_locked(
         if current_session is None or current_session.get("status") == "closed":
             raise KeyError("chat session was not found")
         session = current_session
+        require_managed_agent_session(self, session, work_dir)
         if session.get("steward_context") is not None:
             self.project_contexts.session_context(session)
         if session.get("project_context") is not None:
@@ -821,7 +818,7 @@ def _ensure_adapter_locked(
             objective = manager_agent_objective(
                 str(manager_runtime["runtime_profile"])
             )
-        selected_codex_home = self._check_codex_home(session)
+        selected_codex_home = codex_context.checked_session_home(self.codex_home, session)
         if session.get("session_mode") == CHAT_SESSION_MODE_ATTACHED:
             raise CodexChatAgentError(
                 "The attached host Session must be served by its existing host bridge.",
@@ -1056,7 +1053,7 @@ def _ensure_adapter_locked(
                 self.store.restore_managed_session_if_idle(
                     session_id,
                     upstream_thread_id=adapter.upstream_thread_id,
-                    upstream_mode=self._managed_upstream_mode(session),
+                    upstream_mode=codex_context.managed_upstream_mode(session),
                 )
         except Exception:
             with self.lock:
@@ -1111,7 +1108,8 @@ def submit_turn(
             session = self.store.load_session(session_id)
             if session is None or session.get("status") == "closed":
                 raise KeyError("chat session was not found")
-            self._check_codex_home(session)
+            require_managed_agent_session(self, session, work_dir)
+            codex_context.checked_session_home(self.codex_home, session)
             try:
                 with commit():
                     if loopx_execution and session.get("loopx_tools") is not True:
@@ -1229,6 +1227,7 @@ def steer_active_turn(
         session = self.store.load_session(session_id)
         if session is None or session.get("status") == "closed":
             raise KeyError("chat session was not found")
+        require_managed_agent_session(self, session, Path(session.get("managed_workspace_path") or "."))
         if session.get("steward_context") is not None:
             raise ValueError("bound steward Chat requires its external source admission")
         if session.get("project_context") is not None:
@@ -1935,7 +1934,7 @@ def _transport_retry_binding(
         """Compare the original request and current grants/config, not prose."""
         session = self.store.load_session(session_id) or {}
         turn = self.store.load_turn(session_id, turn_id) or {}
-        self._check_codex_home(session)
+        codex_context.checked_session_home(self.codex_home, session)
         if session.get("project_context") is not None or session.get("steward_context") is not None:
             self.project_contexts.session_context(session)
         if session.get("project_context") is not None:
@@ -2253,7 +2252,8 @@ def resume_session(self, *, session_id: str, work_dir: Path, objective: str) ->
             session = self.store.load_session(session_id)
             if session is None or session.get("status") == "closed":
                 raise KeyError("chat session was not found")
-            self._check_codex_home(session)
+            require_managed_agent_session(self, session, work_dir)
+            codex_context.checked_session_home(self.codex_home, session)
             if session.get("session_mode") == CHAT_SESSION_MODE_ATTACHED:
                 return resume_attached_agent_session(
                     store=self.store,
@@ -2281,7 +2281,7 @@ def resume_session(self, *, session_id: str, work_dir: Path, objective: str) ->
                 with goal_commit():
                     return self.store.restore_managed_session_if_idle(
                         session_id, upstream_thread_id=adapter.upstream_thread_id,
-                        upstream_mode=self._managed_upstream_mode(session))
+                        upstream_mode=codex_context.managed_upstream_mode(session))
             except Exception:
                 with self.lock:
                     owns_adapter = self.adapters.get(session_id) is adapter

loopx/chat_server.py

@@ -644,7 +644,7 @@ def _serve_asset(self, path: str) -> None:
     def _create_session(self) -> None:
         try:
             body = self._read_json()
-            unknown = set(body) - {"goal_id", "agent_id", "agent_model_id", "agent_reasoning_effort", "mode", "context_kind", "project_ref"}
+            unknown = set(body) - {"goal_id", "agent_id", "managed_agent_id", "agent_model_id", "agent_reasoning_effort", "mode", "context_kind", "project_ref"}
             if unknown:
                 raise ValueError("unknown session field")
             goal_id: str | None = _compact_text(body.get("goal_id"), limit=160) or self.server.selected_goal_id or ""
@@ -697,6 +697,7 @@ def _create_session(self) -> None:
                 mode=mode,
                 requested_endpoint=requested_endpoint,
                 agent_model_id=requested_model,
+                **({"managed_agent_id": body["managed_agent_id"]} if "managed_agent_id" in body else {}),
                 **({"agent_reasoning_effort": requested_effort} if requested_effort is not None else {}),
                 **({"project_ref": project_ref} if context_kind == "project" else {}),
             )

loopx/chat_session_open.py

@@ -41,9 +41,13 @@ def open_chat_session(
     agent_model_id: str | None = "",
     agent_reasoning_effort: str | None = None,
     project_ref: str | None = None,
+    managed_agent_id: str | None = None,
 ) -> tuple[dict[str, Any], bool]:
     """Return ``(session, resumed)`` for one entry-point request."""
 
+    if managed_agent_id is not None and context_kind != "goal":
+        raise ValueError("a registered managed Agent requires its local Goal conversation")
+
     if context_kind == "manager":
         return open_manager_session(
             controller=controller,
@@ -73,6 +77,7 @@ def open_chat_session(
         mode=mode,
         channel_id=f"goal.{goal_id}",
         agent_goal_id=goal_id,
+        **({"managed_agent_id": managed_agent_id} if managed_agent_id is not None else {}),
         agent_model_id=agent_model_id or None,
         **({"agent_reasoning_effort": agent_reasoning_effort} if agent_reasoning_effort is not None else {}),
     )