feat(chat): select the steward channel executor explicitly

docs/reference/protocols/manager-evidence-and-continuity-v0.md

@@ -81,10 +81,43 @@ in a `manager.context` event, and supplies it to the executor. Chat prose is
 never the inventory. Normal manager questions no longer silently use a limited
 frontend projection; explicitly choosing status-only still uses that projection.
 
-For Codex, manager defaults are `gpt-6-astra` with `high` reasoning. Set
+Manager defaults are `gpt-6-astra` with `high` reasoning. Set
 `LOOPX_MANAGER_MODEL` and `LOOPX_MANAGER_REASONING_EFFORT` on the Chat service to
-override them. Thread start, resume and turn start explicitly carry the settings;
-worker configuration is unchanged. Capabilities expose the manager defaults.
+override them; an explicit override always wins. Thread start, resume and turn
+start explicitly carry the settings; worker configuration is unchanged.
+Capabilities expose the manager defaults and their source.
+
+### Steward channel host selection
+
+The steward channel **selects** its executor; nothing discovers it. `codex` is
+the shipped endpoint because it is the only transport that can hold an
+interactive steward session today, and `LOOPX_MANAGER_ENDPOINT` re-points that
+default. A configured operator credential (`DEEPSEEK_API_KEY`, endpoint in
+`DEEPSEEK_BASE_URL`) is never a selection signal: discovering a provider key does
+not move the steward channel onto that provider's executor, and it does not move
+the channel's model either. The model follows the endpoint the operator selected,
+so `gpt-6-astra` stays the default while the channel runs on the CLI endpoint. The
+credential is reported as a fact -- the variable name, never the value -- and only
+for the endpoint that actually authenticates with it.
+
+The managed Turn host (`dsh`) runs one bounded work segment per request and has no
+interactive Chat transport, so a session request that names it fails as the typed
+`managed_host_chat_transport_unsupported` host-tool gate instead of an unknown
+endpoint error, in both the Chat service and Lark routing. Promoting `dsh` to the
+steward default is gated on that transport, not on a credential. The steward still
+**drives** managed work on `dsh`: those bounded Turns are the managed execution
+unit described by the LoopX Turn host selection contract, and they are separate
+from the channel the steward answers on.
+
+The Chat capabilities payload carries this resolution in its `manager` block
+(`channel_binding`): the resolved executor endpoint and its source, its executor
+kind in the same vocabulary as the governed Turn surface (`individual` runs on
+one person's CLI login, `managed` on an operator credential), the resolved model
+and its source, whether an operator credential is configured, and
+`available`/`unavailable_reason` when LoopX can prove the selected endpoint cannot
+serve this channel. `available` is `null` when the projection makes no claim. A
+frontend can show which executor and model the steward channel resolved, and why,
+without re-deriving the rule.
 Legacy managed manager sessions retain their logical identity and bounded chat
 history but start a fresh executor thread in the same Codex home on first
 restore. This removes inherited project instructions without importing sessions

examples/loopx-steward-channel-binding-smoke.py

@@ -0,0 +1,195 @@
+#!/usr/bin/env python3
+"""Prove the steward channel selects its executor explicitly and stays put."""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+import tempfile
+from pathlib import Path
+
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO_ROOT))
+
+from loopx.chat_agent import (  # noqa: E402
+    MANAGED_HOST_CHAT_TRANSPORT_UNSUPPORTED,
+    CodexChatAgentError,
+)
+from loopx.chat_manager import (  # noqa: E402
+    MANAGER_ENDPOINT_ENV_VAR,
+    MANAGER_ENDPOINT_SOURCE_EXPLICIT_CONFIG,
+    manager_channel_binding,
+    manager_executor_endpoint_default,
+    manager_model_config,
+    open_manager_session,
+)
+from loopx.chat_runtime import ChatRuntimeController  # noqa: E402
+from loopx.chat_store import ChatSessionStore  # noqa: E402
+
+
+CREDENTIAL_ENV = "DEEPSEEK_API_KEY"
+CREDENTIAL_VALUE = "fixture-operator-credential"
+
+
+def _assert(condition: bool, message: str) -> None:
+    if not condition:
+        raise SystemExit(f"steward channel binding smoke failed: {message}")
+
+
+def _assert_credential_does_not_select() -> dict[str, object]:
+    """A configured provider key must not re-point the steward channel."""
+
+    without_credential = manager_channel_binding({})
+    with_credential = manager_channel_binding({CREDENTIAL_ENV: CREDENTIAL_VALUE})
+    _assert(
+        without_credential["executor_endpoint"] == "codex"
+        and with_credential["executor_endpoint"] == "codex",
+        "the steward channel must keep the shipped CLI endpoint either way",
+    )
+    _assert(
+        without_credential["executor_endpoint_source"]
+        == with_credential["executor_endpoint_source"]
+        == "product_default",
+        "a credential must never become the endpoint source",
+    )
+    _assert(
+        without_credential["model"] == with_credential["model"] == "gpt-6-astra"
+        and with_credential["model_source"] == "vendor_default",
+        "a credential for a provider this channel does not run on must not move the model",
+    )
+    _assert(
+        with_credential["operator_credential_configured"] is True
+        and without_credential["operator_credential_configured"] is False,
+        "the credential must still be reported as a fact",
+    )
+    _assert(
+        CREDENTIAL_VALUE not in json.dumps(with_credential)
+        and with_credential["credential_env_var"] == "",
+        "the binding must never echo a credential value, nor claim one for a CLI endpoint",
+    )
+    _assert(
+        manager_model_config(
+            {CREDENTIAL_ENV: CREDENTIAL_VALUE, "LOOPX_MANAGER_MODEL": "fixture-model"}
+        )
+        == {"model": "fixture-model", "reasoning_effort": "high"},
+        "an explicit model override must win over the shipped default",
+    )
+    return {
+        "without_credential": without_credential,
+        "with_credential": with_credential,
+    }
+
+
+def _assert_explicit_selection_and_managed_host_gate() -> dict[str, object]:
+    """Explicit selection wins; the managed host fails closed as a typed gate."""
+
+    selected = manager_channel_binding({MANAGER_ENDPOINT_ENV_VAR: "dsh"})
+    _assert(
+        selected["executor_endpoint"] == "dsh"
+        and selected["executor_endpoint_source"]
+        == MANAGER_ENDPOINT_SOURCE_EXPLICIT_CONFIG,
+        "an explicit endpoint selection must win over the shipped default",
+    )
+    _assert(
+        selected["executor_kind"] == "managed"
+        and selected["available"] is False
+        and selected["unavailable_reason"] == MANAGED_HOST_CHAT_TRANSPORT_UNSUPPORTED,
+        "the managed host must report its missing Chat transport as a typed reason",
+    )
+    _assert(
+        manager_executor_endpoint_default(
+            {MANAGER_ENDPOINT_ENV_VAR: "dsh", CREDENTIAL_ENV: CREDENTIAL_VALUE}
+        )
+        == "dsh",
+        "the selected endpoint must not depend on the credential",
+    )
+
+    with tempfile.TemporaryDirectory() as gate_root:
+        root = Path(gate_root)
+        runtime = ChatRuntimeController(
+            store=ChatSessionStore(root / "store"), codex_bin="fixture-codex"
+        )
+        try:
+            try:
+                runtime.open_session(
+                    goal_id="loopx-steward-binding-fixture",
+                    agent_id="dsh",
+                    work_dir=root,
+                    objective="fixture",
+                    mode="new",
+                )
+            except CodexChatAgentError as exc:
+                _assert(
+                    exc.error_code == MANAGED_HOST_CHAT_TRANSPORT_UNSUPPORTED
+                    and exc.gate.get("kind") == "host_tool_gate"
+                    and "loopx turn" in exc.gate.get("next_action", ""),
+                    "the managed host must fail closed as a typed host-tool gate",
+                )
+            else:
+                raise SystemExit(
+                    "steward channel binding smoke failed: dsh opened an interactive session"
+                )
+        finally:
+            runtime.close()
+    return selected
+
+
+def _assert_session_opens_the_selected_endpoint() -> str:
+    """The channel opens the endpoint the operator selected, not a credential."""
+
+    opened: list[dict[str, object]] = []
+
+    class _Controller:
+        def open_session(self, **kwargs):
+            opened.append(kwargs)
+            return {"session_id": "fixture-session"}, False
+
+    controller = _Controller()
+    with tempfile.TemporaryDirectory() as work_dir:
+        ambient = os.environ.pop(CREDENTIAL_ENV, None)
+        ambient_endpoint = os.environ.pop(MANAGER_ENDPOINT_ENV_VAR, None)
+        try:
+            open_manager_session(
+                controller=controller,
+                goal_id="loopx-steward-binding-fixture",
+                work_dir=Path(work_dir),
+            )
+            _assert(
+                opened[-1]["agent_id"] == "codex",
+                "without a selection the manager session must open the shipped endpoint",
+            )
+
+            os.environ[CREDENTIAL_ENV] = CREDENTIAL_VALUE
+            open_manager_session(
+                controller=controller,
+                goal_id="loopx-steward-binding-fixture",
+                work_dir=Path(work_dir),
+            )
+            _assert(
+                opened[-1]["agent_id"] == "codex",
+                "a configured credential must not re-point the manager session",
+            )
+        finally:
+            os.environ.pop(CREDENTIAL_ENV, None)
+            if ambient is not None:
+                os.environ[CREDENTIAL_ENV] = ambient
+            if ambient_endpoint is not None:
+                os.environ[MANAGER_ENDPOINT_ENV_VAR] = ambient_endpoint
+    return str(opened[-1]["agent_id"])
+
+
+def main() -> int:
+    payload = {
+        "ok": True,
+        "credential_selection_probe": _assert_credential_does_not_select(),
+        "explicit_selection": _assert_explicit_selection_and_managed_host_gate(),
+        "opened_endpoint": _assert_session_opens_the_selected_endpoint(),
+    }
+    print(json.dumps(payload, ensure_ascii=False, indent=2))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

loopx/capabilities/manager_runtime/machine_profile.py

@@ -201,13 +201,28 @@ def manager_runtime_session_fields(profile: Mapping[str, Any]) -> dict[str, Any]
 def manager_runtime_capability_projection(
     runtime_controller: object,
     model_configuration: Mapping[str, Any],
+    *,
+    channel_binding: Mapping[str, Any] | None = None,
 ) -> dict[str, Any]:
-    """Build the manager section of the shared chat capabilities projection."""
+    """Build the manager section of the shared chat capabilities projection.
+
+    ``channel_binding`` is the resolved steward-channel executor and model
+    binding. The caller owns it, including its credential facts; this projection
+    only carries it into readback so a frontend can show which executor and model
+    the manager channel resolved and why, without re-deriving the rule.
+    """
 
     resolver = getattr(runtime_controller, "manager_runtime_profile", None)
     runtime = (
         resolver()
         if callable(resolver)
         else {**effective_manager_runtime_profile(None), "status": "ready"}
     )
-    return {"scope": "owner_global", **dict(model_configuration), "runtime": runtime}
+    projection: dict[str, Any] = {
+        "scope": "owner_global",
+        **dict(model_configuration),
+        "runtime": runtime,
+    }
+    if channel_binding is not None:
+        projection["channel_binding"] = dict(channel_binding)
+    return projection

loopx/chat_agent.py

@@ -47,6 +47,35 @@ def _host_tool_gate(summary: str, next_action: str) -> dict[str, str]:
     }
 
 
+# The managed Turn host runs one bounded work segment per request and has no
+# interactive Chat transport, so a session request for it is a known outcome
+# rather than an unknown endpoint.
+MANAGED_TURN_HOST_IDS = frozenset({"dsh"})
+MANAGED_HOST_CHAT_TRANSPORT_UNSUPPORTED = "managed_host_chat_transport_unsupported"
+
+
+def agent_endpoint_error(agent_id: str) -> ValueError:
+    """Return the typed error for an Agent id this runtime cannot hold.
+
+    A managed Turn host keeps a typed host-tool gate and an actionable next
+    step, so the steward channel never half-connects to a host it cannot hold.
+    Every other unknown id keeps the existing untyped fallback.
+    """
+
+    if agent_id in MANAGED_TURN_HOST_IDS:
+        return CodexChatAgentError(
+            f"The managed host '{agent_id}' runs bounded LoopX Turns and cannot "
+            "hold an interactive Chat session yet.",
+            error_code=MANAGED_HOST_CHAT_TRANSPORT_UNSUPPORTED,
+            gate=_host_tool_gate(
+                f"'{agent_id}' has no LoopX Chat transport; it is a bounded Turn host.",
+                "Select a chat-capable Agent endpoint for this session, or run "
+                "the managed host through `loopx turn`.",
+            ),
+        )
+    return ValueError(f"unknown Agent endpoint: {agent_id}")
+
+
 def _approval_gate(summary: str) -> dict[str, str]:
     return {
         "kind": "approval_gate",

loopx/chat_lark_api.py

@@ -30,7 +30,12 @@
     CommandRunner,
     default_subprocess_runner,
 )
-from .chat_manager import manager_channel, open_manager_session
+from .chat_agent import CodexChatAgentError
+from .chat_manager import (
+    manager_channel,
+    manager_executor_endpoint_default,
+    open_manager_session,
+)
 from .extensions.lark.goal_channel_contracts import binding_for_goal, goal_from_registry
 from .extensions.lark.goal_channel_targets import goal_channel_target_for_name
 from .history import load_registry
@@ -477,11 +482,11 @@ def _lark_connect(self) -> None:
                 or stored_routing.get("conversation_kind")
                 or "goal"
             )
-            executor_endpoint_id = (
-                _compact_text(body.get("executor_endpoint_id"), limit=100)
-                or stored_routing.get("executor_endpoint_id")
-                or "codex"
-            )
+            executor_endpoint_id = _compact_text(
+                body.get("executor_endpoint_id"), limit=100
+            ) or stored_routing.get("executor_endpoint_id")
+            if conversation_kind == "manager" and not executor_endpoint_id:
+                executor_endpoint_id = manager_executor_endpoint_default()
             session_id: str | None = None
             session_ids_by_agent: dict[str, str] = {}
             if conversation_kind == "manager":
@@ -591,6 +596,11 @@ def _lark_connect(self) -> None:
                     else None,
                     session_id=session_id,
                 )
+        except CodexChatAgentError as exc:
+            self._send_error(
+                str(exc), status=400, gate=exc.gate, error_code=exc.error_code
+            )
+            return
         except ValueError as exc:
             self._send_error(str(exc), status=400, error_code="invalid_lark_connection")
             return