From 8039dc652d360fbd16e0f1ceebc455202ed708e3 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 28 Oct 2026 02:38:05 +0800 Subject: [PATCH] refactor(turn): share managed operation IO and fence session binding Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../2026-10-27-refactor-handoff.md | 14 + .../2026-10-27-refactor-handoff.zh-CN.md | 10 + examples/operation_action_fixtures.py | 2 +- .../turn_driver/codex_operation_host.py | 361 ++---------------- .../turn_driver/codex_sessions.py | 116 +++++- .../turn_driver/dsh_operation_host.py | 180 ++------- .../turn_driver/managed_operation_host.py | 299 +++++++++++++++ .../test_dsh_operation_connection.py | 2 +- .../control_plane/test_dsh_operation_entry.py | 2 +- tests/control_plane/test_dsh_operation_sdk.py | 49 ++- .../control_plane/test_operation_requests.py | 2 +- tests/test_chat_operation_actions.py | 2 +- tests/test_codex_operation_host.py | 82 +++- tests/test_local_delegation.py | 3 +- 14 files changed, 634 insertions(+), 490 deletions(-) create mode 100644 loopx/control_plane/turn_driver/managed_operation_host.py diff --git a/docs/architecture/rfcs/ledger/controller-executor-unified-contract-v0/2026-10-27-refactor-handoff.md b/docs/architecture/rfcs/ledger/controller-executor-unified-contract-v0/2026-10-27-refactor-handoff.md index dd98ecbc8d..276b26155a 100644 --- a/docs/architecture/rfcs/ledger/controller-executor-unified-contract-v0/2026-10-27-refactor-handoff.md +++ b/docs/architecture/rfcs/ledger/controller-executor-unified-contract-v0/2026-10-27-refactor-handoff.md @@ -91,6 +91,20 @@ checks around them. `driver.py::reconcile_failed_turn_session_request` and C1 and explicit migration/replay/rollback; never overwrite old native IDs or borrow continuity between hosts/roles. Read accepted receipts before reverting. +### C3 implementation checkpoint — 2026-10-28 proposal + +Both operation adapters now use shared protocol IO in `managed_operation_host.py` +and select/verify/store through the existing `codex_sessions.py` owner. Native +acceptance, SDK pin/restore and resource cleanup remain transport-owned. Session +paths, wire schemas, prompts and the typed collaboration/admission owners stay +unchanged. A reproduced Codex initialization race is additionally fenced by +comparing the selected document under the existing file lock before storing; +a concurrent replacement survives and no prompt or operation tool is dispatched. +DSH retains its existing initialization fence. Caller adoption, shared IO and +the session boundary roll back together; no state migration is required. This +single-session proposal does not qualify controller/executor roles, generations, +live installation or long-horizon benefit; the original C3 acceptance remains. + ## C4.1 — Typed receipt-to-continuation boundary **Remaining cross-language decision ownership.** `loop_controller.py` owns diff --git a/docs/architecture/rfcs/ledger/controller-executor-unified-contract-v0/2026-10-27-refactor-handoff.zh-CN.md b/docs/architecture/rfcs/ledger/controller-executor-unified-contract-v0/2026-10-27-refactor-handoff.zh-CN.md index 5eca2597b9..7fb4d48ae5 100644 --- a/docs/architecture/rfcs/ledger/controller-executor-unified-contract-v0/2026-10-27-refactor-handoff.zh-CN.md +++ b/docs/architecture/rfcs/ledger/controller-executor-unified-contract-v0/2026-10-27-refactor-handoff.zh-CN.md @@ -73,6 +73,16 @@ DSH schema。Operation adapter 仍使用私有读写 helper，并在周边组装 扩展需要 C1 稳定及显式迁移/replay/rollback；不能覆盖旧 native ID，不能跨 Host/role 借用连续性。回退前读回已接受回执。 +### C3 实现检查点 — 2026-10-28 提案 + +两个 operation adapter 现共用 `managed_operation_host.py` 的协议 IO，并通过原 +`codex_sessions.py` owner 完成 select/verify/store。原生接受、SDK pin/restore 和 +资源清理仍归 transport。会话路径、wire schema、提示正文及 typed collaboration/ +admission owner 保持。另对已复现的 Codex 初始化竞态，在原文件锁内比较选定文档后 +再保存：并发替换得以保留，且不派发 prompt 或 operation tool。DSH 保留原初始化 +保护。调用方采用、公共 IO 和会话边界须一并回退，无需状态迁移。本单会话提案不 +认证 controller/executor 角色、generation、正式安装或长程收益；原 C3 验收保留。 + ## C4.1 — Typed 回执到继续决策边界 **仍存跨语言决策归属。** `loop_controller.py` 拥有 `ValidatedTurnReceipt`、 diff --git a/examples/operation_action_fixtures.py b/examples/operation_action_fixtures.py index 02197a71da..d6e74ee1d0 100644 --- a/examples/operation_action_fixtures.py +++ b/examples/operation_action_fixtures.py @@ -70,7 +70,7 @@ def managed_handler( reasoning_effort="xhigh", ): from loopx.control_plane.turn_driver.codex_cli import _store_codex_cli_session - from loopx.control_plane.turn_driver.codex_operation_host import ( + from loopx.control_plane.turn_driver.managed_operation_host import ( operation_tool_handler, ) diff --git a/loopx/control_plane/turn_driver/codex_operation_host.py b/loopx/control_plane/turn_driver/codex_operation_host.py index 788de38159..1869ee1635 100644 --- a/loopx/control_plane/turn_driver/codex_operation_host.py +++ b/loopx/control_plane/turn_driver/codex_operation_host.py @@ -16,234 +16,22 @@ from typing import Any from ...chat_agent import CodexChatAgentError, CodexChatAgentSession -from ...chat_action_store import ChatActionStore -from ...chat_actions import ChatActionService -from ..collaboration.goal_instance_scope import collaboration_goal_scope -from ..collaboration.operation_handoff import ( - agent_operation_action, - pending_operation_handoffs, -) +from ...extensions.codex_native_child import native_child_observer from ..goals.first_party_host_admission import FirstPartyHostGoalAdmission -from ..effect_runtime import EffectRuntimeRejected +from . import managed_operation_host as operations from .codex_cli import ( _prompt, normalize_codex_stdio_mcp_server, codex_cli_result_schema, ) -from .codex_sessions import ( - _lineage, - _read_codex_cli_session_document, - _codex_session_goal_ref, - _store_codex_cli_session, - load_codex_cli_session, -) -from ...extensions.codex_native_child import native_child_observer +from .codex_sessions import _lineage, select_managed_operation_session from .executor import LOOPX_TURN_HOST_REQUEST_SCHEMA_VERSION from .host_failure import BuiltInHostError from .operation_requests import ( - BoundOperationRequest, operation_request_source, request_reference_instruction, + operation_request_source, request_reference_instruction, ) TRANSPORT = "app-server-operation-tools-v0" -REVISION = "managed-turn-handoff-v0" -OPERATION_TOOL = { - "name": "loopx_operation", - "description": "Read/prepare a scoped operation, consume exact human approval once, or report original evidence. Identity comes from this native connection, never arguments. A receipt is not permission to retry an effect.", - "inputSchema": { - "type": "object", - "properties": { - "action": { - "type": "string", - "enum": [ - "context", - "pending", - "prepare", - "inspect", - "consume", - "report", - ], - }, - "proposal_id": {"type": "string"}, - "consumption_id": {"type": "string"}, - "request": {"type": "object"}, - "outcome": {"type": "object"}, - "cursor": {"type": "string"}, - }, - "required": ["action"], - "additionalProperties": False, - }, -} - - -def operation_tool_handler( - *, - runtime_root: Path, - registry_path: Path, - lineage: Mapping[str, str], - session_id: str, - profile_digest: str, - model: str, - reasoning_effort: str, - source_route: Mapping[str, str] | None = None, - goal_admission: FirstPartyHostGoalAdmission | None = None, - host: str = "codex-cli", - frozen_request: BoundOperationRequest | None = None, -): - """Private closure installed only on a process owned by the Turn driver. - - The caller must already have checked native thread/Turn metadata. This is - not an MCP endpoint or a public actor/proof deserializer. - """ - if host not in {"codex-cli", "dsh"}: - raise ValueError("unsupported owned operation host") - surface = "loopx-managed-dsh" if host == "dsh" else "loopx-managed-codex" - executor = { - "kind": "managed_turn", - "todo_id": lineage["todo_id"], - "session_id": session_id, - "profile_digest": profile_digest, - "revision": REVISION, - "model": model, - "reasoning_effort": reasoning_effort, - **({"host": "dsh"} if host == "dsh" else {}), - } - executor_route = { - **lineage, - "host_surface": surface, - "thread_id": session_id, - "profile_digest": profile_digest, - } - - def handle(tool: str, arguments: Any, native: dict[str, Any]) -> dict[str, Any]: - if tool != "loopx_operation" or not isinstance(arguments, dict): - return {"ok": False, "error": "unsupported_operation_tool"} - if native.get("thread_id") != session_id or not native.get("host_turn_id"): - return {"ok": False, "error": "tool_turn_mismatch"} - allowed = { - "context": {"action"}, - "pending": {"action"}, - "prepare": {"action", "request"}, - "inspect": {"action", "proposal_id"}, - "consume": {"action", "proposal_id", "consumption_id"}, - "report": {"action", "proposal_id", "outcome"}, - } - action = arguments.get("action") - keys = set(arguments) - ({"cursor"} if action == "pending" else set()) - if ( - not isinstance(action, str) - or action not in allowed - or keys != allowed[action] - ): - return {"ok": False, "error": "invalid_operation_arguments"} - try: - if goal_admission is not None: - goal_admission.require_current() - if action == "context": - return { - "ok": True, - **lineage, - "executor": executor, - "authority": "approval_and_first_consumption_required", - "external_write_performed": False, - **({"prepared_request": {"ref": frozen_request.ref}} - if frozen_request is not None else {}), - } - if action == "pending": - with collaboration_goal_scope( - registry_path, - goal_id=lineage["goal_id"], - agents=(lineage["agent_id"],), - ) as scope: - cursor_scope = hashlib.sha256( - json.dumps( - ["managed-operation-v0", str(runtime_root.resolve()), - scope.target(lineage["agent_id"]), executor], - sort_keys=True, separators=(",", ":"), - ).encode() - ).hexdigest() - return { - "ok": True, - **pending_operation_handoffs( - runtime_root, - lineage["goal_id"], - lineage["agent_id"], - registry_path=registry_path, - scope=scope, - cursor=arguments.get("cursor"), - cursor_scope=cursor_scope, - executor_route=executor_route, - ), - } - if action == "prepare": - request = (frozen_request.resolve(arguments["request"]) - if frozen_request is not None else dict(arguments["request"])) - terms = dict(request.get("normalized_parameters") or {}) - if "source_route" in terms and terms["source_route"] != source_route: - raise ValueError("the model cannot retarget the host-selected return audience") - if source_route is not None: - terms["source_route"] = dict(source_route) - for key, expected in { - "goal_id": lineage["goal_id"], - "agent_id": lineage["agent_id"], - "executor": executor, - }.items(): - if key in terms and terms[key] != expected: - raise ValueError( - "operation request cannot select another execution subject" - ) - terms[key] = expected - if request.get("action_kind") != "operation.execute": - raise ValueError("operation tool prepares only typed operations") - request["normalized_parameters"] = terms - service = ChatActionService( - store=ChatActionStore(runtime_root / "chat" / "actions"), - registry_path=registry_path, - ) - return { - "ok": True, - "proposal": service.preview(request), - "execution_allowed": False, - } - actor = { - **lineage, - "host_surface": surface, - "thread_id": session_id, - "profile_digest": profile_digest, - "model": model, - "reasoning_effort": reasoning_effort, - "host_turn_id": native["host_turn_id"], - } - return { - "ok": True, - **agent_operation_action( - runtime_root, - registry_path, - proposal_id=arguments["proposal_id"], - actor=actor, - action=action, - consumption_id=arguments.get("consumption_id"), - outcome=arguments.get("outcome"), - ), - } - except EffectRuntimeRejected as exc: - if exc.diagnostic_code == "operation_source_route_ambiguous": - return {"ok": False, "error": "operation_source_route_ambiguous", - "execution_allowed": False, - "next_action": ( - "Select a registered return audience in the owned DSH host source_route; source routing is not executor authority." - if host == "dsh" else - "Select a registered return audience with --codex-operation-source-route-json in the host invocation; source routing is not executor authority." - )} - return {"ok": False, "error": "operation_admission_rejected", "execution_allowed": False} - except (ValueError, KeyError, TypeError, RuntimeError): - # Private payloads, paths and adapter error text never enter the model tool error. - return { - "ok": False, - "error": "operation_admission_rejected", - "execution_allowed": False, - } - - return handle def run_codex_operation_host( @@ -303,55 +91,18 @@ def run_codex_operation_host( profile_digest = hashlib.sha256( json.dumps(profile, sort_keys=True, separators=(",", ":")).encode() ).hexdigest() - binding = load_codex_cli_session(runtime_root, lineage=lineage) - if goal_admission is not None: - selected = goal_admission.select_state( - read_state=lambda: _read_codex_cli_session_document( - runtime_root, lineage=lineage - ), - goal_ref_of=lambda value: _codex_session_goal_ref(value, lineage=lineage), - ) - binding = dict(selected) if selected is not None else None - session_plan = request.get("session") or {} - if (session_plan.get("context_policy") or {}).get("mode") == "fresh": - binding = None - action = session_plan.get("action") - if ( - action not in {"start_new", "resume"} - or (action == "resume" and not binding) - or (action == "start_new" and binding) - ): - raise ValueError("managed Codex session changed after Turn planning") - if binding and ( - binding.get("operation_transport") != TRANSPORT - or binding.get("operation_profile_digest") != profile_digest - ): - raise ValueError( - "managed operation profile changed; explicitly select a fresh iteration and obtain fresh approval" - ) - if confirmed_operation_id is not None: - # Final admission recheck before native resume. A queued delegation must - # not create/rebind a Session if approval, Todo, profile or lifetime - # changed after the callback. This grants no tool/effect authority. - from ..collaboration.operation_handoff import managed_operation_binding_current - from ..collaboration.operation_wake import require_current_operation_scope - - store = ChatActionStore(runtime_root / "chat" / "actions") - proposal = store.load(confirmed_operation_id) - if proposal is None or binding is None or action != "resume": - raise ValueError("confirmed operation requires its original resumable session") - require_current_operation_scope(registry_path, proposal["normalized_parameters"]) - store._agent_operation_plan( - proposal, action="wake", - binding_current=managed_operation_binding_current( - runtime_root, proposal["normalized_parameters"] - ), - launch_context={"host": "codex-cli", "operation_tools": True, - "iteration_context": (session_plan.get("context_policy") or {}).get("mode")}, - executor_route={**lineage, "host_surface": "loopx-managed-codex", - "thread_id": binding["session_id"], "profile_digest": profile_digest, - "model": model, "reasoning_effort": reasoning_effort}, - ) + selection = select_managed_operation_session( + runtime_root, request=request, lineage=lineage, host="codex-cli", + transport=TRANSPORT, profile_digest=profile_digest, model=model, + reasoning_effort=reasoning_effort, goal_admission=goal_admission, + ) + binding = selection.binding + operations.require_confirmed_operation_resume( + runtime_root=runtime_root, registry_path=registry_path, request=request, + lineage=lineage, binding=binding, host="codex-cli", profile_digest=profile_digest, + model=model, reasoning_effort=reasoning_effort, + confirmed_operation_id=confirmed_operation_id, + ) host_config = ( { "mcp_servers": { @@ -382,7 +133,7 @@ def run_codex_operation_host( model=model, reasoning_effort=reasoning_effort, resume_thread_id=binding["session_id"] if binding else None, - dynamic_tools=[OPERATION_TOOL], + dynamic_tools=[operations.OPERATION_TOOL], host_config=host_config, launch_guard=lambda: launch_scope("codex_operation_process_launch"), response_timeout_sec=30 if timeout_seconds is None else min(timeout_seconds, 30), @@ -390,22 +141,8 @@ def run_codex_operation_host( idle_timeout_sec=180 if timeout_seconds is None else min(timeout_seconds, 180), ) - def store_binding(): - _store_codex_cli_session( - runtime_root, - lineage=lineage, - session_id=session.thread_id, - goal_ref=request.get("goal_ref"), - operation_profile_digest=profile_digest, - operation_model=model, - operation_reasoning_effort=reasoning_effort, - ) - - if goal_admission is None: - store_binding() - else: - goal_admission.accept_result(store_binding) - session.bound_tool_handler = operation_tool_handler( + selection.store(session.thread_id) + session.bound_tool_handler = operations.operation_tool_handler( runtime_root=runtime_root, registry_path=registry_path, lineage=lineage, @@ -419,46 +156,32 @@ def store_binding(): session_id=session.thread_id, profile_digest=profile_digest) if request_source is not None else None), ) - route = {**lineage, "host_surface": "loopx-managed-codex", - "thread_id": session.thread_id, "profile_digest": profile_digest} - continuations = [] - if (runtime_root / "chat" / "actions" / "actions.json").is_file(): - with collaboration_goal_scope( - registry_path, goal_id=lineage["goal_id"], agents=(lineage["agent_id"],), - caller_goal_ref=request.get("goal_ref"), require_active=True, - ) as scope: - pending = pending_operation_handoffs( - runtime_root, lineage["goal_id"], lineage["agent_id"], - registry_path=registry_path, scope=scope, executor_route=route, - cursor_scope=hashlib.sha256(json.dumps(route, sort_keys=True).encode()).hexdigest(), - ) - # Bounded canonical locators only. Private terms still require the - # existing authenticated inspect tool; these locators grant nothing. - continuations = [ - {key: item[key] for key in ("operation_id", "payload_digest", "confirmation_digest", "claim_id")} - for item in pending["items"] if item["status"] == "authorized_pending" - ] + route = operations.operation_executor_route( + lineage, host="codex-cli", session_id=session.thread_id, profile_digest=profile_digest, + ) + continuations = ( + operations.operation_continuations( + runtime_root=runtime_root, registry_path=registry_path, lineage=lineage, + route=route, goal_ref=request.get("goal_ref"), + ) if (runtime_root / "chat" / "actions" / "actions.json").is_file() else [] + ) def on_event(kind: str, event: dict[str, Any]) -> None: if kind != "turn.started": return - # send emits this only after the native turn/start response. A - # launched process, callback ACK or model assertion cannot issue it. - for locator in continuations: - try: - agent_operation_action( - runtime_root, registry_path, proposal_id=locator["operation_id"], - actor={**route, "model": model, "reasoning_effort": reasoning_effort, - "host_turn_id": event.get("upstream_turn_id")}, - action="observe_host_start", turn_key=request["turn_key"], - ) - except (ValueError, KeyError, TypeError, RuntimeError) as exc: - # Stop before dispatching operation tools if confirmation, - # Goal or binding changed while native start was in flight. - raise BuiltInHostError( - "codex_operation_host_start_unrecorded", failure_kind="unknown", - recovery_kind="resume_session", - ) from exc + # Only the authenticated native turn/start response reaches here. + try: + operations.record_operation_start( + runtime_root=runtime_root, registry_path=registry_path, + continuations=continuations, route=route, model=model, + reasoning_effort=reasoning_effort, + host_turn_id=event.get("upstream_turn_id"), turn_key=request["turn_key"], + ) + except (ValueError, KeyError, TypeError, RuntimeError) as exc: + raise BuiltInHostError( + "codex_operation_host_start_unrecorded", failure_kind="unknown", + recovery_kind="resume_session", + ) from exc child_observer = native_child_observer(request, runtime_root=runtime_root, lineage=lineage, registry_path=goal_admission.registry_path if goal_admission is not None else None) @@ -493,7 +216,7 @@ def record_child() -> None: output_schema=codex_cli_result_schema(request), on_event=on_event, turn_start_guard=lambda: launch_scope("codex_operation_turn_start"), - **({"on_native_item": observe_child} if child_observer is not None else {}), + on_native_item=observe_child if child_observer is not None else None, ) return result except CodexChatAgentError as exc: diff --git a/loopx/control_plane/turn_driver/codex_sessions.py b/loopx/control_plane/turn_driver/codex_sessions.py index 725b4d9526..741a2df4a2 100644 --- a/loopx/control_plane/turn_driver/codex_sessions.py +++ b/loopx/control_plane/turn_driver/codex_sessions.py @@ -7,6 +7,7 @@ import os import tempfile from collections.abc import Mapping +from dataclasses import dataclass from pathlib import Path from typing import Any @@ -24,6 +25,7 @@ CODEX_CLI_SESSION_SCHEMA_VERSION = "loopx_codex_cli_session_v1" MANAGED_CLI_SESSION_SCHEMA_VERSION = "loopx_managed_cli_session_v1" SESSION_ID_MAX_CHARS = 256 +_UNCHECKED_SESSION_DOCUMENT = object() def _host_session_schema(host: str) -> str: @@ -93,15 +95,19 @@ def load_codex_cli_session( lineage: Mapping[str, str], session_scope: str = "todo", host: str = "codex-cli", +) -> dict[str, Any] | None: + _host_session_schema(host) + return _session_document_value( + _read_codex_cli_session_document(runtime_root, lineage=lineage, session_scope=session_scope), + lineage=lineage, session_scope=session_scope, host=host, + ) + + +def _session_document_value( + value: object, *, lineage: Mapping[str, str], session_scope: str = "todo", + host: str = "codex-cli", ) -> dict[str, Any] | None: schema = _host_session_schema(host) - path = _session_path(runtime_root, lineage, session_scope) - if not path.exists(): - return None - try: - value = json.loads(path.read_text(encoding="utf-8")) - except (OSError, json.JSONDecodeError): - return None if not isinstance(value, dict): return None if value.get("schema_version") != schema or value.get("host", "codex-cli") != host: @@ -199,6 +205,7 @@ def _store_codex_cli_session( checkpoint_artifact_sha256: str | None = None, checkpoint_artifact_root: str | None = None, checkpoint_model_transition: Mapping[str, str] | None = None, + expected_document: object = _UNCHECKED_SESSION_DOCUMENT, ) -> None: if checkpoint_artifact_sha256 is not None and not BARE_SHA256_PATTERN.fullmatch(checkpoint_artifact_sha256): raise ValueError("checkpoint session source digest is invalid") @@ -210,9 +217,13 @@ def _store_codex_cli_session( with exclusive_file_lock(path): if require_absent and (path.exists() or path.is_symlink()): raise ValueError("session binding already exists; read back the original admission") + _host_session_schema(host) + current = _read_codex_cli_session_document(runtime_root, lineage=lineage, session_scope=session_scope) + if expected_document is not _UNCHECKED_SESSION_DOCUMENT and current != expected_document: + raise ValueError("managed session binding changed during native initialization") # Provenance survives ordinary observations of this exact conversation, # but cannot migrate to a replacement session, profile or Goal lifetime. - prior = load_codex_cli_session(runtime_root, lineage=lineage, session_scope=session_scope, host=host) + prior = _session_document_value(current, lineage=lineage, session_scope=session_scope, host=host) retained_source = None retained_root = None retained_refresh = None @@ -255,6 +266,95 @@ def _store_codex_cli_session( ) +@dataclass(frozen=True) +class ManagedOperationSession: + """Selected file snapshot and its initialization commit, not an authority. + + Both transports retain the existing session file/schema. The Goal admission + owner checks selection and acceptance; the file lock fences replacement + during native initialization. DSH keeps its original replace-only writer; + Codex keeps checkpoint provenance for the exact original conversation. + """ + + runtime_root: Path + lineage: Mapping[str, str] + host: str + binding: dict[str, Any] | None + original: dict[str, Any] | None + goal_ref: Mapping[str, Any] | None + profile_digest: str + model: str + reasoning_effort: str + goal_admission: FirstPartyHostGoalAdmission | None + + def store(self, session_id: str) -> None: + def commit() -> None: + values: dict[str, Any] = dict( + lineage=self.lineage, host=self.host, session_id=session_id, + goal_ref=self.goal_ref, operation_profile_digest=self.profile_digest, + operation_model=self.model, operation_reasoning_effort=self.reasoning_effort, + ) + if self.host == "dsh": + with exclusive_file_lock(_session_path(self.runtime_root, self.lineage)): + if _read_codex_cli_session_document(self.runtime_root, lineage=self.lineage) != self.original: + raise ValueError("DSH binding changed during SDK initialization") + _write_codex_cli_session(self.runtime_root, **values) + else: + _store_codex_cli_session(self.runtime_root, **values, expected_document=self.original) + + if self.goal_admission is None: + commit() + else: + self.goal_admission.accept_result(commit) + + +def select_managed_operation_session( + runtime_root: Path, *, request: Mapping[str, Any], lineage: Mapping[str, str], + host: str, transport: str, profile_digest: str, model: str, reasoning_effort: str, + goal_admission: FirstPartyHostGoalAdmission | None = None, +) -> ManagedOperationSession: + _host_session_schema(host) + # Read Codex's selected document once, through admission when present. The + # same snapshot is normalized for legacy loading and compared at commit. + # DSH retains its pre-selection snapshot to preserve its initialization fence. + if host == "codex-cli": + if goal_admission is None: + original = _read_codex_cli_session_document(runtime_root, lineage=lineage) + binding = _session_document_value(original, lineage=lineage) + else: + original = goal_admission.select_state( + read_state=lambda: _read_codex_cli_session_document(runtime_root, lineage=lineage), + goal_ref_of=lambda value: _codex_session_goal_ref(value, lineage=lineage), + ) + binding = dict(original) if original is not None else None + else: + original = _read_codex_cli_session_document(runtime_root, lineage=lineage) + binding = select_codex_cli_session( + runtime_root, lineage=lineage, host=host, goal_admission=goal_admission, + ) + session_plan = request.get("session") or {} + if (session_plan.get("context_policy") or {}).get("mode") == "fresh": + binding = None + elif host == "dsh" and original is not None and binding is None: + raise ValueError("existing session is not a qualified DSH operation binding; select fresh explicitly") + action = session_plan.get("action") + has_binding = binding is not None if host == "dsh" else bool(binding) + if action not in {"start_new", "resume"} or (action == "resume") != has_binding: + raise ValueError(f"managed {'DSH' if host == 'dsh' else 'Codex'} session changed after Turn planning") + if binding and ( + binding.get("operation_transport") != transport + or binding.get("operation_profile_digest") != profile_digest + ): + raise ValueError( + "DSH operation profile changed; select fresh and obtain fresh approval" if host == "dsh" + else "managed operation profile changed; explicitly select a fresh iteration and obtain fresh approval" + ) + return ManagedOperationSession( + runtime_root, dict(lineage), host, binding, original, request.get("goal_ref"), + profile_digest, model, reasoning_effort, goal_admission, + ) + + def _replace_checkpoint_codex_session( runtime_root: Path, *, lineage: Mapping[str, str], expected: dict[str, Any], session_id: str, refresh: dict[str, Any], diff --git a/loopx/control_plane/turn_driver/dsh_operation_host.py b/loopx/control_plane/turn_driver/dsh_operation_host.py index f6dcfdc9cc..f840aca462 100644 --- a/loopx/control_plane/turn_driver/dsh_operation_host.py +++ b/loopx/control_plane/turn_driver/dsh_operation_host.py @@ -8,6 +8,7 @@ from __future__ import annotations import hashlib +from collections.abc import Callable from contextlib import ExitStack import importlib.metadata import json @@ -17,8 +18,6 @@ from pathlib import Path from typing import Any, Mapping -from ...chat_action_store import ChatActionStore -from ...file_lock import exclusive_file_lock from ...dsh_goal_mode.turn_host_adapter import ( build_result, build_sdk_config, @@ -29,22 +28,10 @@ terminal_error_reason, terminal_output_budget_state, ) -from ..collaboration.goal_instance_scope import collaboration_goal_scope -from ..collaboration.operation_handoff import ( - agent_operation_action, - managed_operation_binding_current, - pending_operation_handoffs, -) from ..goals.first_party_host_admission import FirstPartyHostGoalAdmission -from .codex_operation_host import operation_tool_handler +from . import managed_operation_host as operations from .operation_requests import operation_request_source, request_reference_instruction -from .codex_sessions import ( - _lineage, - _read_codex_cli_session_document, - _session_path, - _write_codex_cli_session, - select_codex_cli_session, -) +from .codex_sessions import _lineage, select_managed_operation_session from .dsh_operation_connection import DshOperationConnection from .executor import LOOPX_TURN_HOST_REQUEST_SCHEMA_VERSION from .host_failure import BuiltInHostError @@ -86,7 +73,7 @@ def _sdk_runtime_pin(runtime_bin: Path | None = None) -> dict[str, Any]: "@deepseek-ai/dsh": manifest["version"], }, "sdk_files": { - str(path): _digest(Path(sdk.locate_file(path))) + str(path): _digest(Path(str(sdk.locate_file(path)))) for path in sdk.files or () if str(path).startswith("deepseek_harness/") and str(path).endswith(".py") @@ -97,7 +84,9 @@ def _sdk_runtime_pin(runtime_bin: Path | None = None) -> dict[str, Any]: }, "npm_closure_sha256": closure.hexdigest(), } - from deepseek_harness_runtime import resolve_bundled_launch_args + # Optional SDK distributions are untyped; their executable/files are pinned + # below and exercised through the real native transport qualification. + from deepseek_harness_runtime import resolve_bundled_launch_args # type: ignore versions = { name: importlib.metadata.version(name) @@ -184,13 +173,13 @@ def _run_dsh_operation_host( or not 0 < timeout_seconds <= 600 ): raise ValueError("DSH operation run requires bounded token and time budgets") - from deepseek_harness import DeepSeekHarness, DeepSeekHarnessConfig + from deepseek_harness import DeepSeekHarness, DeepSeekHarnessConfig # type: ignore lineage = _lineage(request) module = operation_module.resolve(strict=True) patch = cordis.resolve(strict=True) if cordis else None request_source = operation_request_source(operation_request_file, operation_request_sha256) - profile = { + profile: dict[str, Any] = { "transport": DSH_OPERATION_TRANSPORT, "provider": provider, "model": model, @@ -221,30 +210,12 @@ def _run_dsh_operation_host( profile_digest = hashlib.sha256( json.dumps(profile, sort_keys=True, separators=(",", ":")).encode() ).hexdigest() - original = _read_codex_cli_session_document(runtime_root, lineage=lineage) - binding = select_codex_cli_session( - runtime_root, lineage=lineage, host="dsh", goal_admission=goal_admission + selection = select_managed_operation_session( + runtime_root, request=request, lineage=lineage, host="dsh", + transport=DSH_OPERATION_TRANSPORT, profile_digest=profile_digest, model=model, + reasoning_effort=reasoning_effort, goal_admission=goal_admission, ) - session_plan = request.get("session") or {} - fresh = (session_plan.get("context_policy") or {}).get("mode") == "fresh" - if fresh: - binding = None - elif original is not None and binding is None: - raise ValueError( - "existing session is not a qualified DSH operation binding; select fresh explicitly" - ) - action = session_plan.get("action") - if action not in {"start_new", "resume"} or (action == "resume") != ( - binding is not None - ): - raise ValueError("managed DSH session changed after Turn planning") - if binding and ( - binding.get("operation_transport") != DSH_OPERATION_TRANSPORT - or binding.get("operation_profile_digest") != profile_digest - ): - raise ValueError( - "DSH operation profile changed; select fresh and obtain fresh approval" - ) + binding = selection.binding can_restore = profile["runtime"]["versions"]["deepseek-harness-sdk"] == "0.2.1a3" if binding and owner.session_id is None and not can_restore: raise ValueError( @@ -253,55 +224,24 @@ def _run_dsh_operation_host( session_id = binding["session_id"] if binding else "loopx-dsh-" + uuid.uuid4().hex if owner.session_id is not None and owner.session_id != session_id: raise ValueError("an owned DSH connection cannot switch session identity") - route = { - **lineage, - "host_surface": "loopx-managed-dsh", - "thread_id": session_id, - "profile_digest": profile_digest, - } - store = ChatActionStore(runtime_root / "chat" / "actions") - if confirmed_operation_id is not None: - from ..collaboration.operation_wake import require_current_operation_scope - - proposal = store.load(confirmed_operation_id) - if proposal is None or binding is None or action != "resume": - raise ValueError("confirmed operation requires the original DSH session") - parameters = proposal["normalized_parameters"] - require_current_operation_scope(registry_path, parameters) - store._agent_operation_plan( - proposal, - action="wake", - binding_current=managed_operation_binding_current(runtime_root, parameters), - launch_context={ - "host": "dsh", - "operation_tools": True, - "iteration_context": (session_plan.get("context_policy") or {}).get( - "mode" - ), - }, - executor_route={ - **route, - "model": model, - "reasoning_effort": reasoning_effort, - }, - ) + route = operations.operation_executor_route( + lineage, host="dsh", session_id=session_id, profile_digest=profile_digest, + ) + operations.require_confirmed_operation_resume( + runtime_root=runtime_root, registry_path=registry_path, request=request, + lineage=lineage, binding=binding, host="dsh", profile_digest=profile_digest, + model=model, reasoning_effort=reasoning_effort, + confirmed_operation_id=confirmed_operation_id, + ) continuations: list[dict[str, Any]] = [] def native_start(host_turn_id: str) -> None: - for locator in continuations: - agent_operation_action( - runtime_root, - registry_path, - proposal_id=locator["operation_id"], - actor={ - **route, - "model": model, - "reasoning_effort": reasoning_effort, - "host_turn_id": host_turn_id, - }, - action="observe_host_start", - turn_key=request["turn_key"], - ) + operations.record_operation_start( + runtime_root=runtime_root, registry_path=registry_path, + continuations=continuations, route=route, model=model, + reasoning_effort=reasoning_effort, host_turn_id=host_turn_id, + turn_key=request["turn_key"], + ) native_profile = { "provider": provider, @@ -351,7 +291,7 @@ def native_start(host_turn_id: str) -> None: process = getattr(harness.client, "_proc", None) if process is None or process.poll() is not None: raise ValueError("SDK owned process handle is unavailable") - handler = operation_tool_handler( + handler = operations.operation_tool_handler( runtime_root=runtime_root, registry_path=registry_path, lineage=lineage, @@ -367,60 +307,12 @@ def native_start(host_turn_id: str) -> None: if request_source is not None else None), ) - def store_binding() -> None: - with exclusive_file_lock(_session_path(runtime_root, lineage)): - if ( - _read_codex_cli_session_document(runtime_root, lineage=lineage) - != original - ): - raise ValueError("DSH binding changed during SDK initialization") - _write_codex_cli_session( - runtime_root, - lineage=lineage, - host="dsh", - session_id=session_id, - goal_ref=request.get("goal_ref"), - operation_profile_digest=profile_digest, - operation_model=model, - operation_reasoning_effort=reasoning_effort, - ) - - if goal_admission is None: - store_binding() - else: - goal_admission.accept_result(store_binding) + selection.store(session_id) owner.handler = handler - with collaboration_goal_scope( - registry_path, - goal_id=lineage["goal_id"], - agents=(lineage["agent_id"],), - caller_goal_ref=request.get("goal_ref"), - require_active=True, - ) as scope: - pending = pending_operation_handoffs( - runtime_root, - lineage["goal_id"], - lineage["agent_id"], - registry_path=registry_path, - scope=scope, - executor_route=route, - cursor_scope=hashlib.sha256( - json.dumps(route, sort_keys=True).encode() - ).hexdigest(), - ) - continuations.extend( - { - key: item[key] - for key in ( - "operation_id", - "payload_digest", - "confirmation_digest", - "claim_id", - ) - } - for item in pending["items"] - if item["status"] == "authorized_pending" - ) + continuations.extend(operations.operation_continuations( + runtime_root=runtime_root, registry_path=registry_path, lineage=lineage, + route=route, goal_ref=request.get("goal_ref"), + )) def observe(notification: Any) -> None: connection.observe(notification) @@ -470,7 +362,7 @@ def __init__(self) -> None: self.connection: DshOperationConnection | None = None self.harness: Any = None self.session_id: str | None = None - self.handler: Any = None + self.handler: Callable[[str, Any, dict[str, Any]], dict[str, Any]] | None = None self.on_start: Any = None self._entered = False self._lock = threading.Lock() diff --git a/loopx/control_plane/turn_driver/managed_operation_host.py b/loopx/control_plane/turn_driver/managed_operation_host.py new file mode 100644 index 0000000000..94eb6bde15 --- /dev/null +++ b/loopx/control_plane/turn_driver/managed_operation_host.py @@ -0,0 +1,299 @@ +"""Shared IO for the two owned managed-operation transports. + +Native acceptance and process ownership stay in each adapter. Collaboration's +existing typed owner still decides scope, wake, consumption and outcome; this +module transports those requests and returns only canonical pending locators. +""" + +from __future__ import annotations + +import hashlib +import json +from collections.abc import Callable, Mapping, Sequence +from pathlib import Path +from typing import Any + +from ...chat_action_store import ChatActionStore +from ...chat_actions import ChatActionService +from ..collaboration.goal_instance_scope import collaboration_goal_scope +from ..collaboration.operation_handoff import ( + agent_operation_action, + managed_operation_binding_current, + pending_operation_handoffs, +) +from ..collaboration.operation_wake import require_current_operation_scope +from ..effect_runtime import EffectRuntimeRejected +from ..goals.first_party_host_admission import FirstPartyHostGoalAdmission +from .operation_requests import BoundOperationRequest + + +def _operation_surface(host: str) -> str: + if host not in {"codex-cli", "dsh"}: + raise ValueError("unsupported owned operation host") + return "loopx-managed-dsh" if host == "dsh" else "loopx-managed-codex" + + +def operation_executor_route( + lineage: Mapping[str, str], *, host: str, session_id: str, profile_digest: str, +) -> dict[str, str]: + return {**lineage, "host_surface": _operation_surface(host), + "thread_id": session_id, "profile_digest": profile_digest} + + +REVISION = "managed-turn-handoff-v0" +OPERATION_TOOL = { + "name": "loopx_operation", + "description": "Read/prepare a scoped operation, consume exact human approval once, or report original evidence. Identity comes from this native connection, never arguments. A receipt is not permission to retry an effect.", + "inputSchema": { + "type": "object", + "properties": { + "action": { + "type": "string", + "enum": [ + "context", + "pending", + "prepare", + "inspect", + "consume", + "report", + ], + }, + "proposal_id": {"type": "string"}, + "consumption_id": {"type": "string"}, + "request": {"type": "object"}, + "outcome": {"type": "object"}, + "cursor": {"type": "string"}, + }, + "required": ["action"], + "additionalProperties": False, + }, +} + + +def operation_tool_handler( + *, + runtime_root: Path, + registry_path: Path, + lineage: Mapping[str, str], + session_id: str, + profile_digest: str, + model: str, + reasoning_effort: str, + source_route: Mapping[str, str] | None = None, + goal_admission: FirstPartyHostGoalAdmission | None = None, + host: str = "codex-cli", + frozen_request: BoundOperationRequest | None = None, +) -> Callable[[str, Any, dict[str, Any]], dict[str, Any]]: + """Private closure installed only on a process owned by the Turn driver. + + The caller must already have checked native thread/Turn metadata. This is + not an MCP endpoint or a public actor/proof deserializer. + """ + executor_route = operation_executor_route( + lineage, host=host, session_id=session_id, profile_digest=profile_digest, + ) + executor = { + "kind": "managed_turn", + "todo_id": lineage["todo_id"], + "session_id": session_id, + "profile_digest": profile_digest, + "revision": REVISION, + "model": model, + "reasoning_effort": reasoning_effort, + **({"host": "dsh"} if host == "dsh" else {}), + } + + def handle(tool: str, arguments: Any, native: dict[str, Any]) -> dict[str, Any]: + if tool != "loopx_operation" or not isinstance(arguments, dict): + return {"ok": False, "error": "unsupported_operation_tool"} + if native.get("thread_id") != session_id or not native.get("host_turn_id"): + return {"ok": False, "error": "tool_turn_mismatch"} + allowed = { + "context": {"action"}, + "pending": {"action"}, + "prepare": {"action", "request"}, + "inspect": {"action", "proposal_id"}, + "consume": {"action", "proposal_id", "consumption_id"}, + "report": {"action", "proposal_id", "outcome"}, + } + action = arguments.get("action") + keys = set(arguments) - ({"cursor"} if action == "pending" else set()) + if ( + not isinstance(action, str) + or action not in allowed + or keys != allowed[action] + ): + return {"ok": False, "error": "invalid_operation_arguments"} + try: + if goal_admission is not None: + goal_admission.require_current() + if action == "context": + return { + "ok": True, + **lineage, + "executor": executor, + "authority": "approval_and_first_consumption_required", + "external_write_performed": False, + **({"prepared_request": {"ref": frozen_request.ref}} + if frozen_request is not None else {}), + } + if action == "pending": + with collaboration_goal_scope( + registry_path, + goal_id=lineage["goal_id"], + agents=(lineage["agent_id"],), + ) as scope: + cursor_scope = hashlib.sha256( + json.dumps( + ["managed-operation-v0", str(runtime_root.resolve()), + scope.target(lineage["agent_id"]), executor], + sort_keys=True, separators=(",", ":"), + ).encode() + ).hexdigest() + return { + "ok": True, + **pending_operation_handoffs( + runtime_root, + lineage["goal_id"], + lineage["agent_id"], + registry_path=registry_path, + scope=scope, + cursor=arguments.get("cursor"), + cursor_scope=cursor_scope, + executor_route=executor_route, + ), + } + if action == "prepare": + request = (frozen_request.resolve(arguments["request"]) + if frozen_request is not None else dict(arguments["request"])) + terms = dict(request.get("normalized_parameters") or {}) + if "source_route" in terms and terms["source_route"] != source_route: + raise ValueError("the model cannot retarget the host-selected return audience") + if source_route is not None: + terms["source_route"] = dict(source_route) + for key, expected in { + "goal_id": lineage["goal_id"], + "agent_id": lineage["agent_id"], + "executor": executor, + }.items(): + if key in terms and terms[key] != expected: + raise ValueError( + "operation request cannot select another execution subject" + ) + terms[key] = expected + if request.get("action_kind") != "operation.execute": + raise ValueError("operation tool prepares only typed operations") + request["normalized_parameters"] = terms + service = ChatActionService( + store=ChatActionStore(runtime_root / "chat" / "actions"), + registry_path=registry_path, + ) + return { + "ok": True, + "proposal": service.preview(request), + "execution_allowed": False, + } + actor = { + **executor_route, + "model": model, + "reasoning_effort": reasoning_effort, + "host_turn_id": native["host_turn_id"], + } + return { + "ok": True, + **agent_operation_action( + runtime_root, + registry_path, + proposal_id=arguments["proposal_id"], + actor=actor, + action=action, + consumption_id=arguments.get("consumption_id"), + outcome=arguments.get("outcome"), + ), + } + except EffectRuntimeRejected as exc: + if exc.diagnostic_code == "operation_source_route_ambiguous": + return {"ok": False, "error": "operation_source_route_ambiguous", + "execution_allowed": False, + "next_action": ( + "Select a registered return audience in the owned DSH host source_route; source routing is not executor authority." + if host == "dsh" else + "Select a registered return audience with --codex-operation-source-route-json in the host invocation; source routing is not executor authority." + )} + return {"ok": False, "error": "operation_admission_rejected", "execution_allowed": False} + except (ValueError, KeyError, TypeError, RuntimeError): + # Private payloads, paths and adapter error text never enter the model tool error. + return { + "ok": False, + "error": "operation_admission_rejected", + "execution_allowed": False, + } + + return handle + + +def require_confirmed_operation_resume( + *, runtime_root: Path, registry_path: Path, request: Mapping[str, Any], + lineage: Mapping[str, str], binding: Mapping[str, Any] | None, host: str, + profile_digest: str, model: str, reasoning_effort: str, + confirmed_operation_id: str | None, +) -> None: + """Recheck the existing wake authority before either native process starts.""" + if confirmed_operation_id is None: + return + store = ChatActionStore(runtime_root / "chat" / "actions") + proposal = store.load(confirmed_operation_id) + session_plan = request.get("session") or {} + if proposal is None or binding is None or session_plan.get("action") != "resume": + raise ValueError( + "confirmed operation requires the original DSH session" if host == "dsh" + else "confirmed operation requires its original resumable session" + ) + parameters = proposal["normalized_parameters"] + require_current_operation_scope(registry_path, parameters) + store._agent_operation_plan( + proposal, action="wake", + binding_current=managed_operation_binding_current(runtime_root, parameters), + launch_context={"host": host, "operation_tools": True, + "iteration_context": (session_plan.get("context_policy") or {}).get("mode")}, + executor_route={ + **operation_executor_route(lineage, host=host, + session_id=binding["session_id"], profile_digest=profile_digest), + "model": model, "reasoning_effort": reasoning_effort, + }, + ) + + +def operation_continuations( + *, runtime_root: Path, registry_path: Path, lineage: Mapping[str, str], + route: Mapping[str, str], goal_ref: dict[str, str] | None, +) -> list[dict[str, Any]]: + with collaboration_goal_scope( + registry_path, goal_id=lineage["goal_id"], agents=(lineage["agent_id"],), + caller_goal_ref=goal_ref, require_active=True, + ) as scope: + pending = pending_operation_handoffs( + runtime_root, lineage["goal_id"], lineage["agent_id"], + registry_path=registry_path, scope=scope, executor_route=route, + cursor_scope=hashlib.sha256(json.dumps(route, sort_keys=True).encode()).hexdigest(), + ) + # Locators grant nothing. Private terms still require authenticated inspect. + return [ + {key: item[key] for key in ("operation_id", "payload_digest", "confirmation_digest", "claim_id")} + for item in pending["items"] if item["status"] == "authorized_pending" + ] + + +def record_operation_start( + *, runtime_root: Path, registry_path: Path, continuations: Sequence[Mapping[str, Any]], + route: Mapping[str, str], model: str, reasoning_effort: str, + host_turn_id: str | None, turn_key: str, +) -> None: + """Called only after the adapter authenticates native Turn acceptance.""" + for locator in continuations: + agent_operation_action( + runtime_root, registry_path, proposal_id=locator["operation_id"], + actor={**route, "model": model, "reasoning_effort": reasoning_effort, + "host_turn_id": host_turn_id}, + action="observe_host_start", turn_key=turn_key, + ) diff --git a/tests/control_plane/test_dsh_operation_connection.py b/tests/control_plane/test_dsh_operation_connection.py index de8c949b65..c38b31fdf2 100644 --- a/tests/control_plane/test_dsh_operation_connection.py +++ b/tests/control_plane/test_dsh_operation_connection.py @@ -8,7 +8,7 @@ import pytest from examples import operation_action_fixtures as fixtures -from loopx.control_plane.turn_driver.codex_operation_host import operation_tool_handler +from loopx.control_plane.turn_driver.managed_operation_host import operation_tool_handler from loopx.control_plane.turn_driver.codex_sessions import ( _store_codex_cli_session, load_codex_cli_session, diff --git a/tests/control_plane/test_dsh_operation_entry.py b/tests/control_plane/test_dsh_operation_entry.py index f01da5d63a..d8ace2fa18 100644 --- a/tests/control_plane/test_dsh_operation_entry.py +++ b/tests/control_plane/test_dsh_operation_entry.py @@ -114,7 +114,7 @@ def do_POST(self): thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() monkeypatch.setenv("DEEPSEEK_API_KEY", "loopback-fixture-placeholder") - monkeypatch.setenv("DEEPSEEK_BASE_URL", f"http://127.0.0.1:{server.server_port}") + monkeypatch.setenv("DEEPSEEK_BASE_URL", f"http://127.0.0.1:{server.server_port}/anthropic") patch_module = root / "restrict.mjs" patch_module.write_text( "export const name='fixture-operation-only';\nexport function apply(ctx){" diff --git a/tests/control_plane/test_dsh_operation_sdk.py b/tests/control_plane/test_dsh_operation_sdk.py index ce9f655e15..af27a3058a 100644 --- a/tests/control_plane/test_dsh_operation_sdk.py +++ b/tests/control_plane/test_dsh_operation_sdk.py @@ -106,7 +106,7 @@ def values(value): @pytest.mark.parametrize("request_transport", ["inline", "frozen"]) @pytest.mark.parametrize("authority_provider", ["file", "sqlite"]) @pytest.mark.parametrize( - "connection_lifetime", ["retained", "restarted", "missing", "unreadable"] + "connection_lifetime", ["retained", "restarted", "missing", "unreadable", "fresh", "replaced"] ) def test_real_sdk_native_prepare_confirmation_consume_report_and_profile_revocation( tmp_path, monkeypatch, authority_provider, connection_lifetime, request_transport @@ -159,6 +159,8 @@ def operation(): return rows[0] def arguments(index): + if connection_lifetime == "fresh" and index == 4: + return {"action": "context"} if index == 0: return {"action": "context"} if index == 1: @@ -196,7 +198,7 @@ def do_POST(self): assert index < 9, "bounded scripted request budget exceeded" steps.append(index) provider_messages.append(body["messages"]) - if index in {3, 8}: + if index in {3, 8} or (connection_lifetime == "fresh" and index == 5): delta = { "role": "assistant", "content": json.dumps( @@ -351,6 +353,9 @@ def do_POST(self): first = host.run(turn, **options, on_notification=notifications.append) assert first["result_kind"] == "wait", (first, errors) assert len(steps) == 4, (steps, errors) + original_binding = load_codex_cli_session( + options["runtime_root"], lineage=_lineage(turn), host="dsh" + ) proposal = operation() assert proposal["operation"].get("agent_handoff") is None delivered = store.record_operation_delivery( @@ -368,6 +373,46 @@ def do_POST(self): host = hosts.enter_context(OwnedDshOperationHost()) turn["session"]["action"] = "resume" turn["turn_key"] = "sha256:" + "b" * 64 + if connection_lifetime == "fresh": + turn["session"] = {"action": "start_new", "context_policy": {"mode": "fresh"}} + second = host.run(turn, **options, on_notification=notifications.append) + assert second["result_kind"] == "wait", (second, errors) + binding = load_codex_cli_session( + options["runtime_root"], lineage=_lineage(turn), host="dsh" + ) + assert binding["session_id"] != original_binding["session_id"] + assert len(steps) == 6 and not errors + assert operation()["operation"].get("host_start") is None + assert operation()["operation"].get("agent_handoff") is None + contexts = [row for row in tool_results(notifications) if "executor" in row] + assert {row["executor"]["session_id"] for row in contexts} == { + original_binding["session_id"], binding["session_id"], + } + return + if connection_lifetime == "replaced": + from deepseek_harness import DeepSeekHarness + from loopx.control_plane.turn_driver.codex_sessions import _session_path + + original_enter = DeepSeekHarness.__enter__ + processes = [] + replacement = {**original_binding, "session_id": "concurrent-dsh-session"} + path = _session_path(options["runtime_root"], _lineage(turn)) + + def concurrent_enter(client): + result = original_enter(client) + processes.append(result.client._proc) + path.write_text(json.dumps(replacement)) + return result + + monkeypatch.setattr(DeepSeekHarness, "__enter__", concurrent_enter) + with pytest.raises(ValueError, match="binding changed during SDK initialization"): + host.run(turn, **options, confirmed_operation_id=proposal["proposal_id"]) + assert json.loads(path.read_text()) == replacement + assert not host.connection.path.exists() + assert len(processes) == 1 and processes[0].poll() is not None + assert len(steps) == 4 and not errors + assert operation()["operation"].get("host_start") is None + return if connection_lifetime in {"missing", "unreadable"}: original_binding = load_codex_cli_session( options["runtime_root"], lineage=_lineage(turn), host="dsh" diff --git a/tests/control_plane/test_operation_requests.py b/tests/control_plane/test_operation_requests.py index 2da59d276b..55c0c64436 100644 --- a/tests/control_plane/test_operation_requests.py +++ b/tests/control_plane/test_operation_requests.py @@ -8,7 +8,7 @@ import pytest from examples import operation_action_fixtures as fixtures -from loopx.control_plane.turn_driver.codex_operation_host import operation_tool_handler +from loopx.control_plane.turn_driver.managed_operation_host import operation_tool_handler from loopx.control_plane.turn_driver.codex_sessions import _store_codex_cli_session from loopx.control_plane.turn_driver.operation_requests import ( OperationRequestSource, operation_request_source, diff --git a/tests/test_chat_operation_actions.py b/tests/test_chat_operation_actions.py index 8bc4687483..fd489fa71f 100644 --- a/tests/test_chat_operation_actions.py +++ b/tests/test_chat_operation_actions.py @@ -212,7 +212,7 @@ def record(method, *args, **kwargs): def test_managed_prepare_selects_registered_return_audience_without_rebinding_executor(tmp_path: Path) -> None: - from loopx.control_plane.turn_driver.codex_operation_host import operation_tool_handler + from loopx.control_plane.turn_driver.managed_operation_host import operation_tool_handler service, store = _service(tmp_path) registry = json.loads(service.registry_path.read_text()) diff --git a/tests/test_codex_operation_host.py b/tests/test_codex_operation_host.py index 5dbf957629..535ffc4dff 100644 --- a/tests/test_codex_operation_host.py +++ b/tests/test_codex_operation_host.py @@ -31,6 +31,7 @@ import json, sys thread = "owned-app-server-thread" import os +thread = os.environ.get("FAKE_OPERATION_THREAD_ID", thread) turn = os.environ.get("FAKE_OPERATION_TURN_ID", "native-app-server-turn") key = None def emit(value): @@ -103,7 +104,7 @@ def _registered_host_goal(tmp_path: Path) -> Path: def _claimed_native_fixture(tmp_path: Path): from examples import operation_action_fixtures as fixtures - from loopx.control_plane.turn_driver.codex_operation_host import operation_tool_handler + from loopx.control_plane.turn_driver.managed_operation_host import operation_tool_handler service, store = fixtures.service(tmp_path, goal_id="fixture-goal") executable = tmp_path / "fake-codex-operation" @@ -286,15 +287,77 @@ def test_process_launch_without_native_acceptance_cannot_record_operation_start( assert store.load(claimed["proposal_id"])["operation"].get("host_start") is None +@pytest.mark.parametrize("existing_binding", [False, True]) +def test_native_initialization_cannot_overwrite_a_concurrent_session_binding( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, existing_binding: bool, +) -> None: + from loopx.control_plane.turn_driver import codex_operation_host as owner + from loopx.control_plane.turn_driver.codex_sessions import _session_path + + store, _claimed, request, options = _claimed_native_fixture(tmp_path) + path = _session_path(options["runtime_root"], _lineage(request)) + replacement = json.loads(path.read_text()) + replacement["session_id"] = "concurrent-native-session" + if not existing_binding: + path.unlink() + request["session"]["action"] = "start_new" + original_start = owner.CodexChatAgentSession.start + + def concurrent_start(**kwargs): + session = original_start(**kwargs) + # A different admitted launch wins while the real native connection + # initializes. Its binding must survive, with no prompt/tool dispatch. + path.write_text(json.dumps(replacement)) + return session + + monkeypatch.setattr(owner.CodexChatAgentSession, "start", concurrent_start) + prompt = tmp_path / "unexpected-prompt.txt" + monkeypatch.setenv("FAKE_OPERATION_PROMPT_FILE", str(prompt)) + actions = store.path.read_bytes() + with pytest.raises(ValueError, match="binding changed"): + run_codex_operation_host(request, **options) + assert json.loads(path.read_text()) == replacement + assert store.path.read_bytes() == actions + assert not prompt.exists() + + +def test_resume_retains_checkpoint_source_and_fresh_native_identity_does_not( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, +) -> None: + from loopx.control_plane.turn_driver.codex_sessions import _store_codex_cli_session + + store, claimed, request, options = _claimed_native_fixture(tmp_path) + original = load_codex_cli_session(options["runtime_root"], lineage=_lineage(request)) + _store_codex_cli_session( + options["runtime_root"], lineage=_lineage(request), + session_id=original["session_id"], operation_profile_digest=original["operation_profile_digest"], + operation_model="test-model", operation_reasoning_effort="xhigh", + checkpoint_artifact_sha256="c" * 64, checkpoint_artifact_root=str(tmp_path.resolve()), + ) + run_codex_operation_host(request, **options) + resumed = load_codex_cli_session(options["runtime_root"], lineage=_lineage(request)) + assert resumed["session_id"] == original["session_id"] + assert resumed["checkpoint_artifact_sha256"] == "c" * 64 + start = store.load(claimed["proposal_id"])["operation"]["host_start"] + monkeypatch.setenv("FAKE_OPERATION_THREAD_ID", "fresh-native-session") + request["session"] = {"action": "start_new", "context_policy": {"mode": "fresh"}} + run_codex_operation_host(request, **options) + fresh = load_codex_cli_session(options["runtime_root"], lineage=_lineage(request)) + assert fresh["session_id"] == "fresh-native-session" + assert "checkpoint_artifact_sha256" not in fresh + assert "checkpoint_artifact_root" not in fresh + assert store.load(claimed["proposal_id"])["operation"]["host_start"] == start + + def test_start_receipt_failure_aborts_before_operation_tool_dispatch( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: from loopx.chat_action_store import ChatActionStore - from loopx.control_plane.turn_driver import codex_operation_host store, claimed, request, options = _claimed_native_fixture(tmp_path) before = store.path.read_bytes() calls = [] - original = codex_operation_host.operation_tool_handler + from loopx.control_plane.turn_driver import managed_operation_host + original = managed_operation_host.operation_tool_handler def observe(**options): handler = original(**options) @@ -308,7 +371,7 @@ def record(*args): def fail(*args, **kwargs): raise RuntimeError("synthetic receipt write failure") - monkeypatch.setattr(codex_operation_host, "operation_tool_handler", observe) + monkeypatch.setattr(managed_operation_host, "operation_tool_handler", observe) monkeypatch.setattr(ChatActionStore, "record_agent_operation_host_start", fail) with pytest.raises(BuiltInHostError) as error: run_codex_operation_host(request, **options) @@ -373,10 +436,7 @@ def test_registered_source_tokens_survive_owned_native_prepare_and_store_readbac service, ) from loopx.chat_agent import CodexChatAgentSession - from loopx.control_plane.turn_driver.codex_operation_host import ( - OPERATION_TOOL, - operation_tool_handler, - ) + from loopx.control_plane.turn_driver.managed_operation_host import OPERATION_TOOL, operation_tool_handler from loopx.thread_agent_binding import ( bind_thread_agent_in_registry, resolve_registry_thread_agent_binding, @@ -689,9 +749,9 @@ def test_owned_operation_host_reaps_descendants_on_success_and_timeout( def test_live_owned_app_server_native_tool_metadata( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, model: str, effort: str ) -> None: - from loopx.control_plane.turn_driver import codex_operation_host - original = codex_operation_host.operation_tool_handler + from loopx.control_plane.turn_driver import managed_operation_host + original = managed_operation_host.operation_tool_handler calls = [] def observe(**options): @@ -704,7 +764,7 @@ def record(tool, arguments, native): return record - monkeypatch.setattr(codex_operation_host, "operation_tool_handler", observe) + monkeypatch.setattr(managed_operation_host, "operation_tool_handler", observe) request = _request() request["turn_envelope"]["action"]["selected_todo"]["text"] = ( "Non-financial transport qualification only. Call loopx_operation context once. " diff --git a/tests/test_local_delegation.py b/tests/test_local_delegation.py index 44c64121f5..0bfa0250ae 100644 --- a/tests/test_local_delegation.py +++ b/tests/test_local_delegation.py @@ -280,7 +280,8 @@ def test_confirmed_wake_reaches_native_acceptance_through_real_delegation_and_tu from loopx.cli import build_parser from loopx.control_plane.collaboration.operation_wake import dispatch_confirmed_operation_wake from loopx.control_plane.turn_driver.codex_cli import load_codex_cli_session - from loopx.control_plane.turn_driver.codex_operation_host import run_codex_operation_host, operation_tool_handler + from loopx.control_plane.turn_driver.codex_operation_host import run_codex_operation_host + from loopx.control_plane.turn_driver.managed_operation_host import operation_tool_handler from test_codex_operation_host import FAKE_SERVER from test_loopx_turn_codex_cli import _request