From bbbc7c3ef8f0e5add4c2abd2a405e880a689745d Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 11 Nov 2026 16:43:26 +0800 Subject: [PATCH] fix(chat): reconcile adopted steward executor bindings Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../native_chat/conversation_bindings.py | 103 ++++++++++++++++- loopx/chat_server.py | 3 + .../collaboration/conversation_binding.ts | 33 ++++++ tests/test_chat_conversation_bindings.py | 109 ++++++++++++++++++ 4 files changed, 247 insertions(+), 1 deletion(-) diff --git a/loopx/capabilities/native_chat/conversation_bindings.py b/loopx/capabilities/native_chat/conversation_bindings.py index dda5e5aff5..0ce429598c 100644 --- a/loopx/capabilities/native_chat/conversation_bindings.py +++ b/loopx/capabilities/native_chat/conversation_bindings.py @@ -6,8 +6,9 @@ """ from __future__ import annotations -from collections.abc import Callable, Iterator +from collections.abc import Callable, Iterator, Mapping from contextlib import contextmanager +import logging from pathlib import Path from typing import TYPE_CHECKING, Any import uuid @@ -168,6 +169,106 @@ def select_app_steward(self, binding_id: str | None, *, expected_revision: int) raise OSError("App steward selection did not verify") return dict(result["state"]) + def reconcile_steward_executors(self) -> None: + """Publish a previously adopted machine executor at Chat startup.""" + from ...chat_manager import steward_machine_defaults + + controller = self.controller + if controller is None: + return + defaults = steward_machine_defaults(controller) + if (not defaults or defaults.get("status") != "ready" + or defaults.get("source") != "machine_configuration" + or defaults.get("selection_policy") not in {"preferred", "pinned"} + or defaults.get("executor_endpoint") not in {"antigravity", "codex"}): + return + try: + rows = self.read()["bindings"] + if not isinstance(rows, list): + raise ValueError("invalid conversation bindings state") + except (OSError, ValueError, KeyError) as exc: + logging.getLogger(__name__).warning("Steward bindings unavailable for reconciliation (%s)", type(exc).__name__) + return + for row in rows: + if (not isinstance(row, dict) or row.get("context_kind") != "steward" + or row.get("audience", "owner") != "owner" + or row.get("enabled") is not True + or row.get("executor_endpoint_id") not in {"antigravity", "codex"} + or row.get("executor_endpoint_id") == defaults["executor_endpoint"]): + continue + try: + self._reconcile_steward_executor(row, defaults) + except (OSError, ValueError, TypeError, KeyError, RuntimeError) as exc: + logging.getLogger(__name__).warning("Steward executor reconciliation held (%s)", type(exc).__name__) + + def _reconcile_steward_executor(self, row: dict[str, Any], defaults: Mapping[str, Any]) -> None: + from ...chat_manager import manager_executor_allocation, steward_machine_defaults + from ..machine_configuration.builtins import build_builtin_machine_configuration_registry + from ..machine_configuration.store import machine_configuration_revision, read_machine_configuration + from ..steward_executor.allocation import normalize_manager_executor_allocation + from ..steward_executor.machine_defaults import effective_steward_executor_defaults + + store = self.controller.store + binding_id = row["binding_id"] + endpoint = defaults["executor_endpoint"] + channel_prefix = f"manager.external.native.{binding_id}." + candidates = [s for s in store.list_sessions() + if s.get("channel_id", "").startswith(channel_prefix) and s.get("status") != "closed"] + if len(candidates) != 1 or candidates[0].get("agent_id") != endpoint or candidates[0].get("status") != "ready": + return # Defaults alone do not retarget an existing session. + observation = self.observe(row["transport_ref"]) + with exclusive_file_lock(self.path, operation="reconcile_steward_executor"): + current = self.read() + if row not in current["bindings"] or steward_machine_defaults(self.controller) != defaults: + return + session = store.load_session(candidates[0]["session_id"]) + saved = (session or {}).get("steward_context") + stored = (session or {}).get("manager_executor_allocation") + if (not session or not isinstance(saved, dict) or not isinstance(stored, dict) + or session.get("status") != "ready" or session.get("active_turn_id") is not None + or session.get("agent_id") != endpoint or session.get("executor_endpoint_id") != endpoint + or stored.get("selection_policy") != defaults["selection_policy"] + or store.queued_turns(session["session_id"], strict=True, include_reserved=True)): + return + allocation = None + if stored.get("configuration_revision") == defaults.get("configuration_revision"): + try: + allocation = normalize_manager_executor_allocation(stored) + except (TypeError, ValueError): + pass + else: + # A machine configuration transaction records the revision of + # the whole document. Match that exact basis before projecting + # it to the narrower revision used by steward allocations. + configuration = read_machine_configuration(store.root.parent, + registry=build_builtin_machine_configuration_registry()) + if (not configuration or machine_configuration_revision(configuration) != stored.get("configuration_revision") + or effective_steward_executor_defaults(configuration) != defaults): + return + if allocation is None: + allocation = manager_executor_allocation(self.controller, machine_defaults=defaults) + expected = {**self._session_observation(session), "manager_executor_allocation": allocation, + "active_turn_id": session.get("active_turn_id"), "agent_model_id": session.get("agent_model_id"), + "agent_reasoning_effort": session.get("agent_reasoning_effort")} + result = self._core("collaboration.conversation.binding", { + "current": current, "expected_revision": current["revision"], "operation": "reconcile_steward_executor", + "binding_id": binding_id, "observation": observation, "available_projects": self.projects.available(), + "available_goal_ids": self.goal_scope_ids(row), "machine_defaults": dict(defaults), + "session": expected, "queue_empty": True, + }) + if stored != allocation: + store.update_session(session["session_id"], expected_statuses={"ready"}, + expected_upstream_thread_id=session.get("upstream_thread_id"), manager_executor_allocation=allocation) + fresh = store.load_session(session["session_id"]) + if (not fresh or any(fresh.get(key) != value for key, value in expected.items()) + or store.queued_turns(session["session_id"], strict=True, include_reserved=True) + or steward_machine_defaults(self.controller) != defaults): + return + if result["changed"]: + _atomic_write_json(self.path, result["state"], preserve_mode=True) + if self.read() != result["state"]: + raise OSError("steward executor binding publication did not verify") + def app_steward_target(self) -> dict[str, Any]: """Discover within the selected source; opening creates no execution.""" current = self.read() diff --git a/loopx/chat_server.py b/loopx/chat_server.py index ae9abf8ba3..711818a2a4 100644 --- a/loopx/chat_server.py +++ b/loopx/chat_server.py @@ -2042,6 +2042,9 @@ def _wake_goal_context(session: dict[str, object]) -> dict[str, object]: if open_browser: webbrowser.open(url) try: + bindings = server.runtime_controller.project_contexts.conversation_bindings + if bindings is not None: + bindings.reconcile_steward_executors() if external_conversation_factories: server.conversation_transports.start(server) background_api.register_chat_background_services(server) diff --git a/loopx/control_plane/collaboration/conversation_binding.ts b/loopx/control_plane/collaboration/conversation_binding.ts index f176966b4b..3703246b21 100644 --- a/loopx/control_plane/collaboration/conversation_binding.ts +++ b/loopx/control_plane/collaboration/conversation_binding.ts @@ -233,6 +233,39 @@ export function planConversationBinding(params: JsonObject): JsonObject { } if ((current.app_steward_binding_id ?? null) === id) return {changed: false, state: current}; return {changed: true, state: state({...current, revision: current.revision + 1, app_steward_binding_id: id})}; + } else if (params.operation === "reconcile_steward_executor") { + const id = ref(params.binding_id, "steward binding"); + const previous = current.bindings.find(row => row.binding_id === id); + const session = requireJsonObject(params.session, "original steward Session"); + const allocation = requireJsonObject(session.manager_executor_allocation, "steward executor allocation"); + const defaults = requireJsonObject(params.machine_defaults, "steward machine defaults"); + const endpoint = defaults.executor_endpoint; + if (!previous || previous.context_kind !== "steward" || previous.audience === "group" + || !["antigravity", "codex"].includes(String(previous.executor_endpoint_id)) + || !["antigravity", "codex"].includes(String(endpoint)) + || defaults.status !== "ready" || defaults.source !== "machine_configuration" + || !["preferred", "pinned"].includes(String(defaults.selection_policy)) + || typeof defaults.configuration_revision !== "string" + || !/^sha256:[a-f0-9]{64}$/.test(defaults.configuration_revision) + || allocation.schema_version !== "manager_executor_allocation_v0" + || allocation.configuration_revision !== defaults.configuration_revision + || allocation.selection_policy !== defaults.selection_policy || allocation.available !== true + || allocation.configured_endpoint !== endpoint || allocation.executor_endpoint !== endpoint + || allocation.model !== session.agent_model_id || allocation.reasoning_effort !== session.agent_reasoning_effort + || (defaults.executor_model != null && defaults.executor_model !== allocation.model) + || (defaults.executor_reasoning_effort != null && defaults.executor_reasoning_effort !== allocation.reasoning_effort) + || session.agent_id !== endpoint || session.executor_endpoint_id !== endpoint + || session.status !== "ready" || session.active_turn_id !== null || params.queue_empty !== true) { + throw new EffectRuntimeRequestError("reconciliation requires the idle original steward and its adopted machine executor"); + } + const saved = requireJsonObject(session.steward_context, "original steward context"); + rows = current.bindings.map(row => row === previous ? {...row, executor_endpoint_id: endpoint} : row); + resolveBoundConversation({current: {...current, bindings: rows}, binding_id: id, + observation: params.observation, available_projects: params.available_projects, + available_goal_ids: params.available_goal_ids, source_ref: saved.source_ref, + sender_ref: previous.operator_ref, private_human_message: false, origin: "web", + session_context: saved, session}); + if (previous.executor_endpoint_id === endpoint) return {changed: false, state: current}; } else if (params.operation === "configure") { const candidate = binding(params.binding); observed(candidate, params.observation); diff --git a/tests/test_chat_conversation_bindings.py b/tests/test_chat_conversation_bindings.py index 9f802e5d41..b824abebb4 100644 --- a/tests/test_chat_conversation_bindings.py +++ b/tests/test_chat_conversation_bindings.py @@ -729,6 +729,115 @@ def background_inspection(): assert (capture.read_bytes(), store.messages(sid), bindings.path.read_bytes()) == before +@pytest.mark.parametrize("hold", [None, "publication", "queue", "configuration", "operator", "default-only"]) +def test_startup_reconciles_only_an_adopted_steward_executor(steward, monkeypatch, hold): # noqa: F811 + from test_chat_manager_model_adoption import configure + from loopx.capabilities.machine_configuration.builtins import build_builtin_machine_configuration_registry + from loopx.capabilities.machine_configuration.store import machine_configuration_revision, read_machine_configuration + + store, runtime, provider, transport, binding, capture, _ = steward + bindings = transport.bindings + event = provider.event("steward-app", "old-steward-request", "Read the original conversation") + transport.admit("steward-app", event) + request_ref = identity_ref(binding["provider_ref"], event["message_id"]) + request = transport.core.read_request(request_ref) + sid = request["session_id"] + assert runtime.wait_for_turn(session_id=sid, turn_id=request["turn_id"], timeout_sec=10)["status"] == "completed" + transport.reconcile() + bindings.select_app_steward(binding["binding_id"], expected_revision=bindings.read()["revision"]) + before = bindings.read() + stale = {**before, "bindings": [ + {**row, "executor_endpoint_id": "antigravity"} if row["binding_id"] == binding["binding_id"] else row + for row in before["bindings"]]} + bindings.path.write_text(json.dumps(stale)) + configure(store.root.parent, model="gpt-6.1-sol", effort="xhigh") + revision = runtime.steward_executor_defaults()["configuration_revision"] + configuration = read_machine_configuration(store.root.parent, registry=build_builtin_machine_configuration_registry()) + transaction_revision = machine_configuration_revision(configuration) + assert transaction_revision != revision + session = store.update_session(sid, agent_model_id="gpt-6.1-sol", agent_reasoning_effort="xhigh") + session_path = store._session_path(sid) + # A partial previous adoption persisted the endpoint but did not finish the + # binding or write a usable allocation. The actual Session already exists. + session["manager_executor_allocation"] = { + "schema_version": 1, "configuration_revision": transaction_revision if hold != "configuration" else "old", + "selection_policy": "preferred", + } + if hold == "default-only": + session.update(agent_id="antigravity", executor_endpoint_id="antigravity") + session_path.write_text(json.dumps(session)) + if hold == "queue": + store.create_queued_turn(sid, client_turn_id="queued-before-adoption", message="Keep the existing queue", origin="web") + if hold == "operator": + original_observe = bindings.observe + monkeypatch.setattr(bindings, "observe", lambda profile: {**original_observe(profile), "operator_ref": "f" * 24}) + original_turn = store._turn_path(sid, request["turn_id"]).read_bytes() + messages = store.messages(sid) + starts = capture.read_bytes() + writes = list(provider.writes) + + if hold in {None, "publication"}: + with pytest.raises(ValueError, match="canonical bound Session identity changed"): + bindings.app_steward_target() + pending = provider.event("steward-app", "prepared-before-adoption", "Continue in the same conversation") + pending_ref = identity_ref(binding["provider_ref"], pending["message_id"]) + + def interrupt_preparation(*args, **kwargs): + raise OSError("admission interrupted before a Turn") + + with monkeypatch.context() as interruption: + interruption.setattr(transport.core, "_admit_prepared", interrupt_preparation) + with pytest.raises(OSError, match="before a Turn"): + transport.admit("steward-app", pending) + assert transport.core.read_request(pending_ref)["status"] == "prepared" + + if hold == "publication": + from loopx.capabilities.native_chat import conversation_bindings + + def interrupted_publication(*args, **kwargs): + raise OSError("binding publication interrupted") + + with monkeypatch.context() as interruption: + interruption.setattr(conversation_bindings, "_atomic_write_json", interrupted_publication) + bindings.reconcile_steward_executors() + assert bindings.read() == stale + assert store.load_session(sid)["manager_executor_allocation"]["schema_version"] == "manager_executor_allocation_v0" + assert transport.core.read_request(pending_ref)["status"] == "prepared" + + bindings.reconcile_steward_executors() + + assert store.messages(sid) == messages + assert store._turn_path(sid, request["turn_id"]).read_bytes() == original_turn + assert capture.read_bytes() == starts and provider.writes == writes + if hold not in {None, "publication"}: + assert bindings.read() == stale + assert store.load_session(sid)["manager_executor_allocation"] == session["manager_executor_allocation"] + return + assert bindings.read() == {**before, "revision": before["revision"] + 1} + assert bindings.app_steward_target() == {"binding_id": binding["binding_id"], "session_id": sid} + updated = store.load_session(sid) + allocation = updated["manager_executor_allocation"] + assert allocation["schema_version"] == "manager_executor_allocation_v0" + assert allocation["configuration_revision"] == revision + assert updated["steward_context"] == session["steward_context"] + assert updated["upstream_thread_id"] == session["upstream_thread_id"] + snapshot = bindings.path.read_bytes(), session_path.read_bytes() + bindings.reconcile_steward_executors() + assert (bindings.path.read_bytes(), session_path.read_bytes()) == snapshot + + path = transport.root / f"{pending_ref}.json" + transport.reconcile_request(path) + resumed = transport.core.read_request(pending_ref) + assert resumed["status"] == "accepted" and resumed["session_id"] == sid + turn = runtime.wait_for_turn(session_id=sid, turn_id=resumed["turn_id"], timeout_sec=10) + assert turn["status"] == "completed" and turn["client_turn_id"] == f"external-{pending_ref}" + transport.reconcile_request(path) + assert json.loads(path.read_text())["deliveries"]["terminal"]["verified"] is True + transport.reconcile_request(path) + calls = [json.loads(line) for line in capture.read_text().splitlines()] + assert sum(call.get("method") == "turn/start" for call in calls) == 2 + + @pytest.mark.parametrize("change", ["selection", "revocation"]) def test_app_steward_snapshot_rechecks_grant_after_core_discovery(steward, monkeypatch, change): # noqa: F811 store, runtime, provider, transport, binding, capture, _ = steward