From d7a7f4a779e129dad606451fa370d286519edb8f Mon Sep 17 00:00:00 2001 From: Wenjie Zhang Date: Sun, 7 Jun 2026 22:55:20 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E7=B2=BE=E7=AE=80=20Message=20state=20?= =?UTF-8?q?=E4=BA=8B=E4=BB=B6=E8=BD=BD=E8=8D=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../yuxi/services/agent_run_service.py | 174 +++++++++++- backend/server/routers/agent_router.py | 3 +- .../api/test_agent_run_events_router.py | 180 +++++++++++++ .../unit/services/test_agent_run_service.py | 251 ++++++++++++++++++ docs/advanced/api-key-integration.md | 2 + docs/develop-guides/changelog.md | 1 + web/src/apis/agent_api.js | 7 +- web/src/composables/useAgentRunStream.js | 8 +- 8 files changed, 614 insertions(+), 12 deletions(-) create mode 100644 backend/test/integration/api/test_agent_run_events_router.py diff --git a/backend/package/yuxi/services/agent_run_service.py b/backend/package/yuxi/services/agent_run_service.py index 740df31e..4dba3d5a 100644 --- a/backend/package/yuxi/services/agent_run_service.py +++ b/backend/package/yuxi/services/agent_run_service.py @@ -81,6 +81,158 @@ def _format_heartbeat() -> str: return ": heartbeat\n\n" +def _compact_message_dict(message: dict) -> dict: + compact = { + key: message[key] for key in ("id", "role", "content", "type", "message_type") if message.get(key) is not None + } + extra_metadata = message.get("extra_metadata") + if isinstance(extra_metadata, dict) and extra_metadata.get("attachments"): + compact["extra_metadata"] = {"attachments": extra_metadata["attachments"]} + return compact + + +def _compact_semantic_stream_event(stream_event: dict) -> dict: + event_type = stream_event.get("type") + if event_type == "message_delta": + return { + key: stream_event[key] + for key in ("type", "message_id", "content", "reasoning_content", "additional_reasoning_content") + if stream_event.get(key) + } + + if event_type in {"tool_call", "tool_call_delta"}: + compact = { + key: stream_event[key] + for key in ("type", "message_id", "tool_call_id", "name", "args", "args_delta") + if stream_event.get(key) is not None and stream_event.get(key) != "" + } + if stream_event.get("index"): + compact["index"] = stream_event["index"] + return compact + + return {key: value for key, value in stream_event.items() if key not in {"thread_id", "namespace"}} + + +def _compact_tool_stream_event(event: dict) -> dict: + compact = {key: event[key] for key in ("method",) if event.get(key)} + data = event.get("data") + if isinstance(data, dict): + compact_data = { + key: data[key] + for key in ("event", "tool_call_id", "tool_name", "output", "error") + if data.get(key) is not None and data.get(key) != "" + } + if compact_data: + compact["data"] = compact_data + return compact + + +def _compact_stream_chunk(chunk: dict) -> dict: + compact = { + key: chunk[key] + for key in ( + "status", + "run_id", + "parent_run_id", + "message", + "error_type", + "error_message", + "retryable", + "job_try", + "questions", + "interrupt_info", + "source", + "agent_state", + ) + if chunk.get(key) is not None and chunk.get(key) != "" + } + if isinstance(chunk.get("msg"), dict): + compact["msg"] = _compact_message_dict(chunk["msg"]) + if isinstance(chunk.get("stream_event"), dict): + compact["stream_event"] = _compact_semantic_stream_event(chunk["stream_event"]) + if isinstance(chunk.get("event"), dict): + compact["event"] = _compact_tool_stream_event(chunk["event"]) + return compact + + +def _request_id_from_chunk(chunk: object) -> str | None: + if not isinstance(chunk, dict): + return None + request_id = chunk.get("request_id") + if isinstance(request_id, str) and request_id: + return request_id + msg = chunk.get("msg") + extra_metadata = msg.get("extra_metadata") if isinstance(msg, dict) else None + if isinstance(extra_metadata, dict): + request_id = extra_metadata.get("request_id") + if isinstance(request_id, str) and request_id: + return request_id + return None + + +def _request_id_from_payload(payload: object) -> str | None: + if not isinstance(payload, dict): + return None + request_id = payload.get("request_id") + if isinstance(request_id, str) and request_id: + return request_id + request_id = _request_id_from_chunk(payload.get("chunk")) + if request_id: + return request_id + items = payload.get("items") + if isinstance(items, list): + for item in items: + request_id = _request_id_from_chunk(item) + if request_id: + return request_id + return None + + +def _compact_run_event_payload(event_type: str, payload: dict | None) -> dict: + if not isinstance(payload, dict): + return {} + + if event_type == "messages": + compact: dict = {} + if isinstance(payload.get("items"), list): + compact["items"] = [ + _compact_stream_chunk(item) if isinstance(item, dict) else item for item in payload["items"] + ] + if isinstance(payload.get("chunk"), dict): + compact["chunk"] = _compact_stream_chunk(payload["chunk"]) + return compact + + compact = {key: value for key, value in payload.items() if key not in {"chunk", "request_id"}} + if isinstance(payload.get("chunk"), dict): + compact["chunk"] = _compact_stream_chunk(payload["chunk"]) + return compact + + +def _is_empty_agent_state(agent_state: object) -> bool: + if not isinstance(agent_state, dict): + return False + return all(not value for value in agent_state.values()) + + +def _compact_run_event_envelope(envelope: dict) -> dict | None: + event_type = str(envelope.get("event") or "") + payload = envelope.get("payload") + if event_type == "metadata": + return None + if event_type == "custom" and isinstance(payload, dict) and payload.get("name") == "yuxi.agent_state": + state = payload.get("agent_state") + chunk = payload.get("chunk") if isinstance(payload.get("chunk"), dict) else {} + if _is_empty_agent_state(state) or _is_empty_agent_state(chunk.get("agent_state")): + return None + + compact = {key: envelope[key] for key in ("run_id", "thread_id") if key in envelope} + request_id = _request_id_from_payload(payload) + if request_id: + compact["request_id"] = request_id + compact["payload"] = _compact_run_event_payload(event_type, payload) + return compact + + async def create_agent_run_view( *, query: str | None, @@ -245,6 +397,7 @@ async def stream_agent_run_events( run_id: str, after_seq: str, current_uid: str, + verbose: bool = True, ) -> AsyncIterator[str]: started_at = utc_now_naive() last_heartbeat_ts = started_at @@ -294,6 +447,10 @@ async def stream_agent_run_events( last_seq = seq event_type = event.get("event_type") or "message" envelope = event.get("payload") or {} + if not verbose and isinstance(envelope, dict): + envelope = _compact_run_event_envelope(envelope) + if envelope is None: + continue yield _format_sse(envelope, event=event_type, event_id=seq) if event_type == "end": emitted_terminal = True @@ -307,14 +464,17 @@ async def stream_agent_run_events( terminal_seq = await get_last_run_stream_seq(run_id) if terminal_seq in {"", "0-0"}: terminal_seq = None + terminal_envelope = build_run_event_envelope( + run_id=run_id, + thread_id=run.thread_id, + event_type="end", + payload={"status": run.status, "request_id": run.request_id}, + created_at=utc_now_naive().isoformat(), + ) + if not verbose: + terminal_envelope = _compact_run_event_envelope(terminal_envelope) yield _format_sse( - build_run_event_envelope( - run_id=run_id, - thread_id=run.thread_id, - event_type="end", - payload={"status": run.status}, - created_at=utc_now_naive().isoformat(), - ), + terminal_envelope, event="end", event_id=terminal_seq, ) diff --git a/backend/server/routers/agent_router.py b/backend/server/routers/agent_router.py index 8614f409..643148ba 100644 --- a/backend/server/routers/agent_router.py +++ b/backend/server/routers/agent_router.py @@ -286,12 +286,13 @@ async def cancel_agent_run( async def stream_run_events( run_id: str, after_seq: str = "0-0", + verbose: bool = Query(default=True, description="是否返回完整事件载荷；false 时仅返回 UI/客户端消费所需字段"), last_event_id: str | None = Header(default=None, alias="Last-Event-ID"), current_user: User = Depends(get_required_user), ): cursor = last_event_id or after_seq return StreamingResponse( - stream_agent_run_events(run_id=run_id, after_seq=cursor, current_uid=str(current_user.uid)), + stream_agent_run_events(run_id=run_id, after_seq=cursor, current_uid=str(current_user.uid), verbose=verbose), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no"}, ) diff --git a/backend/test/integration/api/test_agent_run_events_router.py b/backend/test/integration/api/test_agent_run_events_router.py new file mode 100644 index 00000000..02b2b043 --- /dev/null +++ b/backend/test/integration/api/test_agent_run_events_router.py @@ -0,0 +1,180 @@ +from __future__ import annotations + +import json +import os +import uuid + +import asyncpg +import pytest +from yuxi.services.run_queue_service import append_run_stream_event, get_redis_client + +pytestmark = [pytest.mark.asyncio, pytest.mark.integration] + + +def _postgres_dsn() -> str: + return os.getenv("POSTGRES_URL", "postgresql+asyncpg://postgres:postgres@postgres:5432/yuxi").replace( + "+asyncpg", "" + ) + + +async def _collect_sse_payloads(response) -> list[tuple[str, dict, str | None]]: + event = "message" + event_id = None + data_lines: list[str] = [] + payloads: list[tuple[str, dict, str | None]] = [] + + async for line in response.aiter_lines(): + if not line: + if data_lines: + payloads.append((event, json.loads("\n".join(data_lines)), event_id)) + event = "message" + event_id = None + data_lines = [] + continue + if line.startswith(":"): + continue + if line.startswith("event:"): + event = line.removeprefix("event:").strip() or "message" + elif line.startswith("id:"): + event_id = line.removeprefix("id:").strip() + elif line.startswith("data:"): + data_lines.append(line.removeprefix("data:").strip()) + + return payloads + + +async def test_run_events_verbose_false_returns_compact_payload(test_client, standard_user): + uid = str(standard_user["user"]["uid"]) + run_id = str(uuid.uuid4()) + thread_id = str(uuid.uuid4()) + request_id = f"req-{uuid.uuid4()}" + + conn = await asyncpg.connect(_postgres_dsn()) + try: + await conn.execute( + """ + INSERT INTO agent_runs (id, thread_id, agent_id, uid, request_id, input_payload, status, run_type) + VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7, $8) + """, + run_id, + thread_id, + "deep-research", + uid, + request_id, + json.dumps({"query": "写一个冒泡排序"}, ensure_ascii=False), + "completed", + "chat", + ) + finally: + await conn.close() + + try: + await append_run_stream_event( + run_id, + "metadata", + { + "request_id": request_id, + "agent_id": "deep-research", + "backend_id": "ChatbotAgent", + "uid": uid, + }, + thread_id=thread_id, + ) + await append_run_stream_event( + run_id, + "custom", + { + "name": "yuxi.agent_state", + "chunk": { + "request_id": request_id, + "response": None, + "thread_id": thread_id, + "status": "agent_state", + "agent_state": { + "todos": [], + "files": {}, + "artifacts": [], + "subagent_runs": [], + }, + "meta": {"uid": uid}, + }, + "agent_state": { + "todos": [], + "files": {}, + "artifacts": [], + "subagent_runs": [], + }, + }, + thread_id=thread_id, + ) + await append_run_stream_event( + run_id, + "messages", + { + "items": [ + { + "request_id": request_id, + "response": "你", + "thread_id": thread_id, + "status": "loading", + "stream_event": { + "type": "tool_call", + "message_id": "msg-1", + "tool_call_id": "call-1", + "name": "ls", + "args": {"path": "/home/gem/user-data/outputs"}, + "thread_id": thread_id, + "namespace": [], + }, + "metadata": { + "langfuse_user_id": uid, + "langgraph_checkpoint_ns": "model:checkpoint", + }, + } + ] + }, + thread_id=thread_id, + ) + await append_run_stream_event( + run_id, + "end", + {"status": "completed", "chunk": {"status": "finished", "request_id": request_id, "meta": {"uid": uid}}}, + thread_id=thread_id, + ) + + async with test_client.stream( + "GET", + f"/api/agent/runs/{run_id}/events", + params={"verbose": "false"}, + headers=standard_user["headers"], + ) as response: + assert response.status_code == 200, response.text + payloads = await _collect_sse_payloads(response) + + assert {event for event, _payload, _event_id in payloads} == {"messages", "end"} + + message_event = next(item for item in payloads if item[0] == "messages") + message_chunk = message_event[1]["payload"]["items"][0] + assert message_event[1]["request_id"] == request_id + assert message_event[2] + assert "request_id" not in message_chunk + assert "metadata" not in message_chunk + assert "response" not in message_chunk + assert "thread_id" not in message_chunk + assert message_chunk["stream_event"]["tool_call_id"] == "call-1" + assert "thread_id" not in message_chunk["stream_event"] + assert "namespace" not in message_chunk["stream_event"] + + end_event = next(item for item in payloads if item[0] == "end") + assert end_event[1]["request_id"] == request_id + assert end_event[1]["payload"]["status"] == "completed" + assert "request_id" not in end_event[1]["payload"]["chunk"] + assert "meta" not in end_event[1]["payload"]["chunk"] + finally: + redis = await get_redis_client() + await redis.delete(f"run:events:{run_id}") + conn = await asyncpg.connect(_postgres_dsn()) + try: + await conn.execute("DELETE FROM agent_runs WHERE id = $1", run_id) + finally: + await conn.close() diff --git a/backend/test/unit/services/test_agent_run_service.py b/backend/test/unit/services/test_agent_run_service.py index dbc90d4c..50ff7c6d 100644 --- a/backend/test/unit/services/test_agent_run_service.py +++ b/backend/test/unit/services/test_agent_run_service.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json from contextlib import asynccontextmanager from types import SimpleNamespace @@ -8,6 +9,13 @@ import pytest import yuxi.services.agent_run_service as agent_run_service +def _sse_data(chunk: str) -> dict: + for line in chunk.splitlines(): + if line.startswith("data: "): + return json.loads(line.removeprefix("data: ")) + raise AssertionError(f"SSE chunk has no data line: {chunk}") + + class _FakeContext: def __init__(self): self.model = "agent-default-model" @@ -121,6 +129,249 @@ async def test_stream_agent_run_events_reads_redis_and_ends_on_end_event(monkeyp assert "id: 1700000000001-0" in chunks[-1] +@pytest.mark.asyncio +async def test_stream_agent_run_events_compacts_verbose_false(monkeypatch: pytest.MonkeyPatch): + @asynccontextmanager + async def fake_session_ctx(): + yield object() + + class Repo: + def __init__(self, db): + self.db = db + + async def get_run_for_user(self, run_id: str, uid: str): + del run_id, uid + return SimpleNamespace(status="completed", thread_id="thread-1") + + async def fake_list_events(run_id: str, *, after_seq: str, limit: int): + del run_id, after_seq, limit + return [ + { + "seq": "1700000000000-0", + "event_type": "metadata", + "payload": { + "schema_version": 1, + "run_id": "run-1", + "thread_id": "thread-1", + "event": "metadata", + "payload": { + "request_id": "req-1", + "agent_id": "deep-research", + "backend_id": "ChatbotAgent", + "uid": "user-1", + }, + "created_at": "2026-05-27T00:00:00+00:00", + }, + "ts": 1700000000000, + }, + { + "seq": "1700000000001-0", + "event_type": "custom", + "payload": { + "schema_version": 1, + "run_id": "run-1", + "thread_id": "thread-1", + "event": "custom", + "payload": { + "name": "yuxi.init", + "chunk": { + "request_id": "req-1", + "response": None, + "thread_id": "thread-1", + "status": "init", + "meta": {"query": "写一个冒泡排序", "uid": "user-1"}, + "msg": { + "role": "user", + "content": "写一个冒泡排序", + "type": "human", + "image_content": "base64-image-data", + "extra_metadata": { + "request_id": "req-1", + "attachments": [], + "debug": "drop-me", + }, + }, + }, + }, + "created_at": "2026-05-27T00:00:00+00:00", + }, + "ts": 1700000000001, + }, + { + "seq": "1700000000002-0", + "event_type": "custom", + "payload": { + "schema_version": 1, + "run_id": "run-1", + "thread_id": "thread-1", + "event": "custom", + "payload": { + "name": "yuxi.agent_state", + "chunk": { + "request_id": "req-1", + "response": None, + "thread_id": "thread-1", + "status": "agent_state", + "agent_state": { + "todos": [], + "files": {}, + "artifacts": [], + "subagent_runs": [], + }, + "meta": {"uid": "user-1"}, + }, + "agent_state": { + "todos": [], + "files": {}, + "artifacts": [], + "subagent_runs": [], + }, + }, + "created_at": "2026-05-27T00:00:00+00:00", + }, + "ts": 1700000000002, + }, + { + "seq": "1700000000003-0", + "event_type": "messages", + "payload": { + "schema_version": 1, + "run_id": "run-1", + "thread_id": "thread-1", + "event": "messages", + "payload": { + "items": [ + { + "request_id": "req-1", + "response": "你", + "thread_id": "thread-1", + "status": "loading", + "stream_event": { + "type": "tool_call", + "message_id": "msg-1", + "tool_call_id": "call-1", + "name": "ls", + "args": {"path": "/home/gem/user-data/outputs"}, + "thread_id": "thread-1", + "namespace": [], + }, + "metadata": { + "langfuse_user_id": "user-1", + "langgraph_checkpoint_ns": "model:checkpoint", + }, + } + ] + }, + "created_at": "2026-05-27T00:00:01+00:00", + }, + "ts": 1700000000003, + }, + { + "seq": "1700000000004-0", + "event_type": "end", + "payload": { + "schema_version": 1, + "run_id": "run-1", + "thread_id": "thread-1", + "event": "end", + "payload": { + "status": "completed", + "chunk": {"status": "finished", "request_id": "req-1", "meta": {"uid": "user-1"}}, + }, + "created_at": "2026-05-27T00:00:02+00:00", + }, + "ts": 1700000000004, + }, + ] + + monkeypatch.setattr(agent_run_service.pg_manager, "get_async_session_context", fake_session_ctx) + monkeypatch.setattr(agent_run_service, "AgentRunRepository", Repo) + monkeypatch.setattr(agent_run_service, "list_run_stream_events", fake_list_events) + + chunks = [] + async for chunk in agent_run_service.stream_agent_run_events( + run_id="run-1", + after_seq="0", + current_uid="user-1", + verbose=False, + ): + chunks.append(chunk) + + assert len(chunks) == 3 + + init_data = _sse_data(chunks[0]) + init_chunk = init_data["payload"]["chunk"] + assert init_data["request_id"] == "req-1" + assert init_data["payload"]["name"] == "yuxi.init" + assert "meta" not in init_chunk + assert "request_id" not in init_chunk + assert "response" not in init_chunk + assert "thread_id" not in init_chunk + assert "image_content" not in init_chunk["msg"] + assert "extra_metadata" not in init_chunk["msg"] + + message_data = _sse_data(chunks[1]) + message_chunk = message_data["payload"]["items"][0] + assert message_data["request_id"] == "req-1" + assert "request_id" not in message_chunk + assert "metadata" not in message_chunk + assert "response" not in message_chunk + assert "thread_id" not in message_chunk + assert message_chunk["stream_event"]["tool_call_id"] == "call-1" + assert "thread_id" not in message_chunk["stream_event"] + assert "namespace" not in message_chunk["stream_event"] + + end_data = _sse_data(chunks[2]) + assert end_data["request_id"] == "req-1" + assert end_data["payload"]["status"] == "completed" + assert "request_id" not in end_data["payload"]["chunk"] + assert "meta" not in end_data["payload"]["chunk"] + + +@pytest.mark.asyncio +async def test_stream_agent_run_events_compact_fallback_end_keeps_request_id(monkeypatch: pytest.MonkeyPatch): + @asynccontextmanager + async def fake_session_ctx(): + yield object() + + class Repo: + def __init__(self, db): + self.db = db + + async def get_run_for_user(self, run_id: str, uid: str): + del run_id, uid + return SimpleNamespace(status="completed", thread_id="thread-1", request_id="req-1") + + async def fake_list_events(run_id: str, *, after_seq: str, limit: int): + del run_id, after_seq, limit + return [] + + async def fake_get_last_run_stream_seq(run_id: str): + del run_id + return "1700000000004-0" + + monkeypatch.setattr(agent_run_service.pg_manager, "get_async_session_context", fake_session_ctx) + monkeypatch.setattr(agent_run_service, "AgentRunRepository", Repo) + monkeypatch.setattr(agent_run_service, "list_run_stream_events", fake_list_events) + monkeypatch.setattr(agent_run_service, "get_last_run_stream_seq", fake_get_last_run_stream_seq) + + chunks = [] + async for chunk in agent_run_service.stream_agent_run_events( + run_id="run-1", + after_seq="0", + current_uid="user-1", + verbose=False, + ): + chunks.append(chunk) + + assert len(chunks) == 1 + assert chunks[0].startswith("event: end") + assert "id: 1700000000004-0" in chunks[0] + data = _sse_data(chunks[0]) + assert data["request_id"] == "req-1" + assert data["payload"] == {"status": "completed"} + + @pytest.mark.asyncio async def test_create_agent_run_persists_input_before_enqueue(monkeypatch: pytest.MonkeyPatch): class FakeResult: diff --git a/docs/advanced/api-key-integration.md b/docs/advanced/api-key-integration.md index 46a17a01..282620f7 100644 --- a/docs/advanced/api-key-integration.md +++ b/docs/advanced/api-key-integration.md @@ -121,6 +121,8 @@ with requests.get(f"{base_url}{run['stream_url']}", headers=headers, stream=True 服务端还会定期发送以 `:` 开头的 heartbeat 注释，客户端应忽略。断线重连时，可以在请求头中传 `Last-Event-ID`，或在 query 参数中传 `after_seq`，服务端会从该序号后继续回放事件。 +事件流默认返回完整载荷，便于排查 LangGraph/Langfuse 运行细节。如果只需要渲染消息、工具调用、工具结果、Agent state 和终止状态，可以在订阅地址追加 `?verbose=false`。精简模式会保留 SSE `event/data/id`、data 中的 `run_id/thread_id/request_id/payload` 以及客户端消费所需字段；同一 data 内的 `request_id` 会外提为单个字段。精简模式还会跳过 `metadata` 和空 `yuxi.agent_state`，并去掉每个 chunk 中重复的 `meta`、`metadata`、`thread_id`、`response`、空 `namespace` 和图片 base64 等调试字段。 + 每次创建 run 都会返回 `request_id`，可用于日志追踪和问题排查。如果需要在多轮对话中使用同一个会话，请复用 `thread_id`，系统会将同一线程的消息串联起来形成连贯的对话上下文。 ## 认证方式 diff --git a/docs/develop-guides/changelog.md b/docs/develop-guides/changelog.md index 7f2e09ae..06c34b91 100644 --- a/docs/develop-guides/changelog.md +++ b/docs/develop-guides/changelog.md @@ -54,6 +54,7 @@ - 工作区文件上传支持多选：`/workspace/upload` 与 Viewer 工作区上传统一使用 `files` 多文件字段，一次最多上传 50 个文件，批量上传失败时清理本次已写入文件。 - 聊天附件新增 MinIO tmp 临时上传、可选 PDF/图片解析、确认后加入线程附件的流程；前端改为弹窗内上传、解析与确认。 - 标准化 Agent run/SSE 执行链路：run 创建时持久化输入消息并提交后入队，worker 统一写入 Redis Stream envelope，SSE 输出 `event/data/id`、心跳注释、`Last-Event-ID` 回放和终止 `end` 事件；前端强制使用 run API 并支持 ask_user_question 中断后以 resume run 恢复；事件 envelope 构造收敛到统一 helper，前端优先使用 envelope 一级 `thread_id` 路由。 +- Agent run SSE 新增 `verbose=false` 精简模式：默认仍返回完整事件载荷；精简模式仅在 SSE 输出前重建最小 payload，跳过 `metadata` 和空 `yuxi.agent_state`，将同一 data 内的 `request_id` 外提为单个字段，移除 chunk 中重复的 `meta`、`metadata`、`thread_id`、`response`、空 `namespace` 和图片 base64 等调试字段，保留消息增量、工具调用、工具结果、非空 Agent state、终止状态和 SSE 游标，前端订阅默认使用精简模式。 - 收敛后端模块边界：文档解析从 `plugins.parser` 移动到 `knowledge.parser`，内容审查从 `plugins.guard` 移动到 `services.guard`。 - 收敛文件服务边界：文件预览判断抽为独立服务，Viewer 文件系统的 workspace 分支复用用户 workspace 服务，线程运行时上下文解析从泛化 `filesystem_service` 拆出为 agent runtime helper。 - 升级 DeepAgents 到 0.6.7 并适配新版文件系统协议：SubAgentMiddleware 改为显式 subagent spec，Skills prompt 补齐新版占位符；sandbox/skills backend 复用新版 `ReadResult`、`GlobResult`、`GrepResult` 等协议类型，文件权限在 backend 层明确区分 skills、uploads、outputs 与 workspace，保留最小 `CustomCompositeBackend` 以避免非 route glob 误扫其他 route；Agent 上下文压缩改为复用 DeepAgents SummarizationMiddleware，历史摘要与大工具结果统一 offload 到 outputs。 diff --git a/web/src/apis/agent_api.js b/web/src/apis/agent_api.js index 925287d4..7e69cdc7 100644 --- a/web/src/apis/agent_api.js +++ b/web/src/apis/agent_api.js @@ -136,11 +136,11 @@ export const agentApi = { * 打开 Run 事件 SSE 连接（调用方负责关闭） * @param {string} runId - run ID * @param {string} afterSeq - 起始 seq/cursor - * @param {Object} options - { signal } + * @param {Object} options - { signal, verbose } * @returns {Promise} */ streamAgentRunEvents: (runId, afterSeq = '0-0', options = {}) => { - const { signal } = options + const { signal, verbose = false } = options const headers = { ...useUserStore().getAuthHeaders() } @@ -148,7 +148,8 @@ export const agentApi = { if (cursor && cursor !== '0-0') { headers['Last-Event-ID'] = cursor } - return fetch(`/api/agent/runs/${runId}/events`, { + const params = new URLSearchParams({ verbose: String(verbose) }) + return fetch(`/api/agent/runs/${runId}/events?${params.toString()}`, { method: 'GET', headers, signal diff --git a/web/src/composables/useAgentRunStream.js b/web/src/composables/useAgentRunStream.js index 64794640..6be234a3 100644 --- a/web/src/composables/useAgentRunStream.js +++ b/web/src/composables/useAgentRunStream.js @@ -305,7 +305,12 @@ export function useAgentRunStream({ }) touchedThreadIds.add(routeThreadId) handleStreamChunk( - { ...chunk, run_id: chunk.run_id || data.run_id || runId, thread_id: routeThreadId }, + { + ...chunk, + request_id: chunk.request_id || data.request_id, + run_id: chunk.run_id || data.run_id || runId, + thread_id: routeThreadId + }, routeThreadId ) }) @@ -320,6 +325,7 @@ export function useAgentRunStream({ handleStreamChunk( { ...payload.chunk, + request_id: payload.chunk.request_id || data.request_id, run_id: payload.chunk.run_id || data.run_id || runId, thread_id: routeThreadId },