mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
fix: preserve existing task notes during parallel additions (#5954)
* fix: preserve task notes during parallel additions * docs: scope task note guidance to continuity module * fix: reserve task note slots using resolved execution keys Signed-off-by: ZJPex <3258236335@qq.com> * refactor: share task-note argument key and align documentation language Signed-off-by: ZJPex <3258236335@qq.com> * fix: ignore malformed sibling arguments in task-note reservations Signed-off-by: ZJPex <3258236335@qq.com> * fix: skip structurally invalid task-note reservations Signed-off-by: ZJPex <3258236335@qq.com> --------- Signed-off-by: ZJPex <3258236335@qq.com>
This commit is contained in:
ZJPex
GitHub
parent
2000bffa8a
commit
8e282c4c24
13
README.md
13
README.md
@ -82,6 +82,7 @@ DeerFlow has newly integrated the intelligent search and crawling toolset indepe
|
||||
- [Sub-Agents](#sub-agents)
|
||||
- [Sandbox \& File System](#sandbox--file-system)
|
||||
- [Context Engineering](#context-engineering)
|
||||
- [Current Task Notes](#current-task-notes)
|
||||
- [Long-Term Memory](#long-term-memory)
|
||||
- [Recommended Models](#recommended-models)
|
||||
- [Embedded Python Client](#embedded-python-client)
|
||||
@ -1934,6 +1935,18 @@ in the composer and in the transcript. There is no automatic history search. See
|
||||
[configuration](backend/docs/CONFIGURATION.md#reading-referenced-conversations)
|
||||
and the [request contract](backend/docs/API.md#referencing-a-previous-conversation).
|
||||
|
||||
### Current Task Notes
|
||||
|
||||
Enable [task notes and history recall](docs/task-continuity.md) with `task_continuity.enabled: true`.
|
||||
A task can retain up to eight notes. Parallel additions beyond the remaining slots return
|
||||
`note_capacity`, preserving existing notes. When artifact-handle resolution is enabled,
|
||||
capacity counts resolved keys; aliases for the same note share one slot.
|
||||
Malformed non-dict sibling arguments do not consume slots or disrupt valid note calls.
|
||||
Notes with invalid keys, content over 750 characters, more than four sources, or malformed
|
||||
source IDs also reserve no slot. Existing keys can still be replaced or deleted. Slots freed
|
||||
by sibling deletions or runtime failures (such as unavailable sources or policy denial)
|
||||
become available in the next batch, when rejected additions can be retried.
|
||||
|
||||
### Long-Term Memory
|
||||
|
||||
The opt-in [DeerMem scope-isolation benchmark](backend/scripts/benchmark/deermem_scope_isolation/README.md)
|
||||
|
||||
10
README_zh.md
10
README_zh.md
@ -74,6 +74,7 @@ DeerFlow 新近集成了 BytePlus 自研的智能搜索与抓取工具集——[
|
||||
- [Sandbox 与文件系统](#sandbox-与文件系统)
|
||||
- [Agentic Browser Control](#agentic-browser-control)
|
||||
- [Context Engineering](#context-engineering)
|
||||
- [当前任务笔记](#当前任务笔记)
|
||||
- [长期记忆](#长期记忆)
|
||||
- [推荐模型](#推荐模型)
|
||||
- [内嵌 Python Client](#内嵌-python-client)
|
||||
@ -797,6 +798,15 @@ Gateway API 调用方可以启用 `read_conversation`，并在一次 run 中提
|
||||
|
||||
无法在请求顶层添加字段的 SDK 客户端可以把同样的列表放在 `context.conversation_references` 中发送，`GET /api/features` 会报告该工具是否启用。启用后，Web UI 输入框会在附件按钮旁边显示一个"引用会话"按钮：最多选择你最近的三个会话，它们只附加到下一条消息上，以 chips 的形式显示在输入框与对话记录里。不会自动搜索历史。参见[配置](backend/docs/CONFIGURATION.md#reading-referenced-conversations)与[请求契约](backend/docs/API.md#referencing-a-previous-conversation)。
|
||||
|
||||
### 当前任务笔记
|
||||
|
||||
当前任务可通过 `task_continuity.enabled: true` 开启[任务笔记与历史回查](docs/task-continuity.md)。
|
||||
任务笔记最多八条；并行新增超出剩余名额时返回 `note_capacity`，保留原有笔记。
|
||||
启用资源句柄解析时按解析后的实际 key 计数，指向同一笔记的别名共用名额。
|
||||
同批调用中格式异常的非字典参数不会占用名额或影响正常笔记调用。
|
||||
无效 key、超过 750 字符的内容、超过四个引用或格式无效的引用 ID 也不会占用名额。
|
||||
已有 key 仍可替换或删除；同批删除及运行时失败（如来源不可读或策略拒绝）释放的名额在下一批可用，可届时重试。
|
||||
|
||||
### 长期记忆
|
||||
|
||||
大多数 agents 会在对话结束后把一切都忘掉，DeerFlow 不一样。
|
||||
|
||||
@ -4,6 +4,11 @@
|
||||
|
||||
*(optional, if `tool_artifacts.enabled && tool_artifacts.resolve_handles_in_args`, default on)* Resolves `art_xxxxxxxx` artifact handles found in tool-call arguments to their real references (`ThreadState.tool_artifacts[].real_ref`) before the tool executes (issue #4676). Runs after the outer receipt layer and before authorization/guardrail, sandbox-audit, read-before-write and progress policies; the same resolved arguments reach policy and execution. Declarative constraints enforce this order even after extension insertion. It mutates only the request args, never the message history. Handles may appear bare, inside backticks, or nested in dict/list args; unknown or expired handles return an error ToolMessage naming the missing handles and up to ten current handles, without executing the tool. This applies even to an empty registry; ordinary concrete arguments and disabled-feature behavior remain unchanged. Both config flags must be on — disabling `tool_artifacts.enabled` stops resolution for threads that still carry registry state.
|
||||
|
||||
Task-note batch reservations read the same resolver's argument view from the
|
||||
call-local `ToolRuntime.state`, using `RESOLVED_TOOL_CALL_ARGS_KEY` from
|
||||
`deerflow.agents.task_continuity.state`. Aliases share a slot by resolved key;
|
||||
message history and checkpoints remain unchanged.
|
||||
|
||||
## Capture
|
||||
|
||||
*(optional, if `tool_artifacts.enabled`, default on)* Captures lightweight artifact references from tool results into `ThreadState.tool_artifacts` (a reducer channel, so entries survive summarization compaction). `tool_artifacts.enabled` gates the whole middleware (capture and consumption tracking); when off, `before_model` returns nothing regardless of registry state. Runs as a `before_model` hook (never wraps a `ToolMessage` in a `Command`). `ThreadState.tool_artifact_processed` stores deduplicated hashes of processed result and settled consuming-call occurrences through an additive reducer. It records empty results too. The ledger is independent of registry retention and survives checkpoint reload, graph recreation and compaction; it is not trimmed with artifacts and grows by a small identity record per processed occurrence. Never replace it with a process-local memo or infer it from currently retained handles. A missing handle gets one retry before its consuming-call occurrence settles, so permanently unknown or evicted handles are scanned at most twice, including across restarts; same-round consumption resolves against state plus pending captures. Capture and consumption updates concatenate list channels so neither is lost. Position-independent in the tool-execution wrap chain: it is a `before_model` hook reading state messages, so it sees the normalized results stored in state regardless of its wrap position (error results are skipped at extraction via `status == "error"`); it is appended after ToolErrorHandlingMiddleware for readability only. Handles are `art_` + 8 hex chars derived deterministically from (thread_id, durable ToolMessage.id, provider tool_call_id, per-result ordinal). Provider call IDs may repeat across turns; graph-assigned message IDs distinguish occurrences and remain stable after compaction. Standalone ID-less hook callers must supply distinct call indices. These short hashes are identifiers, not ownership credentials. Extraction combines all of: structured-content string values under known keys (`file`/`path`/`url`/`task_id`/...) as concrete refs (unknown or non-reference payloads fall back to complete JSON `data` only when they fit 4096 UTF-8 bytes, 1024 nodes and 32 levels; empty and oversized payloads are skipped before unbounded serialization), `file`/`image` content blocks with HTTP(S), Unix-absolute or drive-absolute references (`data:`/`blob:` URIs are rejected so embedded-resource payloads never enter state or tool args), and conservative free-text scanning of both text blocks and plain-string results such as built-in sandbox tool output (sandbox paths + remote file URLs including query/fragment suffixes, gated by `detect_refs_in_text`). The reducer enforces only an absolute ceiling (1000); the configured `tool_artifacts.max_entries` is a per-agent sliding retention window: when a capture round would exceed it, the middleware appends a trailing `{"op": "trim_to", "keep": N}` directive and the reducer evicts the oldest entries (fresh captures are always registered — nothing freezes at the cap), so agents with different configs in one process do not share a cap. Reference entries contain metadata (path/URL, name, type, MIME). The JSON compatibility fallback is an explicitly bounded exception. Concrete path/URL keys reject prose and relative paths; remote task IDs use a separate non-empty, whitespace-free, 256-character token gate. File references on all extraction paths are bounded to 4096 UTF-8 bytes. Structured traversal is bounded before fallback encoding, and valid content-block references are still captured if structured data is rejected.
|
||||
|
||||
@ -12,15 +12,17 @@ from __future__ import annotations
|
||||
|
||||
import re
|
||||
from collections.abc import Awaitable, Callable
|
||||
from dataclasses import replace
|
||||
from typing import override
|
||||
|
||||
from langchain.agents import AgentState
|
||||
from langchain.agents.middleware import AgentMiddleware
|
||||
from langchain_core.messages import ToolMessage
|
||||
from langchain_core.messages import AIMessage, ToolMessage
|
||||
from langgraph.prebuilt.tool_node import ToolCallRequest
|
||||
from langgraph.types import Command
|
||||
|
||||
from deerflow.agents.middlewares.tool_result_meta import normalize_tool_result
|
||||
from deerflow.agents.task_continuity.state import RESOLVED_TOOL_CALL_ARGS_KEY
|
||||
from deerflow.config.tool_artifact_config import ToolArtifactConfig
|
||||
|
||||
_HANDLE_PATTERN = r"(?:`(art_[0-9a-f]{8})`|(?<!\w)(art_[0-9a-f]{8})(?!\w))"
|
||||
@ -85,6 +87,14 @@ class ArtifactResolutionMiddleware(AgentMiddleware[AgentState]):
|
||||
)
|
||||
return normalize_tool_result(message, tool_call_id=message.tool_call_id)
|
||||
|
||||
# Share resolved batch arguments for note admission only in this runtime; preserve message history.
|
||||
if request.tool_call.get("name") == "task_note" and request.runtime is not None:
|
||||
runtime = request.runtime
|
||||
message = next((message for message in reversed(runtime.state.get("messages", [])) if isinstance(message, AIMessage)), None)
|
||||
if message is not None:
|
||||
resolved_calls = {call["id"]: self._resolve_value(call["args"], handle_map) for call in message.tool_calls if call["name"] == "task_note"}
|
||||
request = replace(request, runtime=replace(runtime, state={**runtime.state, RESOLVED_TOOL_CALL_ARGS_KEY: resolved_calls}))
|
||||
|
||||
resolved_args = self._resolve_value(args, handle_map)
|
||||
if resolved_args == args:
|
||||
return request
|
||||
|
||||
@ -1,5 +1,20 @@
|
||||
# Task continuity
|
||||
|
||||
When changing `task_note` capacity or batch receipts, follow the
|
||||
[continuity contract](../../../../../../docs/task-continuity.md) and run the real
|
||||
graph regressions in `backend/tests/test_task_note_capacity.py`; locking one
|
||||
snapshot cannot coordinate parallel Commands. With handle resolution enabled,
|
||||
reservations use the same resolver as execution through the call-local argument
|
||||
view supplied by `ArtifactResolutionMiddleware`. Both modules import
|
||||
`RESOLVED_TOOL_CALL_ARGS_KEY` from `state.py`. This view stays in the current
|
||||
`ToolRuntime.state`, outside messages and checkpoints. Disabled resolution and
|
||||
direct tool graphs retain raw arguments. Skip non-dict sibling arguments before
|
||||
reading note fields so malformed calls cannot break valid sibling receipts.
|
||||
Reservation and execution share `_note_shape_error` for key, content and source
|
||||
structure checks. Skip structurally invalid calls without source lookups during
|
||||
reservation; source availability and policy denial remain runtime outcomes whose
|
||||
unused slots are recalculated in the next batch.
|
||||
|
||||
`history_search` accepts optional `role=user|assistant|tool`, mapped at the tool
|
||||
boundary to stored roles `human|ai|tool`; omission/null keeps all roles. Filter
|
||||
archived JSON payloads before SQLite FTS `LIMIT 8`, and active messages before
|
||||
|
||||
@ -5,6 +5,10 @@ from collections.abc import Sequence
|
||||
|
||||
from langgraph.channels import BinaryOperatorAggregate
|
||||
|
||||
# Call-local resolved arguments shared by artifact resolution and note admission.
|
||||
# This key belongs only to ToolRuntime.state, never checkpointed ThreadState.
|
||||
RESOLVED_TOOL_CALL_ARGS_KEY = "__resolved_tool_call_args"
|
||||
|
||||
MAX_NOTES = 8
|
||||
MAX_NOTE_CHARS = 750
|
||||
MAX_NOTE_SOURCES = 4
|
||||
|
||||
@ -3,12 +3,12 @@
|
||||
import json
|
||||
from typing import Literal
|
||||
|
||||
from langchain_core.messages import ToolMessage
|
||||
from langchain_core.messages import AIMessage, ToolMessage
|
||||
from langchain_core.tools import StructuredTool
|
||||
from langgraph.types import Command
|
||||
|
||||
from deerflow.agents.task_continuity.archive import lookup
|
||||
from deerflow.agents.task_continuity.state import MAX_NOTE_CHARS, MAX_NOTE_SOURCES, MAX_NOTES, NOTE_KEY_PATTERN, SOURCE_ID_PATTERN, normalize_task_notes
|
||||
from deerflow.agents.task_continuity.state import MAX_NOTE_CHARS, MAX_NOTE_SOURCES, MAX_NOTES, NOTE_KEY_PATTERN, RESOLVED_TOOL_CALL_ARGS_KEY, SOURCE_ID_PATTERN, normalize_task_notes
|
||||
from deerflow.tools.types import Runtime
|
||||
from deerflow.utils.file_io import run_file_io
|
||||
|
||||
@ -56,23 +56,70 @@ def _history_read(runtime: Runtime, source_id: str, offset: int = 0) -> str:
|
||||
return json.dumps({**row, "text": text[offset : offset + 4000], "next_offset": offset + 4000 if offset + 4000 < len(text) else None, "status": result["status"]}, ensure_ascii=False)
|
||||
|
||||
|
||||
def _note_shape_error(key: str, content: str, sources: list[str]) -> str | None:
|
||||
"""Share deterministic validation between batch reservation and execution."""
|
||||
if not NOTE_KEY_PATTERN.fullmatch(key) or len(content) > MAX_NOTE_CHARS or len(sources) > MAX_NOTE_SOURCES:
|
||||
return "invalid_note"
|
||||
if any(not isinstance(source_id, str) or not SOURCE_ID_PATTERN.fullmatch(source_id) for source_id in sources):
|
||||
return "invalid_source_id"
|
||||
return None
|
||||
|
||||
|
||||
def _has_note_capacity(runtime: Runtime, notes: dict, key: str) -> bool:
|
||||
"""Reserve new keys from the shared batch snapshot so parallel Commands preserve existing notes."""
|
||||
if key in notes:
|
||||
return True
|
||||
available = MAX_NOTES - len(notes)
|
||||
if available <= 0:
|
||||
return False
|
||||
message = next((message for message in reversed(runtime.state.get("messages", [])) if isinstance(message, AIMessage)), None)
|
||||
if message is None or not any(call["id"] == runtime.tool_call_id for call in message.tool_calls):
|
||||
# Direct tool calls may have no model batch; retain the single-call capacity check.
|
||||
return True
|
||||
reserved: set[str] = set()
|
||||
for call in message.tool_calls:
|
||||
if call["name"] != "task_note":
|
||||
continue
|
||||
# Use middleware-resolved batch arguments when available, matching execution; otherwise use raw arguments.
|
||||
args = runtime.state.get(RESOLVED_TOOL_CALL_ARGS_KEY, {}).get(call["id"], call["args"])
|
||||
if not isinstance(args, dict):
|
||||
continue
|
||||
candidate = args.get("key")
|
||||
content = args.get("content")
|
||||
if not isinstance(candidate, str) or not isinstance(content, str) or not content:
|
||||
continue
|
||||
sources = args.get("source_ids")
|
||||
if sources is None:
|
||||
sources = []
|
||||
if not isinstance(sources, list) or _note_shape_error(candidate, content, sources):
|
||||
continue
|
||||
if candidate not in notes and len(reserved) < available:
|
||||
reserved.add(candidate)
|
||||
# Do not borrow slots from sibling deletions or runtime failures: they have not committed and may be denied.
|
||||
return key in reserved
|
||||
|
||||
|
||||
def _task_note(runtime: Runtime, key: str, content: str, source_ids: list[str] | None = None) -> Command | str:
|
||||
"""Save or replace a short working note for this task; empty content deletes it.
|
||||
|
||||
Keep constraints, decisions, failed attempts, verified facts and next steps
|
||||
before compaction. Maximum 8 keys, 750 characters each and 4 source IDs.
|
||||
Parallel additions reserve slots in tool-call order, skipping structurally invalid notes.
|
||||
Slots freed by sibling deletions or runtime failures are available in the next batch.
|
||||
On note_capacity, replace an existing key or retry after the current batch completes.
|
||||
Notes are model reports, not verified truth or long-term user memory. Cite
|
||||
history_search IDs when possible; uncited notes are explicitly self-reported.
|
||||
"""
|
||||
sources = source_ids or []
|
||||
notes = normalize_task_notes(runtime.state.get("task_notes"))
|
||||
if not NOTE_KEY_PATTERN.fullmatch(key) or len(content) > MAX_NOTE_CHARS or len(sources) > MAX_NOTE_SOURCES:
|
||||
return json.dumps({"error": "invalid_note", "limits": "key: 40 ASCII letters/digits/_/-, content: 750 chars, sources: 4"})
|
||||
if content and key not in notes and len(notes) >= MAX_NOTES:
|
||||
return json.dumps({"error": "note_capacity", "hint": "replace or delete an existing key"})
|
||||
error = _note_shape_error(key, content, sources)
|
||||
if error == "invalid_note":
|
||||
return json.dumps({"error": error, "limits": "key: 40 ASCII letters/digits/_/-, content: 750 chars, sources: 4"})
|
||||
if error:
|
||||
return json.dumps({"error": error})
|
||||
if content and not _has_note_capacity(runtime, notes, key):
|
||||
return json.dumps({"error": "note_capacity", "hint": "replace an existing key, or retry after this batch; deletions and unused reservations free capacity for the next batch"})
|
||||
for source_id in sources:
|
||||
if not SOURCE_ID_PATTERN.fullmatch(source_id):
|
||||
return json.dumps({"error": "invalid_source_id"})
|
||||
try:
|
||||
result = lookup(runtime.state, runtime, source_id=source_id)
|
||||
except ValueError:
|
||||
|
||||
366
backend/tests/test_task_note_capacity.py
Normal file
366
backend/tests/test_task_note_capacity.py
Normal file
@ -0,0 +1,366 @@
|
||||
"""Verify task-note batch capacity and receipts through real tool graphs."""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import threading
|
||||
|
||||
import pytest
|
||||
from langchain.agents import create_agent
|
||||
from langchain.agents.middleware import AgentMiddleware
|
||||
from langchain_core.language_models import BaseChatModel
|
||||
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
|
||||
from langchain_core.outputs import ChatGeneration, ChatResult
|
||||
from langgraph.checkpoint.memory import InMemorySaver
|
||||
from langgraph.prebuilt.tool_node import ToolCallRequest, ToolRuntime
|
||||
from langgraph.types import Command
|
||||
|
||||
from deerflow.agents.middlewares.artifact_resolution_middleware import ArtifactResolutionMiddleware
|
||||
from deerflow.agents.task_continuity.state import RESOLVED_TOOL_CALL_ARGS_KEY
|
||||
from deerflow.agents.task_continuity.tools import task_note
|
||||
from deerflow.agents.thread_state import ThreadState, get_thread_state_schema
|
||||
from deerflow.config.tool_artifact_config import ToolArtifactConfig
|
||||
|
||||
|
||||
class NoteModel(BaseChatModel):
|
||||
calls: list[dict]
|
||||
|
||||
@property
|
||||
def _llm_type(self):
|
||||
return "task-note-capacity-test"
|
||||
|
||||
def bind_tools(self, tools, **kwargs):
|
||||
return self
|
||||
|
||||
def _generate(self, messages, stop=None, run_manager=None, **kwargs):
|
||||
message = AIMessage(content="done") if isinstance(messages[-1], ToolMessage) else AIMessage(content="", tool_calls=self.calls)
|
||||
return ChatResult(generations=[ChatGeneration(message=message)])
|
||||
|
||||
|
||||
def note_call(key, content="new note", *, call_id=None, **kwargs):
|
||||
return {"name": "task_note", "id": call_id or key, "args": {"key": key, "content": content, **kwargs}}
|
||||
|
||||
|
||||
def notebook(count):
|
||||
return {f"keep{i}": {"content": f"original {i}"} for i in range(count)}
|
||||
|
||||
|
||||
def replies(state):
|
||||
return {message.tool_call_id: json.loads(message.content) for message in state["messages"] if isinstance(message, ToolMessage)}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("async_mode", [False, True], ids=["sync", "async"])
|
||||
async def test_parallel_new_notes_preserve_existing_notes_and_report_capacity(async_mode):
|
||||
graph = create_agent(NoteModel(calls=[note_call("new_a"), note_call("new_b")]), tools=[task_note], state_schema=ThreadState)
|
||||
initial = {"messages": [HumanMessage(content="save both notes")], "task_notes": {f"keep{i}": {"content": f"original {i}"} for i in range(7)}}
|
||||
state = await graph.ainvoke(initial) if async_mode else graph.invoke(initial)
|
||||
results = replies(state)
|
||||
|
||||
assert set(state["task_notes"]) == {"keep0", "keep1", "keep2", "keep3", "keep4", "keep5", "keep6", "new_a"}, results
|
||||
assert results["new_a"]["status"] == "saved"
|
||||
assert results["new_b"]["error"] == "note_capacity"
|
||||
assert all(state["task_notes"][f"keep{i}"]["content"] == f"original {i}" for i in range(7))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("async_mode", [False, True], ids=["sync", "async"])
|
||||
@pytest.mark.parametrize("mode", ["full", "delta"])
|
||||
@pytest.mark.parametrize("count", [0, 6, 8])
|
||||
async def test_batch_admission_matches_checkpointed_notebook(async_mode, mode, count):
|
||||
graph = create_agent(NoteModel(calls=[note_call(f"new{i}") for i in range(10)]), tools=[task_note], state_schema=get_thread_state_schema(mode), checkpointer=InMemorySaver())
|
||||
config = {"configurable": {"thread_id": "capacity"}}
|
||||
initial = {"messages": [HumanMessage(content="save notes")], "task_notes": notebook(count)}
|
||||
state = await graph.ainvoke(initial, config) if async_mode else graph.invoke(initial, config)
|
||||
snapshot = await graph.aget_state(config) if async_mode else graph.get_state(config)
|
||||
assert snapshot.values["task_notes"] == state["task_notes"]
|
||||
assert len(state["task_notes"]) == 8
|
||||
assert set(notebook(count)) <= state["task_notes"].keys()
|
||||
for index in range(10):
|
||||
if index < 8 - count:
|
||||
assert replies(state)[f"new{index}"]["status"] == "saved"
|
||||
assert state["task_notes"][f"new{index}"]["content"] == "new note"
|
||||
else:
|
||||
assert replies(state)[f"new{index}"]["error"] == "note_capacity"
|
||||
assert f"new{index}" not in state["task_notes"]
|
||||
|
||||
|
||||
class ReverseCompletion(AgentMiddleware):
|
||||
"""Complete the later call first through middleware to verify admission is scheduling-independent."""
|
||||
|
||||
def __init__(self):
|
||||
self.sync_done = threading.Event()
|
||||
self.async_done = asyncio.Event()
|
||||
self.completed = []
|
||||
|
||||
def wrap_tool_call(self, request, handler):
|
||||
if request.tool_call["id"] == "first":
|
||||
assert self.sync_done.wait(5)
|
||||
result = handler(request)
|
||||
self.completed.append(request.tool_call["id"])
|
||||
self.sync_done.set()
|
||||
return result
|
||||
|
||||
async def awrap_tool_call(self, request, handler):
|
||||
if request.tool_call["id"] == "first":
|
||||
await asyncio.wait_for(self.async_done.wait(), 5)
|
||||
result = await handler(request)
|
||||
self.completed.append(request.tool_call["id"])
|
||||
self.async_done.set()
|
||||
return result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("async_mode", [False, True], ids=["sync", "async"])
|
||||
async def test_admission_uses_call_order_not_completion_order(async_mode):
|
||||
middleware = ReverseCompletion()
|
||||
graph = create_agent(NoteModel(calls=[note_call("new_a", call_id="first"), note_call("new_b", call_id="second")]), tools=[task_note], middleware=[middleware], state_schema=ThreadState)
|
||||
initial = {"messages": [HumanMessage(content="save notes")], "task_notes": notebook(7)}
|
||||
state = await graph.ainvoke(initial) if async_mode else graph.invoke(initial)
|
||||
assert middleware.completed == ["second", "first"]
|
||||
assert replies(state)["first"]["status"] == "saved"
|
||||
assert replies(state)["second"]["error"] == "note_capacity"
|
||||
assert set(state["task_notes"]) == set(notebook(7)) | {"new_a"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("async_mode", [False, True], ids=["sync", "async"])
|
||||
@pytest.mark.parametrize("last_content", ["replacement", ""])
|
||||
async def test_same_new_key_shares_one_slot_and_keeps_ordered_update_delete_semantics(async_mode, last_content):
|
||||
calls = [note_call("new_a", call_id="first"), note_call("new_a", last_content, call_id="last"), note_call("new_b")]
|
||||
graph = create_agent(NoteModel(calls=calls), tools=[task_note], state_schema=ThreadState)
|
||||
initial = {"messages": [HumanMessage(content="update or delete")], "task_notes": notebook(7)}
|
||||
state = await graph.ainvoke(initial) if async_mode else graph.invoke(initial)
|
||||
assert set(notebook(7)) <= state["task_notes"].keys()
|
||||
assert replies(state)["first"]["status"] == "saved"
|
||||
assert replies(state)["last"]["status"] == ("saved" if last_content else "deleted")
|
||||
assert replies(state)["new_b"]["error"] == "note_capacity"
|
||||
if last_content:
|
||||
assert state["task_notes"]["new_a"]["content"] == "replacement"
|
||||
assert len(state["task_notes"]) == 8
|
||||
else:
|
||||
assert set(state["task_notes"]) == set(notebook(7))
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("async_mode", [False, True], ids=["sync", "async"])
|
||||
async def test_full_notebook_allows_replace_delete_and_new_key_in_next_batch(async_mode):
|
||||
saver = InMemorySaver()
|
||||
config = {"configurable": {"thread_id": "next-batch"}}
|
||||
graph = create_agent(NoteModel(calls=[note_call("keep0", "updated"), note_call("keep1", ""), note_call("new_a")]), tools=[task_note], state_schema=ThreadState, checkpointer=saver)
|
||||
initial = {"messages": [HumanMessage(content="update and delete")], "task_notes": notebook(8)}
|
||||
state = await graph.ainvoke(initial, config) if async_mode else graph.invoke(initial, config)
|
||||
assert len(state["task_notes"]) == 7
|
||||
assert state["task_notes"]["keep0"]["content"] == "updated"
|
||||
assert "keep1" not in state["task_notes"]
|
||||
assert replies(state)["keep0"]["status"] == "saved"
|
||||
assert replies(state)["keep1"]["status"] == "deleted"
|
||||
assert replies(state)["new_a"]["error"] == "note_capacity"
|
||||
|
||||
resumed = create_agent(NoteModel(calls=[note_call("new_a", call_id="retry")]), tools=[task_note], state_schema=ThreadState, checkpointer=saver)
|
||||
state = await resumed.ainvoke({"messages": [HumanMessage(content="retry")]}, config) if async_mode else resumed.invoke({"messages": [HumanMessage(content="retry")]}, config)
|
||||
assert len(state["task_notes"]) == 8
|
||||
assert replies(state)["retry"]["status"] == "saved"
|
||||
assert state["task_notes"]["new_a"]["content"] == "new note"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("async_mode", [False, True], ids=["sync", "async"])
|
||||
@pytest.mark.parametrize(
|
||||
("first_call", "error"),
|
||||
[
|
||||
(note_call("invalid", "x" * 751), "invalid_note"),
|
||||
(note_call("invalid", source_ids=["not-a-source"]), "invalid_source_id"),
|
||||
(note_call("invalid", source_ids=["r" + "0" * 32] * 5), "invalid_note"),
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize("resolve_handles", [False, True], ids=["raw", "resolved"])
|
||||
async def test_structurally_invalid_sibling_does_not_reserve_a_slot(async_mode, first_call, error, resolve_handles):
|
||||
middleware = [ArtifactResolutionMiddleware()] if resolve_handles else []
|
||||
graph = create_agent(NoteModel(calls=[first_call, note_call("new_a"), note_call("overflow")]), tools=[task_note], middleware=middleware, state_schema=ThreadState)
|
||||
initial = {"messages": [HumanMessage(content="save notes")], "task_notes": notebook(7)}
|
||||
state = await graph.ainvoke(initial) if async_mode else graph.invoke(initial)
|
||||
assert set(state["task_notes"]) == set(notebook(7)) | {"new_a"}
|
||||
assert replies(state)["invalid"]["error"] == error
|
||||
assert replies(state)["overflow"]["error"] == "note_capacity"
|
||||
assert replies(state)["new_a"]["status"] == "saved"
|
||||
assert state["task_notes"]["new_a"]["content"] == "new note"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_shared_graph_keeps_simultaneous_task_capacity_independent():
|
||||
graph = create_agent(NoteModel(calls=[note_call("new_a"), note_call("new_b")]), tools=[task_note], state_schema=ThreadState, checkpointer=InMemorySaver())
|
||||
|
||||
async def run(user, thread, count):
|
||||
return await graph.ainvoke(
|
||||
{"messages": [HumanMessage(content="save notes")], "task_notes": notebook(count)},
|
||||
{"configurable": {"thread_id": thread}},
|
||||
context={"user_id": user, "thread_id": thread},
|
||||
)
|
||||
|
||||
nearly_full, empty, full = await asyncio.gather(run("alice", "thread-a", 7), run("bob", "thread-b", 0), run("alice", "thread-c", 8))
|
||||
assert set(nearly_full["task_notes"]) == set(notebook(7)) | {"new_a"}
|
||||
assert replies(nearly_full)["new_b"]["error"] == "note_capacity"
|
||||
assert set(empty["task_notes"]) == {"new_a", "new_b"}
|
||||
assert all(result["status"] == "saved" for result in replies(empty).values())
|
||||
assert set(full["task_notes"]) == set(notebook(8))
|
||||
assert all(result["error"] == "note_capacity" for result in replies(full).values())
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("async_mode", [False, True], ids=["sync", "async"])
|
||||
async def test_resolved_handle_can_save_into_empty_notebook(async_mode):
|
||||
calls = [note_call("art_ab12cd34", "check completion")]
|
||||
graph = create_agent(NoteModel(calls=calls), tools=[task_note], middleware=[ArtifactResolutionMiddleware()], state_schema=ThreadState)
|
||||
initial = {
|
||||
"messages": [HumanMessage(content="save task reference")],
|
||||
"tool_artifacts": [{"handle": "art_ab12cd34", "artifact_type": "task", "real_ref": "remote-task-42"}],
|
||||
}
|
||||
state = await graph.ainvoke(initial) if async_mode else graph.invoke(initial)
|
||||
assert replies(state)["art_ab12cd34"]["status"] == "saved"
|
||||
assert set(state["task_notes"]) == {"remote-task-42"}
|
||||
assert state["task_notes"]["remote-task-42"]["content"] == "check completion"
|
||||
assert next(message for message in state["messages"] if isinstance(message, AIMessage)).tool_calls == [{**call, "type": "tool_call"} for call in calls]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("async_mode", [False, True], ids=["sync", "async"])
|
||||
@pytest.mark.parametrize("mode", ["full", "delta"])
|
||||
@pytest.mark.parametrize(
|
||||
("count", "target", "calls", "expected_keys", "saved_ids", "rejected_ids"),
|
||||
[
|
||||
(7, "remote-task-42", [note_call("art_ab12cd34"), note_call("art_ab12cd35", "replacement"), note_call("overflow")], {"remote-task-42"}, ["art_ab12cd34", "art_ab12cd35"], ["overflow"]),
|
||||
(6, "remote-task-42", [note_call("art_ab12cd34"), note_call("remote-task-42", "replacement"), note_call("new_b"), note_call("overflow")], {"remote-task-42", "new_b"}, ["art_ab12cd34", "remote-task-42", "new_b"], ["overflow"]),
|
||||
(7, "keep0", [note_call("art_ab12cd34", "replacement"), note_call("new_b")], {"new_b"}, ["art_ab12cd34", "new_b"], []),
|
||||
(8, "keep0", [note_call("art_ab12cd34", "replacement"), note_call("overflow")], set(), ["art_ab12cd34"], ["overflow"]),
|
||||
(7, "remote-task-42", [note_call("`art_ab12cd34`", call_id="quoted"), note_call("overflow")], {"remote-task-42"}, ["quoted"], ["overflow"]),
|
||||
],
|
||||
ids=["handle-aliases", "concrete-alias", "existing-key", "full-replacement", "backticks"],
|
||||
)
|
||||
async def test_resolved_batch_reserves_distinct_execution_keys(async_mode, mode, count, target, calls, expected_keys, saved_ids, rejected_ids):
|
||||
graph = create_agent(NoteModel(calls=calls), tools=[task_note], middleware=[ArtifactResolutionMiddleware()], state_schema=get_thread_state_schema(mode), checkpointer=InMemorySaver())
|
||||
config = {"configurable": {"thread_id": "resolved-capacity"}}
|
||||
initial = {
|
||||
"messages": [HumanMessage(content="save notes")],
|
||||
"task_notes": notebook(count),
|
||||
"tool_artifacts": [{"handle": handle, "artifact_type": "task", "real_ref": target} for handle in ["art_ab12cd34", "art_ab12cd35"]],
|
||||
}
|
||||
state = await graph.ainvoke(initial, config) if async_mode else graph.invoke(initial, config)
|
||||
snapshot = await graph.aget_state(config) if async_mode else graph.get_state(config)
|
||||
assert snapshot.values["task_notes"] == state["task_notes"]
|
||||
assert set(state["task_notes"]) == set(notebook(count)) | expected_keys
|
||||
for call_id in saved_ids:
|
||||
assert replies(state)[call_id]["status"] == "saved"
|
||||
for call_id in rejected_ids:
|
||||
assert replies(state)[call_id]["error"] == "note_capacity"
|
||||
assert state["task_notes"][target]["content"] == ("replacement" if any(call["args"]["content"] == "replacement" for call in calls) else "new note")
|
||||
assert next(message for message in snapshot.values["messages"] if isinstance(message, AIMessage)).tool_calls == [{**call, "type": "tool_call"} for call in calls]
|
||||
assert RESOLVED_TOOL_CALL_ARGS_KEY not in snapshot.values
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("async_mode", [False, True], ids=["sync", "async"])
|
||||
@pytest.mark.parametrize("resolver_mode", ["absent", "disabled", "resolution-disabled"])
|
||||
async def test_disabled_resolution_reserves_literal_keys(async_mode, resolver_mode):
|
||||
middleware = [] if resolver_mode == "absent" else [ArtifactResolutionMiddleware(ToolArtifactConfig(enabled=resolver_mode != "disabled", resolve_handles_in_args=resolver_mode != "resolution-disabled"))]
|
||||
graph = create_agent(NoteModel(calls=[note_call("art_ab12cd34"), note_call("remote-task-42")]), tools=[task_note], middleware=middleware, state_schema=ThreadState)
|
||||
state_input = {
|
||||
"messages": [HumanMessage(content="save notes")],
|
||||
"task_notes": notebook(7),
|
||||
"tool_artifacts": [{"handle": "art_ab12cd34", "artifact_type": "task", "real_ref": "remote-task-42"}],
|
||||
}
|
||||
state = await graph.ainvoke(state_input) if async_mode else graph.invoke(state_input)
|
||||
assert set(state["task_notes"]) == set(notebook(7)) | {"art_ab12cd34"}
|
||||
assert replies(state)["art_ab12cd34"]["status"] == "saved"
|
||||
assert replies(state)["remote-task-42"]["error"] == "note_capacity"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("async_mode", [False, True], ids=["sync", "async"])
|
||||
@pytest.mark.parametrize("resolve_handles", [False, True], ids=["raw", "resolved"])
|
||||
@pytest.mark.parametrize("malformed_args", [None, '"quoted arguments"', ["not", "a", "mapping"], 42], ids=["null", "string", "list", "number"])
|
||||
async def test_malformed_sibling_arguments_preserve_valid_note_receipts(async_mode, resolve_handles, malformed_args):
|
||||
first_key = "art_ab12cd34" if resolve_handles else "new_a"
|
||||
calls = [note_call("invalid"), note_call(first_key, call_id="first"), note_call("overflow")]
|
||||
message = AIMessage(content="", tool_calls=calls)
|
||||
# Inject malformed internal state after AIMessage validation to test this boundary directly.
|
||||
message.tool_calls[0]["args"] = malformed_args
|
||||
initial_notes = notebook(7)
|
||||
state = {
|
||||
"messages": [message],
|
||||
"task_notes": initial_notes,
|
||||
"tool_artifacts": [{"handle": "art_ab12cd34", "artifact_type": "task", "real_ref": "new_a"}],
|
||||
}
|
||||
middleware = ArtifactResolutionMiddleware() if resolve_handles else None
|
||||
|
||||
def execute(request):
|
||||
return task_note.func(request.runtime, **request.tool_call["args"])
|
||||
|
||||
async def aexecute(request):
|
||||
return await task_note.coroutine(request.runtime, **request.tool_call["args"])
|
||||
|
||||
results = []
|
||||
for call in message.tool_calls[1:]:
|
||||
runtime = ToolRuntime(state=state, context={}, config={}, stream_writer=lambda _: None, tool_call_id=call["id"], store=None)
|
||||
request = ToolCallRequest(tool_call=call, tool=task_note, state=state, runtime=runtime)
|
||||
if async_mode:
|
||||
result = await middleware.awrap_tool_call(request, aexecute) if middleware else await aexecute(request)
|
||||
else:
|
||||
result = middleware.wrap_tool_call(request, execute) if middleware else execute(request)
|
||||
results.append(result)
|
||||
|
||||
saved, rejected = results
|
||||
assert isinstance(saved, Command)
|
||||
assert saved.update["task_notes"] == {"new_a": {"content": "new note", "source_ids": [], "authority": "model_report"}}
|
||||
assert json.loads(saved.update["messages"][0].content)["status"] == "saved"
|
||||
assert json.loads(rejected)["error"] == "note_capacity"
|
||||
assert state["task_notes"] == notebook(7)
|
||||
assert message.tool_calls[0]["args"] == malformed_args
|
||||
assert RESOLVED_TOOL_CALL_ARGS_KEY not in state
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("async_mode", [False, True], ids=["sync", "async"])
|
||||
@pytest.mark.parametrize("field", ["content", "source_ids"])
|
||||
async def test_resolved_invalid_note_shape_does_not_reserve_a_slot(async_mode, field):
|
||||
first = note_call("invalid", "art_ab12cd34") if field == "content" else note_call("invalid", source_ids=["art_ab12cd34"])
|
||||
graph = create_agent(NoteModel(calls=[first, note_call("new_a")]), tools=[task_note], middleware=[ArtifactResolutionMiddleware()], state_schema=ThreadState)
|
||||
initial = {
|
||||
"messages": [HumanMessage(content="save notes")],
|
||||
"task_notes": notebook(7),
|
||||
"tool_artifacts": [{"handle": "art_ab12cd34", "artifact_type": "task", "real_ref": "x" * 751 if field == "content" else "not-a-source"}],
|
||||
}
|
||||
state = await graph.ainvoke(initial) if async_mode else graph.invoke(initial)
|
||||
assert replies(state)["invalid"]["error"] == ("invalid_note" if field == "content" else "invalid_source_id")
|
||||
assert replies(state)["new_a"]["status"] == "saved"
|
||||
assert set(state["task_notes"]) == set(notebook(7)) | {"new_a"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("async_mode", [False, True], ids=["sync", "async"])
|
||||
@pytest.mark.parametrize("source_ids", [None, []])
|
||||
async def test_maximum_length_note_still_reserves_a_slot(async_mode, source_ids):
|
||||
graph = create_agent(NoteModel(calls=[note_call("new_a", "x" * 750, source_ids=source_ids), note_call("overflow")]), tools=[task_note], state_schema=ThreadState)
|
||||
initial = {"messages": [HumanMessage(content="save notes")], "task_notes": notebook(7)}
|
||||
state = await graph.ainvoke(initial) if async_mode else graph.invoke(initial)
|
||||
assert replies(state)["new_a"]["status"] == "saved"
|
||||
assert state["task_notes"]["new_a"]["content"] == "x" * 750
|
||||
assert replies(state)["overflow"]["error"] == "note_capacity"
|
||||
assert set(state["task_notes"]) == set(notebook(7)) | {"new_a"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("async_mode", [False, True], ids=["sync", "async"])
|
||||
async def test_unavailable_sources_keep_reservation_until_next_batch(async_mode):
|
||||
calls = [note_call("unavailable", source_ids=["r" + "0" * 32] * 4), note_call("new_a")]
|
||||
graph = create_agent(NoteModel(calls=calls), tools=[task_note], state_schema=ThreadState)
|
||||
config = {"configurable": {"thread_id": "unavailable-source"}}
|
||||
initial = {"messages": [HumanMessage(content="save notes")], "task_notes": notebook(7)}
|
||||
state = await graph.ainvoke(initial, config) if async_mode else graph.invoke(initial, config)
|
||||
assert replies(state)["unavailable"]["error"] == "source_unavailable"
|
||||
assert replies(state)["new_a"]["error"] == "note_capacity"
|
||||
assert set(state["task_notes"]) == set(notebook(7))
|
||||
retry = create_agent(NoteModel(calls=[note_call("new_a", call_id="retry")]), tools=[task_note], state_schema=ThreadState)
|
||||
state["messages"].append(HumanMessage(content="retry"))
|
||||
state = await retry.ainvoke(state, config) if async_mode else retry.invoke(state, config)
|
||||
assert replies(state)["retry"]["status"] == "saved"
|
||||
assert set(state["task_notes"]) == set(notebook(7)) | {"new_a"}
|
||||
@ -19,12 +19,25 @@ or additional model call is required by the feature itself.
|
||||
The standard lead-agent builders (including custom-agent bootstrap) and
|
||||
`DeerFlowClient` expose three tools through the existing authorization filter:
|
||||
|
||||
- `task_note`: save, replace or delete a named task note. Keep up to eight notes,
|
||||
each with 750 characters and four optional source IDs. A full notebook rejects
|
||||
new keys until an existing key is replaced or deleted. If parallel updates
|
||||
jointly exceed capacity, the reducer retains the last eight insertion-ordered
|
||||
keys; inspect the next injected notebook for the retained entries. Source IDs are checked
|
||||
for availability, not semantic support; all notes remain model reports.
|
||||
- `task_note`: save, replace or delete a named note, with at most eight notes,
|
||||
750 characters and four source IDs per note. Parallel additions reserve the
|
||||
remaining slots for distinct new keys in model tool-call order; repeated keys
|
||||
share a slot. Non-dict sibling arguments are ignored during reservation, so
|
||||
malformed calls neither consume slots nor break valid sibling receipts. Invalid
|
||||
keys, oversized content, excess source IDs and malformed source IDs also reserve
|
||||
no slot; these structural checks use the same resolved arguments as execution.
|
||||
When artifact-handle resolution is enabled, reservations use
|
||||
resolved keys, so handles and concrete keys referring to one note share a slot.
|
||||
Excess additions return `note_capacity` without evicting existing notes or
|
||||
reporting `saved` for a write discarded because of capacity. Existing keys can
|
||||
be replaced at capacity; empty content deletes a note. Valid writes to the same
|
||||
key, including explicit deletions, retain call-order, last-write-wins merging.
|
||||
Reservations use the pre-batch notebook snapshot. Slots freed by sibling
|
||||
deletions or left unused by unavailable-source or policy-denied calls are recalculated in
|
||||
the next batch. On a capacity error, replace an existing key or retry after the
|
||||
current batch completes. This avoids predicting sibling success or serializing
|
||||
other tools. Source IDs establish readability, not semantic support; all notes
|
||||
remain model reports.
|
||||
- `history_search`: keyword search over the current messages and compacted source
|
||||
batches reachable from the current checkpoint. English words and Chinese
|
||||
character bigrams are supported. Returns up to eight 600-character excerpts.
|
||||
@ -112,6 +125,12 @@ lead builders and `DeerFlowClient`.
|
||||
|
||||
## Evidence
|
||||
|
||||
`backend/tests/test_task_note_capacity.py` drives real `create_agent` graphs with
|
||||
a deterministic model, the production `task_note` tool and `ThreadState`. It
|
||||
covers sync/async parallel capacity, receipts, full/delta checkpoints, reverse
|
||||
completion order, repeated keys, deletion followed by retry, and task isolation
|
||||
without calling a live model API.
|
||||
|
||||
[The historical experiment package](experiments/task-continuity-20260912/README.md)
|
||||
contains the original A/B/C/D protocol, scripts and results. Those numbers describe
|
||||
an independent replay prototype under forced compression, not this production
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user