deer-flow/backend/tests/test_extension_stack_wiring.py
Nan Gao 7389331e65
feat(extensions): observe task lifecycle and system model calls (#4684)
* feat(extensions): observe task lifecycle and system model calls

PR 1 (#4636) gave extensions a middleware chain, and a middleware only sees
what passes through the agent graph. Two runtime surfaces stay invisible to
it: when a lead run or a subagent begins and ends, and the DeerFlow-owned
model calls made outside the graph. This slice adds both, with no new
Gateway surface -- routers, services, and the reference extension stay in
PR 3.

Contract (deerflow-extension-api 0.1.1)
---------------------------------------
Two contribution kinds join `middlewares` on the registry:
`task_lifecycle` (`on_task_start` / `on_task_stop`, receiving a `TaskInfo`
and a conservative `TaskOutcome` of completed / aborted / failed) and
`system_model_observer` (`on_system_model_call`, receiving a
`SystemOperationKind`, a `SystemModelRequest` snapshot, and a
`SystemModelResult` carrying either the response or the provider exception
plus a duration).

`SystemModelRequest.messages` normalizes to a tuple at construction. Goal
evaluation and memory extraction pass a message list while title generation
and summarization pass one prompt string, and a bare `str` already satisfies
`Sequence` -- without normalization an observer iterating `request.messages`
would silently walk characters. Copying also makes the frozen snapshot
immutable in fact rather than only by declaration, since observations may run
after the call site returns and keeps mutating its own list.

Registry marks and rollbacks become per-bucket and positional, so an
`install()` that fails after registering two different kinds cannot leave one
of them behind. `needs_task_store` now covers all three kinds: a deployment
that registers only lifecycle hooks still gets a task store.

Task lifecycle
--------------
The lead worker notifies start after the run has started and stop after
completion persistence and the completion hook, but before clearing the
finalizing barrier and publishing the stream end -- holding the barrier
across stop is what keeps a same-thread replacement run from overlapping this
task's lifecycle. Cancellation raised out of the stop notification is
deferred, not propagated in place, so a cancelled run still clears the
barrier and emits its end frame. A subagent with a parent `run_id` wraps its
execution in the same pair inside `finally`, reporting `parent_task_id` so a
delegation tree is reconstructable; a subagent without a `run_id` (embedded
client, standalone LangGraph Server) logs and skips rather than inventing a
parent. Contributors run in registration order inside one shared 3s budget
and every failure is logged and failed open.

System model calls
------------------
Four kinds cover the model calls the middleware chain cannot see: goal
evaluation, memory extraction, title generation, and summarization. Each site
reports both terminal paths without changing the provider exception the host
observes, short-circuits on `has_system_model_observers`, and passes the live
task store when the runtime has one (detached work gets an isolated store).
The sync summarization half stays unobserved on purpose -- it and its only
host caller are the sync side of an async-only runtime, so notifying there
would block a thread on a call site the host never reaches; the reason is
recorded at the call site.

The DeerMem backend must stay vendorable and cannot import the extension API,
so it reports through a new `MemoryCallbacks.on_memory_llm_result` host hook
that the DeerFlow-side callbacks translate into an observation.

Notification loop
-----------------
Extension resources must be touched on the loop that created them, but
subagents can execute on isolated loops and DeerMem runs on a worker thread.
The Gateway registers its serving loop before any runtime dependency starts
and resets it last through the exit stack, so every startup-failure and
cancellation path is covered. Awaited hooks raised on another loop are
dispatched across with `run_coroutine_threadsafe` and awaited under the same
budget; synchronous sites submit fire-and-forget work. Shutdown stops
accepting detached observations before the memory flush -- that flush runs on
a worker thread and can emit memory observations -- while keeping the loop
alive for awaited task hooks until run and subagent drain completes.

Tests
-----
`test_extension_task_lifecycle.py`, `test_extension_subagent_lifecycle.py`,
and `test_extension_system_model_calls.py` cover ordering, fail-open, budget
exhaustion, snapshot binding under a concurrent singleton replacement, the
loop-dispatch and shutdown-suspension paths, and both terminal paths at every
call site. `test_gateway_run_drain_shutdown.py` pins the stop-before-barrier
and drain ordering.

* fix(extensions): decide notification fail-open by origin, observe cancellation

`_notify_each` only guarded `Exception`, so a contributor letting a
`CancelledError` escape — an extension implementing an internal timeout with
cancellation, say — skipped its successors and reached the worker's
deferred-interrupt path, ending an otherwise successful run as cancelled.
Fail-open is about where a failure came from, not its base class: only a
genuine cancellation of the host task increments `Task.cancelling()`, so
propagate on that and contain everything else. `KeyboardInterrupt` /
`SystemExit` still propagate.

`observe_system_model_call` skipped observers on cancellation for the same
base-class reason, leaving goal / title / summarization silent on a terminal
path that is routine — interrupt/rollback admission and shutdown both cancel
the run task, with the provider tokens already spent. Awaiting observers there
is unreliable (a repeated cancel interrupts that await before any of them
runs), so report through the same non-blocking submission the synchronous
memory bridge uses, then propagate the cancellation untouched.

DeerMem keeps `BaseException` around its provider call, now with the reason
recorded: that path runs on a worker thread, where cancelling the awaiting
side never interrupts the running thread, so `CancelledError` cannot arrive
at all. Its host-hook wrapper narrows to `Exception` — only the hook's own
failures are non-fatal, and an observability path must not swallow a process
teardown signal.

* fix(extensions): warn on budget exhaustion, scope observer logs by task, propagate teardown

Review response on #4684:

- The memory observation bridge caught BaseException, which would swallow
  a teardown signal raised while dispatching; it now catches Exception,
  matching the boundary the DeerMem-side call site documents and tests.
- A notification-budget timeout raised mid-hook fell into the generic
  hook-failure path and logged an asyncio-internal traceback; it now logs
  a warning like the pre-hook budget skip, while a TimeoutError a
  contributor raises on its own stays classified as a hook failure.
- System model observer logs passed the operation kind as the task id,
  so log lines said "task goal/title/..."; they now carry the task
  scope id alongside the kind.
2026-08-11 16:33:22 +08:00

425 lines
16 KiB
Python

"""Tests for wiring extension contributions into the real middleware builders."""
from __future__ import annotations
import pytest
from deerflow_extension_api import AgentScope, MiddlewarePlacement, Placement
from langchain.agents.middleware import AgentMiddleware
from deerflow.agents.lead_agent.agent import build_middlewares
from deerflow.config.app_config import AppConfig
from deerflow.config.sandbox_config import SandboxConfig
from deerflow.extensions.isolation import IsolatedMiddleware
from deerflow.extensions.registry import ExtensionRegistry
from deerflow.extensions.stack import PLACEMENT_ANCHORS
def _app_config() -> AppConfig:
# AppConfig.sandbox has no default (`use` is a required field), so a bare
# AppConfig() always fails pydantic validation in this repo. The brief's
# verbatim test code assumed a default-constructible AppConfig; every
# other builder test in this suite (e.g. test_lead_agent_model_resolution.py)
# supplies this same minimal sandbox stanza for the same reason.
return AppConfig(sandbox=SandboxConfig(use="deerflow.sandbox.local:LocalSandboxProvider"))
class _Probe(AgentMiddleware):
def __init__(self, tag: str) -> None:
super().__init__()
self.tag = tag
def _extensions(*placements: MiddlewarePlacement):
class _C:
def contribute_middlewares(self, app_store, ctx):
return placements
registry = ExtensionRegistry()
with registry.attributed_to("demo:install"):
registry.middlewares(_C())
return registry.build()
def _tags(stack):
out = []
for m in stack:
target = m.inner if isinstance(m, IsolatedMiddleware) else m
out.append(target.tag if isinstance(target, _Probe) else type(target).__name__)
return out
def _lead_stack(extensions=None, app_config=None, configurable=None):
return build_middlewares(
config={"configurable": configurable or {}},
model_name="gpt-4o",
app_config=app_config or _app_config(),
extensions=extensions,
)
def test_anchor_table_covers_every_placement():
assert set(PLACEMENT_ANCHORS) == set(Placement)
def test_zero_extensions_leaves_the_stack_unchanged():
baseline = _tags(_lead_stack())
with_empty = _tags(_lead_stack(ExtensionRegistry().build()))
assert baseline == with_empty
assert not any(isinstance(m, IsolatedMiddleware) for m in _lead_stack())
def test_zero_extensions_skip_policy_projection(monkeypatch):
from deerflow.agents.middlewares.tool_error_handling_middleware import (
build_subagent_runtime_middlewares,
)
from deerflow.extensions import policy as policy_module
def _unexpected_projection(app_config):
raise AssertionError("zero-extension path constructed an extension payload")
monkeypatch.setattr(policy_module, "project_host_policy", _unexpected_projection)
empty = ExtensionRegistry().build()
_lead_stack(empty)
build_subagent_runtime_middlewares(app_config=_app_config(), extensions=empty)
def test_zero_extension_composition_reuses_the_built_stack():
from deerflow_extension_api import AgentBuildContext
from deerflow.extensions.stack import compose_with_extensions
middlewares = []
result = compose_with_extensions(
middlewares,
AgentScope.LEAD,
AgentBuildContext(scope=AgentScope.LEAD),
ExtensionRegistry().build(),
)
assert result is middlewares
def test_bound_build_snapshot_is_used_by_lead_and_subagent_fallbacks():
from deerflow.agents.middlewares.tool_error_handling_middleware import (
build_subagent_runtime_middlewares,
)
from deerflow.extensions import bind_agent_build_extensions
lead_capture = _CtxCapture()
subagent_capture = _CtxCapture()
lead_extensions = _extensions_with_contributor(lead_capture)
subagent_extensions = _extensions_with_contributor(subagent_capture)
with bind_agent_build_extensions(lead_extensions):
_lead_stack()
with bind_agent_build_extensions(subagent_extensions):
build_subagent_runtime_middlewares(app_config=_app_config(), model_name="gpt-4o")
assert lead_capture.ctx is not None
assert subagent_capture.ctx is not None
def test_lead_system_middlewares_capture_the_explicit_build_snapshot():
from deerflow.agents.middlewares.title_middleware import TitleMiddleware
registry = ExtensionRegistry()
with registry.attributed_to("observer:install"):
registry.system_model_observer(object())
extensions = registry.build()
stack = _lead_stack(extensions)
title = next(item for item in stack if isinstance(item, TitleMiddleware))
assert title._extensions is extensions
def test_tool_visible_lands_at_the_outermost_position():
stack = _lead_stack(_extensions(MiddlewarePlacement(_Probe("visible"), Placement.TOOL_VISIBLE)))
assert _tags(stack)[0] == "visible"
def test_model_logical_lands_outside_the_retry_middleware():
stack = _lead_stack(_extensions(MiddlewarePlacement(_Probe("decision"), Placement.MODEL_LOGICAL)))
tags = _tags(stack)
assert tags.index("decision") < tags.index("LLMErrorHandlingMiddleware")
def test_tool_raw_lands_inside_tool_error_handling():
stack = _lead_stack(_extensions(MiddlewarePlacement(_Probe("raw"), Placement.TOOL_RAW)))
tags = _tags(stack)
assert tags.index("raw") > tags.index("ToolErrorHandlingMiddleware")
def test_runtime_isolation_failure_is_recorded_after_stack_composition():
from deerflow.extensions import get_runtime_diagnostics, reset_runtime_diagnostics
class _FailingObserver(AgentMiddleware):
def wrap_tool_call(self, request, handler):
raise ValueError("observation exploded")
reset_runtime_diagnostics()
try:
stack = _lead_stack(
_extensions(
MiddlewarePlacement(
_FailingObserver(),
Placement.TOOL_VISIBLE,
)
)
)
isolated = next(middleware for middleware in stack if isinstance(middleware, IsolatedMiddleware))
handler_calls = 0
def handler(request):
nonlocal handler_calls
handler_calls += 1
return "core-result"
assert isolated.wrap_tool_call("request", handler) == "core-result"
diagnostics = get_runtime_diagnostics()
finally:
reset_runtime_diagnostics()
assert handler_calls == 1
assert len(diagnostics) == 1
assert diagnostics[0].source == "demo:install"
assert diagnostics[0].level == "error"
assert "wrap_tool_call" in diagnostics[0].message
def test_build_and_runtime_diagnostics_are_each_recorded_once():
from deerflow.extensions import get_runtime_diagnostics, reset_runtime_diagnostics
class _FailingObserver(AgentMiddleware):
def wrap_model_call(self, request, handler):
raise ValueError("observation exploded")
app_config = _app_config()
app_config.safety_finish_reason.enabled = False
reset_runtime_diagnostics()
try:
stack = _lead_stack(
_extensions(
MiddlewarePlacement(
_FailingObserver(),
Placement.MODEL_PHYSICAL,
)
),
app_config=app_config,
)
isolated = next(middleware for middleware in stack if isinstance(middleware, IsolatedMiddleware))
assert isolated.wrap_model_call("request", lambda request: "core-result") == "core-result"
diagnostics = get_runtime_diagnostics()
finally:
reset_runtime_diagnostics()
assert [diagnostic.level for diagnostic in diagnostics] == ["warning", "error"]
assert sum("fell back to a secondary anchor" in diagnostic.message for diagnostic in diagnostics) == 1
assert sum("wrap_model_call" in diagnostic.message for diagnostic in diagnostics) == 1
def test_lead_only_contribution_is_absent_from_subagent_stack():
from deerflow.agents.middlewares.tool_error_handling_middleware import (
build_subagent_runtime_middlewares,
)
extensions = _extensions(MiddlewarePlacement(_Probe("lead-only"), Placement.STANDARD, scope=AgentScope.LEAD))
stack = build_subagent_runtime_middlewares(app_config=_app_config(), extensions=extensions)
assert "lead-only" not in _tags(stack)
def test_first_subagent_build_resolves_every_lazy_anchor(monkeypatch):
"""A lazy anchor table must be populated before its subagent copy is made.
``dict(dict_subclass)`` bypasses the subclass's ``__iter__``/``__len__``
hooks in CPython. Building a subagent first therefore used to copy an
empty table, then populate only MODEL_PHYSICAL as it installed the
subagent-specific override.
"""
from deerflow.agents.middlewares.tool_error_handling_middleware import (
build_subagent_runtime_middlewares,
)
from deerflow.extensions import stack as stack_module
fresh_table = stack_module._AnchorTable()
monkeypatch.setattr(stack_module._AnchorTable, "_loaded", False)
monkeypatch.setattr(stack_module, "PLACEMENT_ANCHORS", fresh_table)
extensions = _extensions(
MiddlewarePlacement(
_Probe("subagent-standard"),
Placement.STANDARD,
scope=AgentScope.SUBAGENT,
)
)
stack = build_subagent_runtime_middlewares(
app_config=_app_config(),
extensions=extensions,
)
assert "subagent-standard" in _tags(stack)
def test_core_ordering_table_is_enforced_against_a_real_stack(monkeypatch):
"""The default constraint table must be consulted on the REAL built stack.
Task 7 deleted the in-builder guard and the test that forced a real
misordering. Exercising assert_ordering with synthetic classes proves the
function works; it does not prove the composing builder calls it with
core_ordering_constraints() against the stack it actually produces. This
test is the only thing that does.
It patches the default table rather than reordering real middlewares: the
builder imports its middlewares inside the function body, so reordering
them would require stubbing sys.modules entries, which tests the stubbing
more than the wiring. Patching the table asserts exactly the claim at issue.
"""
from deerflow.agents.middlewares.input_sanitization_middleware import InputSanitizationMiddleware
from deerflow.agents.middlewares.tool_error_handling_middleware import ToolErrorHandlingMiddleware
from deerflow.extensions import ordering as ordering_mod
# InputSanitization is the outermost wrapper and ToolErrorHandling sits deep
# in the tail, so demanding the reverse is a constraint the real stack breaks.
impossible = (
ordering_mod.OrderingConstraint(
outer=ToolErrorHandlingMiddleware,
inner=InputSanitizationMiddleware,
reason="deliberately inverted for this test",
),
)
monkeypatch.setattr(ordering_mod, "core_ordering_constraints", lambda: impossible)
with pytest.raises(RuntimeError) as excinfo:
_lead_stack()
message = str(excinfo.value)
assert "ToolErrorHandlingMiddleware" in message
assert "InputSanitizationMiddleware" in message
assert "core middleware order" in message, "with no extension at either participating index the blame must fall on core order"
def test_core_ordering_table_passes_on_the_unmodified_stack():
"""The real stack must satisfy the real constraints — otherwise the test
above would pass for the wrong reason."""
_lead_stack() # must not raise
def test_ordering_violation_raises_and_names_the_extension():
"""A contribution that inverts a core invariant must fail loudly at build
time — the resulting behaviour would otherwise be wrong without an error."""
from deerflow.agents.middlewares.tool_error_handling_middleware import ToolErrorHandlingMiddleware
from deerflow.extensions.ordering import OrderingConstraint, assert_ordering
stack = [ToolErrorHandlingMiddleware(app_config=_app_config()), _Probe("x")]
constraints = (OrderingConstraint(outer=_Probe, inner=ToolErrorHandlingMiddleware, reason="test"),)
with pytest.raises(RuntimeError, match="demo:install"):
assert_ordering(stack, {1: "demo:install"}, constraints)
# --- AgentBuildContext.policy -----------------------------------------------
#
# Extensions read ctx.policy to adapt metering/behaviour to the limits the
# host actually enforces. Both builders must fill it from the resolved
# AppConfig — the field defaults to an all-disabled snapshot, so an omission
# is silent and hands extensions wrong values.
class _CtxCapture:
def __init__(self) -> None:
self.ctx = None
def contribute_middlewares(self, app_store, ctx):
self.ctx = ctx
return ()
def _extensions_with_contributor(contributor):
registry = ExtensionRegistry()
with registry.attributed_to("capture:install"):
registry.middlewares(contributor)
return registry.build()
def _policy_config() -> AppConfig:
from deerflow.config.subagents_config import SubagentsAppConfig
from deerflow.config.token_budget_config import TokenBudgetConfig
return AppConfig(
sandbox=SandboxConfig(use="deerflow.sandbox.local:LocalSandboxProvider"),
token_budget=TokenBudgetConfig(
enabled=True,
max_tokens=12345,
max_input_tokens=1000,
max_output_tokens=2000,
warn_threshold=0.7,
hard_stop_threshold=0.95,
),
subagents=SubagentsAppConfig(
max_total_per_run=7,
token_budget=TokenBudgetConfig(
enabled=True,
max_tokens=54321,
max_input_tokens=3000,
max_output_tokens=4000,
warn_threshold=0.6,
hard_stop_threshold=0.9,
),
),
)
def _assert_projected_policy(policy) -> None:
assert policy.token_budget_enabled is True
assert policy.max_input_tokens == 1000
assert policy.max_output_tokens == 2000
assert policy.max_total_tokens == 12345
assert policy.budget_warn_fraction == 0.7
assert policy.budget_hard_fraction == 0.95
assert policy.max_subagents_per_run is None
def _assert_subagent_policy(policy) -> None:
assert policy.token_budget_enabled is True
assert policy.max_input_tokens == 3000
assert policy.max_output_tokens == 4000
assert policy.max_total_tokens == 54321
assert policy.budget_warn_fraction == 0.6
assert policy.budget_hard_fraction == 0.9
assert policy.max_subagents_per_run is None
def test_lead_build_context_carries_the_projected_host_policy():
capture = _CtxCapture()
_lead_stack(extensions=_extensions_with_contributor(capture), app_config=_policy_config())
assert capture.ctx is not None, "the contributor was never consulted"
_assert_projected_policy(capture.ctx.policy)
def test_subagent_build_context_carries_the_projected_host_policy():
from deerflow.agents.middlewares.tool_error_handling_middleware import build_subagent_runtime_middlewares
capture = _CtxCapture()
build_subagent_runtime_middlewares(
app_config=_policy_config(),
model_name="gpt-4o",
extensions=_extensions_with_contributor(capture),
)
assert capture.ctx is not None, "the contributor was never consulted"
_assert_subagent_policy(capture.ctx.policy)
def test_lead_build_context_projects_the_effective_delegation_override():
capture = _CtxCapture()
_lead_stack(
extensions=_extensions_with_contributor(capture),
app_config=_policy_config(),
configurable={
"subagent_enabled": True,
"max_total_subagents": 3,
},
)
assert capture.ctx is not None
assert capture.ctx.policy.max_subagents_per_run == 3