mirror of
https://github.com/deepseek-ai/deepseek-harness.git
synced 2026-08-29 04:26:38 +00:00
Tianyi Cui
56e038b2e3
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.
124 lines
4.5 KiB
Python
124 lines
4.5 KiB
Python
"""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``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import shutil
|
|
import tempfile
|
|
import threading
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from deepseek_harness import DeepSeekHarness
|
|
|
|
|
|
class MockCompletionHandler(BaseHTTPRequestHandler):
|
|
requests: list[dict[str, Any]] = []
|
|
|
|
def do_POST(self) -> None:
|
|
length = int(self.headers.get("content-length", "0"))
|
|
body = self.rfile.read(length).decode("utf-8")
|
|
self.requests.append({
|
|
"path": self.path,
|
|
"authorization": self.headers.get("authorization"),
|
|
"body": json.loads(body),
|
|
})
|
|
self.send_response(200)
|
|
self.send_header("content-type", "text/event-stream")
|
|
self.end_headers()
|
|
self.wfile.write(b'data: {"choices":[{"delta":{"role":"assistant","content":null,"reasoning_content":""}}]}\n\n')
|
|
self.wfile.write(b'data: {"choices":[{"delta":{"content":"SDK runtime reached the configured HTTP model endpoint."}}]}\n\n')
|
|
self.wfile.write(b'data: {"choices":[{"delta":{"content":""},"finish_reason":"stop"}],"usage":{"prompt_tokens":7,"completion_tokens":9}}\n\n')
|
|
self.wfile.write(b"data: [DONE]\n\n")
|
|
|
|
def log_message(self, _format: str, *_args: object) -> None:
|
|
return
|
|
|
|
|
|
def run_smoke(repo_root: Path, keep_sessions: bool) -> None:
|
|
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"dsh_home={dsh_home}")
|
|
print(f"mock_base_url={base_url}")
|
|
|
|
try:
|
|
with DeepSeekHarness(
|
|
model="sdk-smoke-model",
|
|
cwd=str(repo_root / "python/sdk"),
|
|
runtime_cwd=str(repo_root),
|
|
_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",
|
|
},
|
|
request_timeout_seconds=20,
|
|
shutdown_timeout_seconds=2,
|
|
) as harness:
|
|
result = harness.run(
|
|
"Please reply with a short confirmation and do not call tools.",
|
|
session_id="sdk-smoke-main",
|
|
)
|
|
print(f"final_response={result.final_response}")
|
|
assert "configured HTTP model endpoint" in result.final_response
|
|
assert len(MockCompletionHandler.requests) == 1
|
|
request = MockCompletionHandler.requests[0]
|
|
print(json.dumps(request, ensure_ascii=False, indent=2)[:4000])
|
|
assert request["authorization"] == "Bearer sdk-smoke-key"
|
|
assert request["body"]["model"] == "sdk-smoke-model"
|
|
|
|
jsonl_files = sorted(session_root.rglob("*.jsonl.zstd"))
|
|
assert jsonl_files, f"no Zstandard JSONL sessions were written under {session_root}"
|
|
print("session_jsonl_zstd_files:")
|
|
for path in jsonl_files:
|
|
print(f" {path} bytes={path.stat().st_size}")
|
|
assert path.read_bytes().startswith(bytes.fromhex("28b52ffd"))
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
|
|
if keep_sessions:
|
|
print(f"kept_dsh_home={dsh_home}")
|
|
else:
|
|
shutil.rmtree(dsh_home)
|
|
print("removed temporary dsh home")
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument(
|
|
"--repo-root",
|
|
type=Path,
|
|
default=Path(__file__).resolve().parents[3],
|
|
help="Path to the deepseek-harness checkout.",
|
|
)
|
|
parser.add_argument("--keep-sessions", action="store_true")
|
|
args = parser.parse_args()
|
|
run_smoke(args.repo_root.resolve(), args.keep_sessions)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|