feat(python-sdk): launch dsh profiles from explicit homes

Replace complete-config, session_root, runtime-bin, bridge-bin, and public argv override options with dsh_bin, profile, ordered patches, and dsh_home. Resolve executable/home/patch/cwd paths before spawn, select the sdk profile by default, and fail before launch unless dsh_home or non-empty DSH_HOME is explicit; Python never inherits ~/.dsh silently.

Remove Python-owned DSH_CORDIS_CONFIG, DSH_SESSION_ROOT, and DSH_CWD injection and drop session_root from RunResult. Keep arbitrary argv only as an underscore-prefixed fake-runtime adapter, retain provider/model/token and process controls, and append subprocess stderr to initialization JSON-RPC errors so profile boot failures name their actual plugin cause. Unit and carrier tests cover both exe and Node modes.
This commit is contained in:
Tianyi Cui
2026-08-24 17:28:26 +08:00
parent be7b064504
commit 56e038b2e3
5 changed files with 225 additions and 215 deletions
+19 -11
View File
@@ -1,4 +1,4 @@
"""Drive the repo-source JSON-RPC bin through the SDK and a keyless mock SSE server.
"""Drive the repo-source dsh SDK profile through the SDK and a keyless mock SSE server.
Requires ``pnpm install`` but no build. This manual test is not collected by
pytest; run ``python tests/manual_sdk_agent_smoke.py``.
@@ -16,7 +16,6 @@ from pathlib import Path
from typing import Any
from deepseek_harness import DeepSeekHarness
from deepseek_harness_runtime import bundled_default_config_path
class MockCompletionHandler(BaseHTTPRequestHandler):
@@ -43,15 +42,16 @@ class MockCompletionHandler(BaseHTTPRequestHandler):
def run_smoke(repo_root: Path, keep_sessions: bool) -> None:
session_root = Path(tempfile.mkdtemp(prefix="dsh-sdk-smoke-sessions-"))
runtime_entry = repo_root / "packages/sdk/python-runtime/src/packaged-bin.ts"
dsh_home = Path(tempfile.mkdtemp(prefix="dsh-sdk-smoke-home-"))
session_root = dsh_home / "sessions"
runtime_entry = repo_root / "apps/cli/src/bin.ts"
server = ThreadingHTTPServer(("127.0.0.1", 0), MockCompletionHandler)
thread = threading.Thread(target=server.serve_forever, name="mock-openai-compatible-server", daemon=True)
thread.start()
base_url = f"http://127.0.0.1:{server.server_address[1]}"
print(f"repo_root={repo_root}")
print(f"session_root={session_root}")
print(f"dsh_home={dsh_home}")
print(f"mock_base_url={base_url}")
try:
@@ -59,10 +59,18 @@ def run_smoke(repo_root: Path, keep_sessions: bool) -> None:
model="sdk-smoke-model",
cwd=str(repo_root / "python/sdk"),
runtime_cwd=str(repo_root),
session_root=str(session_root),
cordis=str(bundled_default_config_path()),
launch_args_override=("node", "--import", "tsx", str(runtime_entry)),
_launch_args=(
"node",
"--import",
"tsx",
str(runtime_entry),
"--profile",
"sdk",
),
env={
"DSH_HOME": str(dsh_home),
"DSH_PERMISSION_MODE": "danger-full-access",
"DSH_TELEMETRY_DISABLED": "1",
"DEEPSEEK_BASE_URL": base_url,
"DEEPSEEK_API_KEY": "sdk-smoke-key",
},
@@ -92,10 +100,10 @@ def run_smoke(repo_root: Path, keep_sessions: bool) -> None:
server.server_close()
if keep_sessions:
print(f"kept_session_root={session_root}")
print(f"kept_dsh_home={dsh_home}")
else:
shutil.rmtree(session_root)
print("removed temporary session root")
shutil.rmtree(dsh_home)
print("removed temporary dsh home")
def main() -> None:
+44 -86
View File
@@ -1,4 +1,4 @@
"""Keyless boot tests for the production exe and development node carrier.
"""Keyless boot tests for the production exe and development dsh carrier.
Each carrier skips independently when absent. The dummy API key only satisfies
adapter loading; initialize and shutdown do not call a model.
@@ -6,64 +6,39 @@ adapter loading; initialize and shutdown do not call a model.
from __future__ import annotations
import json
from pathlib import Path
import pytest
from deepseek_harness import DeepSeekHarness, HarnessClient, HarnessConfig
from deepseek_harness.errors import TransportClosedError
from deepseek_harness_runtime import resolve_bundled_launch_args
from deepseek_harness.errors import JsonRpcError, TransportClosedError
from deepseek_harness_runtime import RUNTIME_MODE_ENV_VAR, resolve_bundled_launch_args
_MODES = ("exe", "node")
_REPO_ROOT = Path(__file__).parents[3]
_MINIMAL_CONFIG = _REPO_ROOT / "examples" / "python-sdk-agent" / "minimal.cordis.yml"
# The config must include the JSON-RPC serving plugin.
_CORDIS_YML = """\
- id: sdk-jsonrpc-server
name: '@deepseek-ai/dsh-sdk-jsonrpc-server'
- id: agent-core
name: '@deepseek-ai/dsh-agent-spine-demo'
config:
workspaceContext: false
- id: sessions
name: '@deepseek-ai/dsh-session-persistence-jsonl'
config:
root: './sessions'
- id: session-checkpoints
name: '@deepseek-ai/dsh-session-checkpoint-policy'
- id: subprocess
name: '@deepseek-ai/dsh-subprocess-local'
- id: bash
name: '@deepseek-ai/dsh-bash-local'
config:
cwd: '.'
- id: todo
name: '@deepseek-ai/dsh-tool-todo'
config:
allowParallelInProgress: true
"""
def _launch_args(mode: str) -> tuple[str, ...]:
def _select_mode(mode: str, monkeypatch: pytest.MonkeyPatch) -> None:
try:
return resolve_bundled_launch_args(mode)
resolve_bundled_launch_args(mode)
except FileNotFoundError as exc:
pytest.skip(f"bundled {mode}-mode runtime unavailable on this machine: {exc}")
monkeypatch.setenv(RUNTIME_MODE_ENV_VAR, mode)
def _client(tmp_path: Path, launch_args: tuple[str, ...]) -> HarnessClient:
def _client(tmp_path: Path, mode: str, monkeypatch: pytest.MonkeyPatch, *patches: Path) -> HarnessClient:
_select_mode(mode, monkeypatch)
return HarnessClient(
HarnessConfig(
launch_args_override=launch_args,
dsh_home=str(tmp_path / "home"),
patches=tuple(str(patch) for patch in patches),
cwd=str(tmp_path),
env={
"DSH_CORDIS_CONFIG": "./cordis.yml",
"DSH_SESSION_ROOT": str(tmp_path / "sessions"),
"DSH_CWD": str(tmp_path),
# The lazily mounted adapter requires a key even without a model call.
"DEEPSEEK_API_KEY": "sk-dummy-for-boot",
"DEEPSEEK_BASE_URL": "http://127.0.0.1:9",
"DSH_PERMISSION_MODE": "danger-full-access",
"DSH_TELEMETRY_DISABLED": "1",
},
request_timeout_seconds=120,
)
@@ -71,34 +46,39 @@ def _client(tmp_path: Path, launch_args: tuple[str, ...]) -> HarnessClient:
@pytest.mark.parametrize("mode", _MODES)
def test_bundled_runtime_boots_a_cordis_config(tmp_path: Path, mode: str) -> None:
launch_args = _launch_args(mode)
(tmp_path / "cordis.yml").write_text(_CORDIS_YML)
with _client(tmp_path, launch_args) as client:
def test_bundled_runtime_boots_the_sdk_profile(
tmp_path: Path, mode: str, monkeypatch: pytest.MonkeyPatch
) -> None:
with _client(tmp_path, mode, monkeypatch) as client:
init = client.initialize(provider="deepseek-official", cwd=str(tmp_path), model="deepseek-v4-pro")
assert init.serverInfo is not None
assert init.serverInfo.name == "deepseek-harness-sdk-runtime"
profile = json.loads((tmp_path / "home" / "profiles" / "sdk" / "package.json").read_text())
assert profile["dsh"]["profile"]["bundles"] == [
"@deepseek-ai/dsh-base",
"@deepseek-ai/dsh-sdk-app",
]
@pytest.mark.parametrize("mode", _MODES)
def test_python_sdk_boots_minimal_jsonrpc_config(tmp_path: Path, mode: str) -> None:
launch_args = _launch_args(mode)
model = "minimal-environment-model"
def test_python_sdk_applies_an_ordered_profile_patch(
tmp_path: Path, mode: str, monkeypatch: pytest.MonkeyPatch
) -> None:
_select_mode(mode, monkeypatch)
patch = tmp_path / "persona.patch.yml"
patch.write_text(json.dumps([{
"id": "system-prompt",
"config": {"persona": "Python SDK ordered patch marker."},
}]))
harness = DeepSeekHarness(
model=model,
model="deepseek-v4-pro",
cwd=str(tmp_path),
session_root=str(tmp_path / "sessions"),
cordis=str(_MINIMAL_CONFIG),
env={
"DSH_MODEL": model,
"DSH_CONTEXT_WINDOW": "1000000",
"DSH_SYSTEM_PROMPT": "You are the Python SDK minimal boot test agent.",
},
dsh_home=str(tmp_path / "home"),
patches=(str(patch),),
env={"DSH_PERMISSION_MODE": "danger-full-access"},
api_key="sk-dummy-for-boot",
base_url="http://127.0.0.1:9",
launch_args_override=launch_args,
request_timeout_seconds=120,
)
@@ -107,42 +87,20 @@ def test_python_sdk_boots_minimal_jsonrpc_config(tmp_path: Path, mode: str) -> N
@pytest.mark.parametrize("mode", _MODES)
def test_bundled_runtime_surfaces_unbundled_plugin_failure(tmp_path: Path, mode: str) -> None:
launch_args = _launch_args(mode)
(tmp_path / "cordis.yml").write_text(
"- id: missing\n name: '@deepseek-ai/dsh-does-not-exist'\n"
)
def test_bundled_runtime_surfaces_unbundled_plugin_failure(
tmp_path: Path, mode: str, monkeypatch: pytest.MonkeyPatch
) -> None:
patch = tmp_path / "missing.patch.yml"
patch.write_text(json.dumps([{
"insert": [{"id": "missing", "name": "@deepseek-ai/dsh-does-not-exist"}],
}]))
client = _client(tmp_path, launch_args)
client = _client(tmp_path, mode, monkeypatch, patch)
client.start()
try:
with pytest.raises((TransportClosedError, TimeoutError)) as excinfo:
with pytest.raises((JsonRpcError, TransportClosedError, TimeoutError)) as excinfo:
client.initialize(provider="deepseek-official", cwd=str(tmp_path), model="deepseek-v4-pro")
finally:
client.close()
assert "@deepseek-ai/dsh-does-not-exist" in str(excinfo.value)
@pytest.mark.parametrize("mode", _MODES)
@pytest.mark.parametrize("ambient_config", [None, ""], ids=["unset", "empty-counts-as-absent"])
def test_zero_config_run_injects_bundled_default_cordis_config(
tmp_path: Path, mode: str, ambient_config: str | None, monkeypatch: pytest.MonkeyPatch
) -> None:
_launch_args(mode) # skip early when this carrier is unavailable
monkeypatch.setenv("DSH_RUNTIME_MODE", mode)
if ambient_config is None:
monkeypatch.delenv("DSH_CORDIS_CONFIG", raising=False)
else:
monkeypatch.setenv("DSH_CORDIS_CONFIG", ambient_config)
harness = DeepSeekHarness(
model="deepseek-v4-pro",
cwd=str(tmp_path),
session_root=str(tmp_path / "sessions"),
api_key="sk-dummy-for-boot",
base_url="http://127.0.0.1:9",
request_timeout_seconds=120,
)
with harness:
pass
+98 -68
View File
@@ -9,7 +9,8 @@ from pathlib import Path
import pytest
from deepseek_harness import DeepSeekHarness, HarnessClient, HarnessConfig, Notification, SdkProtocolError
from deepseek_harness import DeepSeekHarness, HarnessClient, HarnessConfig, Notification, RunResult, SdkProtocolError
from deepseek_harness.errors import JsonRpcError
def test_high_level_sdk_runs_turn_and_collects_final_response(tmp_path: Path) -> None:
@@ -95,9 +96,7 @@ for line in sys.stdin:
model="deepseek-v4-flash",
max_tokens=4096,
cwd=str(tmp_path),
cordis=str(tmp_path / "cordis.yml"),
session_root=str(tmp_path / "sessions"),
launch_args_override=(sys.executable, str(script)),
_launch_args=(sys.executable, str(script)),
env={
"ENV_DUMP": str(env_dump),
"INIT_DUMP": str(init_dump),
@@ -113,9 +112,9 @@ for line in sys.stdin:
dumped_env = json.loads(env_dump.read_text())
assert dumped_env["DEEPSEEK_API_KEY"] == "env-key"
assert dumped_env["DEEPSEEK_BASE_URL"] == "http://127.0.0.1:4321"
assert dumped_env["DSH_CWD"] == str(tmp_path)
assert dumped_env["DSH_SESSION_ROOT"] == str(tmp_path / "sessions")
assert dumped_env["DSH_CORDIS_CONFIG"] == str(tmp_path / "cordis.yml")
assert dumped_env["DSH_CWD"] is None
assert dumped_env["DSH_SESSION_ROOT"] is None
assert dumped_env["DSH_CORDIS_CONFIG"] is None
assert json.loads(init_dump.read_text()) == {
"cwd": str(tmp_path),
"provider": "deepseek-official",
@@ -150,7 +149,7 @@ for line in sys.stdin:
seen: list[str] = []
with DeepSeekHarness(
launch_args_override=(sys.executable, str(script)),
_launch_args=(sys.executable, str(script)),
cwd=str(tmp_path),
) as harness:
session = harness.start_session("main")
@@ -188,7 +187,7 @@ for line in sys.stdin:
)
with DeepSeekHarness(
launch_args_override=(sys.executable, str(script)),
_launch_args=(sys.executable, str(script)),
cwd=str(tmp_path),
) as harness:
with pytest.raises(
@@ -224,7 +223,7 @@ for line in sys.stdin:
with DeepSeekHarness(
cwd=".",
runtime_cwd=".",
launch_args_override=(sys.executable, str(script)),
_launch_args=(sys.executable, str(script)),
env={"CAPTURE": str(capture)},
):
pass
@@ -232,7 +231,7 @@ for line in sys.stdin:
expected = str(tmp_path.resolve())
assert json.loads(capture.read_text()) == {
"process": expected,
"environment": expected,
"environment": None,
"wire": expected,
}
@@ -263,7 +262,7 @@ for line in sys.stdin:
)
with DeepSeekHarness(
launch_args_override=(sys.executable, str(script)),
_launch_args=(sys.executable, str(script)),
cwd=str(tmp_path),
) as harness:
result = harness.run("spawn a helper", session_id="main")
@@ -312,7 +311,7 @@ for line in sys.stdin:
seen: list[str] = []
with DeepSeekHarness(
launch_args_override=(sys.executable, str(script)),
_launch_args=(sys.executable, str(script)),
cwd=str(tmp_path),
) as harness:
result = harness.run(
@@ -367,7 +366,7 @@ for line in sys.stdin:
)
with DeepSeekHarness(
launch_args_override=(sys.executable, str(script)),
_launch_args=(sys.executable, str(script)),
cwd=str(tmp_path),
) as harness:
result = harness.run("stay in your lane", session_id="main")
@@ -401,7 +400,7 @@ for line in sys.stdin:
""".strip()
)
with DeepSeekHarness(launch_args_override=(sys.executable, str(script)), cwd=str(tmp_path)) as harness:
with DeepSeekHarness(_launch_args=(sys.executable, str(script)), cwd=str(tmp_path)) as harness:
result = harness.run("one turn", session_id="main")
assert harness.client._notifications.qsize() == 0
@@ -441,7 +440,7 @@ for line in sys.stdin:
""".strip()
)
with DeepSeekHarness(launch_args_override=(sys.executable, str(script)), cwd=str(tmp_path)) as harness:
with DeepSeekHarness(_launch_args=(sys.executable, str(script)), cwd=str(tmp_path)) as harness:
first = harness.run("first turn", session_id="main")
second = harness.run("second turn", session_id="main")
@@ -473,7 +472,7 @@ for line in sys.stdin:
)
with HarnessClient(
HarnessConfig(launch_args_override=(sys.executable, str(script)))
HarnessConfig(_launch_args=(sys.executable, str(script)))
) as client:
init = client.initialize(provider="deepseek-official", cwd="/workspace", model="dsagent")
assert init.serverInfo.name == "fake-dsh"
@@ -612,7 +611,7 @@ for line in sys.stdin:
def broken_filter(_notification: object) -> bool:
raise RuntimeError("bad notification filter")
with HarnessClient(HarnessConfig(launch_args_override=(sys.executable, str(script)))) as client:
with HarnessClient(HarnessConfig(_launch_args=(sys.executable, str(script)))) as client:
client.initialize(provider="deepseek-official", cwd="/workspace", model="dsagent")
with (
client.subscribe_notifications(broken_filter) as broken,
@@ -649,7 +648,7 @@ for line in sys.stdin:
""".strip()
)
with HarnessClient(HarnessConfig(launch_args_override=(sys.executable, str(script)))) as client:
with HarnessClient(HarnessConfig(_launch_args=(sys.executable, str(script)))) as client:
client.initialize(provider="deepseek-official", cwd="/workspace", model="dsagent")
with pytest.raises(ValueError):
client.session_prompt("main", [{"type": "text", "text": "fix it"}])
@@ -677,7 +676,7 @@ for line in sys.stdin:
)
with HarnessClient(
HarnessConfig(launch_args_override=(sys.executable, str(script)))
HarnessConfig(_launch_args=(sys.executable, str(script)))
) as client:
client.initialize(provider="deepseek-official", cwd="/workspace", model="dsagent")
@@ -711,7 +710,7 @@ for line in sys.stdin:
)
with HarnessClient(
HarnessConfig(launch_args_override=(sys.executable, str(script)))
HarnessConfig(_launch_args=(sys.executable, str(script)))
) as client:
init = client.initialize(provider="deepseek-official", cwd="/workspace", model="dsagent")
assert init.serverInfo.name == "fake-dsh"
@@ -731,7 +730,7 @@ time.sleep(60)
with HarnessClient(
HarnessConfig(
launch_args_override=(sys.executable, str(script)),
_launch_args=(sys.executable, str(script)),
request_timeout_seconds=0.1,
)
) as client:
@@ -767,7 +766,7 @@ for line in sys.stdin:
client = HarnessClient(
HarnessConfig(
launch_args_override=(sys.executable, str(script)),
_launch_args=(sys.executable, str(script)),
shutdown_timeout_seconds=0.1,
)
)
@@ -792,6 +791,7 @@ import sys
for line in sys.stdin:
msg = json.loads(line)
if msg.get("method") == "initialize":
print("initialize diagnostic", file=sys.stderr, flush=True)
print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "error": {"code": -32000, "message": "bad initialize"}}), flush=True)
elif msg.get("method") == "shutdown":
print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {}}), flush=True)
@@ -799,14 +799,16 @@ for line in sys.stdin:
""".strip()
)
client = HarnessClient(HarnessConfig(launch_args_override=(sys.executable, str(script))))
client = HarnessClient(HarnessConfig(_launch_args=(sys.executable, str(script))))
client.start()
proc = client._proc
assert proc is not None
with pytest.raises(Exception, match="bad initialize"):
with pytest.raises(JsonRpcError, match="bad initialize") as excinfo:
client.initialize(provider="deepseek-official", cwd=".", model="dsagent")
assert excinfo.value.code == -32000
assert "initialize diagnostic" in str(excinfo.value)
assert proc.wait(timeout=1) is not None
assert client._proc is None
@@ -824,6 +826,16 @@ def test_public_signatures_omit_unsupported_wire_parameters() -> None:
assert "max_tokens" in inspect.signature(HarnessClient.initialize).parameters
assert "client_name" not in HarnessConfig.__dataclass_fields__
assert "client_version" not in HarnessConfig.__dataclass_fields__
assert {"dsh_bin", "profile", "patches", "dsh_home"} <= set(
DeepSeekHarnessConfig.__dataclass_fields__
)
assert {"dsh_bin", "profile", "patches", "dsh_home"} <= set(
HarnessConfig.__dataclass_fields__
)
for removed in ("cordis", "session_root", "runtime_bin", "bridge_bin", "launch_args_override"):
assert removed not in DeepSeekHarnessConfig.__dataclass_fields__
assert removed not in HarnessConfig.__dataclass_fields__
assert "session_root" not in RunResult.__dataclass_fields__
def test_client_close_is_idempotent_before_and_after_start(tmp_path: Path) -> None:
@@ -845,7 +857,7 @@ for line in sys.stdin:
""".strip()
)
client = HarnessClient(HarnessConfig(launch_args_override=(sys.executable, str(script))))
client = HarnessClient(HarnessConfig(_launch_args=(sys.executable, str(script))))
client.start()
client.initialize(provider="deepseek-official", cwd="/workspace", model="dsagent")
client.close()
@@ -865,7 +877,7 @@ sys.exit(42)
with HarnessClient(
HarnessConfig(
launch_args_override=(sys.executable, str(script)),
_launch_args=(sys.executable, str(script)),
request_timeout_seconds=2,
)
) as client:
@@ -897,7 +909,7 @@ with open(os.environ["SEEN"], "w") as seen:
with HarnessClient(
HarnessConfig(
launch_args_override=(sys.executable, str(script)),
_launch_args=(sys.executable, str(script)),
env={"SEEN": str(output)},
)
) as client:
@@ -915,21 +927,22 @@ with open(os.environ["SEEN"], "w") as seen:
json.loads(line)
def _install_fake_bundled_runtime(
def _install_fake_bundled_dsh(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> Path:
"""Install a fake runtime package that records config and serves lifecycle calls.
Returns the fake bundled default config path.
"""
runtime = tmp_path / "dsh-jsonrpc-agent"
) -> None:
"""Install a fake runtime package that records dsh argv and serves lifecycle calls."""
runtime = tmp_path / "dsh.py"
runtime.write_text(
"""#!/usr/bin/env python3
"""
import json
import os
import sys
json.dump({"DSH_CORDIS_CONFIG": os.environ.get("DSH_CORDIS_CONFIG")}, open(os.environ["ENV_DUMP"], "w"))
json.dump({
"argv": sys.argv[1:],
"DSH_HOME": os.environ.get("DSH_HOME"),
"DSH_CORDIS_CONFIG": os.environ.get("DSH_CORDIS_CONFIG"),
}, open(os.environ["ENV_DUMP"], "w"))
for line in sys.stdin:
msg = json.loads(line)
if msg.get("method") == "initialize":
@@ -939,58 +952,75 @@ for line in sys.stdin:
break
""".strip()
)
runtime.chmod(0o755)
default_config = tmp_path / "default-cordis.yml"
module_dir = tmp_path / "deepseek_harness_runtime"
module_dir.mkdir()
(module_dir / "__init__.py").write_text(
f"""
def resolve_bundled_launch_args(mode=None):
return ({str(runtime)!r},)
def bundled_default_config_path():
return {str(default_config)!r}
return ({sys.executable!r}, {str(runtime)!r})
""".strip()
)
monkeypatch.syspath_prepend(str(tmp_path))
monkeypatch.delitem(sys.modules, "deepseek_harness_runtime", raising=False)
return default_config
@pytest.mark.parametrize("ambient_config", [None, ""], ids=["unset", "empty-counts-as-absent"])
def test_client_default_launch_uses_bundled_runtime_and_injects_default_config(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ambient_config: str | None
) -> None:
env_dump = tmp_path / "env.json"
default_config = _install_fake_bundled_runtime(tmp_path, monkeypatch)
if ambient_config is None:
monkeypatch.delenv("DSH_CORDIS_CONFIG", raising=False)
else:
monkeypatch.setenv("DSH_CORDIS_CONFIG", ambient_config)
with HarnessClient(HarnessConfig(env={"ENV_DUMP": str(env_dump)})) as client:
init = client.initialize(provider="deepseek-official", cwd="/workspace", model="deepseek-v4-pro")
assert init.serverInfo.name == "bundled-runtime"
assert json.loads(env_dump.read_text())["DSH_CORDIS_CONFIG"] == str(default_config)
def test_client_respects_explicit_config_over_bundled_default(
def test_client_default_launch_uses_bundled_dsh_sdk_profile_and_explicit_home(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
env_dump = tmp_path / "env.json"
_install_fake_bundled_runtime(tmp_path, monkeypatch)
home = tmp_path / "home"
patch = tmp_path / "sdk.patch.yml"
patch.write_text("[]\n")
_install_fake_bundled_dsh(tmp_path, monkeypatch)
monkeypatch.chdir(tmp_path)
monkeypatch.setenv("DSH_HOME", str(tmp_path / "ambient-home"))
monkeypatch.delenv("DSH_CORDIS_CONFIG", raising=False)
with HarnessClient(HarnessConfig(
profile="sdk",
patches=("sdk.patch.yml",),
dsh_home=str(home),
env={"ENV_DUMP": str(env_dump), "DSH_HOME": str(tmp_path / "env-home")},
)) as client:
init = client.initialize(provider="deepseek-official", cwd="/workspace", model="deepseek-v4-pro")
assert init.serverInfo.name == "bundled-runtime"
assert json.loads(env_dump.read_text()) == {
"argv": ["--profile", "sdk", "--patch", str(patch)],
"DSH_HOME": str(home),
"DSH_CORDIS_CONFIG": None,
}
def test_client_accepts_explicit_environment_dsh_home(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
env_dump = tmp_path / "env.json"
home = tmp_path / "environment-home"
_install_fake_bundled_dsh(tmp_path, monkeypatch)
with HarnessClient(
HarnessConfig(env={"ENV_DUMP": str(env_dump), "DSH_CORDIS_CONFIG": "./explicit.yml"})
HarnessConfig(profile="custom", env={"ENV_DUMP": str(env_dump), "DSH_HOME": str(home)})
) as client:
client.initialize(provider="deepseek-official", cwd="/workspace", model="deepseek-v4-pro")
assert json.loads(env_dump.read_text())["DSH_CORDIS_CONFIG"] == "./explicit.yml"
assert json.loads(env_dump.read_text()) == {
"argv": ["--profile", "custom"],
"DSH_HOME": str(home),
"DSH_CORDIS_CONFIG": None,
}
def test_client_rejects_an_implicit_default_dsh_home(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
_install_fake_bundled_dsh(tmp_path, monkeypatch)
monkeypatch.delenv("DSH_HOME", raising=False)
with pytest.raises(ValueError, match="explicit dsh_home or non-empty DSH_HOME"):
HarnessClient(HarnessConfig(env={})).start()
def test_client_reports_missing_bundled_runtime_dependency(monkeypatch: pytest.MonkeyPatch) -> None:
@@ -998,4 +1028,4 @@ def test_client_reports_missing_bundled_runtime_dependency(monkeypatch: pytest.M
monkeypatch.setattr(sys, "path", [])
with pytest.raises(FileNotFoundError, match="Install deepseek-harness-runtime-bin"):
HarnessClient().start()
HarnessClient(HarnessConfig(dsh_home="/explicit/home")).start()