diff --git a/packages/core/src/opendirector/agent/jiuwenswarm.py b/packages/core/src/opendirector/agent/jiuwenswarm.py index 810ebd2c..e9c2d2e6 100644 --- a/packages/core/src/opendirector/agent/jiuwenswarm.py +++ b/packages/core/src/opendirector/agent/jiuwenswarm.py @@ -1,393 +1,254 @@ -"""One Process CLI invocation per turn, using host tools and schema 0.1.""" +"""Jiuwen Process CLI transport; project operations stay in ToolGateway.""" import asyncio -from contextlib import contextmanager, suppress +from contextlib import suppress import json -import math import os from pathlib import Path -import re import shutil from tempfile import TemporaryDirectory from uuid import uuid4 -from opendirector.agent.process import terminate +from opendirector.agent.context import RunContext +from opendirector.agent.gateway import ToolGateway from opendirector.agent.skill_snapshot import prepare_skills -from opendirector.agent.tools import ROOT, SYSTEM, tools +from opendirector.agent.tools import SYSTEM, tools +from opendirector.agent.process import terminate from opendirector.domain.projects import ProjectError -MAX_INPUT = 1024 * 1024 -MAX_OUTPUT = 8 * 1024 * 1024 -QUERY_TIMEOUT = 30 -RUN_TIMEOUT = 1800 -REQUIRED_FEATURES = ('host_tools', 'cancel', 'stable_event_fields', 'custom_agent') -NATIVE_TOOLS = ('skill_tool', 'list_skill', 'bash', 'read_file', 'write_file', 'edit_file', 'list_dir', 'glob', 'grep') + (('powershell',) if os.name == 'nt' else ()) - - -def native_tool_event(kind, payload, previous=None): - tool = payload.get('tool') - if not isinstance(tool, dict): - raise protocol_error() - name, call_id = tool.get('name'), tool.get('call_id') - # Host callbacks have their own gateway receipts; do not show them twice. - if name not in NATIVE_TOOLS: - return None - if not isinstance(call_id, str) or not call_id: - raise protocol_error() - arguments = tool.get('arguments') - if arguments is None and kind == 'chat.tool_result': - arguments = previous['arguments'] if previous else {} - if isinstance(arguments, str): - try: - arguments = json.loads(arguments, object_pairs_hook=_pairs, parse_float=_number, parse_constant=_number) - except (ValueError, RecursionError) as error: - raise protocol_error() from error - if not isinstance(arguments, dict): - raise protocol_error() - status = tool.get('status') - statuses = {'started'} if kind == 'chat.tool_call' else {'completed', 'failed'} - if not isinstance(status, str) or status not in statuses: - raise protocol_error() - event = dict(type='tool', id='jiuwenswarm:native:' + call_id, name=name, - status='running' if kind == 'chat.tool_call' else 'failed' if status == 'failed' else 'complete', arguments=arguments) - if kind == 'chat.tool_result': - detail = json.dumps(tool.get('result'), ensure_ascii=False) - event.update(result=detail[:65536], resultTruncated=len(detail) > 65536) - if status == 'failed': - event['error'] = 'JiuwenSwarm 原生工具执行失败，请查看工具结果。' - return event - - -@contextmanager -def execution_environment(roots): - """Mount snapshots in the native library without editing the user's library.""" - data = data_directory(executable()) - with TemporaryDirectory(prefix='od-jiuwenswarm-') as temporary: - target = Path(temporary) - config = target / 'config' - config.mkdir() - for name in ('config.yaml', '.env'): - source_file = data / 'config' / name - if source_file.is_file(): - shutil.copyfile(source_file, config / name) - (config / name).chmod(0o600) - if not (config / 'config.yaml').is_file(): - raise ProjectError('JiuwenSwarm 模型配置不存在，请先初始化其配置。', 503) - library = target / 'agent/workspace/skills' - library.mkdir(parents=True) - for root in roots: - shutil.copytree(root, library / root.name) - yield {**os.environ, 'JIUWENSWARM_DATA_DIR': temporary, 'JIUWENSWARM_CONFIG_DIR': str(config)} - - -def data_directory(binary): - source = os.environ.get('JIUWENSWARM_DATA_DIR') - if source: - return Path(source).expanduser().resolve() - local = ROOT / '.toolchain/jiuwenswarm' - return local / 'data' if binary and Path(binary).resolve().is_relative_to(local.resolve()) else Path.home() / '.jiuwenswarm' - def executable(): - configured = os.environ.get('OPENDIRECTOR_JIUWENSWARM_COMMAND') - if configured: - return shutil.which(configured) - binary = shutil.which('jiuwenswarm-process') - if binary: - return binary - local = ROOT / '.toolchain/jiuwenswarm' - candidates = [local / 'venv/bin/jiuwenswarm-process', local / 'venv/Scripts/jiuwenswarm-process.exe'] - candidates += [local / 'run'] if os.name != 'nt' else [] - return next((str(path) for path in candidates if path.is_file() and os.access(path, os.X_OK)), None) - - -def encode(record): - data = (json.dumps(record, ensure_ascii=False, allow_nan=False) + '\n').encode() - if len(data) > MAX_INPUT: - raise ProjectError('JiuwenSwarm 输入超过 1 MiB，请缩短对话或工具结果。', 413) - return data - - -def protocol_error(): - return ProjectError('JiuwenSwarm 协议不匹配或回复不完整，请检查安装版本；不会自动重试。', 502, code='runtime_protocol') - - -def _pairs(pairs): - result = {} - for key, value in pairs: - if key in result: - raise protocol_error() - result[key] = value - return result - - -def _number(value): - number = float(value) - if not math.isfinite(number): - raise protocol_error() - return number - - -class Records: - def __init__(self, request_id, operation=None): - self.request_id, self.operation = request_id, operation - self.sequence, self.session, self.result = 0, None, None - - def accept(self, line): - if self.result is not None or len(line) > MAX_OUTPUT or not line.endswith(b'\n'): - raise protocol_error() - try: - record = json.loads(line, object_pairs_hook=_pairs, parse_float=_number, parse_constant=_number) - if not isinstance(record, dict) or record.get('schema_version') != '0.1' or record.get('request_id') != self.request_id: - raise protocol_error() - if type(record.get('sequence')) is not int or record['sequence'] != self.sequence: - raise protocol_error() - self.sequence += 1 - session = record.get('session_id') - if session is not None: - if not isinstance(session, str) or not session.strip() or self.session is not None and self.session != session: - raise protocol_error() - self.session = session - if record.get('type') == 'event' and self.operation is None: - if not isinstance(record.get('event_type'), str) or not record['event_type']: - raise protocol_error() - return record - expected = 'query_result' if self.operation is not None else 'result' - code = record.get('exit_code') - statuses = {0: 'completed', 1: 'failed', 2: 'failed', 124: 'timed_out', 130: 'cancelled'} - if record.get('type') != expected or session != self.session or type(code) is not int or code not in statuses or record.get('status') != statuses[code]: - raise protocol_error() - if self.operation is not None and (session is not None or self.sequence != 1 or record.get('operation') not in (None, self.operation)): - raise protocol_error() - if code == 0: - if record.get('error') is not None: - raise protocol_error() - if self.operation is not None: - if record.get('operation') != self.operation or not isinstance(record.get('data'), dict): - raise protocol_error() - elif session is None or record.get('output') is not None and not isinstance(record['output'], str): - raise protocol_error() - elif not isinstance(record.get('error'), dict) or not isinstance(record['error'].get('code'), str): - raise protocol_error() - self.result = record - return record - except (ValueError, UnicodeError, RecursionError, TypeError) as error: - raise protocol_error() from error - - def finish(self, exit_code): - if self.result is None or self.result['exit_code'] != exit_code: - raise protocol_error() - if exit_code: - code = self.result['error']['code'] - code = code if re.fullmatch(r'[A-Z_0-9]{1,80}', code) else 'RUNTIME_FAILED' - if self.operation is not None: - raise ProjectError(f'JiuwenSwarm 查询失败（{code}），请检查其安装与模型配置。', 503, code='runtime_query_failed') - raise ProjectError(f'JiuwenSwarm 执行失败（{code}），请检查其模型与配置；已提交的操作不会回滚。', 502, code='runtime_failed') - return self.result - - -async def _stderr(stream): - # Do not expose native diagnostics that may contain credentials. - while await stream.read(8192): - pass - - -async def _start(args, cwd, env=None): - binary = executable() - if binary is None: - raise ProjectError('JiuwenSwarm Process CLI 尚未安装。', 503) - if env is None and Path(binary).resolve().is_relative_to((ROOT / '.toolchain/jiuwenswarm').resolve()): - data = data_directory(binary) - env = {**os.environ, 'JIUWENSWARM_DATA_DIR': str(data), 'JIUWENSWARM_CONFIG_DIR': str(data / 'config')} - try: - return await asyncio.create_subprocess_exec(binary, *args, cwd=cwd, env=env, - stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, start_new_session=True, limit=MAX_OUTPUT) - except OSError as error: - raise ProjectError('JiuwenSwarm Process CLI 无法启动，请检查安装路径。', 503) from error - - -async def query(operation): - request_id = str(uuid4()) - request = encode(dict(schema_version='0.1', type='query', request_id=request_id, - operation=operation, params={}, timeout_seconds=QUERY_TIMEOUT - 5)) - with TemporaryDirectory(prefix='od-jiuwenswarm-query-') as directory: - process = await _start(['--query-json', '-'], directory) - diagnostics = asyncio.create_task(_stderr(process.stderr)) - try: - async with asyncio.timeout(QUERY_TIMEOUT): - process.stdin.write(request) - await process.stdin.drain() - process.stdin.close() - records = Records(request_id, operation) - while line := await process.stdout.readline(): - records.accept(line) - result = records.finish(await process.wait()) - await diagnostics - return result['data'] - except (TimeoutError, ValueError, BrokenPipeError, ConnectionResetError) as error: - raise ProjectError('JiuwenSwarm 查询未完成，请检查安装与配置。', 503) from error - finally: - drain = asyncio.create_task(_stderr(process.stdout)) - try: - await terminate(process, group=os.name != 'nt') - finally: - drain.cancel() - diagnostics.cancel() - await asyncio.gather(drain, diagnostics, return_exceptions=True) - - -async def capabilities(): - result = await query('protocol.capabilities') - features = result.get('features') - if result.get('schema_version') != '0.1' or not isinstance(features, dict) or not all(features.get(name) is True for name in REQUIRED_FEATURES): - raise protocol_error() - return result + return shutil.which('jiuwenswarm-process') async def model_choices(): - result = await query('model.list') - rows = result.get('models') - if not isinstance(rows, list): - raise protocol_error() - choices = [] - for row in rows: - if not isinstance(row, dict) or not isinstance(row.get('selection_key'), str) or not isinstance(row.get('display_name'), str): - raise protocol_error() - choices.append(dict(id=row['selection_key'], label=row['display_name'], efforts=[], - **({'isDefault': True} if row.get('is_default') else {}))) - return choices + data = await query('model.list') + return [dict(id=item['selection_key'], label=item['display_name'], efforts=[], + isDefault=item['is_default']) for item in data['models']] async def readiness(): if executable() is None: return 'not_installed' try: - await capabilities() - return 'ready' if await model_choices() else 'needs_configuration' - except ProjectError as error: - return 'unsupported_version' if error.code == 'runtime_protocol' else 'unknown' + await query('protocol.capabilities') + return 'ready' if (await query('model.list'))['models'] else 'needs_configuration' + except ProjectError: + return 'unknown' -async def _close(process, request_id): - if process.returncode is not None: - return - drain = asyncio.create_task(_stderr(process.stdout)) +async def execute(request, *, cwd, env=None, on_event=None, on_tool_call=None, cancel=None): + binary = executable() + if binary is None: + raise ProjectError('jiuwenswarm-process 尚未安装。', 503) + querying = request['type'] == 'query' + process = await asyncio.create_subprocess_exec(binary, + *(['--query-json', '-'] if querying else ['--run-jsonl']), cwd=cwd, + env={**os.environ, **(env or {})}, stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + start_new_session=True, limit=8 * 1024 * 1024) + + async def write(control): + process.stdin.write((json.dumps(control, ensure_ascii=False) + '\n').encode()) + await process.stdin.drain() + + async def drain_stderr(): + while await process.stderr.read(8192): + pass + + async def watch_cancel(): + await cancel.wait() + with suppress(BrokenPipeError, ConnectionResetError): + await write({'schema_version': '0.1', 'type': 'cancel', 'request_id': request['request_id']}) + + stderr = asyncio.create_task(drain_stderr()) + watcher = asyncio.create_task(watch_cancel()) if cancel is not None else None + result = None + sequence = 0 + session = None try: - async with asyncio.timeout(10): - with suppress(BrokenPipeError, ConnectionResetError, RuntimeError): - process.stdin.write(encode(dict(schema_version='0.1', type='cancel', request_id=request_id))) - await process.stdin.drain() - await process.wait() - except TimeoutError: - await terminate(process, group=os.name != 'nt') + await write(request) + if querying: + process.stdin.close() + while line := await process.stdout.readline(): + record = json.loads(line) + if (not isinstance(record, dict) or not line.endswith(b'\n') + or record.get('schema_version') != '0.1' + or record.get('request_id') != request['request_id'] + or record.get('sequence') != sequence or result is not None): + raise ValueError('Invalid Jiuwen CLI record') + sequence += 1 + if record.get('session_id') is not None: + if session is not None and session != record['session_id']: + raise ValueError('Jiuwen session changed') + session = record['session_id'] + if record['type'] == ('query_result' if querying else 'result'): + result = record + elif record['type'] == 'event' and not querying: + await on_event(record) + if cancel is not None and cancel.is_set(): + continue + if record['event_type'] == 'host_tool.requested': + value = await on_tool_call(record) + await write({'schema_version': '0.1', 'type': 'tool_result', + 'request_id': request['request_id'], 'session_id': session, + 'call_id': record['payload']['call_id'], 'result': value}) + elif record['event_type'] == 'interaction.requested': + raise ProjectError('JiuwenSwarm 需要尚未支持的用户交互，执行已中断。', 502) + else: + raise ValueError('Unexpected Jiuwen CLI record') + code = await process.wait() + await stderr + if result is None or result.get('exit_code') != code: + raise ValueError('Jiuwen CLI missing terminal result or exit code mismatch') + return result finally: - drain.cancel() - await asyncio.gather(drain, return_exceptions=True) + if watcher is not None: + watcher.cancel() + await asyncio.gather(watcher, return_exceptions=True) + process.stdin.close() + await terminate(process, group=True) + stderr.cancel() + await asyncio.gather(stderr, return_exceptions=True) + + +async def query(operation: str) -> dict: + try: + with TemporaryDirectory(prefix='od-jiuwen-query-') as temporary: + async with asyncio.timeout(30): + result = await execute({'schema_version': '0.1', 'type': 'query', + 'request_id': str(uuid4()), 'operation': operation, 'params': {}}, cwd=temporary) + except (ValueError, KeyError, TypeError, TimeoutError, OSError) as error: + raise ProjectError('JiuwenSwarm 查询失败，请检查 CLI 版本与本机配置。', 503) from error + if result['status'] != 'completed': + raise ProjectError('JiuwenSwarm 查询失败，请检查本机配置。', 503) + if operation == 'protocol.capabilities': + data = result['data'] + if data['schema_version'] != '0.1' or not all(data['features'].get(name) for name in ('host_tools', 'cancel', 'run_skills', 'custom_agent')): + raise ProjectError('JiuwenSwarm CLI 协议不支持所需能力，请更新 CLI。', 503) + return result['data'] class JiuwenSwarmRuntime: - async def stream(self, run, gateway): - if set(run.selected_skills) - {item['name'] for item in run.skills}: - raise ProjectError('技能已失效，请重新选择。', 409) + async def stream(self, run: RunContext, gateway: ToolGateway): if run.thinking != 'auto': - raise ProjectError('JiuwenSwarm 当前使用其配置中的推理强度，请选择 Default 强度。', 409) - await capabilities() - work = run.directory / 'agent-runs' / str(run.turn_id) / 'work' + raise ProjectError('JiuwenSwarm 的思考级别由本机模型配置管理，请选择自动。') + await query('protocol.capabilities') + work = (run.directory / 'agent-runs' / str(run.turn_id) / 'work').resolve() work.mkdir(parents=True, exist_ok=False) - roots = await asyncio.to_thread(prepare_skills, work, run.skills) - gateway.roots = roots - functions = [item['function'] for item in tools()] - allowed = [item['name'] for item in functions] + list(NATIVE_TOOLS) - prompt = (f'Execution directory: {work.resolve()}\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') - if run.selected_skills: - prompt += 'Before proceeding, use skill_tool to load these selected skills: ' + ', '.join(run.selected_skills) + '.\n' - prompt += json.dumps(run.messages, ensure_ascii=False) - request_id = str(run.turn_id) - request = dict(schema_version='0.1', type='run', request_id=request_id, mode='agent.work.normal', - input=prompt, timeout_seconds=RUN_TIMEOUT, - agent=dict(name='opendirector', instructions=SYSTEM, tools=allowed), - host_tools=[dict(name=item['name'], description=item['description'], input_schema=item['parameters']) for item in functions], - permissions={'tools': {name: 'allow' for name in allowed}}, skills=run.selected_skills, mcp=[], - workspace=dict(cwd=str(work.resolve()), project_dir=str(work.resolve()), trusted_dirs=[str(work.resolve())])) - if run.runtime_model: - request['model'] = run.runtime_model - payload = encode(request) - with execution_environment(roots) as env: - process = await _start(['--run-jsonl'], work, env) - diagnostics = asyncio.create_task(_stderr(process.stderr)) - records, emitted = Records(request_id), '' - native_calls = {} - trace_path = work.parent.parent / f'{run.turn_id}.jiuwenswarm.jsonl' - try: - async with asyncio.timeout(RUN_TIMEOUT + 30): - process.stdin.write(payload) - await process.stdin.drain() - with trace_path.open('a', encoding='utf-8') as trace: - while line := await process.stdout.readline(): - record = records.accept(line) - if record['type'] == 'result': - continue - kind, value = record['event_type'], record.get('payload', {}) - if kind not in {'chat.delta', 'chat.tool_call', 'chat.tool_result', 'interaction.requested', 'host_tool.requested'}: - continue - if not isinstance(value, dict): - raise protocol_error() - if kind in {'chat.tool_call', 'chat.tool_result'}: - call_id = (value.get('tool') or {}).get('call_id') if isinstance(value.get('tool'), dict) else None - event = native_tool_event(kind, value, native_calls.get(call_id) if isinstance(call_id, str) else None) - if event is not None: - native_calls[call_id] = event - trace.write(json.dumps(event, ensure_ascii=False) + '\n') - trace.flush() - yield event - elif kind == 'chat.delta': - text = value.get('text') - if not isinstance(text, str): - raise protocol_error() - emitted += text - yield {'type': 'delta', 'text': text} - elif kind == 'interaction.requested': - raise ProjectError('JiuwenSwarm 需要交互确认，当前助手暂不支持；本轮已停止，未自动批准。', 409) - elif kind == 'host_tool.requested': - call_id, name, arguments = value.get('call_id'), value.get('name'), value.get('arguments') - if not isinstance(call_id, str) or not call_id or not isinstance(name, str) or not isinstance(arguments, dict) or records.session is None: - raise protocol_error() - event_id = 'jiuwenswarm:' + call_id - yield dict(type='tool', id=event_id, name=name, status='running', arguments=arguments) - outcome = await gateway.call(event_id, name, arguments) - detail = json.dumps(outcome.value, ensure_ascii=False) - event = {**outcome.event(name), 'id': event_id, 'arguments': arguments, - 'result': detail[:65536], 'resultTruncated': len(detail) > 65536} - trace.write(json.dumps(event, ensure_ascii=False) + '\n') - trace.flush() - yield event - response = dict(schema_version='0.1', type='tool_result', request_id=request_id, - session_id=records.session, call_id=call_id) - response.update({'error': detail} if outcome.error else {'result': outcome.value}) - process.stdin.write(encode(response)) - await process.stdin.drain() - result = records.finish(await process.wait()) - await diagnostics - trace.write(json.dumps(dict(type='result', status=result['status'], exit_code=result['exit_code'])) + '\n') - output = result.get('output') or '' - if output and not output.startswith(emitted): - raise protocol_error() - if len(output) > len(emitted): - yield {'type': 'delta', 'text': output[len(emitted):]} - except (TimeoutError, ValueError, BrokenPipeError, ConnectionResetError) as error: - raise ProjectError('JiuwenSwarm 执行中断，已接受的业务操作会保留，请检查回执后继续。', 502) from error - finally: - gateway.closed = True - cleanup = asyncio.create_task(_close(process, request_id)) + gateway.roots = prepare_skills(work, run.skills) + queue = asyncio.Queue(maxsize=128) + cancel = asyncio.Event() + stopping = False + emitted_chars = 0 + terminal_written = False + with TemporaryDirectory(prefix='od-jiuwen-') as temporary: + state = Path(temporary) + config = state / 'config' + config.mkdir() + # Keep credentials out of published execution files and native traces. + home = Path(os.environ.get('JIUWENSWARM_HOME', Path.home())) + source = Path(os.environ.get('JIUWENSWARM_DATA_DIR', home / '.jiuwenswarm')) / 'config' + for name in ('config.yaml', '.env'): + if (source / name).is_file(): + shutil.copyfile(source / name, config / name) + (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) + for root in gateway.roots: + shutil.copytree(root, skill_root / root.name) + gateway.roots += list(skill_root.iterdir()) + functions = [item['function'] for item in tools()] + host_tools = [{'name': item['name'], 'description': item['description'], + 'input_schema': item['parameters']} for item in functions] + request = {'schema_version': '0.1', 'type': 'run', 'request_id': str(run.turn_id), + 'input': (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 intermediate files and outputs into the execution directory. ' + 'Use OpenDirector business tools for project changes.\n' + + json.dumps(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 run.runtime_model: + request['model'] = run.runtime_model + trace_path = work.parent / f'{run.turn_id}.jiuwenswarm.jsonl' + with trace_path.open('a', encoding='utf-8') as trace: + async def emit(event): + if not stopping: + await queue.put(event) + + async def observe(record): + nonlocal emitted_chars + trace.write(json.dumps({'native': record}, ensure_ascii=False) + '\n') + trace.flush() + if record['event_type'] == 'chat.delta': + text = record['payload']['text'] + emitted_chars += len(text) + if text: + await emit({'type': 'delta', 'text': text}) + elif record['event_type'] in {'chat.tool_call', 'chat.tool_result'}: + tool = record['payload']['tool'] + if tool['name'] not in gateway.names: + detail = json.dumps(tool['result'], ensure_ascii=False) + await emit({'type': 'tool', 'id': 'native:' + tool['call_id'], 'name': tool['name'], + 'status': {'started': 'running', 'completed': 'complete', 'failed': 'failed'}[tool['status']], + 'arguments': tool['arguments'], 'result': detail[:65536], + 'resultTruncated': len(detail) > 65536}) + + async def call(record): + payload = record['payload'] + name, arguments = payload['name'], payload['arguments'] + call_id = payload['call_id'] + await emit({'type': 'tool', 'id': call_id, 'name': name, 'status': 'running', 'arguments': arguments}) + outcome = await gateway.call(call_id, name, arguments) + detail = json.dumps(outcome.value, ensure_ascii=False, indent=2) + event = {**outcome.event(name), 'id': call_id, 'arguments': arguments, + 'result': detail[:65536], 'resultTruncated': len(detail) > 65536} + trace.write(json.dumps(event, ensure_ascii=False) + '\n') + trace.flush() + await emit(event) + return outcome.value + + task = asyncio.create_task(execute(request, cwd=str(work), env={'JIUWENSWARM_DATA_DIR': str(state), + 'JIUWENSWARM_CONFIG_DIR': str(config)}, on_event=observe, on_tool_call=call, cancel=cancel)) + pending = None try: - await asyncio.shield(cleanup) - except asyncio.CancelledError: - await cleanup - raise + while True: + pending = asyncio.create_task(queue.get()) + done, _ = await asyncio.wait((task, pending), return_when=asyncio.FIRST_COMPLETED) + if pending in done: + yield pending.result() + if task in done: + break + while not queue.empty(): + yield queue.get_nowait() + result = task.result() + trace.write(json.dumps({'native': result}, ensure_ascii=False) + '\n') + terminal_written = True + if result['status'] != 'completed': + raise ProjectError(f"JiuwenSwarm 执行未完成（{result['status']}），请检查原生轨迹。", 502) + # The terminal output includes the deltas and any unflushed final tail. + remaining = (result.get('output') or '')[emitted_chars:] + if remaining: + yield {'type': 'delta', 'text': remaining} + except (ValueError, KeyError, TypeError, TimeoutError, OSError) as error: + raise ProjectError('JiuwenSwarm 执行失败或需要尚未支持的用户交互，请检查原生轨迹。', 502) from error finally: - diagnostics.cancel() - await asyncio.gather(diagnostics, return_exceptions=True) + stopping = True + if pending is not None: + pending.cancel() + await asyncio.gather(pending, return_exceptions=True) + cancel.set() + if not task.done(): + try: + await asyncio.wait_for(asyncio.shield(task), 10) + except (TimeoutError, ValueError, KeyError, TypeError, OSError): + task.cancel() + with suppress(asyncio.CancelledError, ValueError, KeyError, TypeError, OSError): + result = await task + if not terminal_written: + trace.write(json.dumps({'native': result}, ensure_ascii=False) + '\n') diff --git a/src/backend/workspace.ts b/src/backend/workspace.ts index efb59de9..ea1c6eec 100644 --- a/src/backend/workspace.ts +++ b/src/backend/workspace.ts @@ -8,6 +8,7 @@ import { creationClient, creationPath, objectVersion, + visibleItems, } from "./creationClient.ts"; import { CreationProjection } from "./creationProjection.ts"; import { @@ -180,7 +181,13 @@ export function openExampleWorkspaceProject( } return document; }; - if (existing) return remember(await openWorkspaceProject(existing)); + if (existing) { + const saved = await request(creationPath(existing.id)); + if (visibleItems(saved).size === 0) { + return remember(await importWorkspaceDocument(await createExample())); + } + return remember(await openWorkspaceProject(existing)); + } const blocked = unreadable.find((project) => project.id === preferred); if (blocked) throw new Error(blocked.reason); const example = await createExample(); diff --git a/tests/backendWorkspace.test.ts b/tests/backendWorkspace.test.ts index e49f6db0..b022771d 100644 --- a/tests/backendWorkspace.test.ts +++ b/tests/backendWorkspace.test.ts @@ -88,6 +88,11 @@ function fixture(t: TestContext) { return Response.json({ ok: true }); } const id = path.split("/")[4]!; + if (method === "GET" && path === `/api/creation/projects/${id}` && documents.has(id)) + return Response.json({canvasItems: [ + ...(documents.get(id)?.nodes ?? []).map(node => ({...node, visible: true})), + ...(documents.get(id)?.hiddenNodes ?? []).map(node => ({...node, visible: false})), + ]}); if (path.endsWith("/commands")) { if (offline) throw new TypeError("offline"); if ( @@ -285,7 +290,8 @@ test("existing default example is opened without a stored hint or saving over it test("renamed examples reopen by project identity and deleted examples are recreated", async (t) => { const f = fixture(t); - const prepare = async () => createDocument(EXAMPLE_PROJECT_TITLE); + const prepare = async () => ({...createDocument(EXAMPLE_PROJECT_TITLE), + nodes: [createNode("text", undefined, {text: "保留的示例"})]}); const initial = await openExampleWorkspaceProject( EXAMPLE_PROJECT_TITLE, prepare, @@ -344,8 +350,11 @@ test("an unreadable existing example fails without creating a replacement", asyn const prepare = async () => createDocument(EXAMPLE_PROJECT_TITLE); await openExampleWorkspaceProject(EXAMPLE_PROJECT_TITLE, prepare); f.calls.length = 0; - CreationProjection.prototype.open = async () => { - throw new Error("示例无法读取"); + const fetch = globalThis.fetch; + globalThis.fetch = async (url, options) => { + if (String(url).startsWith("/api/creation/projects/") && !options?.method) + return Response.json({detail: "示例无法读取"}, {status: 503}); + return fetch(url, options); }; await assert.rejects( openExampleWorkspaceProject(EXAMPLE_PROJECT_TITLE, prepare), @@ -370,3 +379,35 @@ test("unknown example creation is not retried with another project identity", as ); assert.equal(f.projects.size, 0); }); + + +test("an empty saved example creates a separate usable example without overwriting it", async (t) => { + const f = fixture(t); + const old = await createWorkspaceProject(EXAMPLE_PROJECT_TITLE, crypto.randomUUID()); + const original = structuredClone(f.documents.get(old.id)); + const reopened = await openExampleWorkspaceProject(EXAMPLE_PROJECT_TITLE, async () => ({ + ...createDocument(EXAMPLE_PROJECT_TITLE), + nodes: [createNode("text", undefined, {text: "示例内容"})], + })); + assert.notEqual(reopened.id, old.id); + assert.equal(reopened.nodes.length, 1); + assert.equal(reopened.title, EXAMPLE_PROJECT_TITLE + " (2)"); + assert.deepEqual(f.documents.get(old.id), original); + assert.equal(f.projects.size, 2); +}); + + +test("a numbered example with only hidden nodes creates a fresh example and preserves hidden content", async (t) => { + const f = fixture(t); + const prepare = async () => ({...createDocument(EXAMPLE_PROJECT_TITLE), + nodes: [createNode("text", undefined, {text: "示例"})]}); + const initial = await openExampleWorkspaceProject(EXAMPLE_PROJECT_TITLE, prepare); + const name = EXAMPLE_PROJECT_TITLE + " (9)"; + f.projects.set(initial.id, {...f.projects.get(initial.id)!, name}); + const hidden = {...initial, title: name, nodes: [], hiddenNodes: initial.nodes}; + f.documents.set(initial.id, hidden); + const reopened = await openExampleWorkspaceProject(EXAMPLE_PROJECT_TITLE, prepare); + assert.notEqual(reopened.id, initial.id); + assert.equal(reopened.nodes.length, 1); + assert.deepEqual(f.documents.get(initial.id), hidden); +}); diff --git a/tests/test_jiuwenswarm_media.py b/tests/test_jiuwenswarm_media.py index e29776b7..f4d9967c 100644 --- a/tests/test_jiuwenswarm_media.py +++ b/tests/test_jiuwenswarm_media.py @@ -93,8 +93,7 @@ def reference(asset): def reply(request): index = len(requests) - 1 if index: - last = request['messages'][-1] - assert last['role'] == 'tool' + last = next(message for message in reversed(request['messages']) if message['role'] == 'tool') result = unpack(last['content']) detail = json.dumps(result) previous = steps[index - 1][0] @@ -192,10 +191,10 @@ def do_POST(self): source_ref = reference(source) snapshot_skill = work / '.agents/skills/media-fixture' steps.extend([ - ('skill_tool', {'skill_name':'media-fixture'}), - ('skill_tool', {'skill_name':'media-fixture', 'relative_file_path':'media.json'}), + ('read_file', {'file_path':str(snapshot_skill / 'SKILL.md')}), + ('read_file', {'file_path':str(snapshot_skill / 'media.json')}), ('read_file', {'file_path':str(snapshot_skill / 'process.py')}), - ('write_file', {'file_path':str(work / 'manifest.json'), 'content':'{"created_by":"native_write_file"}'}), + ('bash', {'command':shlex.join([sys.executable, '-c', 'from pathlib import Path; Path(' + repr(str(work / 'manifest.json')) + ').write_text(' + repr(json.dumps({'created_by':'native_bash'})) + ')'])}), ('bash', {'command':'printf INTENTIONAL_MEDIA_FAILURE; exit 7'}), ('bash', {'command':shlex.join([sys.executable, str(snapshot_skill / 'process.py'), str(source_path), str(work)])}), ('command_schema', {'type':'artifact.publish'}), @@ -229,11 +228,11 @@ def do_POST(self): assert video['metadata']['frameRate'] == 24 and video['metadata']['durationSeconds'] == pytest.approx(1, abs=.05) probe = json.loads(subprocess.check_output(['ffprobe', '-v', 'error', '-show_streams', '-of', 'json', str(directory / video['path'])])) assert probe['streams'][0]['codec_name'] == 'h264' - assert json.loads((work / 'manifest.json').read_text())['created_by'] == 'native_write_file' + assert json.loads((work / 'manifest.json').read_text())['created_by'] == 'native_bash' assert sha256(source_path.read_bytes()).hexdigest() == original_hash assert {p.name:sha256(p.read_bytes()).hexdigest() for p in skill.iterdir()} == skill_hashes activities = events[-1]['conversation']['messages'][-1]['tools'] - assert {'skill_tool', 'read_file', 'write_file', 'bash', 'execute_command'} <= {item['name'] for item in activities} + assert {'read_file', 'bash', 'execute_command'} <= {item['name'] for item in activities} assert any(item['name'] == 'bash' and item['status'] == 'failed' for item in activities) assert sum(item['name'] == 'execute_command' for item in activities) == 4 assert client.delete('/api/creation/sessions/' + session).status_code == 200 @@ -246,7 +245,7 @@ def do_POST(self): for asset in outputs: media = reopened.get(root + '/assets/' + asset['id'] + '/content') assert media.status_code == 200 and media.content == (work / asset['name']).read_bytes() - trace = (directory / 'agent-runs' / f'{turn}.jiuwenswarm.jsonl').read_text() + trace = (directory / 'agent-runs' / turn / f'{turn}.jiuwenswarm.jsonl').read_text() assert 'media-fixture-key' not in trace assert 'MEDIA_PROCESSING_FINISHED' in trace finally: diff --git a/tests/test_jiuwenswarm_runtime.py b/tests/test_jiuwenswarm_runtime.py index abb39448..97f7436e 100644 --- a/tests/test_jiuwenswarm_runtime.py +++ b/tests/test_jiuwenswarm_runtime.py @@ -1,7 +1,6 @@ -"""Real child pipes and authoritative project writes; model execution is simulated.""" import asyncio +from contextlib import aclosing import json -import os from pathlib import Path import sys from uuid import uuid4 @@ -9,339 +8,177 @@ import pytest from fastapi.testclient import TestClient -from opendirector.agent import jiuwenswarm as swarm +from opendirector.agent import jiuwenswarm as jiuwen from opendirector.agent.catalog import model_choices from opendirector.agent.context import RunContext -from opendirector.agent.runtime import AgentRuntimes, runtime_readiness +from opendirector.agent.runtime import AgentRuntimes from opendirector.domain.projects import ProjectError -from opendirector.infrastructure.creation_projects import CreationProjects from opendirector.infrastructure.local_projects import LocalProjects -from opendirector.infrastructure.model_configs import ModelConfigs from opendirector_api.main import create_app -FAKE = r''' -import json, os, sys + +FAKE = r'''import json, os, sys, time from pathlib import Path request = json.loads(sys.stdin.readline()) -request_id = request['request_id'] -session = None -sequence = 0 -mode = os.getenv('SWARM_TEST_MODE', 'success') +assert request['schema_version'] == '0.1' +seq = 0 +sid = None +mode = os.environ.get('JIUWEN_TEST', 'success') def emit(kind, **fields): - global sequence - record = dict(schema_version='0.1', type=kind, request_id=request_id, - session_id=session, sequence=sequence, **fields) - if mode == 'sequence': record['sequence'] += 1 - if mode == 'request': record['request_id'] = 'foreign' - print(json.dumps(record), flush=True) - sequence += 1 -if request['type'] == 'query': - operation = request['operation'] - data = dict(schema_version='0.1', features={name: True for name in - ('host_tools', 'cancel', 'stable_event_fields', 'custom_agent')}) if operation == 'protocol.capabilities' else dict(models=[ - dict(selection_key='local-model#0', display_name='Local model', is_default=True)]) - if mode == 'unsupported': data['features'] = {} - if mode == 'malformed_features': data['features'] = [] - if mode == 'unconfigured' and operation == 'model.list': data['models'] = [] - if mode == 'query_failed' and operation == 'model.list': - emit('query_result', operation=operation, status='failed', exit_code=1, error=dict(code='RUNTIME_FAILED')) - raise SystemExit(1) - emit('query_result', operation=operation, status='completed', exit_code=0, data=data, error=None) + global seq + print(json.dumps(dict(schema_version='0.1', type=kind, sequence=seq, + request_id=request['request_id'], session_id=sid, **fields)), flush=True) + seq += 1 +if sys.argv[1] == '--query-json': + if mode == 'bad_json': + print('invalid', flush=True) + raise SystemExit() + if mode == 'bad_sequence': seq = 2 + if mode == 'stderr': sys.stderr.write('x' * 262144) + data = {'schema_version':'0.1','features':dict.fromkeys(['host_tools','cancel','run_skills','custom_agent'], True)} if request['operation'] == 'protocol.capabilities' else {'models':[{'selection_key':'local#0','display_name':'Local model','is_default':True}]} + emit('query_result',operation=request['operation'],status='completed',exit_code=0,data=data,error=None) raise SystemExit() -assert '--run-jsonl' in sys.argv -assert request['model'] == 'local-model#0' -assert request['agent']['tools'][:4] == ['read_project', 'read', 'command_schema', 'execute_command'] -assert {'skill_tool', 'bash', 'read_file', 'write_file'} <= set(request['agent']['tools']) -assert request['workspace']['cwd'] == str(Path.cwd()) -assert request['skills'] == [] and request['mcp'] == [] -assert 'thinking' not in request -session = 'owned-session' -emit('event', event_type='run.started', payload={}) -emit('event', event_type='future.observation', payload=['unknown payload']) -if mode == 'native': - for kind, status in [('chat.tool_call', 'started'), ('chat.tool_result', 'failed')]: - emit('event', event_type=kind, payload={'tool':dict(call_id='native-check', name='bash', arguments={'command':'false'}, status=status, result={'success':False, 'error':'test failure'})}) -if mode == 'cancel': - Path(os.environ['SWARM_STARTED']).write_text(str(os.getpid())) - control = json.loads(sys.stdin.readline()) - assert control == dict(schema_version='0.1', type='cancel', request_id=request_id) - Path(os.environ['SWARM_CANCELLED']).write_text('cancelled') - emit('result', status='cancelled', exit_code=130, error={'code':'CANCELLED'}) - raise SystemExit(130) +assert sys.argv[1:] == ['--run-jsonl'] +assert not {'mode', 'permissions', 'timeout_seconds'} & request.keys() +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' 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 mode == 'interaction': - emit('event', event_type='interaction.requested', payload={'interaction_id':'permission'}) - control = json.loads(sys.stdin.readline()) - assert control['type'] == 'cancel' + emit('event',event_type='interaction.requested',payload={'interaction_id':'ask-1','interaction':{'source':'ask_user','questions':[{'question':'Choose a source'}]}}) + sys.stdin.readline() + raise SystemExit() +if mode == 'wait': + emit('event',event_type='chat.delta',payload={'text':'Waiting'}) + control=json.loads(sys.stdin.readline()) + assert control['type']=='cancel' + emit('result',status='cancelled',exit_code=130,error={'code':'CANCELLED'},output='Waiting') raise SystemExit(130) -arguments = dict(type='asset.writeText', payload=dict(name='swarm.md', text='Saved through shared tools')) -for index in range(2): - emit('event', event_type='host_tool.requested', payload=dict(call_id='write-once', name='execute_command', arguments=arguments)) - if mode == 'pending': - control = json.loads(sys.stdin.readline()) - assert control['type'] == 'cancel' - Path(os.environ['SWARM_CANCELLED']).write_text('cancelled') - emit('result', status='cancelled', exit_code=130, error={'code':'CANCELLED'}) - raise SystemExit(130) - result = json.loads(sys.stdin.readline()) - assert result['session_id'] == session and result['request_id'] == request_id - assert result['result']['status'] == 'applied' - if index == 0: first = result - else: assert result == first -if mode == 'conflict': - arguments['payload']['text'] = 'Conflicting input' - emit('event', event_type='host_tool.requested', payload=dict(call_id='write-once', name='execute_command', arguments=arguments)) - assert 'error' in json.loads(sys.stdin.readline()) -emit('event', event_type='chat.delta', payload={'text':'Saved.'}) -emit('event', event_type='chat.final', payload={'text':'Saved.'}) -if mode == 'missing': raise SystemExit() -if mode == 'session': session = 'foreign-session' -code = 1 if mode == 'failed' else 0 -emit('result', status='failed' if code else 'completed', exit_code=code, - output='Saved.', error={'code':'TEST_FAILURE'} if code else None) -if mode == 'duplicate': emit('result', status='completed', exit_code=0, error=None) -raise SystemExit(1 if mode == 'exit' else code) +emit('event',event_type='host_tool.requested',payload={'call_id':'write-1','name':'execute_command','arguments':{'type':'asset.writeText','payload':{'name':'plan.md','text':'Native plan'}}}) +control=json.loads(sys.stdin.readline()) +assert control['type']=='tool_result' +assert control['result']['status']=='applied',control +emit('event',event_type='chat.tool_call',payload={'tool':{'call_id':'skill-1','name':'skill_tool','arguments':{'skill_name':'example'},'status':'started','result':None}}) +emit('event',event_type='chat.tool_result',payload={'tool':{'call_id':'skill-1','name':'skill_tool','arguments':None,'status':'completed','result':'Skill body'}}) +delta = '' if mode == 'final_only' else 'Saved.' +if mode in ('final_only', 'separate_final'): + final = 'Final answer.' +elif mode == 'full_final': + final = 'Saved.Final answer.' +else: + final = 'Saved.' +if delta: emit('event',event_type='chat.delta',payload={'text':delta}) +emit('event',event_type='chat.final',payload={'text':final}) +if mode == 'partial': raise SystemExit() +output = final if final.startswith(delta) else delta + final +emit('result',status='completed',exit_code=0,error=None,output=output) ''' @pytest.fixture def cli(tmp_path, monkeypatch): - from contextlib import nullcontext - monkeypatch.setattr(swarm, 'execution_environment', lambda roots: nullcontext(None)) - import opendirector_api.agent as api_agent - async def empty_skills(*args): - return [] - monkeypatch.setattr(api_agent, 'skills', empty_skills) - path = tmp_path / 'jiuwenswarm-process' - path.write_text(f'#!{sys.executable}\n' + FAKE) - path.chmod(0o755) - monkeypatch.setattr(swarm, 'executable', lambda: str(path)) - return path - - -def test_native_tool_activity_and_host_deduplication(cli, tmp_path, monkeypatch): - monkeypatch.setenv('SWARM_TEST_MODE', 'native') - with TestClient(create_app(LocalProjects(tmp_path)), headers={'X-OpenDirector-Request': '1', 'X-OpenDirector-Session': str(uuid4())}) as client: - root = create(client) - final = events(client.post(root + '/assistant/turns', json=dict(id=str(uuid4()), - runtime='jiuwenswarm', runtimeModel='local-model#0', text='Save a plan')))[-1] - assert final['error'] is None - activities = final['conversation']['messages'][-1]['tools'] - assert len(activities) == 2 - assert activities[0]['name'] == 'bash' and activities[0]['status'] == 'failed' - assert 'test failure' in activities[0]['result'] - assert activities[1]['name'] == 'execute_command' and activities[1]['status'] == 'complete' - - -@pytest.mark.parametrize('payload', [{}, {'tool': {'name':'bash', 'call_id':'id', 'arguments':{}, 'status':'unknown'}}, - {'tool': {'name':'bash', 'call_id':'', 'arguments':{}, 'status':'started'}}]) -def test_malformed_native_tool_activity(payload): - with pytest.raises(ProjectError, match='协议'): - swarm.native_tool_event('chat.tool_call', payload) - - -def test_host_tool_observations_do_not_duplicate_gateway_receipts(): - assert swarm.native_tool_event('chat.tool_call', {'tool':{'name':'execute_command'}}) is None - - -def test_native_completion_preserves_start_arguments(): - started = swarm.native_tool_event('chat.tool_call', {'tool':dict(name='bash', call_id='run', - arguments='{"command":"ffprobe clip.mp4"}', status='started')}) - completed = swarm.native_tool_event('chat.tool_result', {'tool':dict(name='bash', call_id='run', - arguments=None, status='completed', result='ok')}, started) - assert completed['arguments'] == started['arguments'] == {'command':'ffprobe clip.mp4'} - - -def test_native_skill_library_uses_copies_and_cleans_private_config(tmp_path, monkeypatch): - data = tmp_path / 'native' - (data / 'config').mkdir(parents=True) - (data / 'config/config.yaml').write_text('model: native-secret') - (data / 'config/.env').write_text('MODEL_KEY=native-secret') - skill = tmp_path / 'project/agent-runs/turn/work/.agents/skills/example' - skill.mkdir(parents=True) - (skill / 'SKILL.md').write_text('Skill snapshot') - monkeypatch.setenv('JIUWENSWARM_DATA_DIR', str(data)) - with swarm.execution_environment([skill]) as env: - private = Path(env['JIUWENSWARM_DATA_DIR']) - assert not private.is_relative_to(tmp_path / 'project') - assert (private / 'config/config.yaml').read_text() == 'model: native-secret' - assert (private / 'config/config.yaml').stat().st_mode & 0o777 == 0o600 - mapped = private / 'agent/workspace/skills/example/SKILL.md' - assert mapped.read_text() == 'Skill snapshot' and not mapped.is_symlink() - mapped.write_text('Native local change') - assert not private.exists() - assert (skill / 'SKILL.md').read_text() == 'Skill snapshot' - assert (data / 'config/config.yaml').read_text() == 'model: native-secret' - assert not list((tmp_path / 'project').rglob('config.yaml')) - - -def test_local_engine_discovery_and_data_directory(tmp_path, monkeypatch): - monkeypatch.delenv('OPENDIRECTOR_JIUWENSWARM_COMMAND', raising=False) - monkeypatch.delenv('JIUWENSWARM_DATA_DIR', raising=False) - monkeypatch.setattr(swarm, 'ROOT', tmp_path) - monkeypatch.setattr(swarm.shutil, 'which', lambda _: None) - local = tmp_path / '.toolchain/jiuwenswarm' - cli = local / 'venv/bin/jiuwenswarm-process' - cli.parent.mkdir(parents=True) - cli.write_text('#!/bin/sh\n') - cli.chmod(0o755) - (local / 'run').write_text('legacy wrapper') - (local / 'run').chmod(0o755) - assert swarm.executable() == str(cli) - assert swarm.data_directory(str(cli)) == local / 'data' - - -def create(client): - pid = str(uuid4()) - root = '/api/creation/projects/' + pid - assert client.post(root + '/commands', json=dict(type='project.create', projectId=pid, - commandId=str(uuid4()), payload=dict(name='Swarm fixture'))).status_code == 200 - assert client.post(root + '/open').status_code == 200 - return root - - -def events(response): - return [json.loads(line[6:]) for line in response.text.splitlines() if line.startswith('data: ')] - - -@pytest.mark.parametrize('mode', ['success', 'conflict', 'missing', 'session', 'duplicate', 'failed', 'exit']) -def test_shared_write_replay_provenance_and_failure_preserve_commits(cli, tmp_path, monkeypatch, mode): - monkeypatch.setenv('SWARM_TEST_MODE', mode) - local = LocalProjects(tmp_path) - turn_id = str(uuid4()) - with TestClient(create_app(local), headers={'X-OpenDirector-Request': '1', 'X-OpenDirector-Session': str(uuid4())}) as client: - root = create(client) - response = client.post(root + '/assistant/turns', json=dict(id=turn_id, - runtime='jiuwenswarm', runtimeModel='local-model#0', text='Save a plan')) - assert response.status_code == 200, response.text - stream = events(response) - final = stream[-1] - assert (final['error'] is None) == (mode in {'success', 'conflict'}) - assert final['conversation']['messages'][-1]['text'] == 'Saved.' - snapshot = client.get(root).json() - assert len(snapshot['assets']) == 1 - asset = snapshot['assets'][0] - assert asset['provenance']['agentRuntime'] == 'jiuwenswarm' - assert asset['provenance']['agentRunId'] == turn_id - assert asset['provenance']['modelSnapshot'] is None - assert (local.root / 'Swarm fixture' / asset['path']).read_text() == 'Saved through shared tools' - assert client.get(root + '/changes').json()['events'][-1]['source'] == 'agent' - activities = final['conversation']['messages'][-1]['tools'] - assert len(activities) == 1 - assert activities[0]['status'] == ('failed' if mode == 'conflict' else 'complete') - - -def test_catalog_readiness_and_early_rejection(cli, tmp_path, monkeypatch): - assert asyncio.run(runtime_readiness('jiuwenswarm')) == 'ready' - assert asyncio.run(model_choices('jiuwenswarm')) == [dict(id='local-model#0', label='Local model', efforts=[], isDefault=True)] - monkeypatch.setenv('SWARM_TEST_MODE', 'unsupported') - assert asyncio.run(runtime_readiness('jiuwenswarm')) == 'unsupported_version' - monkeypatch.setenv('SWARM_TEST_MODE', 'malformed_features') - assert asyncio.run(runtime_readiness('jiuwenswarm')) == 'unsupported_version' - monkeypatch.setenv('SWARM_TEST_MODE', 'query_failed') - assert asyncio.run(runtime_readiness('jiuwenswarm')) == 'unknown' - with pytest.raises(ProjectError, match='查询失败') as error: - asyncio.run(model_choices('jiuwenswarm')) - assert error.value.code == 'runtime_query_failed' - monkeypatch.setenv('SWARM_TEST_MODE', 'unconfigured') - assert asyncio.run(runtime_readiness('jiuwenswarm')) == 'needs_configuration' - with TestClient(create_app(LocalProjects(tmp_path)), headers={'X-OpenDirector-Request': '1', 'X-OpenDirector-Session': str(uuid4())}) as client: - root = create(client) - for options in [dict(skills=['character-replacement']), dict(thinking='high')]: - response = client.post(root + '/assistant/turns', json=dict(id=str(uuid4()), - runtime='jiuwenswarm', text='No write', **options)) - assert response.status_code == 409 - assert client.get(root + '/assistant').json()['messages'] == [] - draft = client.get(root + '/assistant').json() - assert client.put(root + '/assistant/draft', json=dict(conversationId=draft['conversationId'], - runtime='jiuwenswarm', harnessModels={'jiuwenswarm':'local-model#0'})).status_code == 200 - assert client.get(root + '/assistant').json()['harnessModels']['jiuwenswarm'] == 'local-model#0' - - -@pytest.mark.parametrize('mode', ['sequence', 'request']) -def test_malformed_query_identity_is_rejected(cli, monkeypatch, mode): - monkeypatch.setenv('SWARM_TEST_MODE', mode) - with pytest.raises(ProjectError, match='协议'): - asyncio.run(swarm.query('protocol.capabilities')) - - -@pytest.mark.parametrize('mode', ['cancel', 'interaction']) -def test_stop_sends_protocol_cancel_and_collects_child(cli, tmp_path, monkeypatch, mode): - started, cancelled = tmp_path / 'started', tmp_path / 'cancelled' - monkeypatch.setenv('SWARM_TEST_MODE', mode) - monkeypatch.setenv('SWARM_STARTED', str(started)) - monkeypatch.setenv('SWARM_CANCELLED', str(cancelled)) - runtimes = AgentRuntimes(CreationProjects(LocalProjects(tmp_path)), None, ModelConfigs(tmp_path)) - run = RunContext(runtime='jiuwenswarm', project_id=uuid4(), turn_id=uuid4(), - session=str(uuid4()), directory=tmp_path, config=None, messages=[], - selected_skills=[], runtime_model='local-model#0') - async def consume(): - return [event async for event in runtimes.stream(run)] - async def check(): - if mode == 'interaction': - with pytest.raises(ProjectError, match='交互确认'): - await consume() - return - task = asyncio.create_task(consume()) - async with asyncio.timeout(10): - while not started.exists(): - await asyncio.sleep(.02) - pid = int(started.read_text()) - task.cancel() + binary = tmp_path / 'jiuwenswarm-process' + binary.write_text(f'#!{sys.executable}\n' + FAKE) + binary.chmod(0o755) + monkeypatch.setattr(jiuwen, 'executable', lambda: str(binary)) + state = tmp_path / 'native' + (state / 'config').mkdir(parents=True) + (state / 'config/config.yaml').write_text('private-test-key') + monkeypatch.setenv('JIUWENSWARM_DATA_DIR', str(state)) + skill = tmp_path / 'example' + skill.mkdir() + (skill / 'SKILL.md').write_text('---\nname: example\ndescription: Test\n---\nRead resource.txt') + (skill / 'resource.txt').write_text('Native skill resource') + return skill + + +def test_catalog_cli_pipes(cli): + assert asyncio.run(model_choices('jiuwenswarm')) == [ + {'id':'local#0', 'label':'Local model', 'efforts':[], 'isDefault':True}] + + +@pytest.mark.parametrize(('mode', 'expected'), [('success', 'Saved.'), ('partial', 'Saved.'), + ('final_only', 'Final answer.'), ('separate_final', 'Saved.Final answer.'), + ('full_final', 'Saved.Final answer.')]) +def test_project_write_and_trace(cli, tmp_path, monkeypatch, mode, expected): + monkeypatch.setenv('JIUWEN_TEST', mode) + async def skills(*args): + return [{'name':'example','description':'Test','filePath':str(cli/'SKILL.md'),'source':'project'}] + monkeypatch.setattr('opendirector_api.agent.skills', skills) + monkeypatch.setattr('opendirector_api.agent.runtime_available', lambda _: True) + local = LocalProjects(tmp_path / 'project') + with TestClient(create_app(local), headers={'X-OpenDirector-Request':'1', 'X-OpenDirector-Session':str(uuid4())}) as client: + pid = str(uuid4()) + path = '/api/creation/projects/' + pid + assert client.post(path+'/commands', json={'type':'project.create','projectId':pid,'commandId':str(uuid4()),'payload':{'name':'Jiuwen'}}).status_code == 200 + assert client.post(path+'/open').status_code == 200 + tid = str(uuid4()) + response=client.post(path+'/assistant/turns', json={'id':tid,'runtime':'jiuwenswarm','text':'Save plan','skills':['example']}) + assert response.status_code == 200 + events=[json.loads(line[6:]) for line in response.text.splitlines() if line.startswith('data: ')] + assert (events[-1]['error'] is None) == (mode != 'partial'), events + assert events[-1]['conversation']['messages'][-1]['status'] == ('interrupted' if mode == 'partial' else 'complete') + message=events[-1]['conversation']['messages'][-1] + assert message['text']==expected + assert message['tools'][0]['status']=='complete' + assert message['tools'][1]['name']=='skill_tool' + assert message['tools'][1]['status']=='complete' + asset=client.get(path).json()['assets'][0] + assert asset['provenance']['agentRuntime']=='jiuwenswarm' + assert asset['provenance']['agentRunId']==tid + trace=next(local.root.rglob('*.jiuwenswarm.jsonl')).read_text() + assert 'native-session' in trace + assert 'private-test-key' not in trace + assert not list(local.root.rglob('config.yaml')) + + +def test_cancel_collects_child(cli, tmp_path, monkeypatch): + monkeypatch.setenv('JIUWEN_TEST','wait') + run=RunContext(runtime='jiuwenswarm',project_id=uuid4(),turn_id=uuid4(),session='test',directory=tmp_path/'project',config=None,messages=[],selected_skills=['example'],skills=[{'name':'example','filePath':str(cli/'SKILL.md')}]) + async def cancel(): + runtimes=AgentRuntimes(None,None,None) + async def consume(): + async with aclosing(runtimes.stream(run)) as stream: + async for event in stream: + assert event['text']=='Waiting' + asyncio.current_task().cancel() + await asyncio.sleep(0) with pytest.raises(asyncio.CancelledError): - await task - assert cancelled.read_text() == 'cancelled' - with pytest.raises(ProcessLookupError): - os.kill(pid, 0) - asyncio.run(check()) + await asyncio.wait_for(consume(),10) + asyncio.run(cancel()) + assert list(tmp_path.rglob('config.yaml')) == [tmp_path / 'native/config/config.yaml'] + trace = next(tmp_path.rglob('*.jiuwenswarm.jsonl')).read_text() + assert '"status": "cancelled"' in trace -def test_input_bounds_and_discovery(tmp_path, monkeypatch): - with pytest.raises(ProjectError, match='1 MiB'): - swarm.encode({'text': '字' * (1024 * 1024)}) - monkeypatch.setenv('OPENDIRECTOR_JIUWENSWARM_COMMAND', '/missing/swarm') - assert swarm.executable() is None +@pytest.mark.parametrize('mode', ['bad_json', 'bad_sequence']) +def test_query_rejects_invalid_protocol(cli, monkeypatch, mode): + monkeypatch.setenv('JIUWEN_TEST', mode) + with pytest.raises(ProjectError, match='查询失败'): + asyncio.run(jiuwen.query('protocol.capabilities')) -def test_rejects_duplicate_keys_and_unfinished_lines(): - records = swarm.Records('test') - for line in [b'{"sequence":0,"sequence":0}\n', b'{}', b'{"value":NaN}\n']: - with pytest.raises(ProjectError): - records.accept(line) +def test_query_drains_stderr(cli, monkeypatch): + monkeypatch.setenv('JIUWEN_TEST', 'stderr') + assert asyncio.run(jiuwen.query('protocol.capabilities'))['schema_version'] == '0.1' -def test_cancel_waits_for_an_accepted_write_and_preserves_its_receipt(cli, tmp_path, monkeypatch): - from opendirector.domain.creation.commands import parse_command - monkeypatch.setenv('SWARM_TEST_MODE', 'pending') - cancelled = tmp_path / 'cancelled' - monkeypatch.setenv('SWARM_CANCELLED', str(cancelled)) - store = CreationProjects(LocalProjects(tmp_path)) - pid, session, turn = uuid4(), str(uuid4()), uuid4() - assert store.execute(parse_command(dict(type='project.create', commandId=uuid4(), projectId=pid, - payload=dict(name='Accepted write')))).status == 'applied' - store.open(pid, session) - runtimes = AgentRuntimes(store, None, ModelConfigs(tmp_path)) - original = runtimes.business.execute - async def check(): - accepted, release = asyncio.Event(), asyncio.Event() - async def delayed(*args, **kwargs): - accepted.set() - await release.wait() - return await original(*args, **kwargs) - monkeypatch.setattr(runtimes.business, 'execute', delayed) - run = RunContext(runtime='jiuwenswarm', project_id=pid, turn_id=turn, session=session, - directory=store.root / 'Accepted write', config=None, messages=[], selected_skills=[], runtime_model='local-model#0') - async def consume(): - return [event async for event in runtimes.stream(run)] - task = asyncio.create_task(consume()) - await asyncio.wait_for(accepted.wait(), 10) - task.cancel() - async with asyncio.timeout(10): - while not cancelled.exists(): - await asyncio.sleep(.02) - assert not task.done() - release.set() - with pytest.raises(asyncio.CancelledError): - await task - asyncio.run(check()) - assert len(store.snapshot(pid).assets) == 1 - assert store.snapshot(pid).assets[0].provenance.agentRuntime == 'jiuwenswarm' +def test_interaction_is_recorded_without_approval(cli, tmp_path, monkeypatch): + monkeypatch.setenv('JIUWEN_TEST', 'interaction') + run = RunContext(runtime='jiuwenswarm', project_id=uuid4(), turn_id=uuid4(), session='test', + directory=tmp_path/'project', config=None, messages=[], selected_skills=['example'], + skills=[{'name':'example', 'filePath':str(cli/'SKILL.md')}]) + async def consume(): + runtimes = AgentRuntimes(None, None, None) + with pytest.raises(ProjectError, match='用户交互'): + async for _ in runtimes.stream(run): + pass + asyncio.run(consume()) + trace = next(tmp_path.rglob('*.jiuwenswarm.jsonl')).read_text() + assert 'interaction.requested' in trace and 'Choose a source' in trace