[feat] agent: resume native harness sessions across turns

apps/api/src/opendirector_api/agent.py

@@ -282,7 +282,7 @@ async def produce() -> None:
                     run = RunContext(runtime=value.runtime, project_id=project_id, turn_id=value.id,
                         session=session, directory=directory, config=config, messages=context,
                         selected_skills=value.skills, thinking=value.thinking, model_selections=model_selections,
-                        runtime_model=value.runtimeModel, skills=available)
+                        runtime_model=value.runtimeModel, skills=available, conversation_id=UUID(saved['conversationId']))
                     async with aclosing(runtimes.stream(run)) as stream:
                         async for event in stream:
                             if event['type'] == 'delta':

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

@@ -1,11 +1,14 @@
 """Shared input contract for SDK, CLI and service-backed runtimes."""
+import json
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Literal
 from uuid import UUID
 
 from pydantic import JsonValue
 
+from opendirector.infrastructure.local_projects import write_json
+
 RuntimeId = Literal['default', 'opencode', 'codex', 'claude', 'jiuwenswarm', 'deepseek']
 AgentEvent = dict[str, JsonValue]
 
@@ -21,6 +24,28 @@ class RunContext:
     messages: list[dict[str, JsonValue]]
     selected_skills: list[str]
     model_selections: dict[str, str | None] = field(default_factory=dict)
+    conversation_id: UUID | None = None
     thinking: str = 'auto'
     runtime_model: str | None = None
     skills: list[dict[str, str]] = field(default_factory=list)
+
+    def native_session_path(self) -> Path | None:
+        if self.conversation_id is None:
+            return None
+        return self.directory / 'agent-runs' / f'{self.runtime}-sessions' / str(self.conversation_id)
+
+    def native_session_id(self, selection: dict) -> str | None:
+        directory = self.native_session_path()
+        if directory is None or not (directory / 'session.json').is_file():
+            return None
+        saved = json.loads((directory / 'session.json').read_text(encoding='utf-8'))
+        if saved['selection'] == selection and saved['messages'] == len(self.messages) - 1:
+            return saved['sessionId']
+        return None
+
+    def save_native_session(self, session_id: str, selection: dict) -> None:
+        directory = self.native_session_path()
+        if directory is not None:
+            directory.mkdir(parents=True, exist_ok=True)
+            write_json(directory / 'session.json', {'sessionId': session_id, 'selection': selection,
+                                                    'messages': len(self.messages) + 1})

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

@@ -57,13 +57,18 @@ async def request(self, method, params):
             return record['result']
         raise ProjectError('DeepSeek Harness 在回复完成前退出。', 502)
 
-    async def initialize(self, work, servers):
+    async def initialize(self, work, servers, session_id=None):
         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']
+        params = {'cwd': str(work), 'mcpServers': servers}
+        if session_id:
+            result = await self.request('session/resume', {**params, 'sessionId': session_id})
+            self.session = session_id
+        else:
+            result = await self.request('session/new', params)
+            self.session = result['sessionId']
         return result.get('configOptions', [])
 
     async def close(self):
@@ -138,19 +143,20 @@ async def model_choices(selected=None):
             await terminate(process, group=True)
 
 
-async def run(process, work, prompt, model, thinking, server, observe):
+async def run(process, work, blocks, model, thinking, server, observe, session_id=None):
     client = Client(process, observe)
     try:
         async with asyncio.timeout(30):
-            await client.initialize(work, [server])
+            await client.initialize(work, [server], session_id)
             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}]})
+            'prompt': [{'type': 'text', 'text': block} for block in blocks]})
         if result['stopReason'] != 'end_turn':
             raise ProjectError(f"DeepSeek Harness 执行未完成（{result['stopReason']}）。", 502)
+        return client.session
     finally:
         async def cleanup():
             if client.session:

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

@@ -141,9 +141,11 @@ async def stream(self, run: RunContext, gateway: ToolGateway):
         stopping = False
         emitted_chars = 0
         terminal_written = False
+        selection = {'model': run.runtime_model, 'thinking': run.thinking}
+        session_id = run.native_session_id(selection)
         with TemporaryDirectory(prefix='od-jiuwen-') as temporary:
-            state = Path(temporary)
-            config = state / 'config'
+            state = run.native_session_path() or Path(temporary)
+            config = Path(temporary) / 'config'
             config.mkdir()
             # Keep credentials out of published execution files and native traces.
             home = Path(os.environ.get('JIUWENSWARM_HOME', Path.home()))
@@ -154,10 +156,15 @@ async def stream(self, run: RunContext, gateway: ToolGateway):
                     (config / name).chmod(0o600)
             if not (config / 'config.yaml').is_file():
                 raise ProjectError('请先运行 jiuwenswarm-init 并配置模型。', 503)
-            skill_root = state / 'agent/workspace/skills'
-            skill_root.mkdir(parents=True)
+            skill_root = Path(temporary) / 'skills'
+            skill_root.mkdir()
             for root in gateway.roots:
                 shutil.copytree(root, skill_root / root.name)
+            skill_link = state / 'agent/workspace/skills'
+            skill_link.parent.mkdir(parents=True, exist_ok=True)
+            if skill_link.is_symlink():
+                skill_link.unlink()
+            skill_link.symlink_to(skill_root, target_is_directory=True)
             gateway.roots += list(skill_root.iterdir())
             functions = [item['function'] for item in tools()]
             host_tools = [{'name': item['name'], 'description': item['description'],
@@ -168,11 +175,13 @@ async def stream(self, run: RunContext, gateway: ToolGateway):
                     'Read source assets at their supplied absolute paths; do not move or modify them. '
                     'Write intermediate files and outputs into the execution directory. '
                     'Use OpenDirector business tools for project changes.\n'
-                    + json.dumps(run.messages, ensure_ascii=False)),
+                    + json.dumps(run.messages[-1:] if session_id else run.messages, ensure_ascii=False)),
                 'agent': {'name': 'opendirector', 'instructions': SYSTEM,
                     'tools': [item['name'] for item in functions] + ['skill_tool', 'list_skill', 'read_file', 'bash']},
                 'skills': run.selected_skills, 'mcp': [], 'host_tools': host_tools,
                 'workspace': {'cwd': str(work), 'project_dir': str(work)}}
+            if session_id:
+                request['session_id'] = session_id
             if run.runtime_model:
                 request['model'] = run.runtime_model
             trace_path = work.parent / f'{run.turn_id}.jiuwenswarm.jsonl'
@@ -234,6 +243,9 @@ async def call(record):
                         if reason:
                             raise ProjectError(f"JiuwenSwarm 执行未完成（{result['status']}）：{reason}", 502)
                         raise ProjectError(f"JiuwenSwarm 执行未完成（{result['status']}），请检查原生轨迹。", 502)
+                    if session_id and session_id != result['session_id']:
+                        raise ProjectError('JiuwenSwarm 返回了不同的会话 ID。', 502)
+                    run.save_native_session(result['session_id'], selection)
                     # The terminal output includes the deltas and any unflushed final tail.
                     remaining = (result.get('output') or '')[emitted_chars:]
                     if remaining:

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

@@ -112,6 +112,10 @@ async def stream(self, run: RunContext, gateway: ToolGateway):
         directory, config, context = run.directory, run.config, run.messages
         selected, thinking = run.selected_skills, run.thinking
         runtime, runtime_model, catalog = ('opencode' if run.runtime == 'default' else run.runtime), run.runtime_model, run.skills
+        selection = {'model': runtime_model, 'thinking': thinking}
+        if config:
+            selection.update(config=config.get('id'), modelId=config['modelId'])
+        session_id = run.native_session_id(selection)
         queue = asyncio.Queue(maxsize=128)
         pending = set()
         closed = False
@@ -196,20 +200,26 @@ async def connection(reader, writer):
                     args += ['-c', f'model_reasoning_effort={json.dumps(thinking)}']
                 elif thinking != 'auto' and runtime == 'claude':
                     args += ['--effort', thinking]
-                prompt = (f'Execution directory: {work}\n'
+                if session_id:
+                    if runtime == 'codex':
+                        args += ['resume', session_id, '-']
+                    else:
+                        args += ['--session' if runtime == 'opencode' else '--resume', session_id]
+                execution_context = (f'Execution directory: {work}\n'
                     'Use inherited PATH to discover installed programs with command -v. '
                     'Read source assets at their supplied absolute paths; do not move or modify them. '
                     'Write all intermediate files and outputs into the execution directory. '
-                    'Use OpenDirector business tools for project changes.\n'
-                    + json.dumps(context, ensure_ascii=False))
+                    'Use OpenDirector business tools for project changes.')
+                prompt = execution_context + '\n' + json.dumps(context[-1:] if session_id else context, ensure_ascii=False)
+                skill_instruction = ''
                 if selected:
                     name = selected[0]
                     if runtime == 'codex':
                         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
+                        skill_instruction = f'Use the native skill tool to load {name} from {work / ".agents/skills"} before proceeding.'
                     else:
                         prompt = f'Use the native skill tool to load {name} before proceeding.\n' + prompt
                 try:
@@ -232,13 +242,32 @@ async def drain_stderr():
                     process.stdin.close()
                 async def output():
                     try:
+                        reported_session = None
                         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)
+                            session_work = run.native_session_path() or work
+                            session_work.mkdir(parents=True, exist_ok=True)
+                            blocks = [execution_context] if session_id else [SYSTEM, execution_context]
+                            if skill_instruction:
+                                blocks.append(skill_instruction)
+                            if not session_id and context[:-1]:
+                                blocks.append('Conversation history\n' + json.dumps(context[:-1], ensure_ascii=False))
+                            if context:
+                                blocks.append('Current user request\n' + context[-1]['content'])
+                            reported_session = await deepseek.run(process, session_work, blocks, runtime_model, thinking,
+                                                            deepseek_server, observe, session_id)
                         else:
                             while line := await process.stdout.readline():
                                 raw = json.loads(line)
+                                if runtime == 'codex' and raw['type'] == 'thread.started':
+                                    reported_session = raw.get('thread_id')
+                                elif runtime == 'opencode':
+                                    reported_session = raw.get('sessionID') or reported_session
+                                elif runtime == 'claude':
+                                    reported_session = raw.get('session_id') or reported_session
+                                if session_id and reported_session and session_id != reported_session:
+                                    raise ProjectError('Harness 返回了不同的会话 ID。', 502)
                                 await queue.put({'native': raw})
                         if await process.wait():
                             await stderr_task
@@ -247,6 +276,8 @@ async def observe(raw):
                                 diagnostic = diagnostic.replace(config['apiKey'], '[redacted]')
                             await queue.put({'native': {'type': 'diagnostic', 'stderr': diagnostic}})
                             raise ProjectError(f'{runtime} 进程退出（{process.returncode}），请检查 CLI 登录和配置。', 502)
+                        if reported_session:
+                            run.save_native_session(reported_session, selection)
                     finally:
                         await queue.put(None)
                 pump = asyncio.create_task(output())

tests/test_deepseek_runtime.py

@@ -29,7 +29,9 @@ def result(req, value): write(dict(id=req['id'], result=value))
 def read(): return json.loads(sys.stdin.readline())
 req=read(); assert req['method']=='initialize'
 result(req, dict(protocolVersion=0 if mode=='bad_version' else 1))
-req=read(); assert req['method']=='session/new'
+req=read(); assert req['method'] in {'session/new','session/resume'}
+resumed=req['method']=='session/resume'
+if resumed: assert req['params']['sessionId']==sid
 servers=req['params']['mcpServers']
 model=dict(id='model',currentValue='route/default',options=[dict(group='route',options=[dict(value='route/default',name='Default model')])])
 if mode=='efforts':
@@ -47,9 +49,17 @@ def read(): return json.loads(sys.stdin.readline())
  elif req['method']=='session/cancel':
   Path(os.environ['DEEPSEEK_CANCELLED']).write_text('cancelled') if 'DEEPSEEK_CANCELLED' in os.environ else None
  elif req['method']=='session/prompt':
-  text=req['params']['prompt'][0]['text']
-  assert 'Use OpenDirector' in text and str(Path.cwd()) in text
+  blocks=[block['text'] for block in req['params']['prompt']]
+  if resumed:
+   assert blocks[0].startswith('Execution directory:')
+   assert not any('Conversation history' in block or 'You are OpenDirector' in block for block in blocks)
+   assert 'Continue' in blocks[-1]
+  else:
+   assert blocks[0].startswith('You are OpenDirector') and str(Path.cwd()) in blocks[1]
+   assert 'Continue' not in blocks[-1]
+  text=blocks[-1]
   if mode!='wait':
+   assert text.startswith('Current user request' + chr(10))
    assert 'Summarize the delivered result and material limitations.' in text
    assert 'User model preferences for this turn:' in text and 'selected-image' in text
   patch=json.loads(Path(sys.argv[sys.argv.index('--patch')+1]).read_text())
@@ -127,6 +137,13 @@ def test_acp_shared_write_and_native_trace(cli, tmp_path, monkeypatch, mode):
         trace=next(local.root.rglob('*.deepseek.jsonl')).read_text()
         assert 'owned-session' in trace and 'session/update' in trace
         assert 'OPENDIRECTOR_AGENT_TOKEN' not in trace
+        if mode == 'success':
+            response = client.post(path+'/assistant/turns', json=dict(
+                id=str(uuid4()), runtime='deepseek', runtimeModel='route/default', text='Continue',
+                skills=['example'], modelSelections={'image': 'selected-image'}))
+            events = [json.loads(line[6:]) for line in response.text.splitlines() if line.startswith('data: ')]
+            assert events[-1]['error'] is None, events
+
 
 
 def test_cancel_acp_before_collecting_child(cli, tmp_path, monkeypatch):

tests/test_jiuwenswarm_runtime.py

@@ -43,13 +43,17 @@ def emit(kind, **fields):
 assert 'trusted_dirs' not in request['workspace']
 state = Path(os.environ['JIUWENSWARM_DATA_DIR'])
 assert sorted(p.name for p in (state/'agent/workspace/skills').iterdir()) == ['example']
-assert 'private-test-key' in (state/'config/config.yaml').read_text()
+assert 'private-test-key' in (Path(os.environ['JIUWENSWARM_CONFIG_DIR'])/'config.yaml').read_text()
 assert 'private-test-key' not in json.dumps(request)
 assert '*' not in request['agent']['tools']
 assert 'skill_tool' in request['agent']['tools']
 assert request['mcp'] == []
 assert request['skills'] == ['example']
 sid='native-session'
+if 'Continue plan' in request['input']:
+ assert request['session_id']==sid and 'Save plan' not in request['input']
+else:
+ assert 'session_id' not in request
 if mode == 'quota':
  emit('result',status='failed',exit_code=1,error={'code':'RUNTIME_ERROR','message':'HTTP 429: Token Plan quota exceeded (2056)'},output=None)
  raise SystemExit(1)
@@ -139,6 +143,12 @@ async def skills(*args):
         assert 'native-session' in trace
         assert 'private-test-key' not in trace
         assert not list(local.root.rglob('config.yaml'))
+        if mode == 'success':
+            response = client.post(path+'/assistant/turns', json={
+                'id':str(uuid4()), 'runtime':'jiuwenswarm', 'text':'Continue plan', 'skills':['example']})
+            events = [json.loads(line[6:]) for line in response.text.splitlines() if line.startswith('data: ')]
+            assert events[-1]['error'] is None, events
+
 
 
 def test_cancel_collects_child(cli, tmp_path, monkeypatch):