deepseek-harness/python/sdk/tests/manual_sdk_agent_smoke.py

"""Manual keyless smoke: drive the repo-source jsonrpc-agent bin (node + tsx).

Runs the SDK against `packages/ui/jsonrpc-agent/src/bin.ts` executed from the
repo checkout (requires `pnpm install`; no build, no API key — the model
endpoint is a local mock SSE server). The bin only boots the supplied
cordis.yml — the stdio JSON-RPC server itself comes from the config's
`@deepseek-ai/dsh-jsonrpc` entry — so the runtime package's default cordis.yml
is passed explicitly. Not collected by pytest; run it directly:
`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
from deepseek_harness_runtime import bundled_default_config_path


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:
    session_root = Path(tempfile.mkdtemp(prefix="dsh-sdk-smoke-sessions-"))
    runtime_entry = repo_root / "packages/ui/jsonrpc-agent/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"mock_base_url={base_url}")

    try:
        with DeepSeekHarness(
            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)),
            env={
                "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"turn_status={result.status}")
        print(f"final_response={result.final_response}")
        assert result.status == "ok", result
        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"))
        assert jsonl_files, f"no jsonl sessions were written under {session_root}"
        print("session_jsonl_files:")
        for path in jsonl_files:
            print(f"  {path} bytes={path.stat().st_size}")
            with path.open("r", encoding="utf-8") as handle:
                first_line = handle.readline().strip()
            if first_line:
                print(f"  first_line={first_line[:500]}")
    finally:
        server.shutdown()
        server.server_close()

    if keep_sessions:
        print(f"kept_session_root={session_root}")
    else:
        shutil.rmtree(session_root)
        print("removed temporary session root")


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()