diff --git a/apps/api/src/opendirector_api/agent.py b/apps/api/src/opendirector_api/agent.py index c1a5a9ba..79274bb7 100644 --- a/apps/api/src/opendirector_api/agent.py +++ b/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': diff --git a/packages/core/src/opendirector/agent/context.py b/packages/core/src/opendirector/agent/context.py index 40181bee..c70429ff 100644 --- a/packages/core/src/opendirector/agent/context.py +++ b/packages/core/src/opendirector/agent/context.py @@ -1,4 +1,5 @@ """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 @@ -6,6 +7,8 @@ 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}) diff --git a/packages/core/src/opendirector/agent/deepseek.py b/packages/core/src/opendirector/agent/deepseek.py index c9a28b1a..83010733 100644 --- a/packages/core/src/opendirector/agent/deepseek.py +++ b/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: diff --git a/packages/core/src/opendirector/agent/jiuwenswarm.py b/packages/core/src/opendirector/agent/jiuwenswarm.py index 2d284da1..25937a03 100644 --- a/packages/core/src/opendirector/agent/jiuwenswarm.py +++ b/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: diff --git a/packages/core/src/opendirector/agent/native.py b/packages/core/src/opendirector/agent/native.py index e28ea138..a76ec55a 100644 --- a/packages/core/src/opendirector/agent/native.py +++ b/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,12 +200,18 @@ 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': @@ -209,7 +219,7 @@ async def connection(reader, writer): 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()) diff --git a/tests/test_deepseek_runtime.py b/tests/test_deepseek_runtime.py index 5c8df48c..3c5f6e7c 100644 --- a/tests/test_deepseek_runtime.py +++ b/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): diff --git a/tests/test_jiuwenswarm_runtime.py b/tests/test_jiuwenswarm_runtime.py index eced83bc..9150346c 100644 --- a/tests/test_jiuwenswarm_runtime.py +++ b/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): diff --git a/tests/test_native_runtime.py b/tests/test_native_runtime.py index e9ea6c1f..63ecaa92 100644 --- a/tests/test_native_runtime.py +++ b/tests/test_native_runtime.py @@ -23,6 +23,11 @@ from pathlib import Path prompt = sys.stdin.read() assert 'Segment plan' not in prompt +if 'Continue plan' in prompt: + assert 'Save a plan' not in prompt + runtime = os.environ['TEST_RUNTIME'] + resume_args = {'codex': ['resume', 'test-native', '-'], 'claude': ['--resume', 'test-native']} + assert sys.argv[1:] == resume_args.get(runtime, ['--session', 'test-native']) assert list(Path('.agents/skills').glob('*/SKILL.md')) assert str(Path.cwd()) in prompt subprocess.run(['sh', '-c', 'printf \"native execution works\" > environment-check.txt'], check=True) @@ -111,6 +116,10 @@ def command(kind, work, socket, token, model, model_url, catalog, **kwargs): work = next(local.root.rglob(f'agent-runs/{turn["id"]}/work')) assert work.is_dir() assert (work / 'environment-check.txt').read_text() == 'native execution works' + response = client.post(path + '/assistant/turns', json={**turn, 'id': str(uuid4()), 'text': 'Continue plan'}) + events = [json.loads(line[6:]) for line in response.text.splitlines() if line.startswith('data: ')] + assert events[-1]['error'] is None, events + def test_native_commands_keep_auth_and_restrict_tools(tmp_path, monkeypatch): @@ -191,6 +200,15 @@ def model(request): assert len(client.get(path).json()['assets']) == 1 assert len(requests) == 3 assert 'server-only-key' not in next(local.root.rglob('*.opencode.jsonl')).read_text() + response = client.post(path + '/assistant/turns', json={**turn, 'id': str(uuid4()), 'text': 'Continue the plan.'}) + events = [json.loads(line[6:]) for line in response.text.splitlines() if line.startswith('data: ')] + assert events[-1]['error'] is None, events + assert len(requests) == 4 + messages = requests[-1]['messages'] + assert 'OpenCode ready.' in json.dumps(messages[:-1]) + assert 'Continue the plan.' in json.dumps(messages[-1]) + assert 'Use the skill and save a plan.' not in json.dumps(messages[-1]) + @pytest.mark.parametrize('runtime', ['default', 'opencode', 'codex', 'claude'])