fix(turn-driver): say what the managed runtime verdict is a claim about

docs/integrations/deepseek-harness-connector.md

@@ -197,6 +197,16 @@ line, `deepseek-v4-flash@high` in the shipped shape, with the provider prepended
 only when it is not the shipped one -- it is one line because every plan carries
 it, and the agent-facing output budget is a contract.
 
+`runtime_probe` states what the launchability verdict is a claim about: the
+module it probed (`deepseek_harness`) and `scope: "probing_interpreter"`. The dsh
+SDK is an optional dependency of the interpreter that answered, so one machine
+can hold a service environment where it resolves and a checkout environment
+where it does not, and the same readback then differs between them. Run
+`loopx doctor` to see which interpreter answered (`python.executable`) before
+provisioning a runtime the machine may already have; the probe never reports an
+absolute path itself, because this readback is carried into the Turn execution
+payload.
+
 `run-once --execute` fails closed on that verdict: status `unavailable`, no host
 invocation, no journal write, and no quota spend, with
 `dsh_runtime_unavailable`, `operator_credential_unconfigured`, or

examples/loopx-turn-managed-executor-binding-smoke.py

@@ -27,20 +27,24 @@
 from loopx.cli import main as cli_main  # noqa: E402
 from loopx.control_plane.turn_driver import executor as turn_executor  # noqa: E402
 from loopx.control_plane.turn_driver.host_binding import (  # noqa: E402
+    DSH_RUNTIME_MODULE,
     DSH_RUNTIME_UNAVAILABLE,
     EXECUTOR_KIND_INDIVIDUAL,
     EXECUTOR_KIND_MANAGED,
+    MANAGED_RUNTIME_PROBE_SCHEMA_VERSION,
     OPERATOR_CREDENTIAL_UNCONFIGURED,
     REMEDY_CONFIGURE_DSH_RUNTIME,
     REMEDY_CONFIGURE_OPERATOR_CREDENTIAL,
     REMEDY_SELECT_INDIVIDUAL_HOST,
+    RUNTIME_PROBE_SCOPE_INTERPRETER,
 )
 
 
 GOAL_ID = "loopx-turn-managed-executor-fixture"
 AGENT_ID = "codex-managed-executor-fixture"
 TODO_ID = "todo_managedexec01"
 CREDENTIAL_ENV = "DEEPSEEK_API_KEY"
+CREDENTIAL_VALUE = "sk-fixture-operator"
 RUNTIME_MODULE = "deepseek_harness"
 
 
@@ -239,6 +243,29 @@ def _managed_binding(payload: dict[str, Any]) -> dict[str, Any]:
     return binding
 
 
+def _expect_probe(binding: dict[str, Any], *, available: bool) -> None:
+    """Pin the scope of the launchability verdict at the CLI boundary.
+
+    ``managed_executor_binding`` answers from the interpreter that probes, and
+    one machine can hold a service environment where the dsh SDK resolves and a
+    checkout environment where it does not. The verdict is only actionable when
+    the readback says which environment answered, so the public payload this
+    smoke reads has to carry it -- for the plan, for the fail-closed refusal,
+    and for every executor kind, so no reader branches on the field's absence.
+    """
+
+    probe = binding["runtime_probe"]
+    assert probe["schema_version"] == MANAGED_RUNTIME_PROBE_SCHEMA_VERSION, probe
+    assert probe["module"] == DSH_RUNTIME_MODULE, probe
+    assert probe["scope"] == RUNTIME_PROBE_SCOPE_INTERPRETER, probe
+    assert probe["available"] is available, probe
+    # This readback is carried into the Turn execution payload, so it stays
+    # public-safe: no absolute path and no credential value.
+    serialized = json.dumps(probe)
+    assert "/" not in serialized, serialized
+    assert CREDENTIAL_VALUE not in serialized, serialized
+
+
 def main() -> int:
     with tempfile.TemporaryDirectory(
         prefix="loopx-turn-managed-executor-"
@@ -254,17 +281,17 @@ def main() -> int:
         assert exit_code == 0, payload
         assert payload["host"]["kind"] == "codex-cli", payload
         uncredentialed_default = payload["managed_executor"]
-        assert (
-            uncredentialed_default["executor_kind"] == EXECUTOR_KIND_INDIVIDUAL
-        ), uncredentialed_default
+        assert uncredentialed_default["executor_kind"] == EXECUTOR_KIND_INDIVIDUAL, (
+            uncredentialed_default
+        )
         assert uncredentialed_default["operator_credential_bound"] is False, (
             uncredentialed_default
         )
         assert uncredentialed_default["available"] is None, uncredentialed_default
 
         # 2. The credential resolves and authenticates the managed default.
         with (
-            _operator_credential("sk-fixture-operator"),
+            _operator_credential(CREDENTIAL_VALUE),
             _harness_runtime(available=True),
         ):
             exit_code, payload = _run_cli(_plan_command(registry, runtime, project))
@@ -274,23 +301,27 @@ def main() -> int:
         assert bound["credential_env"] == CREDENTIAL_ENV, bound
         assert bound["operator_credential_bound"] is True, bound
         assert bound["available"] is True, bound
+        _expect_probe(bound, available=True)
 
         # 3. With the runtime genuinely missing the same plan reports the other
-        #    typed reason rather than promising a launch.
+        #    typed reason rather than promising a launch, and the unchanged
+        #    verdict travels with the scope it was answered at.
         with (
-            _operator_credential("sk-fixture-operator"),
+            _operator_credential(CREDENTIAL_VALUE),
             _harness_runtime(available=False),
         ):
             exit_code, payload = _run_cli(_plan_command(registry, runtime, project))
         assert exit_code == 0, payload
         missing = _managed_binding(payload)
         assert missing["available"] is False, missing
         assert missing["unavailable_reason"] == DSH_RUNTIME_UNAVAILABLE, missing
+        _expect_probe(missing, available=False)
 
         # 4. An explicit individual host stays selected even while the operator
-        #    credential is configured, and makes no launch claim.
+        #    credential is configured, and makes no launch claim. It probes no
+        #    runtime, and the field is still present as an explicit ``None``.
         with (
-            _operator_credential("sk-fixture-operator"),
+            _operator_credential(CREDENTIAL_VALUE),
             _harness_runtime(available=True),
         ):
             exit_code, payload = _run_cli(
@@ -302,6 +333,7 @@ def main() -> int:
         assert individual["executor_kind"] == EXECUTOR_KIND_INDIVIDUAL, individual
         assert individual["available"] is None, individual
         assert individual["operator_credential_bound"] is False, individual
+        assert individual["runtime_probe"] is None, individual
 
         # 5. Executing an explicitly selected managed host without the
         #    credential fails closed: typed status, no host invocation, no
@@ -328,12 +360,16 @@ def main() -> int:
         ], refusal
         assert refusal["remediation_host"] == "codex-cli", refusal
         assert refusal["remediation_env_vars"] == [CREDENTIAL_ENV], refusal
+        # The operator reads the refusal, so the refusal is where the scope has
+        # to be visible: this environment answered "the credential is missing",
+        # not "the runtime is missing".
+        _expect_probe(refusal["managed_executor"], available=True)
         _expect_no_effects(refusal)
         _expect_no_journal(refusal, runtime)
 
         # 6. The same refusal covers a provably unlaunchable runtime.
         with (
-            _operator_credential("sk-fixture-operator"),
+            _operator_credential(CREDENTIAL_VALUE),
             _harness_runtime(available=False),
         ):
             exit_code, refusal = _run_cli(
@@ -353,6 +389,10 @@ def main() -> int:
             REMEDY_CONFIGURE_DSH_RUNTIME,
             REMEDY_SELECT_INDIVIDUAL_HOST,
         ], refusal
+        # The refusal that sends an operator to provision a runtime names the
+        # environment that could not import it, so the same readback cannot be
+        # taken for a machine-level fact.
+        _expect_probe(refusal["managed_executor"], available=False)
         _expect_no_effects(refusal)
         _expect_no_journal(refusal, runtime)
 

loopx/chat_manager.py

@@ -485,12 +485,15 @@ def manager_channel_binding(
     executor_kind = MANAGER_ENDPOINT_KINDS.get(endpoint, "")
     credential_env = ""
     execution_profile: str | None = None
+    runtime_probe: dict[str, Any] | None = None
     if executor_kind == MANAGER_EXECUTOR_KIND_MANAGED:
         managed = managed_executor_binding(endpoint, environ=environ)
         credential_env = str(managed.get("credential_env") or "")
         execution_profile = managed.get("execution_profile")
         available: bool | None = managed.get("available")
         unavailable_reason: str | None = managed.get("unavailable_reason")
+        if isinstance(managed.get("runtime_probe"), Mapping):
+            runtime_probe = dict(managed["runtime_probe"])
     else:
         available, unavailable_reason = None, None
     model, model_source = manager_model_resolution(
@@ -523,6 +526,11 @@ def manager_channel_binding(
         "execution_profile": execution_profile,
         "available": available,
         "unavailable_reason": unavailable_reason,
+        # Which environment answered the launchability verdict, quoted from the
+        # governed Turn surface's own readback. A surface that shows
+        # `dsh_runtime_unavailable` without this cannot tell an operator which
+        # of this machine's interpreters is missing the runtime.
+        "runtime_probe": runtime_probe,
         "model": model,
         "model_source": model_source,
         **manager_channel_session_mode_readback(session),

loopx/control_plane/turn_driver/host_binding.py

@@ -73,6 +73,13 @@
 # launchability fact this projection checks without side effects.
 DSH_RUNTIME_MODULE = "deepseek_harness"
 DSH_RUNTIME_UNAVAILABLE = "dsh_runtime_unavailable"
+# The launchability verdict is answered by the interpreter that probes, and two
+# environments on one machine disagree: a checkout venv without the SDK answers
+# "unavailable" for a machine whose service venv has it. The readback says which
+# environment answered, so an operator can tell a machine-level gap from a
+# process-level one instead of re-provisioning a runtime that is already there.
+MANAGED_RUNTIME_PROBE_SCHEMA_VERSION = "managed_runtime_probe_v0"
+RUNTIME_PROBE_SCOPE_INTERPRETER = "probing_interpreter"
 # A managed host is billed to the operator's own endpoint. Without the operator
 # credential (or an explicit injected runner) LoopX cannot authenticate that
 # endpoint, so it refuses instead of letting the managed default consume
@@ -133,6 +140,26 @@ def dsh_runtime_importable(
         return False
 
 
+def managed_runtime_probe(*, runtime_available: bool) -> dict[str, Any]:
+    """State what the runtime verdict is a claim about, and what it probed.
+
+    ``scope`` is the claim's limit: the answer describes the interpreter that
+    probed, not the machine, and two environments on one machine can answer
+    differently. The probing interpreter itself is deliberately not reported
+    here -- this readback is carried into the Turn execution payload, and the
+    dsh adapter's boundary forbids publishing a local absolute path into LoopX
+    state. An operator compares environments through ``doctor``'s
+    ``python.executable`` instead.
+    """
+
+    return {
+        "schema_version": MANAGED_RUNTIME_PROBE_SCHEMA_VERSION,
+        "module": DSH_RUNTIME_MODULE,
+        "scope": RUNTIME_PROBE_SCOPE_INTERPRETER,
+        "available": bool(runtime_available),
+    }
+
+
 def _managed_unavailable_remediation(reason: str | None) -> list[str]:
     """Name the operator-reachable exits from one managed refusal.
 
@@ -175,6 +202,10 @@ def managed_executor_binding(
     with it, the provider claims to authenticate. It is ``None`` for every
     non-managed executor because neither the profile nor the credential belongs
     to an individual or generic host.
+
+    ``runtime_probe`` states what the ``dsh_runtime_unavailable`` verdict is a
+    claim about, so a reader does not take a process-level answer for a
+    machine-level fact.
     """
 
     if host == MANAGED_HOST:
@@ -212,6 +243,9 @@ def managed_executor_binding(
             "unavailable_remediation": _managed_unavailable_remediation(
                 unavailable_reason
             ),
+            "runtime_probe": managed_runtime_probe(
+                runtime_available=runtime_available,
+            ),
         }
     return {
         "schema_version": MANAGED_EXECUTOR_BINDING_SCHEMA_VERSION,
@@ -228,6 +262,9 @@ def managed_executor_binding(
         "available": None,
         "unavailable_reason": None,
         "unavailable_remediation": [],
+        # The same field exists for every executor kind so a reader never
+        # branches on its presence; only a managed executor probes a runtime.
+        "runtime_probe": None,
     }
 
 

tests/test_manager_channel_binding.py

@@ -45,6 +45,9 @@
 from loopx.chat_server import ChatHTTPServer, ChatRequestHandler
 from loopx.chat_store import ChatSessionStore
 from loopx.control_plane.turn_driver import host_binding
+from loopx.control_plane.turn_driver.host_binding import (
+    RUNTIME_PROBE_SCOPE_INTERPRETER,
+)
 from loopx.extensions.lark.cli_resolution import LarkCliResolution
 
 
@@ -651,6 +654,10 @@ def test_a_channel_without_a_session_reads_as_unbound(monkeypatch):
         MANAGER_CHANNEL_SESSION_MODE_SOURCE_UNBOUND
     )
     assert binding["session_status"] is None
+    # The channel quotes the governed Turn surface's probe scope, so a surface
+    # showing `dsh_runtime_unavailable` can say which environment answered.
+    assert binding["runtime_probe"]["scope"] == RUNTIME_PROBE_SCOPE_INTERPRETER
+    assert binding["runtime_probe"]["available"] is True
 
 
 def test_an_unrecognized_session_mode_is_named_rather_than_coerced():

tests/test_turn_managed_executor_binding.py

@@ -2,21 +2,26 @@
 
 from __future__ import annotations
 
+import json
+
 import pytest
 
 from loopx.control_plane.turn_driver.host_binding import (
+    DSH_RUNTIME_MODULE,
     DSH_RUNTIME_UNAVAILABLE,
     EXECUTOR_KIND_GENERIC,
     EXECUTOR_KIND_INDIVIDUAL,
     EXECUTOR_KIND_MANAGED,
     INDIVIDUAL_TURN_HOST,
     MANAGED_EXECUTOR_BINDING_SCHEMA_VERSION,
+    MANAGED_RUNTIME_PROBE_SCHEMA_VERSION,
     MANAGED_TURN_HOST,
     OPERATOR_CREDENTIAL_UNCONFIGURED,
     REMEDY_CONFIGURE_DSH_RUNTIME,
     REMEDY_CONFIGURE_OPERATOR_CREDENTIAL,
     REMEDY_CORRECT_EXECUTION_PROFILE,
     REMEDY_SELECT_INDIVIDUAL_HOST,
+    RUNTIME_PROBE_SCOPE_INTERPRETER,
     managed_executor_unavailable_payload,
     managed_executor_binding,
     resolve_default_turn_host,
@@ -59,9 +64,45 @@
         "available": True,
         "unavailable_reason": None,
         "unavailable_remediation": [],
+        "runtime_probe": {
+            "schema_version": MANAGED_RUNTIME_PROBE_SCHEMA_VERSION,
+            "module": DSH_RUNTIME_MODULE,
+            "scope": RUNTIME_PROBE_SCOPE_INTERPRETER,
+            "available": True,
+        },
     }
 
 
+def test_the_runtime_verdict_states_what_it_is_a_claim_about():
+    """A process-level answer must not read as a machine-level fact."""
+
+    present = managed_executor_binding(
+        "dsh",
+        environ={"DEEPSEEK_API_KEY": "sk-operator"},
+        module_probe=_RUNTIME,
+    )
+    absent = managed_executor_binding(
+        "dsh",
+        environ={"DEEPSEEK_API_KEY": "sk-operator"},
+        module_probe=_NO_RUNTIME,
+    )
+    individual = managed_executor_binding("codex-cli")
+
+    # The probe follows the same seam the verdict does, so a caller that
+    # injects one gets both facts from one answer.
+    assert present["runtime_probe"]["available"] is True
+    assert absent["runtime_probe"]["available"] is False
+    assert absent["runtime_probe"]["scope"] == RUNTIME_PROBE_SCOPE_INTERPRETER
+    assert absent["runtime_probe"]["module"] == DSH_RUNTIME_MODULE
+    # An individual executor probes no runtime, and the field is still present
+    # so no reader branches on its absence.
+    assert individual["runtime_probe"] is None
+    # This readback travels into the Turn execution payload, so it must stay
+    # public-safe: no absolute path, no credential value.
+    serialized = json.dumps(absent["runtime_probe"])
+    assert "/" not in serialized and "sk-operator" not in serialized, serialized
+
+
 def test_managed_executor_fails_closed_when_the_runtime_is_missing():
     binding = managed_executor_binding(
         "dsh",