fix(dsh): inject managed machine credential into Harness

loopx/cli_commands/turn.py

@@ -186,13 +186,14 @@ def handle_turn_command(
         # whether that host can launch here, so a caller never has to infer it
         # from the host id. The explicit runner hook is the one launchability
         # fact only this command layer knows.
+        operator_environ = operator_provider_environ(runtime_root)
         payload["managed_executor"] = managed_executor_binding(
             args.host,
             # The credential a managed Turn authenticates with is this
             # machine's resolved pair, not whatever the invoking shell happens
             # to export: the readback above the launch and the launch itself
             # have to name the same credential.
-            environ=operator_provider_environ(runtime_root),
+            environ=operator_environ,
             dsh_runner_configured=bool(getattr(args, "dsh_runner", None)),
             provider=getattr(args, "dsh_provider", None),
             model=getattr(args, "dsh_model", None),
@@ -988,6 +989,7 @@ def resolve_built_in_session_binding(
                 host_runner = build_dsh_host_runner(
                     args,
                     workspace=project,
+                    environ=operator_environ,
                 )
 
             def post_settlement_reward_memory(

loopx/cli_commands/turn_dsh_host.py

@@ -6,6 +6,10 @@
 from typing import Any
 
 from ..dsh_goal_mode.turn_host_adapter import DshHostConfig, run_dsh_host
+from ..control_plane.operator_credential import (
+    OPERATOR_CREDENTIAL_ENV_VARS,
+    OPERATOR_ENDPOINT_ENV_VAR,
+)
 
 
 DshHostRunner = Callable[[Mapping[str, Any]], dict[str, Any]]
@@ -15,10 +19,20 @@ def build_dsh_host_runner(
     args: argparse.Namespace,
     *,
     workspace: Path,
+    environ: Mapping[str, str],
 ) -> DshHostRunner:
     """Bind CLI-owned DSH options to the in-process Turn host adapter."""
+    # The Turn planner and this host launch receive the same already-resolved
+    # environment. Forward only the operator provider pair: unrelated service
+    # variables are not part of the child host's credential contract.
+    credential = {
+        name: str(environ[name])
+        for name in (*OPERATOR_CREDENTIAL_ENV_VARS, OPERATOR_ENDPOINT_ENV_VAR)
+        if environ and environ.get(name)
+    }
     config = DshHostConfig(
         workspace=workspace,
+        env=credential,
         **{
             key: value
             for key, value in {

loopx/dsh_goal_mode/turn_host_adapter.py

@@ -380,6 +380,9 @@ class DshHostConfig:
     runtime_bin: str | None = None
     request_timeout_seconds: float | None = None
     dsh_runner: Path | None = None
+    # Only the resolved operator provider pair belongs in the child runtime;
+    # the caller deliberately excludes unrelated service environment values.
+    env: Mapping[str, str] | None = None
 
     def resolved_profile(self) -> dict[str, Any]:
         """Return the profile fields this attempt would use, with their source."""
@@ -458,23 +461,24 @@ def _execute_turn_host_request(
             if config.dsh_runner is not None
             else run_dsh_turn
         )
-        outcome = normalize_runner_outcome(
-            runner(
-                prompt=prompt,
-                session_id=session_id,
-                workspace=workspace,
-                # Preserve the established runner keyword while mapping the
-                # path to the current SDK's explicit dsh_home field.
-                session_root=dsh_home,
-                provider=str(profile["provider"]),
-                model=str(profile["model"]),
-                reasoning_effort=str(profile["reasoning_effort"]),
-                max_tokens=output_token_budget["max_tokens"],
-                cordis=config.cordis,
-                runtime_bin=config.runtime_bin,
-                request_timeout_seconds=config.request_timeout_seconds,
-            )
-        )
+        runner_arguments: dict[str, Any] = {
+            "prompt": prompt,
+            "session_id": session_id,
+            "workspace": workspace,
+            # Preserve the established runner keyword while mapping the path
+            # to the current SDK's explicit dsh_home field.
+            "session_root": dsh_home,
+            "provider": str(profile["provider"]),
+            "model": str(profile["model"]),
+            "reasoning_effort": str(profile["reasoning_effort"]),
+            "max_tokens": output_token_budget["max_tokens"],
+            "cordis": config.cordis,
+            "runtime_bin": config.runtime_bin,
+            "request_timeout_seconds": config.request_timeout_seconds,
+        }
+        if config.env is not None and config.dsh_runner is None:
+            runner_arguments["env"] = dict(config.env)
+        outcome = normalize_runner_outcome(runner(**runner_arguments))
     except DshHostResultError as exc:
         raise BuiltInHostError(
             "dsh_host_result_rejected",

tests/test_dsh_goal_mode.py

@@ -11,6 +11,7 @@
 import pytest
 
 from loopx import dsh_goal_mode
+from loopx.cli_commands import turn_dsh_host
 from loopx.control_plane.quota.turn_envelope import (
     turn_envelope_action_signature_document,
 )
@@ -196,6 +197,78 @@ def run_fake_dsh_turn(**kwargs: object) -> str:
     assert session_ids[0] == session_ids[2]
 
 
+def test_dsh_host_forwards_the_resolved_credential_to_the_runtime(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    calls: list[dict[str, object]] = []
+
+    def run_fake_dsh_turn(**kwargs: object) -> str:
+        calls.append(kwargs)
+        return '{"result_kind":"wait"}'
+
+    monkeypatch.setattr(turn_host_adapter, "run_dsh_turn", run_fake_dsh_turn)
+    credential = {
+        "DEEPSEEK_API_KEY": "fixture-machine-key",
+        "DEEPSEEK_BASE_URL": "https://provider.invalid",
+    }
+    config = turn_host_adapter.DshHostConfig(
+        workspace=tmp_path,
+        env=credential,
+    )
+
+    turn_host_adapter.run_dsh_host(_signed_request(), config=config)
+
+    assert calls[0]["env"] == credential
+    assert calls[0]["env"] is not credential
+
+
+@pytest.mark.parametrize(
+    "resolved,expected",
+    [
+        ({"DEEPSEEK_API_KEY": "fixture-machine-key"},
+         {"DEEPSEEK_API_KEY": "fixture-machine-key"}),
+        ({"DEEPSEEK_API_KEY": "fixture-env-key",
+          "DEEPSEEK_BASE_URL": "https://provider.invalid",
+          "UNRELATED_SERVICE_SECRET": "must-not-travel"},
+         {"DEEPSEEK_API_KEY": "fixture-env-key",
+          "DEEPSEEK_BASE_URL": "https://provider.invalid"}),
+        ({"UNRELATED_SERVICE_SECRET": "must-not-travel"}, {}),
+    ],
+)
+def test_turn_runner_projects_only_the_resolved_operator_provider_pair(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    resolved: dict[str, str],
+    expected: dict[str, str],
+) -> None:
+    captured: list[turn_host_adapter.DshHostConfig] = []
+
+    def capture(_request: object, *, config: turn_host_adapter.DshHostConfig) -> dict:
+        captured.append(config)
+        return {"ok": True}
+
+    monkeypatch.setattr(turn_dsh_host, "run_dsh_host", capture)
+    args = SimpleNamespace(
+        dsh_provider=None,
+        dsh_model=None,
+        dsh_reasoning_effort=None,
+        dsh_max_tokens=16_384,
+        dsh_home=None,
+        dsh_cordis=None,
+        dsh_runtime_bin=None,
+        timeout_seconds=60,
+        dsh_runner=None,
+    )
+    runner = turn_dsh_host.build_dsh_host_runner(
+        args,
+        workspace=tmp_path,
+        environ=resolved,
+    )
+
+    assert runner({}) == {"ok": True}
+    assert dict(captured[0].env or {}) == expected
+
+
 def test_dsh_goal_mode_is_a_first_class_subpackage() -> None:
     # The subpackage owns the adapter constants and API surface, following the
     # pi/opencode/kunluncode goal-mode packaging pattern.