perf(kernel): read the steward endpoint vocabulary without the chat layer

loopx/capabilities/steward_executor/machine_defaults.py

@@ -27,13 +27,41 @@
 from pathlib import Path
 from typing import Any
 
+from ...control_plane.turn_driver.host_binding import (
+    EXECUTOR_KIND_MANAGED,
+    MANAGED_TURN_HOST,
+)
 from ...reasoning_effort import REASONING_EFFORTS
 from ..machine_configuration.contract import (
     MACHINE_CONFIGURATION_SCHEMA,
     MachineConfigurationNamespace,
     machine_configuration_revision,
 )
 
+# The endpoint an operator selects to run the steward on the managed executor,
+# and the endpoint the channel runs when nothing is selected. Neither name is
+# "default" beyond that: the shipped default is stated once, in the channel
+# resolution in ``loopx.chat_manager``, so a reader cannot mistake the pair for
+# two competing defaults. The vocabulary lives here, beside the namespace that
+# validates it, so reading machine configuration never loads the chat layer;
+# ``loopx.chat_manager`` re-exports every name unchanged.
+MANAGER_ENDPOINT_MANAGED = MANAGED_TURN_HOST
+MANAGER_ENDPOINT_INDIVIDUAL = "codex"
+# Executor kinds name where this channel's model work is billed and bounded
+# rather than which adapter is launched, and they use the same vocabulary as the
+# governed Turn surface: an individual executor runs on one person's own CLI
+# login, a managed executor on an operator-supplied credential. Which kind an
+# endpoint is decides whether an operator credential belongs to it at all; the
+# mere presence of a credential decides nothing.
+MANAGER_EXECUTOR_KIND_INDIVIDUAL = "individual"
+MANAGER_EXECUTOR_KIND_MANAGED = EXECUTOR_KIND_MANAGED
+MANAGER_ENDPOINT_KINDS = {
+    MANAGER_ENDPOINT_INDIVIDUAL: MANAGER_EXECUTOR_KIND_INDIVIDUAL,
+    # The managed host is billed to the operator's own endpoint, not to one
+    # person's CLI login, so it is reached only by selecting it.
+    MANAGER_ENDPOINT_MANAGED: MANAGER_EXECUTOR_KIND_MANAGED,
+}
+
 STEWARD_EXECUTOR_NAMESPACE = "steward_executor"
 STEWARD_EXECUTOR_SCHEMA_V0 = "steward_executor_machine_defaults_v0"
 STEWARD_EXECUTOR_SCHEMA = "steward_executor_machine_defaults_v1"
@@ -57,14 +85,11 @@
 def steward_executor_endpoints() -> frozenset[str]:
     """Return the executor ids this namespace accepts for the steward channel.
 
-    The chat layer owns the endpoint vocabulary, so this reads it from the same
-    owner that resolves the channel instead of restating a second list here.
-    The import stays local because the chat layer resolves *through* this
-    namespace, and a module-level import would close that loop.
+    The channel resolution and this namespace read one endpoint vocabulary,
+    defined above, instead of restating a second list here.
     """
 
-    manager = importlib.import_module("loopx.chat_manager")
-    return frozenset(manager.MANAGER_ENDPOINT_KINDS)
+    return frozenset(MANAGER_ENDPOINT_KINDS)
 
 
 def steward_reasoning_efforts() -> tuple[str, ...]:

loopx/chat_manager.py

@@ -19,10 +19,14 @@
     REASONING_EFFORTS,
     managed_execution_profile,
 )
-from .control_plane.turn_driver.host_binding import (
-    EXECUTOR_KIND_MANAGED,
-    MANAGED_TURN_HOST,
-    managed_executor_binding,
+from .control_plane.turn_driver.host_binding import managed_executor_binding
+# Every name stays importable from this module, which tests and smokes use.
+from .capabilities.steward_executor.machine_defaults import (  # noqa: F401
+    MANAGER_ENDPOINT_INDIVIDUAL,
+    MANAGER_ENDPOINT_KINDS,
+    MANAGER_ENDPOINT_MANAGED,
+    MANAGER_EXECUTOR_KIND_INDIVIDUAL,
+    MANAGER_EXECUTOR_KIND_MANAGED,
 )
 from .capabilities.manager_context.answer_contract import (
     classify_manager_answer_shape,
@@ -214,12 +218,11 @@ def is_manager_channel(value: Any) -> bool:
 # always says which executor it is running on and why.
 MANAGER_CHANNEL_BINDING_SCHEMA_VERSION = "manager_channel_binding_v0"
 MANAGER_ENDPOINT_ENV_VAR = "LOOPX_MANAGER_ENDPOINT"
-# The endpoint an operator selects to run the steward on the managed executor,
-# and the endpoint the channel runs when nothing is selected. Neither name is
-# "default" beyond that: the shipped default is stated once, in the resolution
-# below, so a reader cannot mistake the pair for two competing defaults.
-MANAGER_ENDPOINT_MANAGED = MANAGED_TURN_HOST
-MANAGER_ENDPOINT_INDIVIDUAL = "codex"
+# The endpoint and executor-kind vocabulary (`MANAGER_ENDPOINT_MANAGED`,
+# `MANAGER_ENDPOINT_INDIVIDUAL`, `MANAGER_EXECUTOR_KIND_*` and
+# `MANAGER_ENDPOINT_KINDS`) is imported above from the `steward_executor`
+# namespace that validates it, so reading machine configuration never loads
+# this module. The shipped default is still stated once, in the resolution below.
 MANAGER_ENDPOINT_SOURCE_PRODUCT_DEFAULT = "product_default"
 MANAGER_ENDPOINT_SOURCE_EXPLICIT_CONFIG = "explicit_config"
 # The operator's persistent machine choice. It outranks the service environment
@@ -233,20 +236,6 @@ def is_manager_channel(value: Any) -> bool:
 # steward has exactly one such decision, and it is not conditional on a
 # credential, which is why no credential branch appears beside it.
 MANAGER_ENDPOINT_DEFAULT_REASON_STEWARD_CHANNEL_DEFAULT = "steward_channel_default"
-# Executor kinds name where this channel's model work is billed and bounded
-# rather than which adapter is launched, and they use the same vocabulary as the
-# governed Turn surface: an individual executor runs on one person's own CLI
-# login, a managed executor on an operator-supplied credential. Which kind an
-# endpoint is decides whether an operator credential belongs to it at all; the
-# mere presence of a credential decides nothing.
-MANAGER_EXECUTOR_KIND_INDIVIDUAL = "individual"
-MANAGER_EXECUTOR_KIND_MANAGED = EXECUTOR_KIND_MANAGED
-MANAGER_ENDPOINT_KINDS = {
-    MANAGER_ENDPOINT_INDIVIDUAL: MANAGER_EXECUTOR_KIND_INDIVIDUAL,
-    # The managed host is billed to the operator's own endpoint, not to one
-    # person's CLI login, so it is reached only by selecting it.
-    MANAGER_ENDPOINT_MANAGED: MANAGER_EXECUTOR_KIND_MANAGED,
-}
 
 MANAGER_MODEL_ENV_VAR = "LOOPX_MANAGER_MODEL"
 MANAGER_MODEL_DEFAULT = "gpt-6.1-sol"

tests/test_cli_entrypoint.py

@@ -368,6 +368,57 @@ def test_turn_handler_import_defers_optional_workflows() -> None:
 	assert completed.returncode == 0, completed.stderr
 
 
+def test_steward_endpoint_vocabulary_is_read_without_the_chat_layer() -> None:
+	script = """
+import sys
+from loopx.capabilities.steward_executor import (
+    normalize_steward_executor_machine_defaults,
+    steward_executor_endpoints,
+)
+
+assert steward_executor_endpoints() == frozenset({"codex", "dsh"})
+normalize_steward_executor_machine_defaults(
+    {"schema_version": "steward_executor_machine_defaults_v0", "executor_endpoint": "dsh"}
+)
+for name in ("loopx.chat_manager", "loopx.chat_agent", "loopx.chat_store"):
+    assert name not in sys.modules, name
+
+import loopx.chat_manager as manager
+from loopx.capabilities.steward_executor import machine_defaults
+for name in (
+    "MANAGER_ENDPOINT_INDIVIDUAL", "MANAGER_ENDPOINT_MANAGED", "MANAGER_ENDPOINT_KINDS",
+    "MANAGER_EXECUTOR_KIND_INDIVIDUAL", "MANAGER_EXECUTOR_KIND_MANAGED",
+):
+    assert getattr(manager, name) is getattr(machine_defaults, name), name
+assert manager.MANAGER_ENDPOINT_KINDS == {"codex": "individual", "dsh": "managed"}
+"""
+	completed = run_isolated_script(script)
+	assert completed.returncode == 0, completed.stderr
+
+
+@pytest.mark.parametrize("command", [
+	["heartbeat-prompt", "--thin", "--codex-app", "--goal-id", "perf-goal", "--agent-id", "perf-agent",
+		"--agent-scope", "fixture", "--available-capability", "shell"],
+	["quota", "should-run", "--goal-id", "perf-goal", "--agent-id", "perf-agent",
+		"--available-capability", "shell", "--codex-app", "--turn-instance-id", "2026-10-31T00:00:00Z"],
+	["status", "--goal-id", "perf-goal"],
+	["refresh-state", "--goal-id", "perf-goal", "--agent-id", "perf-agent",
+		"--next-action", "Continue bounded fixture work.", "--no-global-sync", "--dry-run"],
+], ids=["heartbeat-prompt", "should-run", "status", "refresh-state"])
+def test_heartbeat_chain_routes_do_not_load_the_chat_layer(tmp_path, monkeypatch, command):
+	from tests.control_plane.canonical_authority_fixture import initialize_canonical_source_fixture, isolate_sqlite_runtime
+
+	isolate_sqlite_runtime(tmp_path, monkeypatch)
+	registry, runtime = write_command_fixture(tmp_path)
+	state = registry.parent.parent / ".codex/goals/perf-goal/ACTIVE_GOAL_STATE.md"
+	initialize_canonical_source_fixture(runtime, "perf-goal", state_path=state, provider="sqlite")
+	argv = ["--registry", str(registry), "--runtime-root", str(runtime), "--format", "json", *command]
+	result = run_cli_main("loopx.entrypoint", argv,
+		forbidden_modules=("loopx.cli", "loopx.chat_manager", "loopx.chat_agent", "loopx.chat_store"))
+	assert result.returncode == 0, result.stderr
+	assert json.loads(result.stdout)["ok"] is True
+
+
 def test_turn_facade_preserves_every_export_owner_identity() -> None:
 	script = """
 from importlib import import_module