From 1dd6d056278536af1cb4fc60347f4ca40fcb629b Mon Sep 17 00:00:00 2001 From: Nan Gao Date: Tue, 3 Nov 2026 15:06:52 +0800 Subject: [PATCH] fix(extensions): hand compaction observers the live task store and name the task in CompactionEvent (#7831) * feat(extension-api): add task identity fields to CompactionEvent (0.2.10) CompactionEvent carried the transform, its hashes and its counts, but nothing that names the task, run or thread the compaction happened in. An observer that wants to relate a compaction to the run that produced it had to match hashes across sessions, which is unreliable. Append three optional fields, task_id, run_id and thread_id (str | None, default None). Existing keyword and positional construction is unchanged, so already-released extensions keep working. The host fills them in the next commit. Bump deerflow-extension-api to 0.2.10 and the harness's exact pin with it. * fix(extensions): hand compaction observers the live task store and its identity The summarization middleware already recovers the live task store from the runtime to observe its summary system call, and the registry allocates that store precisely because a compaction observer is registered. notify_context_compacted() then discarded it and handed every observer a detached store, with a docstring claiming there was no live task. - The lead run worker and the subagent executor now seed every task store they allocate with the task's TaskInfo, not only when a lifecycle contributor is registered; the lifecycle notification reuses the seeded object. A subagent without a parent run_id still gets no TaskInfo. - compact_state() and acompact_state() pass the store from task_store_from_runtime() to _record_compaction(). - notify_context_compacted() takes an optional task_store. A live store goes to the observers and fills the event's task_id/run_id/thread_id from its TaskInfo; without one (manual /compact runs outside any run) observers still receive a detached store and the ids stay None. For a background subagent the store keeps its public scope id (the provider tool_call_id) while TaskInfo.task_id stays the execution id, as lifecycle contributors already see it; the event follows TaskInfo. * docs(extensions): document compaction event identity and seeded TaskInfo The extensions guide and the developer manual still said compaction observers always receive a detached store because no task is live at the call site. Describe what the host does now: lead runs and subagents with a parent run seed every task store with TaskInfo, compaction observers get the live store, and CompactionEvent's task_id/run_id/thread_id come from it; only manual /compact falls back to a detached store with None ids. Adds the 0.2.10 row to the contract version table (PR number pending). * docs(changelog): note compaction observers receive the live task identity * docs(extensions): list compaction observers among task-store triggers * docs: reference #7831 for the compaction event identity change * docs(extensions): backfill contract history rows 0.2.5 to 0.2.9 The version table jumped from 0.2.4 to 0.2.10 although 0.2.5-0.2.9 shipped. Add those rows, attributing names merged without a bump to the first release that carried them (#5796 in 0.2.5, #6920 and #5645 in 0.2.8, #7729 in 0.2.10). The page intro now says it documents 0.2.4 plus the 0.2.10 CompactionEvent fields, and points to the history for the rest. --- CHANGELOG.md | 10 + CHANGELOG_zh.md | 7 + .../deerflow_extension_api/__init__.py | 2 +- .../deerflow_extension_api/compaction.py | 10 + backend/packages/extension-api/pyproject.toml | 2 +- .../middlewares/summarization_middleware.py | 12 +- .../harness/deerflow/extensions/AGENTS.md | 13 +- .../harness/deerflow/extensions/notify.py | 24 +- .../harness/deerflow/runtime/runs/worker.py | 25 +- .../harness/deerflow/subagents/executor.py | 27 +- backend/packages/harness/pyproject.toml | 2 +- .../tests/test_compaction_event_identity.py | 286 ++++++++++++++++++ .../test_context_compaction_observation.py | 6 +- backend/tests/test_extension_api_contracts.py | 2 +- backend/tests/test_subagent_executor.py | 69 +++++ backend/uv.lock | 2 +- .../en/harness/extensions/observers.mdx | 10 +- .../en/harness/extensions/reference.mdx | 17 +- .../content/en/harness/extensions/runtime.mdx | 2 +- .../zh/harness/extensions/observers.mdx | 10 +- .../zh/harness/extensions/reference.mdx | 17 +- .../content/zh/harness/extensions/runtime.mdx | 2 +- 22 files changed, 495 insertions(+), 62 deletions(-) create mode 100644 backend/tests/test_compaction_event_identity.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 8505a8f218..a7f99f474b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,15 @@ This section accumulates work toward the **2.3.0** milestone ### Fixed +- **extensions:** Context-compaction observers now learn which task a + compaction happened in. Inside a lead run or subagent they receive that + task's store instead of a detached one, and `CompactionEvent` gains + `task_id`, `run_id` and `thread_id` (default `None`), copied from the + `TaskInfo` the host now seeds into every task store even when no lifecycle + contributor is registered. A manual `/compact` still gets a detached store + and `None` ids. `deerflow-extension-api` is now 0.2.10; existing + extensions keep working. ([#7831]) + - **channels:** A DingTalk reply in AI card mode (`card_template_id`) is no longer lost when a card request fails, for example without card permission or during a DingTalk outage. dingtalk-stream logs a failed card request and @@ -15998,3 +16007,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#7772]: https://github.com/bytedance/deer-flow/pull/7772 [#7781]: https://github.com/bytedance/deer-flow/pull/7781 [#7792]: https://github.com/bytedance/deer-flow/pull/7792 +[#7831]: https://github.com/bytedance/deer-flow/pull/7831 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index d1dd152155..e43319a185 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -20,6 +20,12 @@ ### 修复 +- **扩展：** 上下文压缩观察者现在能知道压缩发生在哪个任务里。在 Lead 运行或子 Agent + 内部，观察者收到的是该任务的存储，而不再是分离存储；`CompactionEvent` 新增 + `task_id`、`run_id`、`thread_id`（默认 `None`），取自宿主放进每个任务存储的 + `TaskInfo`，即使没有注册生命周期贡献者也会放入。手动 `/compact` 仍收到分离存储， + id 为 `None`。`deerflow-extension-api` 升级到 0.2.10，已有扩展不受影响。([#7831]) + - **渠道：** 钉钉 AI 卡片模式（`card_template_id`）下，卡片请求失败（例如缺少卡片权限或钉钉服务异常）时 不再丢失回复。dingtalk-stream 会记录失败的卡片请求后按成功返回，因此文本形式的运行中提示和最终答案的 `sampleMarkdown` 兜底都不会执行。现在渠道自行发送卡片请求，并在失败时回退。([#7781]) @@ -12745,3 +12751,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#7772]: https://github.com/bytedance/deer-flow/pull/7772 [#7781]: https://github.com/bytedance/deer-flow/pull/7781 [#7792]: https://github.com/bytedance/deer-flow/pull/7792 +[#7831]: https://github.com/bytedance/deer-flow/pull/7831 diff --git a/backend/packages/extension-api/deerflow_extension_api/__init__.py b/backend/packages/extension-api/deerflow_extension_api/__init__.py index d8d5c56f7b..4d735e4b59 100644 --- a/backend/packages/extension-api/deerflow_extension_api/__init__.py +++ b/backend/packages/extension-api/deerflow_extension_api/__init__.py @@ -102,7 +102,7 @@ from deerflow_extension_api.state import ExtensionData #: Contract version. Before 1.0, minors may break and patches are additive. #: From 1.0 on, bump the major for breaking changes. -API_VERSION = "0.2.9" +API_VERSION = "0.2.10" __all__ = [ "AGENT_RUNS_CONTEXT_KEY", diff --git a/backend/packages/extension-api/deerflow_extension_api/compaction.py b/backend/packages/extension-api/deerflow_extension_api/compaction.py index 3a4c233c3e..23547d171e 100644 --- a/backend/packages/extension-api/deerflow_extension_api/compaction.py +++ b/backend/packages/extension-api/deerflow_extension_api/compaction.py @@ -26,6 +26,13 @@ text later reaches a message (the ``durable_context_data`` block), but will not equal ``output_content_hash``. A consumer must join a compaction to what came after it through this event alone, never by re-hashing a later projection of the same text. + +``task_id`` / ``run_id`` / ``thread_id`` name the task the compaction happened in. The +host copies them from the ``TaskInfo`` it seeds into the live task store — the same +store the observer receives — so a consumer can key a compaction on +``(thread_id, run_id)`` instead of matching hashes across sessions. They stay ``None`` +when no task identity exists: a manual ``/compact`` outside any run (observers then +receive a detached store) or a subagent executed without a parent run. """ from __future__ import annotations @@ -46,6 +53,9 @@ class CompactionEvent: output_content_hash: str compacted_message_count: int kept_message_count: int + task_id: str | None = None + run_id: str | None = None + thread_id: str | None = None class ContextCompactionObserver(Protocol): diff --git a/backend/packages/extension-api/pyproject.toml b/backend/packages/extension-api/pyproject.toml index 399d992635..8f4030ce34 100644 --- a/backend/packages/extension-api/pyproject.toml +++ b/backend/packages/extension-api/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "deerflow-extension-api" -version = "0.2.9" +version = "0.2.10" description = "Public contracts for DeerFlow extensions" requires-python = ">=3.12" # Keep the contract package import-light and independent from the host. Public diff --git a/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py b/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py index 13a3f6279c..cac88e6ff4 100644 --- a/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py +++ b/backend/packages/harness/deerflow/agents/middlewares/summarization_middleware.py @@ -13,7 +13,7 @@ from dataclasses import dataclass from time import monotonic from typing import Any, Protocol, override, runtime_checkable -from deerflow_extension_api import CompactionEvent, canonical_hash +from deerflow_extension_api import CompactionEvent, ExtensionData, canonical_hash, task_store_from_runtime from langchain.agents import AgentState from langchain.agents.middleware import SummarizationMiddleware from langchain_core.messages import AnyMessage, HumanMessage, RemoveMessage, SystemMessage, get_buffer_string @@ -803,6 +803,7 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware): summary: str, compacted_message_count: int, kept_message_count: int, + task_store: ExtensionData | None, ) -> None: event = CompactionEvent( transform_kind=_COMPACTION_TRANSFORM_KIND, @@ -812,7 +813,7 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware): compacted_message_count=compacted_message_count, kept_message_count=kept_message_count, ) - notify_context_compacted(event, extensions=self._extensions) + notify_context_compacted(event, extensions=self._extensions, task_store=task_store) def _record_summary_telemetry(self, runtime: Runtime, previous_summary: str | None, summary: str, summarized_count: int, *, llm_call_skipped: bool = False) -> None: """Count successful summary results, unchanged results, and no-op cache reuse. @@ -894,6 +895,7 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware): summary=summary, compacted_message_count=len(messages_to_summarize), kept_message_count=len(preserved_messages), + task_store=task_store_from_runtime(runtime), ) task_history = None if self._task_continuity_config is not None: @@ -921,13 +923,12 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware): if prepared is None: return None messages_to_summarize, preserved_messages, previous_summary, total_tokens = prepared - from deerflow_extension_api import task_store_from_runtime - + task_store = task_store_from_runtime(runtime) source_content_hashes = self._freeze_compaction_sources(messages_to_summarize) summary = await self._asummarize_with( messages_to_summarize, previous_summary=previous_summary, - task_store=task_store_from_runtime(runtime), + task_store=task_store, runtime=runtime, ) if summary is None: @@ -944,6 +945,7 @@ class DeerFlowSummarizationMiddleware(SummarizationMiddleware): summary=summary, compacted_message_count=len(messages_to_summarize), kept_message_count=len(preserved_messages), + task_store=task_store, ) task_history = None if self._task_continuity_config is not None: diff --git a/backend/packages/harness/deerflow/extensions/AGENTS.md b/backend/packages/harness/deerflow/extensions/AGENTS.md index 121eee95a3..79ae3a903e 100644 --- a/backend/packages/harness/deerflow/extensions/AGENTS.md +++ b/backend/packages/harness/deerflow/extensions/AGENTS.md @@ -234,10 +234,12 @@ both sides when the extension must observe both synchronous and asynchronous exe paths. Lead runs and subagents allocate an `ExtensionData` task store only when middleware, -task-lifecycle, or system-model observation is registered; services and routers are +task-lifecycle, system-model or compaction observation is registered; services and routers are app-scoped and do not allocate one. Middleware and system-call sites recover the live store through `EXTENSION_TASK_STORE_KEY` / `task_store_from_runtime()`; lifecycle -contributors receive that same store directly. Each task resolves the immutable +contributors receive that same store directly. Both hosts seed each store with the task's +`TaskInfo` whether or not a lifecycle contributor exists (subagents only with a parent +`run_id`), so any hook holding the store can name its task. Each task resolves the immutable loaded-extension snapshot once and binds that same object through task-store allocation, hooks, and synchronous agent construction, so a concurrent singleton replacement cannot mix two extension generations without changing the LangGraph graph-factory ABI. The @@ -306,9 +308,10 @@ permanently empty. `notify_context_compacted()` is a synchronous, fire-and-forget entry point — both the sync and async compaction paths call it without an `await` — that dispatches to the same registered extension-notification loop system-model-call cancellation uses, reusing `_notify_each`'s per-observer fail-open -containment. There is no live task to attach at that call site, so observers receive a -detached task store, the same fallback `notify_system_model_call` uses when its caller -supplies none. +containment. Both compaction paths pass the live store from `task_store_from_runtime()`; +observers receive it and the event's `task_id`/`run_id`/`thread_id` come from its seeded +`TaskInfo`. Manual `/compact` runs outside any task, so it falls back to a detached store +with `None` ids, as `notify_system_model_call` does. Gateway services start in registration order after the persistence engine and session factory are ready. Ungranted services share an `ExtensionRuntimeDeps` snapshot containing the diff --git a/backend/packages/harness/deerflow/extensions/notify.py b/backend/packages/harness/deerflow/extensions/notify.py index c652c7a32f..02af39363c 100644 --- a/backend/packages/harness/deerflow/extensions/notify.py +++ b/backend/packages/harness/deerflow/extensions/notify.py @@ -4,6 +4,7 @@ from __future__ import annotations import asyncio import concurrent.futures +import dataclasses import logging import threading import time @@ -529,7 +530,11 @@ def dispatch_system_model_observation( coro.close() -def notify_context_compacted(event: CompactionEvent, extensions: LoadedExtensions | None = None) -> None: +def notify_context_compacted( + event: CompactionEvent, + extensions: LoadedExtensions | None = None, + task_store: ExtensionData | None = None, +) -> None: """Fan a completed compaction out to observers, fire-and-forget. The compaction seam sits in the summarization middleware's ``before_model`` / @@ -539,10 +544,12 @@ def notify_context_compacted(event: CompactionEvent, extensions: LoadedExtension loop the same non-blocking way a synchronous system-model-call cancellation does, reusing the same fail-open cancellation containment inside ``_notify_each``. - There is no live task for this hook to attach observers to (unlike lifecycle or - system-model-call notification, which run from an awaited call site holding the - real task store), so observers receive a detached store — the same fallback - ``notify_system_model_call`` uses when its caller has none. + Inside a lead run or subagent the caller passes the live task store it recovered + from the runtime context — the same store the summary system call is observed + with. Observers receive it, and the event's ``task_id``/``run_id``/``thread_id`` + are filled from the ``TaskInfo`` the host seeded into it. A compaction outside any + task (manual ``/compact``) passes none, so observers receive a detached store — + the same fallback ``notify_system_model_call`` uses — and the ids stay ``None``. """ resolved = extensions if resolved is None: @@ -553,7 +560,12 @@ def notify_context_compacted(event: CompactionEvent, extensions: LoadedExtension if not observers: return app_store = resolved.app_store - task_store = ExtensionData("detached") + if task_store is None: + task_store = ExtensionData("detached") + else: + info = task_store.get(TaskInfo) + if info is not None: + event = dataclasses.replace(event, task_id=info.task_id, run_id=info.run_id, thread_id=info.thread_id) what = f"compaction ({event.transform_kind})" dispatch_system_model_observation( _notify_each( diff --git a/backend/packages/harness/deerflow/runtime/runs/worker.py b/backend/packages/harness/deerflow/runtime/runs/worker.py index a35759115f..fc4fe59409 100644 --- a/backend/packages/harness/deerflow/runtime/runs/worker.py +++ b/backend/packages/harness/deerflow/runtime/runs/worker.py @@ -1238,23 +1238,26 @@ async def run_agent( task_id = lead_task_id(run_id) if extensions.needs_task_store: - task_store = ExtensionData(task_id) - - if extensions.has_task_lifecycle: - task_info = TaskInfo( + identity = TaskInfo( task_id=task_id, run_id=run_id, thread_id=thread_id, kind="lead", agent_name=record.assistant_id, ) - assert task_store is not None - await notify_task_start( - extensions, - task_store, - task_info, - timeout=_EXTENSION_TASK_NOTIFY_TIMEOUT_SECONDS, - ) + task_store = ExtensionData(task_id) + # Seed the identity for every hook handed this store, not only + # lifecycle contributors: compaction events are named from it. + task_store.set(identity) + # A lifecycle contributor always implies a task store. + if extensions.has_task_lifecycle: + task_info = identity + await notify_task_start( + extensions, + task_store, + task_info, + timeout=_EXTENSION_TASK_NOTIFY_TIMEOUT_SECONDS, + ) if not record.ownership_lost and thread_store is not None: try: diff --git a/backend/packages/harness/deerflow/subagents/executor.py b/backend/packages/harness/deerflow/subagents/executor.py index 6b726cb9e2..2661713a08 100644 --- a/backend/packages/harness/deerflow/subagents/executor.py +++ b/backend/packages/harness/deerflow/subagents/executor.py @@ -1748,17 +1748,22 @@ class SubagentExecutor: task_info: TaskInfo | None = None if loaded_extensions.needs_task_store: task_store = ExtensionData(result.external_task_id or result.task_id) - if loaded_extensions.has_task_lifecycle and self.run_id: - task_info = TaskInfo( - task_id=result.task_id, - run_id=self.run_id, - thread_id=self.thread_id or "", - kind="subagent", - parent_task_id=lead_task_id(self.run_id), - agent_name=self.config.name, - ) - assert task_store is not None - elif loaded_extensions.has_task_lifecycle: + if self.run_id: + identity = TaskInfo( + task_id=result.task_id, + run_id=self.run_id, + thread_id=self.thread_id or "", + kind="subagent", + parent_task_id=lead_task_id(self.run_id), + agent_name=self.config.name, + ) + # Seed the identity for every hook handed this store, not only + # lifecycle contributors: compaction events are named from it. + task_store.set(identity) + # A lifecycle contributor always implies a task store. + if loaded_extensions.has_task_lifecycle: + task_info = identity + if loaded_extensions.has_task_lifecycle and not self.run_id: logger.debug( "[trace=%s] Subagent %s has no run_id; skipping extension task lifecycle", self.trace_id, diff --git a/backend/packages/harness/pyproject.toml b/backend/packages/harness/pyproject.toml index 435bec3f5d..0fafb4312b 100644 --- a/backend/packages/harness/pyproject.toml +++ b/backend/packages/harness/pyproject.toml @@ -13,7 +13,7 @@ dependencies = [ # the contract version it implements, extensions declare ranges. A range # here would let pip resolve a newer contract package than this harness # implements, making newer extensions look supported at runtime. - "deerflow-extension-api==0.2.9", + "deerflow-extension-api==0.2.10", "dotenv>=0.9.9", "exa-py>=1.0.0", "httpx>=0.28.0", diff --git a/backend/tests/test_compaction_event_identity.py b/backend/tests/test_compaction_event_identity.py new file mode 100644 index 0000000000..2050f6abd6 --- /dev/null +++ b/backend/tests/test_compaction_event_identity.py @@ -0,0 +1,286 @@ +"""A compaction event names the task, run and thread it happened in. + +The host already holds the live task store when summarization runs: the summary +system call is observed with it. A compaction observer that cannot name its task +can only correlate a compaction with the run that produced it by matching hashes +across sessions, which is unreliable, so the host hands the observer the store it +already has and copies the seeded ``TaskInfo`` identity onto the event. +""" + +from __future__ import annotations + +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import pytest +from deerflow_extension_api import EXTENSION_TASK_STORE_KEY, CompactionEvent, ExtensionData, TaskInfo +from langchain_core.messages import AIMessage, HumanMessage +from langgraph.checkpoint.memory import InMemorySaver + +from deerflow.agents.middlewares.summarization_middleware import DeerFlowSummarizationMiddleware +from deerflow.extensions import notify +from deerflow.extensions.registry import ExtensionRegistry +from deerflow.runtime import context_compaction +from deerflow.runtime.runs.manager import RunManager +from deerflow.runtime.runs.schemas import RunStatus +from deerflow.runtime.runs.worker import RunContext, run_agent + + +class _CompactionRecorder: + def __init__(self) -> None: + self.calls: list[tuple[ExtensionData, CompactionEvent]] = [] + self._arrived = asyncio.Event() + + async def on_context_compacted(self, app_store, task_store, event): + self.calls.append((task_store, event)) + self._arrived.set() + + async def received(self) -> list[tuple[ExtensionData, CompactionEvent]]: + # Dispatch is fire-and-forget onto the notify loop, so the observer may + # still be pending when the compaction call returns. + await asyncio.wait_for(self._arrived.wait(), 5) + return self.calls + + +class _LifecycleRecorder: + def __init__(self) -> None: + self.infos: list[TaskInfo] = [] + + async def on_task_start(self, app_store, task_store, info): + self.infos.append(info) + + async def on_task_stop(self, app_store, task_store, info, outcome): + return None + + +def _extensions(observer, *, lifecycle=None): + registry = ExtensionRegistry() + with registry.attributed_to("identity:install"): + registry.context_compaction_observer(observer) + if lifecycle is not None: + registry.task_lifecycle(lifecycle) + return registry.build() + + +def _messages() -> list: + return [ + HumanMessage(content="user-1"), + AIMessage(content="assistant-1"), + HumanMessage(content="user-2"), + AIMessage(content="assistant-2"), + ] + + +def _middleware(extensions) -> DeerFlowSummarizationMiddleware: + model = MagicMock() + model.invoke.return_value = SimpleNamespace(text="compressed summary") + model.ainvoke = AsyncMock(return_value=SimpleNamespace(text="compressed summary")) + model.with_config.return_value = model + return DeerFlowSummarizationMiddleware( + model=model, + trigger=("messages", 4), + keep=("messages", 2), + token_counter=len, + extensions=extensions, + ) + + +def _identity(event: CompactionEvent) -> tuple[str | None, str | None, str | None]: + return event.task_id, event.run_id, event.thread_id + + +# --- The contract stays additive ------------------------------------------- + + +def test_an_event_built_the_old_way_still_constructs_without_identity(): + keyword = CompactionEvent( + transform_kind="summarization", + transform_version="1", + source_content_hashes=("h1",), + output_content_hash="h2", + compacted_message_count=1, + kept_message_count=1, + ) + positional = CompactionEvent("summarization", "1", ("h1",), "h2", 1, 1) + + assert keyword == positional + assert _identity(keyword) == (None, None, None) + + +# --- Dispatch ---------------------------------------------------------------- + + +class TestNotifyHandsOverTheLiveStore: + @pytest.mark.asyncio + async def test_a_live_store_reaches_the_observer_and_names_the_event(self, monkeypatch): + monkeypatch.setattr(notify, "_notify_loop", asyncio.get_running_loop()) + observer = _CompactionRecorder() + store = ExtensionData("run-1") + store.set(TaskInfo(task_id="run-1", run_id="run-1", thread_id="thread-1", kind="lead")) + + notify.notify_context_compacted(CompactionEvent("summarization", "1", (), "h", 0, 0), extensions=_extensions(observer), task_store=store) + + [(received_store, event)] = await observer.received() + assert received_store is store + assert _identity(event) == ("run-1", "run-1", "thread-1") + + @pytest.mark.asyncio + async def test_a_store_without_task_info_is_handed_over_without_identity(self, monkeypatch): + # A subagent with no parent run allocates a store but never seeds TaskInfo. + monkeypatch.setattr(notify, "_notify_loop", asyncio.get_running_loop()) + observer = _CompactionRecorder() + store = ExtensionData("sub-1") + + notify.notify_context_compacted(CompactionEvent("summarization", "1", (), "h", 0, 0), extensions=_extensions(observer), task_store=store) + + [(received_store, event)] = await observer.received() + assert received_store is store + assert _identity(event) == (None, None, None) + + @pytest.mark.asyncio + async def test_no_store_falls_back_to_detached(self, monkeypatch): + monkeypatch.setattr(notify, "_notify_loop", asyncio.get_running_loop()) + observer = _CompactionRecorder() + + notify.notify_context_compacted(CompactionEvent("summarization", "1", (), "h", 0, 0), extensions=_extensions(observer)) + + [(received_store, event)] = await observer.received() + assert received_store.scope_id == "detached" + assert _identity(event) == (None, None, None) + + +# --- End to end through the hosts that allocate the store ----------------- + + +class _CompactingLeadAgent: + """Stands in for the lead graph: runs the real summarization middleware once + against the runtime context ``run_agent`` hands the graph.""" + + def __init__(self, middleware: DeerFlowSummarizationMiddleware) -> None: + self.middleware = middleware + self.task_store: ExtensionData | None = None + + async def astream(self, graph_input, config=None, stream_mode=None, subgraphs=False): + context = (config or {}).get("context") + self.task_store = context.get(EXTENSION_TASK_STORE_KEY) + assert await self.middleware.abefore_model({"messages": _messages()}, SimpleNamespace(context=context)) is not None + yield {"messages": []} + + +def _bridge(): + return SimpleNamespace(publish=AsyncMock(), publish_end=AsyncMock(), cleanup=AsyncMock()) + + +async def _run_lead(extensions, agent): + manager = RunManager() + record = await manager.create("thread-identity", assistant_id="custom-agent") + await run_agent( + _bridge(), + manager, + record, + ctx=RunContext(checkpointer=InMemorySaver(), extensions=extensions), + agent_factory=lambda *, config: agent, + graph_input={}, + config={}, + ) + assert record.status is RunStatus.success + return record + + +class TestLeadRunCompaction: + @pytest.mark.asyncio + async def test_async_compaction_names_the_run_it_happened_in(self, monkeypatch): + monkeypatch.setattr(notify, "_notify_loop", asyncio.get_running_loop()) + observer = _CompactionRecorder() + lifecycle = _LifecycleRecorder() + extensions = _extensions(observer, lifecycle=lifecycle) + agent = _CompactingLeadAgent(_middleware(extensions)) + + record = await _run_lead(extensions, agent) + + [(store, event)] = await observer.received() + [info] = lifecycle.infos + assert store is agent.task_store + assert store.scope_id == info.task_id == record.run_id + assert store.get(TaskInfo) == info + assert _identity(event) == (info.task_id, info.run_id, info.thread_id) == (record.run_id, record.run_id, "thread-identity") + + @pytest.mark.asyncio + async def test_identity_does_not_depend_on_a_lifecycle_contributor(self, monkeypatch): + """Registering a compaction observer alone allocates the store; the host + must not drop the identity just because no lifecycle contributor asked.""" + monkeypatch.setattr(notify, "_notify_loop", asyncio.get_running_loop()) + observer = _CompactionRecorder() + extensions = _extensions(observer) + assert extensions.needs_task_store and not extensions.has_task_lifecycle + agent = _CompactingLeadAgent(_middleware(extensions)) + + record = await _run_lead(extensions, agent) + + [(store, event)] = await observer.received() + assert store is agent.task_store + assert store.scope_id == record.run_id + assert store.get(TaskInfo) == TaskInfo( + task_id=record.run_id, + run_id=record.run_id, + thread_id="thread-identity", + kind="lead", + agent_name="custom-agent", + ) + assert _identity(event) == (record.run_id, record.run_id, "thread-identity") + + +class TestSyncCompaction: + @pytest.mark.asyncio + async def test_sync_compaction_carries_the_live_task_identity(self, monkeypatch): + monkeypatch.setattr(notify, "_notify_loop", asyncio.get_running_loop()) + observer = _CompactionRecorder() + store = ExtensionData("run-sync") + info = TaskInfo(task_id="run-sync", run_id="run-sync", thread_id="thread-sync", kind="lead") + store.set(info) + runtime = SimpleNamespace(context={"thread_id": "thread-sync", EXTENSION_TASK_STORE_KEY: store}) + middleware = _middleware(_extensions(observer)) + + # The synchronous graph path runs before_model off the serving loop. + result = await asyncio.to_thread(middleware.before_model, {"messages": _messages()}, runtime) + + assert result is not None + [(received_store, event)] = await observer.received() + assert received_store is store + assert _identity(event) == (info.task_id, info.run_id, info.thread_id) + + +class _Accessor: + def __init__(self, values: dict) -> None: + self.snapshot = SimpleNamespace( + values=values, + config={"configurable": {"thread_id": "thread-manual", "checkpoint_id": "ckpt-old", "checkpoint_ns": ""}}, + metadata={"step": 4}, + ) + + async def aget(self, _config): + return self.snapshot + + async def aget_tuple(self, _config): + return SimpleNamespace(checkpoint={"channel_values": dict(self.snapshot.values)}) + + async def aupdate(self, config, values, *, as_node=None): + return {"configurable": {"thread_id": "thread-manual", "checkpoint_ns": "", "checkpoint_id": "ckpt-compacted"}} + + +class TestManualCompaction: + @pytest.mark.asyncio + async def test_manual_compact_outside_any_run_falls_back_to_a_detached_store(self, monkeypatch): + monkeypatch.setattr(notify, "_notify_loop", asyncio.get_running_loop()) + observer = _CompactionRecorder() + middleware = _middleware(_extensions(observer)) + monkeypatch.setattr(context_compaction, "_create_compaction_middleware", lambda **_kwargs: middleware) + + result = await context_compaction.compact_thread_context(_Accessor({"messages": _messages()}), "thread-manual", app_config=SimpleNamespace()) + + assert result.compacted + [(store, event)] = await observer.received() + assert store.scope_id == "detached" + assert store.get(TaskInfo) is None + assert _identity(event) == (None, None, None) diff --git a/backend/tests/test_context_compaction_observation.py b/backend/tests/test_context_compaction_observation.py index 690ae3800f..908d987b19 100644 --- a/backend/tests/test_context_compaction_observation.py +++ b/backend/tests/test_context_compaction_observation.py @@ -133,7 +133,7 @@ class TestSummarizationEmitsTheEvent: monkeypatch.setattr( summarization_middleware, "notify_context_compacted", - lambda event, extensions=None: events.append(event), + lambda event, extensions=None, task_store=None: events.append(event), ) middleware = _middleware() @@ -159,7 +159,7 @@ class TestSummarizationEmitsTheEvent: monkeypatch.setattr( summarization_middleware, "notify_context_compacted", - lambda event, extensions=None: events.append(event), + lambda event, extensions=None, task_store=None: events.append(event), ) # A trigger threshold far above the message count never fires, so # compaction never runs and the record half is never reached. @@ -204,7 +204,7 @@ class TestAnInstallWithNoObserverPaysNothing: from deerflow.extensions import get_agent_build_extensions events = [] - monkeypatch.setattr(summarization_middleware, "notify_context_compacted", lambda event, extensions=None: events.append(event)) + monkeypatch.setattr(summarization_middleware, "notify_context_compacted", lambda event, extensions=None, task_store=None: events.append(event)) unobserved = replace(get_agent_build_extensions(), context_compaction_observers=()) result = await _middleware(extensions=unobserved).abefore_model({"messages": _messages()}, _runtime()) diff --git a/backend/tests/test_extension_api_contracts.py b/backend/tests/test_extension_api_contracts.py index d3152e2b87..f06cd38a35 100644 --- a/backend/tests/test_extension_api_contracts.py +++ b/backend/tests/test_extension_api_contracts.py @@ -342,7 +342,7 @@ def test_runtime_api_version_matches_the_installed_contract_package(): """Every additive contract slice bumps both gates together.""" from importlib.metadata import version - assert API_VERSION == "0.2.9" + assert API_VERSION == "0.2.10" assert API_VERSION == version("deerflow-extension-api") diff --git a/backend/tests/test_subagent_executor.py b/backend/tests/test_subagent_executor.py index 26234513b1..a62499c305 100644 --- a/backend/tests/test_subagent_executor.py +++ b/backend/tests/test_subagent_executor.py @@ -31,6 +31,7 @@ from unittest.mock import AsyncMock, MagicMock, patch import pytest from packaging.version import Version +from deerflow.agents.middlewares.summarization_middleware import DeerFlowSummarizationMiddleware from deerflow.agents.middlewares.tool_receipt_middleware import ToolReceiptMiddleware from deerflow.runtime.journal import RunJournal from deerflow.sandbox.lease import SandboxLeaseManager @@ -7853,3 +7854,71 @@ class TestSubagentCancellationLifecycle: assert holder.execution_done_event.is_set() assert capacity.snapshot().running == 0 assert lifecycle.stops == [(lifecycle.starts[0], TaskOutcome.ABORTED)] + + +class TestSubagentCompactionIdentity: + """A compaction inside a subagent names the subagent's own task. + + The summarization middleware is imported at module level: this module + replaces ``deerflow.agents.middlewares`` with a double while the executor + runs, so only an already-imported module is the real one. + """ + + @pytest.mark.anyio + @pytest.mark.parametrize( + "run_id,external_task_id", + [("parent-run", None), ("parent-run", "call-1"), (None, None)], + ids=["sync", "background", "no-parent-run"], + ) + async def test_compaction_carries_the_seeded_task_identity(self, classes, base_config, monkeypatch, run_id, external_task_id): + from deerflow_extension_api import EXTENSION_TASK_STORE_KEY, TaskInfo + + from deerflow.extensions import notify + from deerflow.extensions.registry import ExtensionRegistry + + monkeypatch.setattr(notify, "_notify_loop", asyncio.get_running_loop()) + received, arrived = [], asyncio.Event() + + class Observer: + async def on_context_compacted(self, app_store, task_store, event): + received.append((task_store, event)) + arrived.set() + + registry = ExtensionRegistry() + with registry.attributed_to("identity:test"): + registry.context_compaction_observer(Observer()) + extensions = registry.build() + model = MagicMock() + model.ainvoke = AsyncMock(return_value=SimpleNamespace(text="compressed summary")) + model.with_config.return_value = model + middleware = DeerFlowSummarizationMiddleware(model=model, trigger=("messages", 4), keep=("messages", 2), token_counter=len, extensions=extensions) + bound = {} + + class Agent: + async def astream(self, *args, context, **kwargs): + bound["store"] = context[EXTENSION_TASK_STORE_KEY] + messages = [classes["HumanMessage"](content="user-1"), classes["AIMessage"](content="assistant-1"), classes["HumanMessage"](content="user-2"), classes["AIMessage"](content="assistant-2")] + assert await middleware.abefore_model({"messages": messages}, SimpleNamespace(context=context)) is not None + yield {"messages": [classes["AIMessage"](content="done")]} + + executor = classes["SubagentExecutor"](config=base_config, tools=[], thread_id="sub-thread", run_id=run_id, extensions=extensions) + monkeypatch.setattr(executor, "_build_initial_state", AsyncMock(return_value=({}, [], None))) + monkeypatch.setattr(executor, "_create_agent", AsyncMock(return_value=Agent())) + holder = classes["SubagentResult"](task_id="exec-1", trace_id="trace", status=classes["SubagentStatus"].PENDING, external_task_id=external_task_id) + + result = await executor._aexecute("Task", holder) + + assert result.status is classes["SubagentStatus"].COMPLETED + await asyncio.wait_for(arrived.wait(), 5) + [(store, event)] = received + assert store is bound["store"] + # The store keeps its public scope contract: the provider correlation id + # when there is one, otherwise the execution id. + assert store.scope_id == (external_task_id or "exec-1") + if run_id is None: + assert store.get(TaskInfo) is None + assert (event.task_id, event.run_id, event.thread_id) == (None, None, None) + return + info = store.get(TaskInfo) + assert info == TaskInfo(task_id="exec-1", run_id=run_id, thread_id="sub-thread", kind="subagent", parent_task_id=run_id, agent_name=base_config.name) + assert (event.task_id, event.run_id, event.thread_id) == (info.task_id, info.run_id, info.thread_id) diff --git a/backend/uv.lock b/backend/uv.lock index c0ee723f79..7444972402 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -967,7 +967,7 @@ extensions = [] [[package]] name = "deerflow-extension-api" -version = "0.2.9" +version = "0.2.10" source = { editable = "packages/extension-api" } [[package]] diff --git a/frontend/src/content/en/harness/extensions/observers.mdx b/frontend/src/content/en/harness/extensions/observers.mdx index e1c862861b..12ac377287 100644 --- a/frontend/src/content/en/harness/extensions/observers.mdx +++ b/frontend/src/content/en/harness/extensions/observers.mdx @@ -14,7 +14,7 @@ Middleware sees each model and tool call inside an agent. The four contributions | `TaskLifecycleContributor` | `registry.task_lifecycle()` | Start and stop of every lead run and subagent | async, awaited | The task store | | `SystemModelCallObserver` | `registry.system_model_observer()` | After each DeerFlow-owned model call | async | The task store, or detached | | `AgentAssemblyObserver` | `registry.agent_assembly_observer()` | At the end of every agent construction | **sync** | The app store only | -| `ContextCompactionObserver` | `registry.context_compaction_observer()` | After each summarization | async, fire-and-forget | Detached | +| `ContextCompactionObserver` | `registry.context_compaction_observer()` | After each summarization | async, fire-and-forget | The task store, or detached | All the examples on this page come from one extension that registers all four: @@ -312,10 +312,13 @@ Summarization replaces many messages with one summary. Afterwards, nothing in st | `output_content_hash` | `canonical_hash` of the summary text | | `compacted_message_count` | How many messages were removed | | `kept_message_count` | How many messages were kept | +| `task_id` | The task the compaction happened in, or `None` outside a task | +| `run_id` | That task's run, or `None` | +| `thread_id` | That task's thread, or `None` | To match an event against messages you hold, hash exactly the same way: `canonical_hash(message.content)`, passing the content itself. Never stringify it first, because multimodal content is a list of dicts and `str()` depends on key order. Do not try to rediscover the summary later by hashing what the model is shown: the prompt carries a bounded, escaped rendering of the summary, whose hash will not match `output_content_hash`. -The notification is fire-and-forget. It is dispatched to the notification loop without blocking the model turn, and the observer receives a **detached** store, because there is no live task at that call site. When no compaction observer is registered, the host skips the hashing pass too. +The notification is fire-and-forget. It is dispatched to the notification loop without blocking the model turn. Inside a lead run or subagent, the observer receives that task's store, and `task_id`, `run_id`, and `thread_id` are copied from the `TaskInfo` the host keeps in it, so you can key a compaction on `(thread_id, run_id)` instead of matching hashes. A manual `/compact` runs outside any task, so the observer receives a **detached** store and the three ids are `None`. A subagent started without a parent run passes its own store but has no `TaskInfo`, so its ids are `None` too. When no compaction observer is registered, the host skips the hashing pass too. ### Example @@ -323,12 +326,13 @@ The notification is fire-and-forget. It is dispatched to the notification loop w class CompactionLogger: async def on_context_compacted(self, app_store: ExtensionData, task_store: ExtensionData, event: CompactionEvent) -> None: logger.info( - "%s v%s folded %d message(s) into summary %s, kept %d", + "%s v%s folded %d message(s) into summary %s, kept %d (run %s)", event.transform_kind, event.transform_version, event.compacted_message_count, event.output_content_hash[:12], event.kept_message_count, + event.run_id or "none", ) ``` diff --git a/frontend/src/content/en/harness/extensions/reference.mdx b/frontend/src/content/en/harness/extensions/reference.mdx index e2d61341ea..5a0ed43fcd 100644 --- a/frontend/src/content/en/harness/extensions/reference.mdx +++ b/frontend/src/content/en/harness/extensions/reference.mdx @@ -7,7 +7,7 @@ import { Callout } from "nextra/components"; # Reference -This page lists every name in `deerflow_extension_api.__all__` for contract version **0.2.4**. Import them from the package root: +This page documents every name in `deerflow_extension_api.__all__` as of contract version **0.2.4**, plus the `CompactionEvent` identity fields added in 0.2.10. The other names added since 0.2.4 are listed under [Version history](#version-history) but are not described on this page yet. Import any name from the package root: ```python from deerflow_extension_api import ExtensionRegistry, MiddlewarePlacement, Placement @@ -143,7 +143,7 @@ class TaskLifecycleContributor(Protocol): ### `TaskInfo` -Frozen dataclass. +Frozen dataclass. The host also stores it in every task store it allocates (for a subagent, only when it has a parent run), so `task_store.get(TaskInfo)` names the task from any hook that receives the store. | Field | Type | Default | | ---------------- | ----------------------------- | ------- | @@ -260,8 +260,11 @@ Frozen dataclass, captured while both sides of the transform still exist. | `output_content_hash` | `str` | | `compacted_message_count` | `int` | | `kept_message_count` | `int` | +| `task_id` | `str \| None = None` | +| `run_id` | `str \| None = None` | +| `thread_id` | `str \| None = None` | -Hashes are `canonical_hash(message.content)` of the content passed directly, never a stringified copy. To match a message to an event, hash its `content` the same way. +The three ids come from the `TaskInfo` in the task store the observer receives. They are `None` when there is no task identity: a manual `/compact`, or a subagent without a parent run. Hashes are `canonical_hash(message.content)` of the content passed directly, never a stringified copy. To match a message to an event, hash its `content` the same way. ## Services @@ -502,7 +505,7 @@ Frozen dataclass describing one non-secret deployment setting. ### `API_VERSION` -The host's contract version as a dotted string, `"0.2.4"` for the contract this page describes. Always equal to the package version in `backend/packages/extension-api/pyproject.toml`. +The host's contract version as a dotted string, such as `"0.2.10"`. Always equal to the package version in `backend/packages/extension-api/pyproject.toml`. ## Compatibility rules @@ -528,5 +531,11 @@ The host's contract version as a dotted string, `"0.2.4"` for the contract this | 0.2.2 | [#5647](https://github.com/bytedance/deer-flow/pull/5647) | Experimental `registry.plugin()` with `PluginContribution`, `BrowserModule`, `BackendAction`, `ModelTool`, `ActionContext`, `ToolContext`, `SettingsField` | | 0.2.3 | [#5727](https://github.com/bytedance/deer-flow/pull/5727), [#5685](https://github.com/bytedance/deer-flow/pull/5685) | Request-scoped run evidence: `resolve_run_evidence_reader`, `require_run_evidence_reader`, `RUN_EVIDENCE_READER_RESOLVER_KEY` (#5727). Packaged browser resources: `BrowserAssets`, and `PluginContribution.frontend` also accepts it (#5685) | | 0.2.4 | [#5842](https://github.com/bytedance/deer-flow/pull/5842) | Plugin resource authorization: `require_plugin_management`, `arequire_plugin_management`, `EXTENSION_PLUGIN_AUTHZ_RESOLVER_KEY`, `EXTENSION_PLUGIN_AUTHZ_RESOLVER_ASYNC_KEY` | +| 0.2.5 | [#5796](https://github.com/bytedance/deer-flow/pull/5796), [#6190](https://github.com/bytedance/deer-flow/pull/6190) | Bounded host model calls: `ExtensionRuntimeDeps.model_invoker` with `ModelInvoker`, `ModelMessage`, `ModelInvocationRequest`, `ModelInvocationResult`, `ModelUsage`, and the errors `ModelInvocationError`, `ModelInvocationUnavailable`, `ModelInvocationUnauthorized`, `ModelInvocationFailed`, `ModelOutputValidationError` (#5796). Host-bound Agent run control: `AgentRuns`, `AgentRun`, `AgentRunError`, `resolve_agent_runs`, `require_agent_runs`, `AGENT_RUNS_RESOLVER_KEY`, `AGENT_RUNS_CONTEXT_KEY`, `ActionContext.agent_runs` (#6190) | +| 0.2.6 | [#6861](https://github.com/bytedance/deer-flow/pull/6861) | `PluginToolResult` and `ModelTool.response_format`, so a plugin tool can return a JSON artifact with its text | +| 0.2.7 | [#6904](https://github.com/bytedance/deer-flow/pull/6904) | `ExtensionRegistry.knowledge_provider()` for read-only knowledge retrieval providers | +| 0.2.8 | [#6920](https://github.com/bytedance/deer-flow/pull/6920), [#5645](https://github.com/bytedance/deer-flow/pull/5645), [#7075](https://github.com/bytedance/deer-flow/pull/7075) | `AgentRuns.configuration_evidence()`, `read_artifact()`, `configuration_revision()`, and `ActionContext.interaction` (#6920). Completed-run evidence and host-owned updates of existing custom skills: `ExtensionRuntimeDeps.completed_run_evidence` and `skill_mutations`, with `CompletedRunEvidenceReader`, `CompletedRunSnapshot`, `CompletedRunEvent`, `CompletedRunEventPage`, `EvidenceLimits`, `SkillMutationService`, `MutationCapabilities`, `Proposal`, `ProposalBundle`, `BundleFile`, `Operation`, `AssessmentRef`, `AssetRevision`, `HostCheckResult`, `SkillRevisionView`, and the shared `HostCapabilityError` (#5645). The `durable` argument of `configuration_revision()` (#7075) | +| 0.2.9 | [#7138](https://github.com/bytedance/deer-flow/pull/7138) | `ExecutionFence`, and the `execution_fence` argument of `AgentRuns.start()` and `cancel()` | +| 0.2.10 | [#7729](https://github.com/bytedance/deer-flow/pull/7729), [#7831](https://github.com/bytedance/deer-flow/pull/7831) | `AgentRuns.send_input()`, `list_inputs()`, `authorize_control()`, `dispatch_input()` (#7729). `CompactionEvent.task_id`, `run_id`, `thread_id`; compaction observers receive the live task store, and every task store carries its `TaskInfo` (#7831) | No release has removed a public name. diff --git a/frontend/src/content/en/harness/extensions/runtime.mdx b/frontend/src/content/en/harness/extensions/runtime.mdx index f0d39eea58..9fca4f1eb5 100644 --- a/frontend/src/content/en/harness/extensions/runtime.mdx +++ b/frontend/src/content/en/harness/extensions/runtime.mdx @@ -107,7 +107,7 @@ The app store is shared by every extension in the process. It is safe because st ### Detached stores -Some notifications have no live task. Memory extraction normally runs on a background thread after the run, and compaction observers are always dispatched outside the turn. These receive a fresh store whose `scope_id` is `"detached"`. Anything written to it is discarded as soon as the notification finishes, so write to the app store if the value must outlive the call. +Some notifications have no live task. Memory extraction normally runs on a background thread after the run, and a manual `/compact` runs outside any run. These receive a fresh store whose `scope_id` is `"detached"`. Anything written to it is discarded as soon as the notification finishes, so write to the app store if the value must outlive the call. ## ExtensionData diff --git a/frontend/src/content/zh/harness/extensions/observers.mdx b/frontend/src/content/zh/harness/extensions/observers.mdx index 5641dce407..8d21fbcc7a 100644 --- a/frontend/src/content/zh/harness/extensions/observers.mdx +++ b/frontend/src/content/zh/harness/extensions/observers.mdx @@ -14,7 +14,7 @@ import { Callout } from "nextra/components"; | `TaskLifecycleContributor` | `registry.task_lifecycle()` | 每次 Lead 运行和子 Agent 的开始与结束 | 异步，被等待 | 任务存储 | | `SystemModelCallObserver` | `registry.system_model_observer()` | 每次 DeerFlow 自有模型调用之后 | 异步 | 任务存储，或分离存储 | | `AgentAssemblyObserver` | `registry.agent_assembly_observer()` | 每次 Agent 构建结束时 | **同步** | 仅应用存储 | -| `ContextCompactionObserver` | `registry.context_compaction_observer()` | 每次摘要压缩之后 | 异步，即发即弃 | 分离存储 | +| `ContextCompactionObserver` | `registry.context_compaction_observer()` | 每次摘要压缩之后 | 异步，即发即弃 | 任务存储，或分离存储 | 本页所有示例都来自同一个注册了全部四种贡献的扩展： @@ -309,10 +309,13 @@ class ContextCompactionObserver(Protocol): | `output_content_hash` | 摘要文本的 `canonical_hash` | | `compacted_message_count` | 移除了多少条消息 | | `kept_message_count` | 保留了多少条消息 | +| `task_id` | 压缩发生在哪个任务里；不在任务内时为 `None` | +| `run_id` | 该任务所属的运行，或 `None` | +| `thread_id` | 该任务所属的线程，或 `None` | 要把事件和你手里的消息对应起来，必须用完全相同的方式哈希：`canonical_hash(message.content)`，直接传入 content 本身。绝不要先转成字符串，因为多模态 content 是 dict 列表，`str()` 的结果取决于键顺序。也不要试图事后通过对模型看到的内容做哈希来找回摘要：提示词里带的是摘要经过截断和转义的渲染版本，它的哈希不会等于 `output_content_hash`。 -这个通知是即发即弃的。它被分派到通知循环，不会阻塞模型这一轮；观察者收到的是**分离**存储，因为在这个调用点没有活跃任务。没有注册压缩观察者时，宿主连哈希这一步也会跳过。 +这个通知是即发即弃的。它被分派到通知循环，不会阻塞模型这一轮。在 Lead 运行或子 Agent 内部，观察者收到的是该任务的存储，`task_id`、`run_id`、`thread_id` 取自宿主放在其中的 `TaskInfo`，因此你可以按 `(thread_id, run_id)` 关联一次压缩，而不必去匹配哈希。手动 `/compact` 不在任何任务之内，所以观察者收到的是**分离**存储，三个 id 都是 `None`。没有父运行就启动的子 Agent 会传入它自己的存储，但其中没有 `TaskInfo`，所以 id 同样是 `None`。没有注册压缩观察者时，宿主连哈希这一步也会跳过。 ### 示例 @@ -320,12 +323,13 @@ class ContextCompactionObserver(Protocol): class CompactionLogger: async def on_context_compacted(self, app_store: ExtensionData, task_store: ExtensionData, event: CompactionEvent) -> None: logger.info( - "%s v%s folded %d message(s) into summary %s, kept %d", + "%s v%s folded %d message(s) into summary %s, kept %d (run %s)", event.transform_kind, event.transform_version, event.compacted_message_count, event.output_content_hash[:12], event.kept_message_count, + event.run_id or "none", ) ``` diff --git a/frontend/src/content/zh/harness/extensions/reference.mdx b/frontend/src/content/zh/harness/extensions/reference.mdx index 997c2140b4..4a05bce2c8 100644 --- a/frontend/src/content/zh/harness/extensions/reference.mdx +++ b/frontend/src/content/zh/harness/extensions/reference.mdx @@ -7,7 +7,7 @@ import { Callout } from "nextra/components"; # 参考 -本页列出契约版本 **0.2.4** 中 `deerflow_extension_api.__all__` 的每一个名称。请从包根导入： +本页说明契约版本 **0.2.4** 时 `deerflow_extension_api.__all__` 中的每一个名称，以及 0.2.10 为 `CompactionEvent` 新增的身份字段。0.2.4 之后新增的其他名称列在[版本历史](#版本历史)中，本页尚未展开说明。所有名称都从包根导入： ```python from deerflow_extension_api import ExtensionRegistry, MiddlewarePlacement, Placement @@ -143,7 +143,7 @@ class TaskLifecycleContributor(Protocol): ### `TaskInfo` -冻结的 dataclass。 +冻结的 dataclass。宿主还会把它放进自己分配的每个任务存储（子 Agent 仅在有父运行时），所以任何拿到该存储的钩子都能用 `task_store.get(TaskInfo)` 确定所属任务。 | 字段 | 类型 | 默认值 | | ---------------- | ----------------------------- | ------- | @@ -260,8 +260,11 @@ class ContextCompactionObserver(Protocol): | `output_content_hash` | `str` | | `compacted_message_count` | `int` | | `kept_message_count` | `int` | +| `task_id` | `str \| None = None` | +| `run_id` | `str \| None = None` | +| `thread_id` | `str \| None = None` | -哈希是对直接传入的内容计算的 `canonical_hash(message.content)`，从不是字符串化后的副本。要把某条消息与事件对应起来，请用同样的方式哈希它的 `content`。 +这三个 id 取自观察者收到的任务存储里的 `TaskInfo`。没有任务身份时它们为 `None`：手动 `/compact`，或没有父运行的子 Agent。哈希是对直接传入的内容计算的 `canonical_hash(message.content)`，从不是字符串化后的副本。要把某条消息与事件对应起来，请用同样的方式哈希它的 `content`。 ## 服务 @@ -501,7 +504,7 @@ class ReleasePolicyProvider(Protocol): ### `API_VERSION` -宿主的契约版本，为点分字符串，本页描述的契约是 `"0.2.4"`。始终等于 `backend/packages/extension-api/pyproject.toml` 中的包版本。 +宿主的契约版本，为点分字符串，例如 `"0.2.10"`。始终等于 `backend/packages/extension-api/pyproject.toml` 中的包版本。 ## 兼容性规则 @@ -527,5 +530,11 @@ class ReleasePolicyProvider(Protocol): | 0.2.2 | [#5647](https://github.com/bytedance/deer-flow/pull/5647) | 实验性的 `registry.plugin()`，以及 `PluginContribution`、`BrowserModule`、`BackendAction`、`ModelTool`、`ActionContext`、`ToolContext`、`SettingsField` | | 0.2.3 | [#5727](https://github.com/bytedance/deer-flow/pull/5727)、[#5685](https://github.com/bytedance/deer-flow/pull/5685) | 请求级运行证据：`resolve_run_evidence_reader`、`require_run_evidence_reader`、`RUN_EVIDENCE_READER_RESOLVER_KEY`（#5727）。打包的浏览器资源：`BrowserAssets`，`PluginContribution.frontend` 也接受它（#5685） | | 0.2.4 | [#5842](https://github.com/bytedance/deer-flow/pull/5842) | 插件资源授权：`require_plugin_management`、`arequire_plugin_management`、`EXTENSION_PLUGIN_AUTHZ_RESOLVER_KEY`、`EXTENSION_PLUGIN_AUTHZ_RESOLVER_ASYNC_KEY` | +| 0.2.5 | [#5796](https://github.com/bytedance/deer-flow/pull/5796)、[#6190](https://github.com/bytedance/deer-flow/pull/6190) | 受限的宿主模型调用：`ExtensionRuntimeDeps.model_invoker`，以及 `ModelInvoker`、`ModelMessage`、`ModelInvocationRequest`、`ModelInvocationResult`、`ModelUsage` 和错误类型 `ModelInvocationError`、`ModelInvocationUnavailable`、`ModelInvocationUnauthorized`、`ModelInvocationFailed`、`ModelOutputValidationError`（#5796）。宿主绑定的 Agent 运行控制：`AgentRuns`、`AgentRun`、`AgentRunError`、`resolve_agent_runs`、`require_agent_runs`、`AGENT_RUNS_RESOLVER_KEY`、`AGENT_RUNS_CONTEXT_KEY`、`ActionContext.agent_runs`（#6190） | +| 0.2.6 | [#6861](https://github.com/bytedance/deer-flow/pull/6861) | `PluginToolResult` 和 `ModelTool.response_format`，插件工具可以在文本之外再返回一个 JSON 产物 | +| 0.2.7 | [#6904](https://github.com/bytedance/deer-flow/pull/6904) | `ExtensionRegistry.knowledge_provider()`，用于注册只读的知识检索提供方 | +| 0.2.8 | [#6920](https://github.com/bytedance/deer-flow/pull/6920)、[#5645](https://github.com/bytedance/deer-flow/pull/5645)、[#7075](https://github.com/bytedance/deer-flow/pull/7075) | `AgentRuns.configuration_evidence()`、`read_artifact()`、`configuration_revision()`，以及 `ActionContext.interaction`（#6920）。已完成运行的证据，以及由宿主执行的已有自定义技能更新：`ExtensionRuntimeDeps.completed_run_evidence` 和 `skill_mutations`，以及 `CompletedRunEvidenceReader`、`CompletedRunSnapshot`、`CompletedRunEvent`、`CompletedRunEventPage`、`EvidenceLimits`、`SkillMutationService`、`MutationCapabilities`、`Proposal`、`ProposalBundle`、`BundleFile`、`Operation`、`AssessmentRef`、`AssetRevision`、`HostCheckResult`、`SkillRevisionView` 和共用的 `HostCapabilityError`（#5645）。`configuration_revision()` 的 `durable` 参数（#7075） | +| 0.2.9 | [#7138](https://github.com/bytedance/deer-flow/pull/7138) | `ExecutionFence`，以及 `AgentRuns.start()` 和 `cancel()` 的 `execution_fence` 参数 | +| 0.2.10 | [#7729](https://github.com/bytedance/deer-flow/pull/7729)、[#7831](https://github.com/bytedance/deer-flow/pull/7831) | `AgentRuns.send_input()`、`list_inputs()`、`authorize_control()`、`dispatch_input()`（#7729）。`CompactionEvent.task_id`、`run_id`、`thread_id`；压缩观察者收到活跃的任务存储，每个任务存储都带有自己的 `TaskInfo`（#7831） | 至今没有任何版本移除过公开名称。 diff --git a/frontend/src/content/zh/harness/extensions/runtime.mdx b/frontend/src/content/zh/harness/extensions/runtime.mdx index 040061933f..2aa4d57602 100644 --- a/frontend/src/content/zh/harness/extensions/runtime.mdx +++ b/frontend/src/content/zh/harness/extensions/runtime.mdx @@ -107,7 +107,7 @@ plugins: ### 分离存储 -有些通知没有活跃任务。记忆抽取通常在运行结束后于后台线程执行，上下文压缩观察者总是在轮次之外分派。它们会收到一个全新的存储，`scope_id` 为 `"detached"`。写进去的东西在通知结束后立即丢弃，所以需要比调用活得更久的值请写进应用存储。 +有些通知没有活跃任务。记忆抽取通常在运行结束后于后台线程执行，手动 `/compact` 则不在任何运行之内。它们会收到一个全新的存储，`scope_id` 为 `"detached"`。写进去的东西在通知结束后立即丢弃，所以需要比调用活得更久的值请写进应用存储。 ## ExtensionData