diff --git a/docs/product/periodic-reports.md b/docs/product/periodic-reports.md index 5575aee217..d2898f4f42 100644 --- a/docs/product/periodic-reports.md +++ b/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. diff --git a/loopx/capabilities/native_chat/codex_context.py b/loopx/capabilities/native_chat/codex_context.py index dd7fb4d7d7..f416b273a7 100644 --- a/loopx/capabilities/native_chat/codex_context.py +++ b/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. diff --git a/loopx/capabilities/native_chat/steward_managed_mode.py b/loopx/capabilities/native_chat/steward_managed_mode.py index e2ced4e6c9..44873d7ebf 100644 --- a/loopx/capabilities/native_chat/steward_managed_mode.py +++ b/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) diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py index 37f6996785..3738c3301f 100644 --- a/loopx/chat_runtime.py +++ b/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 diff --git a/loopx/chat_server.py b/loopx/chat_server.py index 9f07fcac4b..49be93e939 100644 --- a/loopx/chat_server.py +++ b/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 {}), ) diff --git a/loopx/chat_session_open.py b/loopx/chat_session_open.py index 658a9a517e..21ebd12d63 100644 --- a/loopx/chat_session_open.py +++ b/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 {}), ) diff --git a/loopx/chat_store.py b/loopx/chat_store.py index 5f3a661476..413c118e93 100644 --- a/loopx/chat_store.py +++ b/loopx/chat_store.py @@ -257,6 +257,8 @@ def create_session( *, goal_id: str | None, goal_instance_id: str | None = None, + managed_agent_id: str | None = None, + managed_workspace_path: str | None = None, agent_id: str, adapter_kind: str, upstream_thread_id: str, @@ -338,6 +340,14 @@ def create_session( channel_id or f"goal.{goal_id}", field="channel_id", ) + if managed_agent_id is not None: + from .control_plane.collaboration import conversation_scope + if not managed_workspace_path or not Path(managed_workspace_path).is_absolute() or conversation_scope({"goal_id": normalized_goal_id, "managed_agent_id": managed_agent_id, + "channel_id": normalized_channel_id, "session_mode": normalized_mode, + "project_context": project_context, "steward_context": steward_context})["kind"] != "owner_goal": + raise ValueError("managed Agent identity requires its exact local Core channel") + elif managed_workspace_path is not None: + raise ValueError("a managed workspace requires its registered Agent identity") session_dir = self._session_dir(token) session_dir.mkdir(parents=True, exist_ok=False, mode=0o700) (session_dir / "turns").mkdir(mode=0o700) @@ -345,6 +355,7 @@ def create_session( "schema_version": CHAT_SESSION_SCHEMA_VERSION, "session_id": token, "goal_id": normalized_goal_id, + **({"managed_agent_id": managed_agent_id, "managed_workspace_path": managed_workspace_path} if managed_agent_id is not None else {}), **({"project_context": project_context} if project_context is not None else {}), **({"steward_context": steward_context} if steward_context is not None else {}), **( @@ -434,6 +445,8 @@ def update_session(self, session_id: str, **changes: Any) -> dict[str, Any]: payload = self.load_session(session_id) if payload is None: raise KeyError("chat session was not found") + if payload.get("managed_agent_id") is not None and "goal_id" in changes and changes["goal_id"] != payload["goal_id"]: + raise ValueError("a registered Agent Session cannot change its original Goal") allowed = { "status", "active_turn_id", @@ -1897,6 +1910,8 @@ def public_session(self, payload: dict[str, Any]) -> dict[str, Any]: "active_turn_id", "last_error_code", "created_at", "updated_at", "last_activity_at", ) } | { + **({"managed_agent_id": payload["managed_agent_id"], "goal_instance_id": payload.get("goal_instance_id")} + if payload.get("managed_agent_id") is not None else {}), "transcript_revision": ( hashlib.sha256(str(revision).encode("ascii")).hexdigest() if revision is not None else None diff --git a/loopx/control_plane/collaboration/__init__.py b/loopx/control_plane/collaboration/__init__.py index 6ae9525506..f1f8d3a9f0 100644 --- a/loopx/control_plane/collaboration/__init__.py +++ b/loopx/control_plane/collaboration/__init__.py @@ -20,7 +20,7 @@ def conversation_scope(session: dict[str, Any], *, origin: str | None = None) -> # Core distinguishes an absent legacy Goal field from an explicit project null. return effect_runtime_result("collaboration.conversation.scope", { **{key: session[key] for key in ( - "channel_id", "goal_id", "project_context", "steward_context", + "channel_id", "goal_id", "project_context", "steward_context", "managed_agent_id", "session_mode", ) if key in session}, **({"origin": origin} if origin is not None else {}), }) diff --git a/loopx/control_plane/collaboration/conversation_scope.ts b/loopx/control_plane/collaboration/conversation_scope.ts index 42c93cc973..b19cba7f96 100644 --- a/loopx/control_plane/collaboration/conversation_scope.ts +++ b/loopx/control_plane/collaboration/conversation_scope.ts @@ -1,5 +1,18 @@ /** Pure identity contract shared with presentation. Host grant observation and * filesystem IO remain outside this classifier. */ +export function goalAgentConversationIdentity(input: Record): Record { + const goal = input.goal_id, agent = input.managed_agent_id; + if (typeof goal !== "string" || !/^[A-Za-z0-9][A-Za-z0-9._-]{0,159}$/.test(goal) + || goal === "loopx-manager" || typeof agent !== "string" + || !/^[A-Za-z0-9][A-Za-z0-9._-]{0,119}$/.test(agent)) { + throw new Error("select an exact registered Goal and Agent"); + } + // Core's route key also contains goal_id. The separate Agent channel keeps + // legacy Goal conversations unassigned, without conflating an executor id + // with the registered work identity or encoding ambiguous Goal/Agent pairs. + return {managed_agent_id: agent, channel_id: `agent.${agent}`}; +} + export function normalizeProjectContext(value: unknown): Record { if (!value || typeof value !== "object" || Array.isArray(value)) throw new Error("invalid project conversation context"); const context = value as Record; @@ -152,6 +165,16 @@ export function resolveConversationScope(input: Record): Conver if (input.project_context != null) { return {kind: "unavailable", goal_ids: [], private_conversation: false}; } + if (input.managed_agent_id != null) { + try { + const identity = goalAgentConversationIdentity(input); + if (channel === identity.channel_id && input.session_mode === "managed_runtime" + && input.steward_context == null && (input.origin === undefined || input.origin === "web")) { + return {kind: "owner_goal", goal_ids: [String(goal)], private_conversation: true}; + } + } catch { /* Inconsistent recovered identity grants no context. */ } + return {kind: "unavailable", goal_ids: [], private_conversation: false}; + } if (channel === "manager") { return {kind: "owner_portfolio", goal_ids: null, private_conversation: true}; } diff --git a/loopx/control_plane/collaboration/steward_managed_mode.ts b/loopx/control_plane/collaboration/steward_managed_mode.ts index 50d6ea2a10..1b43896d94 100644 --- a/loopx/control_plane/collaboration/steward_managed_mode.ts +++ b/loopx/control_plane/collaboration/steward_managed_mode.ts @@ -9,7 +9,22 @@ import type { JsonObject } from "../runtime_decode.ts"; import { requireJsonObject, requireNonEmptyString } from "../runtime_decode.ts"; import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; import { ENVELOPED_SHA256_PATTERN } from "../content_digest.ts"; -import {resolveConversationScope} from "./conversation_scope.ts"; +import {goalAgentConversationIdentity, resolveConversationScope} from "./conversation_scope.ts"; + +/** Registration and current Goal lifetime are observed by the host, never by + * a message or the selected executor/model. This creates no scheduling grant. */ +export function admitManagedSessionIdentity(params: JsonObject): JsonObject { + const goal = requireJsonObject(params.goal, "original registered Goal"); + const identity = goalAgentConversationIdentity({goal_id: goal.goal_id, managed_agent_id: params.managed_agent_id}); + if (params.goal_active !== true || !Array.isArray(goal.registered_agents) + || !goal.registered_agents.includes(identity.managed_agent_id) + || goal.goal_instance_id != null && (typeof goal.goal_instance_id !== "string" + || !/^[A-Za-z0-9._-]{1,160}$/.test(goal.goal_instance_id)) + || params.workspace_path !== goal.workspace_path) { + throw new EffectRuntimeRequestError("the managed Agent must belong to the active local Goal and workspace"); + } + return {...identity, goal_instance_id: goal.goal_instance_id ?? null}; +} /** Local owner Sessions reuse the same governed Turn host. Registration is a * selection, not a grant: the native Turn owner still admits the actual work. @@ -23,11 +38,13 @@ export function admitManagedSessionTurn(params: JsonObject): JsonObject { const turn = params.turn == null ? null : requireJsonObject(params.turn, "current Chat Turn"); if (resolveConversationScope(session).kind !== "owner_goal" || session.project_context != null || session.steward_context != null - || session.session_mode !== "managed_runtime" || session.agent_id !== "codex" + || session.session_mode !== "managed_runtime" || session.agent_id !== "codex" || session.executor_endpoint_id !== "codex" || session.adapter_kind !== "codex_app_server" || session.upstream_mode === "goal" || session.status === "closed" || params.goal_active !== true || goal.goal_id !== session.goal_id || goal.goal_instance_id !== (session.goal_instance_id ?? null) || !Array.isArray(goal.registered_agents) || !goal.registered_agents.includes(actor.agent_id) + || session.managed_agent_id !== actor.agent_id + || session.managed_workspace_path !== goal.workspace_path || typeof goal.workspace_path !== "string" || !goal.workspace_path) { throw new EffectRuntimeRequestError("select an active local Goal's managed Codex conversation and a registered Agent"); } diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 6cb94c3ba0..3c110f4271 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -629,6 +629,7 @@ export function createEffectRuntimeHandlers( ["collaboration.steward.managed_mode.plan", lazyHandler(() => import("./collaboration/steward_managed_mode.ts"), ({planStewardManagedMode}) => planStewardManagedMode)], ["collaboration.steward.managed_turn.admit", lazyHandler(() => import("./collaboration/steward_managed_mode.ts"), ({admitStewardManagedTurn}) => admitStewardManagedTurn)], ["collaboration.session.managed_turn.admit", lazyHandler(() => import("./collaboration/steward_managed_mode.ts"), ({admitManagedSessionTurn}) => admitManagedSessionTurn)], + ["collaboration.session.managed_identity", lazyHandler(() => import("./collaboration/steward_managed_mode.ts"), ({admitManagedSessionIdentity}) => admitManagedSessionIdentity)], ["collaboration.steward.monitor.plan", lazyHandler(() => import("./collaboration/steward_managed_mode.ts"), ({planStewardMonitor}) => planStewardMonitor)], ["collaboration.steward.report.plan", lazyHandler(() => import("./collaboration/steward_reports.ts"), ({planStewardReport}) => planStewardReport)], ["collaboration.steward.monitor.observation", lazyHandler(() => import("./collaboration/steward_managed_mode.ts"), ({observeStewardPortfolio}) => observeStewardPortfolio)], diff --git a/tests/control_plane_ts/steward_reports.test.ts b/tests/control_plane_ts/steward_reports.test.ts index d35e7e6b21..fba1f7f141 100644 --- a/tests/control_plane_ts/steward_reports.test.ts +++ b/tests/control_plane_ts/steward_reports.test.ts @@ -1,7 +1,35 @@ import assert from "node:assert/strict"; import test from "node:test"; import {planStewardReport} from "../../loopx/control_plane/collaboration/steward_reports.ts"; -import {admitStewardManagedTurn, bindStewardManagedActor} from "../../loopx/control_plane/collaboration/steward_managed_mode.ts"; +import {admitManagedSessionIdentity, admitManagedSessionTurn, admitStewardManagedTurn, bindStewardManagedActor} from "../../loopx/control_plane/collaboration/steward_managed_mode.ts"; +import {resolveConversationScope} from "../../loopx/control_plane/collaboration/conversation_scope.ts"; + +test("a managed work identity is registration-bound and never chosen from the executor or legacy history", () => { + const goal = {goal_id: "project", goal_instance_id: "original", registered_agents: ["writer", "reviewer"], workspace_path: "/workspace"}; + const input = {goal, goal_active: true, workspace_path: "/workspace", managed_agent_id: "writer"}; + const identity = admitManagedSessionIdentity(input); + assert.deepEqual(identity, {managed_agent_id: "writer", channel_id: "agent.writer", goal_instance_id: "original"}); + assert.notDeepEqual(admitManagedSessionIdentity({...input, managed_agent_id: "reviewer"}), identity); + for (const changed of [{managed_agent_id: "codex"}, {managed_agent_id: ["writer"]}, {goal_active: false}, + {workspace_path: "/other"}, {goal: {...goal, goal_instance_id: {unverified: true}}}]) { + assert.throws(() => admitManagedSessionIdentity({...input, ...changed})); + } + const session = {...identity, goal_id: goal.goal_id, session_id: "a".repeat(32), session_mode: "managed_runtime", + agent_id: "codex", executor_endpoint_id: "codex", adapter_kind: "codex_app_server", managed_workspace_path: "/workspace"}; + assert.equal(resolveConversationScope(session).kind, "owner_goal"); + assert.equal(resolveConversationScope({...session, channel_id: "goal.project"}).kind, "unavailable"); + assert.equal(resolveConversationScope({...session, origin: "lark"}).kind, "unavailable"); + const request = {operation: "native_turn", host: "codex-app-server", work_kind: "native_turn", + turn_instance_id: "one-operation", todo_id: null, validation_command: null, + actor: {schema_version: "loopx_managed_session_actor_v0", goal_id: goal.goal_id, goal_instance_id: "original", + agent_id: "writer", session_id: session.session_id, workspace_path: "/workspace"}}; + const params = {goal, goal_active: true, session, request, queued_turn_count: 0}; + assert.deepEqual(admitManagedSessionTurn(params).actor, request.actor); + for (const changed of [{managed_agent_id: "reviewer"}, {managed_agent_id: undefined, channel_id: "goal.project"}, + {goal_instance_id: "replaced"}, {managed_workspace_path: "/other"}, {executor_endpoint_id: "other"}]) { + assert.throws(() => admitManagedSessionTurn({...params, session: {...session, ...changed}})); + } +}); function original() { const binding = {binding_id: "b".repeat(24), transport_ref: "private-steward", provider_ref: "c".repeat(24), diff --git a/tests/test_native_steward_turn_host.py b/tests/test_native_steward_turn_host.py index 7a2d6edd73..17d7d9276c 100644 --- a/tests/test_native_steward_turn_host.py +++ b/tests/test_native_steward_turn_host.py @@ -30,7 +30,8 @@ def managed_fixture(steward, tmp_path, monkeypatch, provider="file", *, validato goal = load_goal_from_registry(runtime.registry_path, actor["goal_id"]) workspace = Path(goal["repo"]) session, _ = runtime.open_session(goal_id=actor["goal_id"], agent_id="codex", work_dir=workspace, - objective="Observe the original task", channel_id=f"goal.{actor['goal_id']}", mode="new") + objective="Observe the original task", channel_id=f"goal.{actor['goal_id']}", mode="new", + managed_agent_id=actor["agent_id"]) turn, _ = runtime.submit_turn(session_id=session["session_id"], client_turn_id="original-question", message="Inspect this source and retain our original conversation.", work_dir=workspace, objective="Observe the original task") assert finish(runtime, turn)["status"] == "completed" @@ -54,6 +55,109 @@ def managed_fixture(steward, tmp_path, monkeypatch, provider="file", *, validato return ManagedSessionTurn(runtime), session, actor, added["todo_id"], observation, capture +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_registered_agents_have_distinct_durable_sessions_without_adopting_legacy_history(steward, tmp_path, monkeypatch, provider): + service, first, actor, _, _, capture = managed_fixture(steward, tmp_path, monkeypatch, provider) + store, runtime, *_ = steward + from loopx.chat_session_open import open_chat_session + from loopx.agent_registry import load_goal_from_registry + cli(runtime, "configure-goal", "--goal-id", actor["goal_id"], "--registered-agent", actor["agent_id"], + "--registered-agent", "peer", "--execute") + workspace = Path(load_goal_from_registry(runtime.registry_path, actor["goal_id"])["repo"]) + def opened(agent=None, mode="resume_latest"): + return open_chat_session(controller=runtime, context_kind="goal", goal_id=actor["goal_id"], + work_dir=workspace, objective="Independent work contexts", mode=mode, managed_agent_id=agent) + legacy, _ = opened() + assert not service.inspect(legacy["session_id"])["available"] + legacy_turn, _ = runtime.submit_turn(session_id=legacy["session_id"], client_turn_id="legacy-question", + message="Keep this unmapped history intact", work_dir=workspace, objective="Legacy conversation") + assert finish(runtime, legacy_turn)["status"] == "completed" + old_request = service._request(legacy["session_id"], actor["agent_id"], "old-unknown") + old = store.accept_managed_turn(legacy["session_id"], client_turn_id="managed.old-unknown", + message="Retained pre-binding operation", origin="native_loopx", loopx_execution=True, loopx_request=old_request).turn + old_calls = capture.read_bytes() + assert service.inspect(legacy["session_id"], operation_id="old-unknown")["turn"]["turn_id"] == old["turn_id"] + assert service.submit(legacy["session_id"], agent_id=actor["agent_id"], operation_id="old-unknown") == (old, False) + assert capture.read_bytes() == old_calls # Readback of unknown work is never a replay. + legacy_history = store.messages(legacy["session_id"]) + fake = Path(runtime.codex_bin) + # The base synthetic host uses one fixed id. Give newly started threads a + # distinct id and honor the exact original id on thread/resume/read. + fake.write_text(fake.read_text().replace('"durable-thread"', + 'request.get("params", {}).get("threadId", "new-peer-thread")')) + peer, resumed = opened("peer") + assert not resumed and peer["managed_agent_id"] == "peer" and peer["agent_id"] == "codex" + assert peer["session_id"] != first["session_id"] != legacy["session_id"] + assert peer["upstream_thread_id"] != first["upstream_thread_id"] + assert store.messages(peer["session_id"]) == [] + # An attached source is a peer record, never the managed executor's target. + attached = store.create_session(goal_id=actor["goal_id"], agent_id="peer", adapter_kind="codex_app_server", + upstream_thread_id="original-attached", session_mode="attached_host", host_surface="codex_app") + before_calls = capture.read_bytes() + for _ in range(2): + same, resumed = opened("peer") + assert resumed and same["session_id"] == peer["session_id"] + same, resumed = opened(actor["agent_id"]) + assert resumed and same["session_id"] == first["session_id"] + resumed_calls = [json.loads(line) for line in capture.read_bytes()[len(before_calls):].splitlines()] + assert not [row for row in resumed_calls if row["method"] in {"thread/start", "turn/start"}] + for row in (first, peer): + runtime.adapters.pop(row["session_id"]).close_session() + same, resumed = opened(row["managed_agent_id"]) + assert resumed and same["session_id"] == row["session_id"] + assert same["upstream_thread_id"] == row["upstream_thread_id"] + assert opened()[0]["session_id"] == legacy["session_id"] + assert store.messages(legacy["session_id"]) == legacy_history + assert store.load_session(attached["session_id"]) == attached + with pytest.raises(ValueError): + service.submit(first["session_id"], agent_id="peer", operation_id="wrong-registered-agent") + with pytest.raises(ValueError): + store.update_session(first["session_id"], managed_agent_id="peer") + with pytest.raises(ValueError): + store.update_session(first["session_id"], goal_id="another-goal") + public = store.public_session(first) + assert public["managed_agent_id"] == actor["agent_id"] and public["executor_endpoint_id"] == "codex" + + +@pytest.mark.parametrize("fault", ["unregistered", "revoked", "stopped", "lifetime", "workspace", "project", "channel", "executor"]) +def test_managed_identity_rejections_do_not_create_or_resume_a_host(steward, tmp_path, monkeypatch, fault): + _, session, actor, _, _, capture = managed_fixture(steward, tmp_path, monkeypatch) + store, runtime, *_ = steward + from loopx.agent_registry import load_goal_from_registry + workspace = Path(load_goal_from_registry(runtime.registry_path, actor["goal_id"])["repo"]) + args = dict(goal_id=actor["goal_id"], agent_id="codex", work_dir=workspace, + objective="Keep original identity", mode="resume_latest", managed_agent_id=actor["agent_id"]) + if fault == "unregistered": + args["managed_agent_id"] = "unregistered" + elif fault in {"revoked", "stopped", "lifetime"}: + registry = json.loads(runtime.registry_path.read_text()) + goal = next(row for row in registry["goals"] if row["id"] == actor["goal_id"]) + if fault == "revoked": goal["coordination"]["registered_agents"] = ["peer"] + elif fault == "stopped": goal["status"] = "stopped" + else: goal["goal_instance_id"] = "replaced-lifetime" + runtime.registry_path.write_text(json.dumps(registry)) + elif fault == "workspace": args["work_dir"] = tmp_path + elif fault == "project": args["project_ref"] = "0" * 24 + elif fault == "channel": args["channel_id"] = "manager.external.other" + elif fault == "executor": args["agent_id"] = "other-executor" + before_calls, before_sessions = capture.read_bytes(), store.list_sessions() + # A changed Goal lifetime starts a fresh registered conversation only on + # an explicit new/resume request; the old Session cannot execute again. + if fault == "lifetime": + with pytest.raises(ValueError, match="identity changed"): + runtime._ensure_adapter_locked(store.load_session(session["session_id"]), work_dir=workspace, objective="old") + else: + with pytest.raises(ValueError): runtime.open_session(**args) + if fault in {"revoked", "stopped", "lifetime"}: + with pytest.raises(ValueError): + runtime.submit_turn(session_id=session["session_id"], client_turn_id="revoked-request", + message="Must not be accepted under a stale identity", work_dir=workspace, objective="old") + assert store.turn_for_client(session["session_id"], "revoked-request") is None + with pytest.raises(ValueError): + runtime.resume_session(session_id=session["session_id"], work_dir=workspace, objective="old") + assert capture.read_bytes() == before_calls and store.list_sessions() == before_sessions + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) def test_local_managed_session_reuses_native_turn_and_declared_validation(steward, tmp_path, monkeypatch, provider): service, session, actor, todo_id, observation, capture = managed_fixture(steward, tmp_path, monkeypatch, provider) @@ -160,6 +264,7 @@ def test_managed_turn_http_correlates_readback_without_accepting_commands(stewar store, runtime, *_ = steward server = ChatHTTPServer(("127.0.0.1", 0), ChatRequestHandler) server.runtime_controller, server.chat_store, server.verbose = runtime, store, False + server.registry_path, server.selected_goal_id = runtime.registry_path, actor["goal_id"] threading.Thread(target=server.serve_forever, daemon=True).start() route = f"/api/chat/sessions/{session['session_id']}/native-turn" @@ -176,6 +281,19 @@ def request(body=None, suffix="", origin=None): try: before = capture.read_bytes() + # Exercise the shipped producer, not only a hand-built Session fixture. + conn = http.client.HTTPConnection(*server.server_address, timeout=20) + conn.request("POST", "/api/chat/sessions", json.dumps({"goal_id": actor["goal_id"], + "managed_agent_id": actor["agent_id"], "mode": "resume_latest"}), + headers={"Content-Type": "application/json"}) + opened = conn.getresponse() + opened_body = json.loads(opened.read()) + conn.close() + assert opened.status == 200 and opened_body["session_id"] == session["session_id"], opened_body + assert opened_body["session"]["managed_agent_id"] == actor["agent_id"] + assert not [json.loads(line) for line in capture.read_bytes()[len(before):].splitlines() + if json.loads(line)["method"] in {"thread/start", "turn/start"}] + before = capture.read_bytes() assert request()[1]["available"] body = {"agent_id": actor["agent_id"], "operation_id": "http-original", "todo_id": todo_id} assert request(body, origin="https://foreign.example")[0] == 403