fix(chat): reconcile adopted steward executor bindings

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

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)

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

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