diff --git a/packages/ui/jsonrpc/README.md b/packages/ui/jsonrpc/README.md index dbc3fddcd0..95ebdb0b00 100644 --- a/packages/ui/jsonrpc/README.md +++ b/packages/ui/jsonrpc/README.md @@ -20,4 +20,4 @@ The plugin owns the PROTOCOL-level exit: a `shutdown` request is answered first ## Wire notes -`initialize.serverInfo.name` is the wire-stable `deepseek-harness-sdk-runtime` (SDK clients key on it, independent of this package's name). The `initialize` params `sessionRoot`, `systemPrompt`, and `clientInfo`, and the `session/prompt` param `profile`, are accepted for wire compatibility but currently unused — persistence roots and the deployment persona come from the `cordis.yml` (see the TODO in [`src/server.ts`](src/server.ts)). +`initialize.serverInfo.name` is the wire-stable `deepseek-harness-sdk-runtime` (SDK clients key on it, independent of this package's name). Persistence roots and the deployment persona come from `cordis.yml`; the wire exposes only parameters the server applies. diff --git a/packages/ui/jsonrpc/src/server.ts b/packages/ui/jsonrpc/src/server.ts index ac0f10f03c..1b193aaf1e 100644 --- a/packages/ui/jsonrpc/src/server.ts +++ b/packages/ui/jsonrpc/src/server.ts @@ -13,6 +13,7 @@ */ import type { Context } from 'cordis' +import { resolve } from 'node:path' import type { ContentBlock } from '@deepseek-ai/dsh-llm' import type { AgentHandle } from '@deepseek-ai/dsh-agent' import { AgentId } from '@deepseek-ai/dsh-agent' @@ -27,17 +28,6 @@ export interface InitializeParams { cwd: string /** Model name every SDK-created agent runs on (see {@link HarnessSdkServer.initialize} for adapter fallback). */ model: string - /** Accepted for SDK wire compatibility; unused — persistence roots come from the `cordis.yml`. */ - sessionRoot?: string - /** - * Accepted for SDK wire compatibility; currently NOT applied — the deployment - * persona comes from the `cordis.yml` system-prompt config. TODO(jsonrpc): - * map this onto a per-runtime system-prompt section once a per-agent override - * seam exists. - */ - systemPrompt?: string - /** Accepted for SDK wire compatibility; unused diagnostic client identity. */ - clientInfo?: { name?: string; version?: string } } /** Result of the `initialize` request: the server's identity for the SDK handshake. */ @@ -52,8 +42,6 @@ export interface SessionPromptParams { sessionId: string /** The prompt content blocks, sent verbatim as the user message. */ contentBlocks: ContentBlock[] - /** Accepted for SDK wire compatibility; unused — profiles are not a harness concept. */ - profile?: string } /** Result of a `session/prompt` request: the prompt ran to turn settle (outcome rides on `session.finished`). */ @@ -132,7 +120,7 @@ export class HarnessSdkServer { agentId: String(info.id), ...(parentSessionId === undefined ? {} : { parentSessionId }), childSessionId, - status: info.stopReason === 'completed' || info.stopReason === 'max-tokens' ? 'ok' : 'error', + status: info.stopReason === 'completed' ? 'ok' : 'error', stopReason: info.stopReason, ...(info.lastAssistantMessage === undefined ? {} : { lastAssistantMessage: info.lastAssistantMessage }), }) @@ -148,7 +136,7 @@ export class HarnessSdkServer { * @returns the server identity for the handshake. */ async initialize(params: InitializeParams): Promise { - this.cwd = params.cwd + this.cwd = resolve(params.cwd) this.model = params.model if (!this.llmFiber && !this.hasAdapterFor(this.model)) { this.llmFiber = await this.ctx.plugin(LlmDeepSeek, { models: [this.model] }) @@ -193,12 +181,28 @@ export class HarnessSdkServer { this.shuttingDown = true const pendingCreations = [...this.sessionCreations.values()] await Promise.allSettled(pendingCreations) + this.sessionCreations.clear() const records = [...this.sessions.values()] this.sessions.clear() - await Promise.all(records.map(rec => rec.handle.dispose())) - await this.llmFiber?.dispose() + this.subagentSessions.clear() + const failures: unknown[] = [] + while (this.disposers.length > 0) { + try { + this.disposers.pop()?.() + } catch (error) { + failures.push(error) + } + } + const teardownResults = await Promise.allSettled([ + ...records.map(rec => Promise.resolve().then(() => rec.handle.dispose())), + ...(this.llmFiber === undefined ? [] : [Promise.resolve().then(() => this.llmFiber?.dispose())]), + ]) this.llmFiber = undefined - while (this.disposers.length > 0) this.disposers.pop()?.() + failures.push(...teardownResults + .filter((result): result is PromiseRejectedResult => result.status === 'rejected') + .map(result => result.reason as unknown)) + if (failures.length === 1) throw failures[0] + if (failures.length > 1) throw new AggregateError(failures, 'SDK server teardown failed') return {} } @@ -251,7 +255,7 @@ export class HarnessSdkServer { private finishedStatus(reason: TurnEndReason | undefined): 'ok' | 'error' { if (!reason) return 'error' - return reason.kind === 'completed' || reason.kind === 'max-tokens' ? 'ok' : 'error' + return reason.kind === 'completed' ? 'ok' : 'error' } private hasAdapterFor(model: string): boolean { diff --git a/packages/ui/jsonrpc/tests/server.spec.ts b/packages/ui/jsonrpc/tests/server.spec.ts index 6b04f4b913..8d4d791841 100644 --- a/packages/ui/jsonrpc/tests/server.spec.ts +++ b/packages/ui/jsonrpc/tests/server.spec.ts @@ -108,15 +108,12 @@ describe('HarnessSdkServer', () => { const init = await server.handleRequest('initialize', { cwd: storageDir, model: 'dsagent-model', - sessionRoot: storageDir, - systemPrompt: 'Custom SDK instructions.', }) as { serverInfo: { name: string } } expect(init.serverInfo.name).toBe('deepseek-harness-sdk-runtime') await server.handleRequest('session/prompt', { sessionId: 'main', contentBlocks: [{ type: 'text', text: 'fix it' }], - profile: 'build', }) expect(llmServer.requests).toHaveLength(1) @@ -307,7 +304,7 @@ describe('HarnessSdkServer', () => { agentId: 'fallback-child-agent', parentSessionId: 'fallback-parent', childSessionId: 'fallback-child-session', - status: 'ok', + status: 'error', stopReason: 'max-tokens', lastAssistantMessage: [], }, @@ -386,7 +383,7 @@ describe('HarnessSdkServer', () => { } expect(server.finishedStatus(undefined)).toBe('error') - expect(server.finishedStatus({ kind: 'max-tokens' })).toBe('ok') + expect(server.finishedStatus({ kind: 'max-tokens' })).toBe('error') expect(server.finishedStatus({ kind: 'error' })).toBe('error') await server.shutdown() } finally { @@ -462,4 +459,63 @@ describe('HarnessSdkServer', () => { expect(retryHandle.dispose).toHaveBeenCalledOnce() await expect(server.getOrCreateSession('after-shutdown')).rejects.toThrow('SDK server is shutting down') }) + + it('resolves a relative cwd before creating the session', async () => { + const create = vi.fn<(options: unknown) => Promise>() + .mockResolvedValue({ agent: {} as Agent, dispose: () => Promise.resolve() }) + const ctx = { + on: vi.fn(() => () => undefined), + agents: { create, get: () => undefined }, + get: () => ({ models: () => ['model'] }), + } as unknown as Context + const server = new HarnessSdkServer(ctx, new FakeTransport()) as unknown as { + initialize(params: { cwd: string; model: string }): Promise + getOrCreateSession(sessionId: string): Promise + shutdown(): Promise> + } + + await server.initialize({ cwd: '.', model: 'model' }) + await server.getOrCreateSession('relative') + + expect(create).toHaveBeenCalledWith(expect.objectContaining({ meta: { cwd: process.cwd() } })) + await server.shutdown() + }) + + it('settles every teardown and aggregates multiple failures', async () => { + const firstDispose = vi.fn(() => { throw new Error('first teardown failed') }) + const secondDispose = vi.fn(() => Promise.reject(new Error('second teardown failed'))) + const ctx = { + on: vi.fn(() => () => undefined), + agents: { create: vi.fn(), get: () => undefined }, + get: () => undefined, + } as unknown as Context + const server = new HarnessSdkServer(ctx, new FakeTransport()) as unknown as { + sessions: Map + shutdown(): Promise> + } + server.sessions.set('first', { handle: { agent: {} as Agent, dispose: firstDispose }, lastTurnEnd: undefined }) + server.sessions.set('second', { handle: { agent: {} as Agent, dispose: secondDispose }, lastTurnEnd: undefined }) + + await expect(server.shutdown()).rejects.toThrow('SDK server teardown failed') + expect(firstDispose).toHaveBeenCalledOnce() + expect(secondDispose).toHaveBeenCalledOnce() + }) + + it('continues teardown after a subscription disposer fails', async () => { + let subscription = 0 + const listenerFailure = new Error('listener teardown failed') + const on = vi.fn(() => { + subscription += 1 + return subscription === 1 ? () => { throw listenerFailure } : () => undefined + }) + const ctx = { + on, + agents: { create: vi.fn(), get: () => undefined }, + get: () => undefined, + } as unknown as Context + const server = new HarnessSdkServer(ctx, new FakeTransport()) + + await expect(server.shutdown()).rejects.toBe(listenerFailure) + expect(on).toHaveBeenCalledTimes(4) + }) }) diff --git a/python/sdk/src/deepseek_harness/api.py b/python/sdk/src/deepseek_harness/api.py index dca522bf5d..c45a70ee33 100644 --- a/python/sdk/src/deepseek_harness/api.py +++ b/python/sdk/src/deepseek_harness/api.py @@ -23,14 +23,11 @@ class DeepSeekHarnessConfig: runtime_cwd: str | None = None session_root: str | None = None cordis: str | None = None - system_prompt: str | None = None env: dict[str, str] = field(default_factory=dict) runtime_bin: str | None = None launch_args_override: tuple[str, ...] | None = None request_timeout_seconds: float | None = None shutdown_timeout_seconds: float | None = 1.0 - client_name: str = "deepseek_harness_python_sdk" - client_version: str = "0.0.0-dev" base_url: str | None = None api_key: str | None = None @@ -52,8 +49,9 @@ class DeepSeekHarness: if config is not None and kwargs: raise TypeError("pass either DeepSeekHarnessConfig or keyword options, not both") self.config = config or DeepSeekHarnessConfig(**kwargs) - cwd = self.config.cwd or str(Path.cwd()) - runtime_cwd = self.config.runtime_cwd or cwd + cwd = str(Path(self.config.cwd or Path.cwd()).resolve()) + runtime_cwd = str(Path(self.config.runtime_cwd).resolve()) if self.config.runtime_cwd is not None else cwd + self._cwd = cwd env = dict(self.config.env) if self.config.session_root is not None: env["DSH_SESSION_ROOT"] = self.config.session_root @@ -73,8 +71,6 @@ class DeepSeekHarness: env=env, request_timeout_seconds=self.config.request_timeout_seconds, shutdown_timeout_seconds=self.config.shutdown_timeout_seconds, - client_name=self.config.client_name, - client_version=self.config.client_version, ) ) self._initialized = False @@ -95,10 +91,8 @@ class DeepSeekHarness: return self._client.start() self._client.initialize( - cwd=self.config.cwd or str(Path.cwd()), + cwd=self._cwd, model=self.config.model, - session_root=self.config.session_root, - system_prompt=self.config.system_prompt, ) self._initialized = True @@ -115,10 +109,9 @@ class DeepSeekHarness: input: str | list[JsonObject], *, session_id: str | None = None, - profile: str | None = None, on_notification: Callable[[Notification], None] | None = None, ) -> TurnResult: - return self.start_session(session_id).run(input, profile=profile, on_notification=on_notification) + return self.start_session(session_id).run(input, on_notification=on_notification) class Session: @@ -130,7 +123,6 @@ class Session: self, input: str | list[JsonObject], *, - profile: str | None = None, on_notification: Callable[[Notification], None] | None = None, ) -> TurnResult: content_blocks = normalize_input(input) @@ -156,7 +148,6 @@ class Session: self.harness.client.session_prompt( self.id, content_blocks, - profile=profile, on_notification=collect, notification_subscription=subscription, ) diff --git a/python/sdk/src/deepseek_harness/client.py b/python/sdk/src/deepseek_harness/client.py index 5a6c28d3a2..e81dd6f30a 100644 --- a/python/sdk/src/deepseek_harness/client.py +++ b/python/sdk/src/deepseek_harness/client.py @@ -9,6 +9,7 @@ import time import uuid from collections import deque from dataclasses import dataclass +from pathlib import Path from typing import Callable, Literal, TypeAlias, TypeVar from pydantic import BaseModel @@ -31,8 +32,6 @@ class HarnessConfig: env: dict[str, str] | None = None request_timeout_seconds: float | None = None shutdown_timeout_seconds: float | None = 1.0 - client_name: str = "deepseek_harness_python_sdk" - client_version: str = "0.0.0-dev" class HarnessClient: @@ -75,7 +74,7 @@ class HarnessClient: stderr=subprocess.PIPE, text=True, encoding="utf-8", - cwd=self.config.cwd, + cwd=None if self.config.cwd is None else str(Path(self.config.cwd).resolve()), env=env, bufsize=1, ) @@ -90,18 +89,22 @@ class HarnessClient: self.request("shutdown", None, response_model=_ShutdownResponse, timeout_seconds=self.config.shutdown_timeout_seconds) except Exception as exc: self._stderr_lines.append(f"shutdown request failed: {exc}") - self._proc = None if proc.stdin: try: proc.stdin.close() except Exception as exc: self._stderr_lines.append(f"stdin close failed: {exc}") - try: - if proc.poll() is None: + if proc.poll() is None: + try: proc.terminate() - proc.wait(timeout=2) - except Exception: + except ProcessLookupError: + pass + try: + proc.wait(timeout=self.config.shutdown_timeout_seconds) + except subprocess.TimeoutExpired: proc.kill() + proc.wait() + self._proc = None self._fail_waiters(self._runtime_closed_error("DeepSeek Harness runtime closed")) if self._reader_thread and self._reader_thread.is_alive(): self._reader_thread.join(timeout=0.5) @@ -113,35 +116,26 @@ class HarnessClient: *, cwd: str, model: str, - session_root: str | None = None, - system_prompt: str | None = None, ) -> InitializeResponse: payload: JsonObject = { - "clientInfo": { - "name": self.config.client_name, - "version": self.config.client_version, - }, - "cwd": cwd, + "cwd": str(Path(cwd).resolve()), "model": model, } - if session_root is not None: - payload["sessionRoot"] = session_root - if system_prompt is not None: - payload["systemPrompt"] = system_prompt - return self.request("initialize", payload, response_model=InitializeResponse) + try: + return self.request("initialize", payload, response_model=InitializeResponse) + except BaseException: + self.close() + raise def session_prompt( self, session_id: str, content_blocks: list[JsonObject], *, - profile: str | None = None, on_notification: Callable[[Notification], None] | None = None, notification_subscription: "NotificationSubscription | None" = None, ) -> None: payload: JsonObject = {"sessionId": session_id, "contentBlocks": content_blocks} - if profile is not None: - payload["profile"] = profile self.request( "session/prompt", payload, diff --git a/python/sdk/tests/manual_sdk_agent_smoke.py b/python/sdk/tests/manual_sdk_agent_smoke.py index 81ee36cfc9..d259d7dab7 100644 --- a/python/sdk/tests/manual_sdk_agent_smoke.py +++ b/python/sdk/tests/manual_sdk_agent_smoke.py @@ -66,7 +66,6 @@ def run_smoke(repo_root: Path, keep_sessions: bool) -> None: runtime_cwd=str(repo_root), session_root=str(session_root), cordis=str(bundled_default_config_path()), - system_prompt="You are running a Python SDK smoke test.", launch_args_override=("node", "--import", "tsx", str(runtime_entry)), env={ "DEEPSEEK_BASE_URL": base_url, @@ -78,7 +77,6 @@ def run_smoke(repo_root: Path, keep_sessions: bool) -> None: result = harness.run( "Please reply with a short confirmation and do not call tools.", session_id="sdk-smoke-main", - profile="build", ) print(f"turn_status={result.status}") print(f"final_response={result.final_response}") diff --git a/python/sdk/tests/test_client.py b/python/sdk/tests/test_client.py index 08fa13a331..0f66ee92e8 100644 --- a/python/sdk/tests/test_client.py +++ b/python/sdk/tests/test_client.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import inspect import sys import threading import time @@ -121,6 +122,45 @@ for line in sys.stdin: assert seen == ["subagent.started", "session.finished"] +def test_relative_cwd_is_absolute_in_process_environment_and_wire( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + script = tmp_path / "capture_cwd.py" + capture = tmp_path / "cwd.json" + script.write_text( + """ +import json +import os +import sys + +for line in sys.stdin: + msg = json.loads(line) + if msg.get("method") == "initialize": + json.dump({"process": os.getcwd(), "environment": os.environ.get("DSH_CWD"), "wire": msg["params"]["cwd"]}, open(os.environ["CAPTURE"], "w")) + print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"serverInfo": {"name": "fake-runtime"}}}), flush=True) + elif msg.get("method") == "shutdown": + print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {}}), flush=True) + break +""".strip() + ) + monkeypatch.chdir(tmp_path) + + with DeepSeekHarness( + cwd=".", + runtime_cwd=".", + launch_args_override=(sys.executable, str(script)), + env={"CAPTURE": str(capture)}, + ): + pass + + expected = str(tmp_path.resolve()) + assert json.loads(capture.read_text()) == { + "process": expected, + "environment": expected, + "wire": expected, + } + + def test_session_run_includes_subagent_finished_for_parent_session(tmp_path: Path) -> None: script = tmp_path / "fake_runtime.py" script.write_text( @@ -293,7 +333,7 @@ for line in sys.stdin: init = client.initialize(cwd="/workspace", model="dsagent") assert init.serverInfo.name == "fake-dsh" - client.session_prompt("main", [{"type": "text", "text": "fix it"}], profile="build") + client.session_prompt("main", [{"type": "text", "text": "fix it"}]) notification = client.next_notification() assert notification.method == "llm/request" assert notification.payload["requestId"] == "req-1" @@ -339,7 +379,7 @@ for line in sys.stdin: with HarnessClient(HarnessConfig(launch_args_override=(sys.executable, str(script)))) as client: client.initialize(cwd="/workspace", model="dsagent") with pytest.raises(ValueError): - client.session_prompt("main", [{"type": "text", "text": "fix it"}], profile="build") + client.session_prompt("main", [{"type": "text", "text": "fix it"}]) def test_client_routes_bridge_requests_and_sends_responses(tmp_path: Path) -> None: @@ -434,9 +474,12 @@ def test_client_close_times_out_when_shutdown_does_not_respond(tmp_path: Path) - script.write_text( """ import json +import signal import sys import time +signal.signal(signal.SIGTERM, signal.SIG_IGN) + for line in sys.stdin: msg = json.loads(line) if msg.get("method") == "initialize": @@ -453,10 +496,56 @@ for line in sys.stdin: ) ) client.start() + proc = client._proc + assert proc is not None client.initialize(cwd="/workspace", model="dsagent") start = time.monotonic() client.close() assert time.monotonic() - start < 2 + assert proc.poll() is not None + assert client._proc is None + + +def test_initialize_failure_reaps_started_runtime(tmp_path: Path) -> None: + script = tmp_path / "rejecting_runtime.py" + script.write_text( + """ +import json +import sys + +for line in sys.stdin: + msg = json.loads(line) + if msg.get("method") == "initialize": + 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) + break +""".strip() + ) + + client = HarnessClient(HarnessConfig(launch_args_override=(sys.executable, str(script)))) + client.start() + proc = client._proc + assert proc is not None + + with pytest.raises(Exception, match="bad initialize"): + client.initialize(cwd=".", model="dsagent") + + assert proc.wait(timeout=1) is not None + assert client._proc is None + + +def test_public_signatures_omit_unsupported_wire_parameters() -> None: + from deepseek_harness import DeepSeekHarnessConfig, Session + + assert "session_root" not in inspect.signature(HarnessClient.initialize).parameters + assert "system_prompt" not in inspect.signature(HarnessClient.initialize).parameters + assert "profile" not in inspect.signature(HarnessClient.session_prompt).parameters + assert "profile" not in inspect.signature(DeepSeekHarness.run).parameters + assert "profile" not in inspect.signature(Session.run).parameters + assert "system_prompt" not in DeepSeekHarnessConfig.__dataclass_fields__ + assert "client_name" not in HarnessConfig.__dataclass_fields__ + assert "client_version" not in HarnessConfig.__dataclass_fields__ def test_client_close_is_idempotent_before_and_after_start(tmp_path: Path) -> None: