mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-11-12 22:20:33 +00:00
fix(gateway): keep server-owned checkpoint keys out of client thread metadata (#8019)
* fix(gateway): keep LangGraph checkpoint keys out of client thread metadata POST /api/threads spread client metadata over the root checkpoint's step/source/writes/parents. A non-mapping parents broke every history read and a non-integer step broke every later turn, leaving deletion as the only way out. Server values now come after the client's. * docs(changelog): reference #8019 for thread-create checkpoint metadata * fix(gateway): keep server-owned checkpoint keys out of client thread metadata The root checkpoint still copied client keys the server reads back as trusted. A non-mapping counters_since_delta_snapshot failed every state update and turn in both checkpoint modes, and deerflow_agent_name was copied into the next state rewrite as the server-authored agent binding. Drop a named set of checkpoint-owned keys before the spread so the fix no longer relies on dict ordering alone. --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
parent
81de0761ed
commit
8853e68638
10
CHANGELOG.md
10
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
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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, {})
|
||||
|
||||
@ -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
|
||||
{
|
||||
|
||||
@ -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
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user