diff --git a/CHANGELOG.md b/CHANGELOG.md index 1816c47f8c..9c4da50c77 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -68,6 +68,15 @@ This section accumulates work toward the **2.3.0** milestone ### Fixed +- **gateway:** Creating a thread with checkpoint-owned keys such as `step`, + `parents` or `counters_since_delta_snapshot` in its metadata no longer + leaves the thread broken. Those keys used to land on the thread's first + checkpoint, so a bad `parents` made the thread's history fail with 500 for + good, and a bad `step` or delta counter made every later turn fail. A + `deerflow_agent_name` key was also copied into the next state update as if + the server had set it. The first checkpoint now takes none of these keys + from the request. ([#8019]) + - **mcp:** Two enabled MCP servers could bind the same tool name (server `a` exposing `b_x` and server `a_b` exposing `x` both bind `a_b_x`). Both declarations reached the model and the provider silently kept only one. @@ -16392,6 +16401,7 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#7993]: https://github.com/bytedance/deer-flow/pull/7993 [#7998]: https://github.com/bytedance/deer-flow/issues/7998 [#8006]: https://github.com/bytedance/deer-flow/pull/8006 +[#8019]: https://github.com/bytedance/deer-flow/pull/8019 [#8043]: https://github.com/bytedance/deer-flow/pull/8043 [#8046]: https://github.com/bytedance/deer-flow/pull/8046 [#8057]: https://github.com/bytedance/deer-flow/pull/8057 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index e585e132e2..205878d99c 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -33,6 +33,12 @@ ### 修复 +- **网关:** 创建对话时在元数据中带上 `step`、`parents`、`counters_since_delta_snapshot` + 等由检查点自身维护的键,不再让该对话损坏。此前这些键会写入对话的首个检查点,错误的 `parents` + 会让该对话的历史记录永久返回 500,错误的 `step` 或增量计数器会让之后每一轮对话都失败; + `deerflow_agent_name` 还会被当作服务端写入的值复制到下一次状态更新中。现在首个检查点不再从 + 请求中接收这些键。([#8019]) + - **通道:** 企业微信不再丢弃 2–10 MB 的图片。此前适配器把图片上限定为 2 MB(同一协议中语音的限制), 超出即在上传前被跳过,消息到达时缺少图片。现在图片上限与平台的 10 MB 对齐;普通文件 20 MB 的限制不变。([#8174]) @@ -12891,6 +12897,7 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写,核心包含子 [#7992]: https://github.com/bytedance/deer-flow/pull/7992 [#7993]: https://github.com/bytedance/deer-flow/pull/7993 [#8006]: https://github.com/bytedance/deer-flow/pull/8006 +[#8019]: https://github.com/bytedance/deer-flow/pull/8019 [#8043]: https://github.com/bytedance/deer-flow/pull/8043 [#8046]: https://github.com/bytedance/deer-flow/pull/8046 [#8064]: https://github.com/bytedance/deer-flow/pull/8064 diff --git a/backend/app/gateway/routers/threads.py b/backend/app/gateway/routers/threads.py index 436564b08a..1231c500a8 100644 --- a/backend/app/gateway/routers/threads.py +++ b/backend/app/gateway/routers/threads.py @@ -65,7 +65,7 @@ from deerflow.runtime.context_compaction import ( ThreadCompactionResult, compact_thread_context, ) -from deerflow.runtime.context_keys import checkpoint_agent_binding_metadata +from deerflow.runtime.context_keys import CHECKPOINT_AGENT_NAME_METADATA_KEY, checkpoint_agent_binding_metadata from deerflow.runtime.events.message_seq import stamp_messages_with_seq from deerflow.runtime.goal import ( DEFAULT_MAX_GOAL_CONTINUATIONS, @@ -129,6 +129,13 @@ def _checkpoint_mode_http_error(exc: Exception, thread_id: str) -> HTTPException # ``deerflow_origin`` marks a thread the server created for a schedule or an # extension; only the server sets it (``create_thread`` below, run admission). _SERVER_RESERVED_METADATA_KEYS: frozenset[str] = frozenset({"owner_id", "user_id", THREAD_PROJECT_METADATA_KEY, DEERFLOW_ORIGIN_KEY}) +# Checkpoint metadata the server writes and later reads back as trusted: +# LangGraph's lineage keys and delta counters (``step`` drives the next turn, +# ``parents`` the history walk, and ``counters_since_delta_snapshot`` is read +# by every run and state update) and the agent binding that gates memory +# writes. ``create_thread`` copies client metadata into the root checkpoint +# without them; the thread row keeps the client's metadata as sent. +_CHECKPOINT_OWNED_METADATA_KEYS: frozenset[str] = frozenset({"step", "source", "writes", "parents", "created_at", "counters_since_delta_snapshot", CHECKPOINT_AGENT_NAME_METADATA_KEY}) _SIDECAR_METADATA_KEY = "deerflow_sidecar" _BRANCH_METADATA_KEY = "deerflow_branch" _BRANCH_TITLE_SEQUENCE_METADATA_KEY = "branch_title_sequence" @@ -1065,12 +1072,14 @@ async def _create_thread_locked(body: ThreadCreateRequest, request: Request, thr # Write an empty checkpoint so state endpoints work immediately config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}} try: + # Client metadata cannot supply checkpoint-owned keys; see + # ``_CHECKPOINT_OWNED_METADATA_KEYS``. ckpt_metadata = { + **{key: value for key, value in metadata.items() if key not in _CHECKPOINT_OWNED_METADATA_KEYS}, "step": -1, "source": "input", "writes": None, "parents": {}, - **metadata, "created_at": now, } await checkpointer.aput(config, empty_checkpoint(), ckpt_metadata, {}) diff --git a/backend/docs/API.md b/backend/docs/API.md index de08fa3202..5af4a0a093 100644 --- a/backend/docs/API.md +++ b/backend/docs/API.md @@ -188,6 +188,11 @@ only letters, digits and hyphens; any other value is rejected with `422`. Branches preserve valid stored IDs, including their original spelling. If a legacy source has an invalid ID, its new branch stores `null` and uses the lead agent. +`metadata` is stored on the thread as sent and copied into its first +checkpoint, except for the keys the server owns there: `step`, `source`, +`writes`, `parents`, `created_at`, `counters_since_delta_snapshot` and +`deerflow_agent_name`. + **Response:** ```json { diff --git a/backend/tests/test_threads_router.py b/backend/tests/test_threads_router.py index e429e04afc..c26c82c583 100644 --- a/backend/tests/test_threads_router.py +++ b/backend/tests/test_threads_router.py @@ -5678,6 +5678,58 @@ def test_create_thread_marks_a_host_origin_from_request_state() -> None: assert response.json()["metadata"] == {"title": "t", DEERFLOW_ORIGIN_KEY: {"kind": "extension", "namespace": "community.teams"}} +@pytest.mark.parametrize( + "forged", + [ + {"parents": "x"}, + {"step": "zero"}, + {"source": "loop", "writes": {"runtime_run_duration": {"run-1": 1.0}}}, + {"counters_since_delta_snapshot": "x"}, + {CHECKPOINT_AGENT_NAME_METADATA_KEY: "forged-agent"}, + ], + ids=["parents", "step", "source-writes", "delta-counters", "agent-binding"], +) +def test_create_thread_keeps_server_checkpoint_metadata(monkeypatch, forged) -> None: + """Client metadata cannot supply the keys the server reads back from the root checkpoint. + + A bad ``parents`` breaks every history read, a bad ``step`` or delta + counter breaks every later turn and state update, and a forged agent + binding is copied into the next state rewrite as if the server wrote it. + """ + from deerflow.runtime.events.store.memory import MemoryRunEventStore + + app, _store, checkpointer = _build_thread_app() + app.state.run_event_store = MemoryRunEventStore() + graph_factory = _wire_extension_agent(monkeypatch, app, checkpointer, "full") + config = {"configurable": {"thread_id": "forged-meta", "checkpoint_ns": ""}} + + with TestClient(app) as client: + created = client.post("/api/threads", json={"thread_id": "forged-meta", "metadata": {**forged, "title": "mine"}}) + assert created.status_code == 200, created.text + + root = asyncio.run(checkpointer.aget_tuple(config)).metadata + assert {key: root[key] for key in ("step", "source", "writes", "parents")} == {"step": -1, "source": "input", "writes": None, "parents": {}} + assert "counters_since_delta_snapshot" not in root + assert CHECKPOINT_AGENT_NAME_METADATA_KEY not in root + assert root["title"] == "mine" + for response in (client.get("/api/threads/forged-meta"), client.get("/api/threads/forged-meta/state"), client.post("/api/threads/forged-meta/history", json={})): + assert response.status_code == 200, response.text + + updated = client.post("/api/threads/forged-meta/state", json={"values": {"title": "renamed"}}) + assert updated.status_code == 200, updated.text + assert CHECKPOINT_AGENT_NAME_METADATA_KEY not in asyncio.run(checkpointer.aget_tuple(config)).metadata + + graph = graph_factory() + graph.checkpointer = checkpointer + asyncio.run(graph.ainvoke({"messages": [HumanMessage(content="hi")]}, config)) + + with TestClient(app) as client: + history = client.post("/api/threads/forged-meta/history", json={}) + assert history.status_code == 200, history.text + assert history.json()[-1]["metadata"]["step"] == -1 + assert all(isinstance(entry["metadata"]["step"], int) for entry in history.json()) + + def _unread_app(tmp_path, *, owner_check_passes: bool = True): from uuid import UUID