[feat] agent: unify queries and improve trace presentation

.github/workflows/ci.yml

@@ -72,6 +72,9 @@ jobs:
           curl --fail -H 'X-OpenDirector-Request: 1' "$CREATION_API_URL/api/creation-test-health"
           curl --silent --fail "$CREATION_WEB_URL/" >/dev/null
           node scripts/verify_creation_browser.mjs
+          node tests/browser/assistant_activity.mjs
+          node tests/browser/canvas_generation_reference.mjs
+          node tests/browser/created_node_focus.mjs
           node tests/browser/reference_reconnect.mjs
           node tests/browser/context_menu_dismissal.mjs
           node tests/browser/mp4_export.mjs

apps/api/src/opendirector_api/agent.py

@@ -252,6 +252,7 @@ async def turn(project_id: UUID, value: Turn, x_opendirector_session: UUID = Hea
             async def produce() -> None:
                 text, status, error, accepted = '', 'complete', None, False
                 activities = []
+                plan = None
                 try:
                     user = {'id': str(value.id), 'role': 'user', 'text': value.text, 'content': value.content,
                         'assets': [{'id': str(assets[key].id), 'name': assets[key].name, 'type': assets[key].metadata.mimeType,
@@ -282,6 +283,8 @@ async def produce() -> None:
                         async for event in stream:
                             if event['type'] == 'delta':
                                 text += event['text']
+                            if event['type'] == 'plan':
+                                plan = event['entries']
                             if event['type'] == 'tool':
                                 previous = next((item for item in activities if item['id'] == event['id']), None)
                                 activity = {**{key: item for key, item in event.items() if key != 'type'},
@@ -308,7 +311,8 @@ async def produce() -> None:
                         saved = await asyncio.to_thread(conversations.update, project_id, finish={
                             'id': str(uuid5(value.id, 'assistant')), 'role': 'assistant', 'text': text, 'status': status,
                             'model': model,
-                            'runtime': value.runtime, 'thinking': value.thinking, 'tools': activities}, conversation_id=value.conversationId)
+                            'runtime': value.runtime, 'thinking': value.thinking, 'tools': activities,
+                            **({'plan': plan} if plan is not None else {})}, conversation_id=value.conversationId)
                     else:
                         saved = None
                     if asyncio.current_task().cancelling():

apps/api/src/opendirector_api/creation.py

@@ -6,7 +6,7 @@
 from urllib.parse import quote, unquote
 
 from fastapi import APIRouter, Header, Query, Request
-from fastapi.responses import Response, StreamingResponse
+from fastapi.responses import FileResponse, Response, StreamingResponse
 from starlette.background import BackgroundTask
 from starlette.concurrency import run_in_threadpool
 
@@ -85,10 +85,22 @@ def projection(project_id: UUID):
         return canvas.projection(project_id)
 
     @router.post('/projects/{project_id}/queries')
-    def query(project_id: UUID, value: CreationQuery):
+    async def query(project_id: UUID, value: CreationQuery):
         if value.projectId != project_id:
             raise ProjectError('查询不属于当前工程。')
-        return canvas.query(value)
+        result = await canvas.query(value)
+        if value.type == 'asset.preview' and 'preview' in result:
+            result['preview'].pop('path')
+            result['preview']['url'] = f'/api/creation/projects/{project_id}/assets/{value.objectId}/preview'
+        return result
+
+    @router.get('/projects/{project_id}/assets/{asset_id}/preview')
+    async def asset_preview(project_id: UUID, asset_id: UUID):
+        target = projects.local.state_dir / 'previews' / str(project_id)
+        result = await run_in_threadpool(assets.preview, project_id, asset_id, target)
+        if 'preview' not in result:
+            return result
+        return FileResponse(result['preview']['path'], media_type=result['preview']['mimeType'])
 
     @router.get('/projects/{project_id}/history')
     def operations(project_id: UUID):

docs/architecture/agent-runtime.md

@@ -37,7 +37,7 @@ Default 与 OpenCode CLI 分别配置。外部 CLI 不读取 OpenDirector 保存
 
 适配层为每轮创建 `agent-runs/<turn-id>/work/`，启动 harness 并连接本轮 MCP 工具入口。原生工具以此目录为工作目录，通过继承的 PATH 查找 FFmpeg、ffprobe 等程序。
 
-业务工具分为工程查询 `read_project`、技能或文本素材读取 `read`、命令契约查询 `command_schema` 和命令执行 `execute_command`。Gateway 按调用身份保留任务和结果：相同身份、相同输入复用结果，不同输入拒绝执行。
+业务工具分为工程与模型概览 `read_project`、对象／状态／素材查询 `query`、技能或文本素材读取 `read`、命令契约查询 `command_schema` 和命令执行 `execute_command`。Gateway 按调用身份保留任务和结果：相同身份、相同输入复用结果，不同输入拒绝执行。
 
 新文件通过 `artifact.publish` 登记，再根据需要调用 `canvas.show` 展示。发布回执、素材登记和画布展示分别表达，不以创建节点代替文件交付。
 

docs/architecture/unified-project.md

@@ -110,6 +110,8 @@ UI、Agent 和后台完成写入共享可恢复变更通道。通知提示对象
 
 自动聚焦仅发生在节点切换或输入面板重新打开时。参数调整、保存回执、后台及 Agent 更新不抢走用户正在操作的控件焦点。输入框持有焦点时，方向键、删除与选择快捷键作用于文本；多选、拖动和无生成提示词的节点沿用画布交互。
 
+同步新增的画布节点进入可见区域时，视口可居中展示新节点，并避让侧栏与 AI 对话面板；批量新增按整体范围展示。跟随不修改节点位置、选择或输入焦点。用户拖动、缩放、多选或编辑时不自动移动视口，结束操作后不补做已跳过的跟随。已有节点的状态更新不触发重新定位。
+
 焦点属于 ViewState，不改变业务对象版本、生成意图或采用关系。
 
 ## 模型与渲染边界

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

@@ -166,6 +166,11 @@ def events(record, calls):
         return []
     update = record['params']['update']
     kind = update['sessionUpdate']
+    if kind == 'agent_thought_chunk':
+        return [{'type': 'thinking'}]
+    if kind == 'plan':
+        return [{'type': 'plan', 'entries': [
+            {'content': entry['content'], 'status': entry['status']} for entry in update['entries']]}]
     if kind == 'agent_message_chunk' and update['content']['type'] == 'text':
         return [{'type': 'delta', 'text': update['content']['text']}]
     if kind not in {'tool_call', 'tool_call_update'}:
@@ -181,4 +186,4 @@ def events(record, calls):
         'result': detail[:65536], 'resultTruncated': len(detail) > 65536}
     calls[identity] = event
     # Shared business activity is reported from authoritative gateway receipts.
-    return [] if any(name in {f'opendirector.{tool}', f'mcp__opendirector__{tool}'} for tool in ('read_project', 'read_canvas_item', 'preview_asset', 'read', 'command_schema', 'execute_command')) else [event]
+    return [] if any(name in {f'opendirector.{tool}', f'mcp__opendirector__{tool}'} for tool in ('read_project', 'query', 'read', 'command_schema', 'execute_command')) else [event]

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

@@ -61,7 +61,7 @@ async def execute(self, call_id: str, name: str, arguments: dict[str, JsonValue]
                     raise ProjectError('工具调用标识或名称无效。')
                 arguments = TypeAdapter(dict[str, JsonValue]).validate_python(arguments, strict=True)
                 schema = next(item['function']['parameters'] for item in tools() if item['function']['name'] == name)
-                if set(arguments) - set(schema['properties']) or set(schema.get('required', [])) - set(arguments):
+                if 'properties' in schema and (set(arguments) - set(schema['properties']) or set(schema.get('required', [])) - set(arguments)):
                     raise ProjectError('工具参数缺失或包含未定义字段。')
                 value = await self.business.execute(self.run.project_id, self.run.turn_id, self.run.session,
                     call_id, name, arguments, self.roots, self.run.config,
@@ -91,6 +91,7 @@ async def execute(self, call_id: str, name: str, arguments: dict[str, JsonValue]
     async def close(self) -> None:
         self.closed = True
         for identity, task in self.calls.values():
-            if json.loads(identity)[0] == 'read_canvas_item':
+            name, arguments = json.loads(identity)
+            if name == 'query' and arguments.get('type') == 'canvas.status':
                 task.cancel()
         await asyncio.gather(*(task for _, task in self.calls.values()), return_exceptions=True)

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

@@ -146,7 +146,7 @@ async def connection(reader, writer):
                         outcome = await gateway.call('mcp:' + key, params['name'], params.get('arguments', {}))
                         result = {'content': [{'type': 'text', 'text': json.dumps(outcome.value, ensure_ascii=False)}],
                                   'isError': bool(outcome.error)}
-                        if params['name'] == 'preview_asset' and not outcome.error:
+                        if params['name'] == 'query' and params.get('arguments', {}).get('type') == 'asset.preview' and not outcome.error:
                             preview = outcome.value.get('preview')
                             if preview:
                                 content = base64.b64encode(await asyncio.to_thread(Path(preview['path']).read_bytes)).decode()

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

@@ -1,5 +1,6 @@
 """Business commands shared by every agent runtime."""
 import asyncio
+import shutil
 from pathlib import Path
 from uuid import UUID, uuid5
 
@@ -10,9 +11,11 @@
 from opendirector.application.creation_canvas import CreationCanvas
 from opendirector.domain.creation.assets import AssetProvenance
 from opendirector.domain.creation.base import ObjectVersion
-from opendirector.domain.creation.commands import AssetPreviewQuery, CanvasItemQuery, COMMAND_MODELS, parse_command
+from opendirector.domain.creation.commands import Query, COMMAND_MODELS, parse_command
 from opendirector.domain.creation.models import ModelSnapshot
 from opendirector.domain.projects import ProjectError
+from opendirector.infrastructure.creation_projects import directory
+from opendirector.infrastructure.local_projects import child
 
 ROOT = Path(__file__).resolve().parents[5]
 EXCLUDED = {'project.create', 'project.trash', 'project.restore', 'asset.import', 'job.checkpoint',
@@ -45,13 +48,16 @@ def tools() -> list[dict]:
                          {'type': 'object', 'properties': {}, 'additionalProperties': False}),
               tool_schema('read', 'Read a discovered skill file/resource, or text asset via asset:<UUID>. No scripts.',
                          {'type': 'object', 'properties': {'path': {'type': 'string'}}, 'required': ['path'], 'additionalProperties': False})]
-    result += [tool_schema('read_canvas_item', 'Read a canvas item, generation status and current asset; optionally wait for changes.',
-                           CanvasItemQuery.model_json_schema()),
-               tool_schema('preview_asset', 'Preview an asset by ID: image thumbnail, video first frame, audio first 30 seconds or text excerpt.',
-                           AssetPreviewQuery.model_json_schema())]
+    query_schema = TypeAdapter(Query).json_schema()
+    query_schema['type'] = 'object'
+    for definition in query_schema['$defs'].values():
+        for field in ('queryId', 'projectId'):
+            definition['properties'].pop(field)
+            definition['required'].remove(field)
+    result.append(tool_schema('query', 'Query objects, jobs, assets or versions. canvas.status can wait; asset.preview provides media for native viewing.', query_schema))
     command_type = {'type': 'string', 'enum': list(COMMANDS)}
-    result.append(tool_schema('command_schema', 'Read the exact input schema before executing a project command.',
-        {'type': 'object', 'properties': {'type': command_type}, 'required': ['type'], 'additionalProperties': False}))
+    result.append(tool_schema('command_schema', 'Read command inputs; pass modelConfigId for a compact schema constrained to the chosen model.',
+        {'type': 'object', 'properties': {'type': command_type, 'modelConfigId': {'type': 'string', 'format': 'uuid', 'description': 'Chosen generation model; narrows media kind and supported parameters.'}}, 'required': ['type'], 'additionalProperties': False}))
     versions = TypeAdapter(list[ObjectVersion]).json_schema()
     definitions = versions.pop('$defs')
     result.append(tool_schema('execute_command', 'Execute a validated project command. Read command_schema first. IDs and project scope come from the host.',
@@ -74,15 +80,22 @@ async def execute(self, project_id: UUID, turn_id: UUID, session: str, call_id:
             snapshot = await asyncio.to_thread(self.projects.snapshot, project_id)
             return {'project': snapshot.model_dump(mode='json'), 'models': self.configs.list(), 'commands': list(COMMANDS),
                 'commandMetadata': {kind: {'description': model.__doc__} for kind, model in COMMANDS.items()}}
-        if name == 'read_canvas_item':
-            query = CanvasItemQuery.model_validate(arguments)
-            return await CreationCanvas(self.projects).wait_item_status(project_id, query.itemId, query.waitSeconds)
-        if name == 'preview_asset':
-            query = AssetPreviewQuery.model_validate(arguments)
-            with self.projects.local.transaction() as db:
-                directory, _, _ = self.projects._load(db, project_id)
-            target = directory / 'agent-runs' / str(turn_id) / 'work' / 'previews'
-            return await asyncio.to_thread(self.assets.preview, project_id, query.assetId, target)
+        if name == 'query':
+            if {'projectId', 'queryId'} & arguments.keys():
+                raise ProjectError('查询身份与工程范围由宿主提供。')
+            query = TypeAdapter(Query).validate_python({**arguments, 'projectId': project_id,
+                'queryId': uuid5(turn_id, call_id)})
+            result = await CreationCanvas(self.projects).query(query)
+            if query.type == 'asset.preview' and 'preview' in result:
+                with self.projects.local.transaction() as db:
+                    project_directory, _, _ = self.projects._load(db, project_id)
+                target = directory(project_directory / 'agent-runs' / str(turn_id) / 'work' / 'previews')
+                target.mkdir(parents=True, exist_ok=True)
+                source = Path(result['preview']['path'])
+                destination = child(target, source.name)
+                await asyncio.to_thread(shutil.copyfile, source, destination)
+                result['preview']['path'] = str(destination.resolve())
+            return result
         if name == 'read':
             path = TypeAdapter(str).validate_python(arguments['path'], strict=True)
             if path.startswith('asset:'):
@@ -111,6 +124,55 @@ async def execute(self, project_id: UUID, turn_id: UUID, session: str, call_id:
             for field in ('commandId', 'projectId'):
                 schema['properties'].pop(field)
             schema['required'] = [key for key in schema['required'] if key not in {'commandId', 'projectId'}]
+            if arguments.get('modelConfigId'):
+                config_id = str(UUID(arguments['modelConfigId']))
+                config = next((value for value in self.configs.list() if value['id'] == config_id), None)
+                if config is None:
+                    raise ProjectError('所选模型不可用。')
+                media_kind = 'text' if config['kind'] == 'agent' else config['kind']
+                def narrow(value: JsonValue) -> None:
+                    if isinstance(value, dict):
+                        discriminator = value.get('discriminator', {})
+                        mapping = discriminator.get('mapping', {})
+                        if discriminator.get('propertyName') == 'kind' and media_kind in mapping:
+                            value['$ref'] = mapping[media_kind]
+                            value.pop('oneOf')
+                            value.pop('discriminator')
+                        for child in value.values():
+                            narrow(child)
+                    elif isinstance(value, list):
+                        for child in value:
+                            narrow(child)
+                narrow(schema)
+                parameters = schema.get('$defs', {}).get('VideoParameters')
+                if parameters and media_kind == 'video':
+                    for field, choices in [('videoMode', config['inputRules']['videoModes']),
+                                           ('generateAudio', config['inputRules']['generateAudioOptions'])]:
+                        prop = parameters['properties'][field]
+                        prop['enum'] = choices
+                        if prop['default'] not in choices:
+                            prop.pop('default')
+                            parameters.setdefault('required', []).append(field)
+                            spec = schema['$defs']['VideoGenerationSpec']
+                            if 'parameters' not in spec.get('required', []):
+                                spec.setdefault('required', []).append('parameters')
+            definitions = schema.pop('$defs', {})
+            used: set[str] = set()
+            def references(value: JsonValue) -> None:
+                if isinstance(value, dict):
+                    if '$ref' in value:
+                        key = value['$ref'].rsplit('/', 1)[1]
+                        if key not in used:
+                            used.add(key)
+                            references(definitions[key])
+                    for child in value.values():
+                        references(child)
+                elif isinstance(value, list):
+                    for child in value:
+                        references(child)
+            references(schema)
+            if used:
+                schema['$defs'] = {key: value for key, value in definitions.items() if key in used}
             return schema
         if name != 'execute_command':
             raise ProjectError('未知业务工具。')
@@ -129,9 +191,9 @@ async def execute(self, project_id: UUID, turn_id: UUID, session: str, call_id:
                 provenance = AssetProvenance(source='agent_written', agentRunId=turn_id, agentRuntime=native_runtime)
             if command.type == 'artifact.publish':
                 with self.projects.local.transaction() as db:
-                    directory, _, _ = self.projects._load(db, project_id)
+                    project_directory, _, _ = self.projects._load(db, project_id)
                 receipt = await asyncio.to_thread(publish_artifact, self.assets, command,
-                    execution_directory=directory / 'agent-runs' / str(turn_id) / 'work', session=session, provenance=provenance)
+                    execution_directory=project_directory / 'agent-runs' / str(turn_id) / 'work', session=session, provenance=provenance)
             else:
                 receipt = await asyncio.to_thread(self.assets.execute, command, session=session, source='agent', provenance=provenance)
         else: