[feat] agent: add DeepSeek Harness CLI and native execution traces

apps/api/src/opendirector_api/agent.py

@@ -34,7 +34,7 @@ class Draft(BaseModel):
     assets: list[dict[str, JsonValue]] = Field(default_factory=list, max_length=64)
     runtime: Runtime = 'default'
     runtimeModel: str | None = Field(default=None, max_length=200)
-    harnessModels: dict[Literal["opencode", "codex", "claude", "jiuwenswarm"], Annotated[str, Field(max_length=200)] | None] = Field(default_factory=dict)
+    harnessModels: dict[Literal["opencode", "codex", "claude", "jiuwenswarm", "deepseek"], Annotated[str, Field(max_length=200)] | None] = Field(default_factory=dict)
     thinking: Literal['auto', 'low', 'medium', 'high'] = 'auto'
     modelSelections: dict[str, str | None] = Field(default_factory=dict)
     skills: list[str] = Field(default_factory=list, max_length=1)
@@ -195,8 +195,11 @@ async def runtime_status(project_id: UUID, runtime: Runtime):
         return {**item, 'status': await runtime_readiness(runtime)}
 
     @router.get('/runtimes/{runtime}/models')
-    async def runtime_models(project_id: UUID, runtime: Literal['opencode', 'codex', 'claude', 'jiuwenswarm']):
+    async def runtime_models(project_id: UUID, runtime: Literal['opencode', 'codex', 'claude', 'jiuwenswarm', 'deepseek'], selected: str | None = None):
         await asyncio.to_thread(conversations.path, project_id)
+        if runtime == 'deepseek':
+            from opendirector.agent.deepseek import model_choices as deepseek_models
+            return await deepseek_models(selected)
         return await model_choices(runtime)
 
     @router.get('/skills')

docs/architecture/agent-runtime.md

@@ -13,6 +13,7 @@ Harness 指承担模型调用、工具执行和上下文管理的 Agent 执行
 | Codex | PATH 中的 codex | 原生认证与模型目录；隔离启动前解析所选模型或默认模型 |
 | Claude Code | PATH 中的 claude | 原生认证、环境变量与初始化模型目录 |
 | JiuwenSwarm | PATH 中的 jiuwenswarm-process，通过 stdin/stdout JSONL 直接调用 | 本机 Jiuwen 模型目录、配置与环境变量 |
+| DeepSeek Harness | PATH 中的 dsh，以 acp profile 运行标准 JSON-RPC | CLI 原生 profile、模型目录与认证 |
 
 Default 与 OpenCode CLI 分别配置。外部 CLI 不读取 OpenDirector 保存的模型凭证。进程继承服务端环境，包括 PATH 和第三方 API 配置。
 
@@ -60,5 +61,6 @@ Default 将模型配置中的输入模态传入 bundled OpenCode。外部 harnes
 | Codex | workspace-write 沙箱，开放 Shell 和业务工具 |
 | Claude Code | 明确允许 Skill、Bash、Read 和业务工具 |
 | JiuwenSwarm | 明确允许 skill_tool、list_skill、read_file、bash 和业务工具 |
+| DeepSeek Harness | 沿用原生工具与权限；Skill 限于本轮快照，业务工具通过 MCP 接入，未承接的审批不自动批准 |
 
 工具允许列表与操作系统沙箱提供不同约束。业务工具独立校验工程范围和输入；原生文件访问权限不授予直接改写工程存储的业务权限。

packages/core/pyproject.toml

@@ -6,7 +6,7 @@ build-backend = "setuptools.build_meta"
 name = "opendirector-core"
 version = "0.1.0"
 requires-python = ">=3.11"
-dependencies = ["filelock>=3.16,<4", "httpx>=0.27,<1", "anyio>=4,<5", "pydantic>=2,<3", "Pillow>=12,<14"]
+dependencies = ["PyYAML>=6,<7", "filelock>=3.16,<4", "httpx>=0.27,<1", "anyio>=4,<5", "pydantic>=2,<3", "Pillow>=12,<14"]
 
 [tool.setuptools.packages.find]
 where = ["src"]

packages/core/src/opendirector/agent/catalog.py

@@ -50,6 +50,9 @@ async def opencode_fallback(binary: str, configured_providers: list[str]) -> str
 
 
 async def model_choices(runtime: str) -> list[dict]:
+    if runtime == 'deepseek':
+        from opendirector.agent.deepseek import model_choices as deepseek_models
+        return await deepseek_models()
     if runtime == 'jiuwenswarm':
         from opendirector.agent.jiuwenswarm import model_choices as swarm_models
         return await swarm_models()

packages/core/src/opendirector/agent/context.py

@@ -6,7 +6,7 @@
 
 from pydantic import JsonValue
 
-RuntimeId = Literal['default', 'opencode', 'codex', 'claude', 'jiuwenswarm']
+RuntimeId = Literal['default', 'opencode', 'codex', 'claude', 'jiuwenswarm', 'deepseek']
 AgentEvent = dict[str, JsonValue]
 
 

packages/core/src/opendirector/agent/deepseek.py

@@ -0,0 +1,184 @@
+"""DeepSeek Harness CLI over standard ACP JSON-RPC; no Python SDK required."""
+import asyncio
+from contextlib import suppress
+import json
+import os
+from pathlib import Path
+import shutil
+import sys
+from tempfile import TemporaryDirectory
+
+import yaml
+
+from opendirector.agent.process import terminate
+from opendirector.agent.tools import ROOT
+from opendirector.domain.projects import ProjectError
+
+
+class Client:
+    def __init__(self, process, observe=None):
+        self.process = process
+        self.observe = observe
+        self.sequence = 0
+        self.session = None
+
+    async def write(self, record):
+        self.process.stdin.write((json.dumps(record, ensure_ascii=False) + '\n').encode())
+        await self.process.stdin.drain()
+
+    async def request(self, method, params):
+        self.sequence += 1
+        identity = self.sequence
+        await self.write({'jsonrpc': '2.0', 'id': identity, 'method': method, 'params': params})
+        while line := await self.process.stdout.readline():
+            record = json.loads(line)
+            if not line.endswith(b'\n') or not isinstance(record, dict) or record.get('jsonrpc') != '2.0':
+                raise ProjectError('DeepSeek Harness ACP 协议不匹配。', 502)
+            if self.observe:
+                await self.observe(record)
+            if 'method' in record:
+                if 'id' not in record:
+                    continue
+                if record['method'] == 'session/request_permission':
+                    # Native policy requests never imply an unattended grant.
+                    options = record['params']['options']
+                    rejected = next((item for item in options if item['kind'] == 'reject_once'), None)
+                    outcome = {'outcome': 'selected', 'optionId': rejected['optionId']} if rejected else {'outcome': 'cancelled'}
+                    response = {'jsonrpc': '2.0', 'id': record['id'], 'result': {'outcome': outcome}}
+                else:
+                    response = {'jsonrpc': '2.0', 'id': record['id'],
+                        'error': {'code': -32601, 'message': 'Client method is not supported'}}
+                await self.write(response)
+                continue
+            if record.get('id') != identity:
+                raise ProjectError('DeepSeek Harness ACP 回复标识不匹配。', 502)
+            if 'error' in record:
+                raise ProjectError(f"DeepSeek Harness 请求失败（{method}），请检查 CLI 模型与配置。", 502)
+            return record['result']
+        raise ProjectError('DeepSeek Harness 在回复完成前退出。', 502)
+
+    async def initialize(self, work, servers):
+        result = await self.request('initialize', {'protocolVersion': 1,
+            'clientInfo': {'name': 'opendirector', 'version': '0.1.0'}, 'clientCapabilities': {}})
+        if result['protocolVersion'] != 1:
+            raise ProjectError('DeepSeek Harness 不支持 ACP v1，请更新 CLI。', 503)
+        result = await self.request('session/new', {'cwd': str(work), 'mcpServers': servers})
+        self.session = result['sessionId']
+        return result.get('configOptions', [])
+
+    async def close(self):
+        if self.session:
+            await self.request('session/close', {'sessionId': self.session})
+        self.process.stdin.close()
+        await self.process.wait()
+
+
+def arguments(private, skill_root=None):
+    binary = shutil.which('dsh')
+    if binary is None:
+        raise ProjectError('DeepSeek Harness 尚未安装，请安装 dsh CLI。', 503)
+    home = Path(os.environ.get('DSH_HOME', Path.home() / '.dsh')).expanduser()
+    desktop = home / 'profiles/desktop/cordis.patch.yml'
+    acp = home / 'profiles/acp/cordis.patch.yml'
+    try:
+        desktop_rows = (yaml.safe_load(desktop.read_text(encoding='utf-8')) or []) if desktop.is_file() else []
+        acp_rows = (yaml.safe_load(acp.read_text(encoding='utf-8')) or []) if acp.is_file() else []
+    except yaml.YAMLError as error:
+        raise ProjectError('无法读取 DeepSeek Harness 原生模型配置。', 503) from error
+    overridden = {row.get('id') for row in acp_rows if 'config' in row}
+    rows = [row for row in desktop_rows if row.get('id', '').startswith('llm-')
+        and row['id'] not in overridden and 'config' in row]
+    default = next((row['config'] for row in desktop_rows if row.get('id') == 'agent-default-model'), None)
+    if default and 'acp' not in overridden:
+        rows.append({'id': 'acp', 'config': {key: default[key] for key in ('provider', 'model')}})
+    rows.append({'id': 'skill-filesystem', 'config': {'includeDefaultRoots': False,
+        'customSkillDirs': [str(skill_root)] if skill_root else [], 'watch': False}})
+    patch = private / 'deepseek.patch.json'
+    patch.write_text(json.dumps(rows, ensure_ascii=False), encoding='utf-8')
+    patch.chmod(0o600)
+    return [binary, '--profile', 'acp', '--patch', str(patch)]
+
+
+def command(work, socket, token, bridge_env=None):
+    args = arguments(Path(socket).parent, work / '.agents/skills')
+    server = {'name': 'opendirector', 'command': sys.executable,
+        'args': [str(ROOT / 'scripts/agent-mcp.py')],
+        'env': [{'name': key, 'value': value} for key, value in
+            {**(bridge_env or {'OPENDIRECTOR_AGENT_SOCKET': socket}), 'OPENDIRECTOR_AGENT_TOKEN': token}.items()]}
+    return args, dict(os.environ), server
+
+
+async def model_choices(selected=None):
+    with TemporaryDirectory(prefix='od-deepseek-query-') as temporary:
+        process = await asyncio.create_subprocess_exec(*arguments(Path(temporary)), cwd=temporary,
+            stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.DEVNULL, start_new_session=True, limit=8 * 1024 * 1024)
+        client = Client(process)
+        try:
+            async with asyncio.timeout(30):
+                options = await client.initialize(Path(temporary), [])
+                default_value = next(item for item in options if item['id'] == 'model')['currentValue']
+                if selected:
+                    result = await client.request('session/set_config_option', {'sessionId': client.session, 'configId': 'model', 'value': selected})
+                    options = result['configOptions']
+                model = next(item for item in options if item['id'] == 'model')
+                reasoning = next((item for item in options if item['id'] == 'reasoning_effort'), {})
+                efforts = [item['value'] for item in reasoning.get('options', []) if item['value'] in {'low', 'medium', 'high'}]
+                rows = []
+                for group in model['options']:
+                    for item in group.get('options', [group]):
+                        rows.append({'id': item['value'], 'label': item['name'],
+                            'efforts': efforts if item['value'] == model['currentValue'] else [],
+                            'isDefault': item['value'] == default_value})
+                await client.close()
+                return rows
+        except (TimeoutError, OSError, ValueError, KeyError, StopIteration) as error:
+            raise ProjectError('DeepSeek Harness 模型查询失败，请检查 CLI 配置。', 503) from error
+        finally:
+            await terminate(process, group=True)
+
+
+async def run(process, work, prompt, model, thinking, server, observe):
+    client = Client(process, observe)
+    try:
+        async with asyncio.timeout(30):
+            await client.initialize(work, [server])
+            if model:
+                await client.request('session/set_config_option', {'sessionId': client.session, 'configId': 'model', 'value': model})
+            if thinking != 'auto':
+                await client.request('session/set_config_option', {'sessionId': client.session, 'configId': 'reasoning_effort', 'value': thinking})
+        result = await client.request('session/prompt', {'sessionId': client.session,
+            'prompt': [{'type': 'text', 'text': prompt}]})
+        if result['stopReason'] != 'end_turn':
+            raise ProjectError(f"DeepSeek Harness 执行未完成（{result['stopReason']}）。", 502)
+    finally:
+        async def cleanup():
+            if client.session:
+                await client.write({'jsonrpc': '2.0', 'method': 'session/cancel', 'params': {'sessionId': client.session}})
+            await client.close()
+        with suppress(TimeoutError, OSError, ProjectError, ValueError):
+            await asyncio.wait_for(cleanup(), 5)
+        await terminate(process, group=True)
+
+
+def events(record, calls):
+    if record.get('method') != 'session/update':
+        return []
+    update = record['params']['update']
+    kind = update['sessionUpdate']
+    if kind == 'agent_message_chunk' and update['content']['type'] == 'text':
+        return [{'type': 'delta', 'text': update['content']['text']}]
+    if kind not in {'tool_call', 'tool_call_update'}:
+        return []
+    identity = update['toolCallId']
+    previous = calls.get(identity, {})
+    name = update.get('title', previous.get('name', 'tool'))
+    arguments = update.get('rawInput', previous.get('arguments', {}))
+    detail = json.dumps(update.get('content', []), ensure_ascii=False)
+    event = {'type': 'tool', 'id': 'deepseek:' + identity, 'name': name,
+        'arguments': arguments, 'status': {'pending': 'running', 'in_progress': 'running',
+            'completed': 'complete', 'failed': 'failed'}[update.get('status', 'in_progress')],
+        'result': detail[:65536], 'resultTruncated': len(detail) > 65536}
+    calls[identity] = event
+    # Shared business activity is reported from authoritative gateway receipts.
+    return [] if any(name in {f'opendirector.{tool}', f'mcp__opendirector__{tool}'} for tool in ('read_project', 'read', 'command_schema', 'execute_command')) else [event]

packages/core/src/opendirector/agent/native.py

@@ -9,6 +9,7 @@
 from tempfile import TemporaryDirectory
 
 from opendirector.agent.model_gateway import ModelGateway
+from opendirector.agent import deepseek
 from opendirector.agent.context import RunContext
 from opendirector.agent.gateway import ToolGateway
 from opendirector.agent.process import terminate
@@ -25,7 +26,7 @@ def executable(runtime: str) -> str | None:
             if path.is_file():
                 return str(path)
         return None
-    return shutil.which(runtime)
+    return shutil.which('dsh' if runtime == 'deepseek' else runtime)
 
 
 def command(runtime: str, work: Path, socket: str, token: str, model: str | None,
@@ -174,9 +175,13 @@ async def connection(reader, writer):
             model_server = await asyncio.start_server(model.connect, '127.0.0.1', 0) if model else None
             try:
                 model_url = f'http://127.0.0.1:{model_server.sockets[0].getsockname()[1]}' if model_server else ''
-                args, env = command(run.runtime, work, endpoint, token,
-                    config['modelId'] if config else runtime_model, model_url, catalog, bridge_env=bridge_env,
-                    **({'input_modalities': config.get('inputModalities', ['text'])} if run.runtime == 'default' else {}))
+                deepseek_server = None
+                if runtime == 'deepseek':
+                    args, env, deepseek_server = deepseek.command(work, endpoint, token, bridge_env)
+                else:
+                    args, env = command(run.runtime, work, endpoint, token,
+                        config['modelId'] if config else runtime_model, model_url, catalog, bridge_env=bridge_env,
+                        **({'input_modalities': config.get('inputModalities', ['text'])} if run.runtime == 'default' else {}))
                 if thinking != 'auto' and runtime == 'codex':
                     args += ['-c', f'model_reasoning_effort={json.dumps(thinking)}']
                 elif thinking != 'auto' and runtime == 'claude':
@@ -193,6 +198,8 @@ async def connection(reader, writer):
                         prompt = f'${name}\n' + prompt
                     elif runtime == 'claude':
                         prompt = f'/opendirector:{name}\n' + prompt
+                    elif runtime == 'deepseek':
+                        prompt = f'Use the native skill tool to load {name} from {work / ".agents/skills"} before proceeding.\n' + prompt
                     else:
                         prompt = f'Use the native skill tool to load {name} before proceeding.\n' + prompt
                 process = await asyncio.create_subprocess_exec(*args, cwd=work, env=env,
@@ -204,14 +211,20 @@ async def drain_stderr():
                         stderr_tail.extend(chunk)
                         del stderr_tail[:-8192]
                 stderr_task = asyncio.create_task(drain_stderr())
-                process.stdin.write(prompt.encode() + b'\n')
-                await process.stdin.drain()
-                process.stdin.close()
+                if runtime != 'deepseek':
+                    process.stdin.write(prompt.encode() + b'\n')
+                    await process.stdin.drain()
+                    process.stdin.close()
                 async def output():
                     try:
-                        while line := await process.stdout.readline():
-                            raw = json.loads(line)
-                            await queue.put({'native': raw})
+                        if runtime == 'deepseek':
+                            async def observe(raw):
+                                await queue.put({'native': raw})
+                            await deepseek.run(process, work, SYSTEM + '\n' + prompt, runtime_model, thinking, deepseek_server, observe)
+                        else:
+                            while line := await process.stdout.readline():
+                                raw = json.loads(line)
+                                await queue.put({'native': raw})
                         if await process.wait():
                             await stderr_task
                             diagnostic = stderr_tail.decode('utf-8', errors='replace').replace(token, '[redacted]')
@@ -224,12 +237,13 @@ async def output():
                 pump = asyncio.create_task(output())
                 run_dir = directory / 'agent-runs'
                 run_dir.mkdir(exist_ok=True)
+                native_calls = {}
                 with (run_dir / f'{turn_id}.{runtime}.jsonl').open('a', encoding='utf-8') as trace:
                     while (event := await queue.get()) is not None:
                         trace.write(json.dumps(event, ensure_ascii=False) + '\n')
                         trace.flush()
                         if 'native' in event:
-                            for item in text_events(runtime, event['native']):
+                            for item in (deepseek.events(event['native'], native_calls) if runtime == 'deepseek' else text_events(runtime, event['native'])):
                                 yield item
                         else:
                             yield event
@@ -246,6 +260,11 @@ async def output():
                     for task in model.active:
                         task.cancel()
                     await asyncio.gather(*model.active, return_exceptions=True)
+                if runtime == 'deepseek' and pump and not pump.done():
+                    while not queue.empty():
+                        queue.get_nowait()
+                    pump.cancel()
+                    await asyncio.gather(pump, return_exceptions=True)
                 if process is not None:
                     await terminate(process, group=True)
                 while not queue.empty():