diff --git a/CHANGELOG.md b/CHANGELOG.md index 41766c34df..4724d5403f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,12 @@ This section accumulates work toward the **2.3.0** milestone plaintext, which with authorization disabled (the default) is every signed-in user. Non-secret knobs, `api_key_env` names and unset values are unchanged. ([#7665]) +- **gateway:** `GET /threads/{thread_id}/token-usage` now resolves the live + config in a worker thread. `build_context_usage` called the accessor inline, + and `get_app_config()` revalidates on every call — it digests the whole + `config.yaml`, then `extensions_config.json` — so a config edit landing just + before a thread view stalled the Gateway event loop that runs, streams and + every other thread share. The response is unchanged. ([#7727]) ## [2.2.0] — 2026-10-29 @@ -15934,3 +15940,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#7659]: https://github.com/bytedance/deer-flow/pull/7659 [#7665]: https://github.com/bytedance/deer-flow/pull/7665 [#7723]: https://github.com/bytedance/deer-flow/pull/7723 +[#7727]: https://github.com/bytedance/deer-flow/pull/7727 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index a87009dbee..06297fbcc1 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -24,6 +24,10 @@ 此前它们按加载结果原样返回配置,而 `$VAR` 引用此时已被解析,因此 DeerMem 提取模型的密钥 或 Honcho 的 `api_key` 会以明文返回给所有拥有 `memory:read` 的调用方;在默认关闭授权时, 这就是所有已登录用户。非机密配置项、`api_key_env` 变量名和未设置的值保持不变。([#7665]) +- **网关:** `GET /threads/{thread_id}/token-usage` 现在在工作线程里解析实时配置。 + 此前 `build_context_usage` 直接内联调用该访问器,而 `get_app_config()` 每次调用都会重新校验 + ——它先对整个 `config.yaml` 取摘要,再对 `extensions_config.json` 做同样的事——所以在有人打开 + 会话视图前刚好改了配置,就会让运行、消息流与其他线程共用的网关事件循环整段停顿。响应内容不变。([#7727]) ## [2.2.0] — 2026-10-29 @@ -12701,3 +12705,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写,核心包含子 [#7659]: https://github.com/bytedance/deer-flow/pull/7659 [#7665]: https://github.com/bytedance/deer-flow/pull/7665 [#7723]: https://github.com/bytedance/deer-flow/pull/7723 +[#7727]: https://github.com/bytedance/deer-flow/pull/7727 diff --git a/backend/app/gateway/context_usage.py b/backend/app/gateway/context_usage.py index ef0d6730cc..ccbc8e70c5 100644 --- a/backend/app/gateway/context_usage.py +++ b/backend/app/gateway/context_usage.py @@ -66,7 +66,10 @@ def build_context_usage_payload(*, token_count: int, max_context_tokens: int | N async def build_context_usage(request: Request, thread_id: str, run_store: Any, user_id: str | None = None) -> dict[str, Any] | None: """Return approximate usage for the latest materialized thread checkpoint.""" try: - app_config = get_config() + # ``get_config`` revalidates ``config.yaml`` and + # ``extensions_config.json`` on every call (stat + full-file hash), so + # reading it here would put that IO on the loop every thread view pays. + app_config = await asyncio.to_thread(get_config) except HTTPException: return None diff --git a/backend/tests/blocking_io/test_context_usage_config_read.py b/backend/tests/blocking_io/test_context_usage_config_read.py new file mode 100644 index 0000000000..165f5be6d5 --- /dev/null +++ b/backend/tests/blocking_io/test_context_usage_config_read.py @@ -0,0 +1,132 @@ +"""Regression anchor: ``/threads/{id}/token-usage`` must read the live config off the loop. + +``build_context_usage`` resolves the model capacity for its payload through +``app.gateway.deps.get_config``, which routes to ``get_app_config()``. That +loader revalidates on every call — it stats and hashes the whole +``config.yaml``, then does the same for ``extensions_config.json`` (see +**Config Caching** in ``deerflow/config/AGENTS.md``) — and reparses both after +an edit. ``GET /threads/{thread_id}/token-usage`` builds this payload on every +request, while runs, streams and other threads share the same Gateway loop, so +a view requested just after a config edit made that request stall the loop for +the whole reload. The token count a few lines below the read was already +offloaded with ``asyncio.to_thread``; the config read now joins it. + +Each anchor performs a real filesystem read and asserts the capacity that only +the bytes on disk can produce, so it cannot pass because the read was stubbed +away, and it records the calling thread so a dropped wrapper is attributed. +""" + +from __future__ import annotations + +import asyncio +import threading +from collections.abc import AsyncIterator +from pathlib import Path +from types import SimpleNamespace +from typing import Any + +import pytest +import pytest_asyncio +from blockbuster import BlockingError +from fastapi import HTTPException +from langchain_core.messages import HumanMessage + +from app.gateway import context_usage + +pytestmark = pytest.mark.asyncio + +_CAPACITY_ON_DISK = 4096 +_CONFIG_TEXT = f"model: disk-model\ntoken_capacity: {_CAPACITY_ON_DISK}\n" + + +class _RecordingConfigLoader: + """``get_config`` stand-in that reads from disk, and remembers where it ran.""" + + def __init__(self, path: Path) -> None: + self.path = path + self.calls = 0 + self.on_loop_thread = False + + def __call__(self) -> SimpleNamespace: + self.calls += 1 + self.on_loop_thread = threading.current_thread() is threading.main_thread() + name, capacity = self.path.read_text(encoding="utf-8").splitlines() + model_name = name.split(":", 1)[1].strip() + return SimpleNamespace( + models=[SimpleNamespace(name=model_name)], + get_model_config=lambda requested: SimpleNamespace(context_window=int(capacity.split(":", 1)[1])) if requested == model_name else None, + ) + + +class _Accessor: + async def aget(self, config: dict[str, Any]) -> SimpleNamespace: + return SimpleNamespace(values={"messages": [HumanMessage(content="context window probe")]}) + + +async def _stub_accessor_builder(request, *, thread_id: str): + return _Accessor(), {"configurable": {"thread_id": thread_id}} + + +class _RunStore: + async def list_by_thread(self, thread_id: str, *, limit: int = 1, user_id: str | None = None): + return [{"model_name": "disk-model"}] + + +def _write_capacity_file(path: Path) -> Path: + path.write_text(_CONFIG_TEXT, encoding="utf-8") + return path + + +@pytest_asyncio.fixture +async def loader(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[_RecordingConfigLoader]: + # The strict gate wraps setup as well as the call, so the fixture's own + # write goes through a worker. + config_file = await asyncio.to_thread(_write_capacity_file, tmp_path / "capacity.yaml") + recorder = _RecordingConfigLoader(config_file) + monkeypatch.setattr(context_usage, "get_config", recorder) + monkeypatch.setattr(context_usage, "build_thread_checkpoint_state_accessor", _stub_accessor_builder) + yield recorder + await asyncio.to_thread(config_file.unlink) + + +async def test_build_context_usage_reads_capacity_off_the_loop(loader: _RecordingConfigLoader) -> None: + payload = await context_usage.build_context_usage(SimpleNamespace(), "thread-1", _RunStore()) + + assert payload is not None + # The capacity only the file on disk can produce: the read was not stubbed away. + assert payload["max_context_tokens"] == _CAPACITY_ON_DISK + assert payload["token_count"] > 0 + assert payload["percentage"] == pytest.approx(round(payload["token_count"] / _CAPACITY_ON_DISK * 100, 1)) + assert loader.calls == 1 + assert not loader.on_loop_thread, "the live config read ran on the event loop thread" + + +async def test_offloaded_read_still_reports_an_unavailable_config(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Control: moving the read to a worker must not turn a 503 into a payload.""" + + def raise_unavailable() -> SimpleNamespace: + raise HTTPException(status_code=503, detail="Configuration not available") + + monkeypatch.setattr(context_usage, "get_config", raise_unavailable) + monkeypatch.setattr(context_usage, "build_thread_checkpoint_state_accessor", _stub_accessor_builder) + + assert await context_usage.build_context_usage(SimpleNamespace(), "thread-1", _RunStore()) is None + + +async def test_inline_config_read_actually_trips_the_gate(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Meta-check: the gate is live around this module, so the anchor above is not vacuous. + + The strict gate only trips when the blocking read sits under an ``app`` or + ``deerflow`` frame, so this calls the loader ``deps.get_config`` routes to. + That accessor maps every failure to 503, which would hide the + ``BlockingError``, hence the loader itself. + """ + from deerflow.config.app_config import get_app_config, reset_app_config + + monkeypatch.setenv("DEER_FLOW_CONFIG_PATH", str(tmp_path / "capacity.yaml")) + await asyncio.to_thread(reset_app_config) + try: + with pytest.raises(BlockingError): + get_app_config() + finally: + await asyncio.to_thread(reset_app_config)